Move tiles of interest set out of Redis and into S3

tilequeue/command.py

@@ -615,7 +615,7 @@ def tilequeue_process(cfg, peripherals):
 
     data_fetch = DataFetch(
         feature_fetcher, sqs_input_queue, sql_data_fetch_queue, io_pool,
-        peripherals.redis_cache_index, logger, cfg.metatile_zoom, cfg.max_zoom)
+        logger, cfg.metatile_zoom, cfg.max_zoom)
 
     data_processor = ProcessAndFormatData(
         post_process_data, formats, sql_data_fetch_queue, processor_queue,
@@ -874,59 +874,48 @@ def tilequeue_seed(cfg, peripherals):
     tile_queue = peripherals.queue
     # suppresses checking the in flight list while seeding
     tile_queue.is_seeding = True
-    redis_cache_index = peripherals.redis_cache_index
     enqueuer = ThreadedEnqueuer(tile_queue, cfg.seed_n_threads, logger)
 
     # based on cfg, create tile generator
     tile_generator = make_seed_tile_generator(cfg)
 
     queue_buf_size = 1024
     tile_queue_queue = Queue.Queue(queue_buf_size)
-    redis_queue = Queue.Queue(queue_buf_size)
-
-    # updating sqs and updating redis happen in background threads
-    def redis_add():
-        redis_coords_buf = []
-        done = False
-        while not done:
-            coord = redis_queue.get()
-            if coord is None:
-                done = True
-            else:
-                redis_coords_buf.append(coord)
-            if len(redis_coords_buf) >= 1024 or (done and redis_coords_buf):
-                redis_cache_index.index_coords(redis_coords_buf)
-                del redis_coords_buf[:]
 
+    # updating sqs happens in background threads
     def tile_queue_enqueue():
         enqueuer(tile_queue_queue)
 
-    logger.info('Sqs ... ')
+    logger.info('Enqueueing to SQS ... ')
     thread_enqueue = threading.Thread(target=tile_queue_enqueue)
     thread_enqueue.start()
 
-    if cfg.seed_should_add_to_tiles_of_interest:
-        logger.info('Tiles of interest ...')
-        thread_redis = threading.Thread(target=redis_add)
-        thread_redis.start()
-
     n_coords = 0
     for coord in tile_generator:
         tile_queue_queue.put(coord)
-        if cfg.seed_should_add_to_tiles_of_interest:
-            redis_queue.put(coord)
         n_coords += 1
         if n_coords % 100000 == 0:
             logger.info('%d enqueued' % n_coords)
 
     tile_queue_queue.put(None)
-    if cfg.seed_should_add_to_tiles_of_interest:
-        redis_queue.put(None)
-        thread_redis.join()
-        logger.info('Tiles of interest ... done')
 
     thread_enqueue.join()
-    logger.info('Sqs ... done')
+    logger.info('Enqueueing to SQS ... done')
+
+    if cfg.seed_should_add_to_tiles_of_interest:
+        logger.info('Adding to Redis ... ')
+
+        toi_set = peripherals.redis_cache_index.fetch_tiles_of_interest()
+
+        tile_generator = make_seed_tile_generator(cfg)
+        for coord in tile_generator:
+            coord_int = coord_marshall_int(coord)
+            toi_set.add(coord_int)
+
+        peripherals.redis_cache_index.set_tiles_of_interest(toi_set)
+
+        logger.info('Adding to Redis ... done')
+
     logger.info('Seeding tiles ... done')
     logger.info('%d coordinates enqueued' % n_coords)
 
@@ -1082,13 +1071,7 @@ def tilequeue_prune_tiles_of_interest(cfg, peripherals):
                 len(toi_to_remove))
     peripherals.stats.gauge('gardener.removed', len(toi_to_remove))
 
-    def delete_tile_of_interest(s3_parts, coord_ints):
-        # Remove from the redis toi set
-        removed = peripherals.redis_cache_index.remove_tiles_of_interest(
-            coord_ints)
-
-        logger.info('Removed %s tiles from redis', removed)
-
+    def delete_from_s3(s3_parts, coord_ints):
         # Remove from S3
         s3 = boto.connect_s3(cfg.aws_access_key_id, cfg.aws_secret_access_key)
         buk = s3.get_bucket(s3_parts['bucket'], validate=False)
@@ -1115,7 +1098,7 @@ def delete_tile_of_interest(s3_parts, coord_ints):
                     len(toi_to_remove))
 
         for coord_ints in grouper(toi_to_remove, 1000):
-            delete_tile_of_interest(s3_parts, coord_ints)
+            delete_from_s3(s3_parts, coord_ints)
 
         logger.info('Removing %s tiles from TOI and S3 ... done',
                     len(toi_to_remove))
@@ -1130,17 +1113,6 @@ def delete_tile_of_interest(s3_parts, coord_ints):
         logger.info('Skipping TOI add step because there are '
                     'no tiles to add')
     else:
-        logger.info('Adding %s tiles to Redis TOI set ...',
-                    len(toi_to_add))
-
-        for coord_ints in grouper(toi_to_add, 1000):
-            added = peripherals.redis_cache_index.add_tiles_of_interest(
-                coord_ints)
-            logger.info('Added %s tiles to redis', added)
-
-        logger.info('Adding %s tiles to Redis TOI set ... done',
-                    len(toi_to_add))
-
         logger.info('Enqueueing %s tiles to SQS ...',
                     len(toi_to_add))
 
@@ -1153,6 +1125,16 @@ def delete_tile_of_interest(s3_parts, coord_ints):
         logger.info('Enqueueing %s tiles to SQS ... done',
                     len(toi_to_add))
 
+    if toi_to_add or toi_to_remove:
+        logger.info('Setting new tiles of interest ... ')
+
+        peripherals.redis_cache_index.set_tiles_of_interest(new_toi)
+
+        logger.info('Setting new tiles of interest ... done')
+    else:
+        logger.info('Tiles of interest did not change, '
+                    'so not setting new tiles of interest')
+
     logger.info('Pruning tiles of interest ... done')
     time_overall.stop()
 

tilequeue/worker.py

@@ -158,12 +158,11 @@ class DataFetch(object):
 
     def __init__(
             self, fetcher, input_queue, output_queue, io_pool,
-            redis_cache_index, logger, metatile_zoom, max_zoom):
+            logger, metatile_zoom, max_zoom):
         self.fetcher = fetcher
         self.input_queue = input_queue
         self.output_queue = output_queue
         self.io_pool = io_pool
-        self.redis_cache_index = redis_cache_index
         self.logger = logger
         self.metatile_zoom = metatile_zoom
         self.max_zoom = max_zoom