(migrator, reindexer, iterable, max_retry, num_workers)
Migrate in parallel with worker pool

    When running in steady state, failed doc will be retried up to the
    max retry limit. Documents awaiting retry and all documents that
    started the migration process but did not finish will be saved and
    retried on the next run if the migration is stopped before it
    completes.
Migrate in parallel with worker pool
Migrate in parallel with worker pool
def _migrator_with_worker_pool(migrator, reindexer, iterable, max_retry, num_workers):
    """Migrate in parallel with worker pool

    When running in steady state, failed doc will be retried up to the
    max retry limit. Documents awaiting retry and all documents that
    started the migration process but did not finish will be saved and
    retried on the next run if the migration is stopped before it
    completes.
    """
    def work_on(doc, key, retry_count):
        try:
            ok = migrator.migrate(doc)
            assert ok, "run_with_worker_pool expects success!"
        except Exception:
            err = traceback.format_exc().strip()
            print("Error processing blob:\n{}".format(err))
            if retry_count < max_retry:
                print("will retry {}".format(key))
                retry_blobs[key] += 1
                queue.put(doc)
                return
            migrator.save_backup(doc, "too many retries")
            print("too many retries {}".format(key))
        retry_blobs.pop(key, None)

    def retry_loop():
        for doc in queue:
            enqueue_doc(doc)

    def enqueue_doc(doc):
        key = reindexer.get_key(doc)
        retry_count = retry_blobs.setdefault(key, 0)
        # pool.spawn will block until a worker is available
        pool.spawn(work_on, doc, key, retry_count)
        # Returning True here means the underlying iterator will think
        # this doc has been processed successfully. Therefore we must
        # process this doc before the process exits or save it to be
        # processed on the next run.
        return True

    queue = LifoQueue()
    loop = gevent.spawn(retry_loop)
    pool = Pool(size=num_workers)

    class gmigrator:
        migrate = staticmethod(enqueue_doc)

    with migrator:
        retry_blobs = iterable.get_iterator_detail("retry_blobs") or {}
        for key in list(retry_blobs):
            queue.put(reindexer.load(key))
        try:
            yield gmigrator
        finally:
            try:
                print("waiting for workers to stop... (Ctrl+C to abort)")
                queue.put(StopIteration)
                loop.join()
                while not pool.join(timeout=10):
                    print("waiting for {} workers to stop...".format(len(pool)))
            finally:
                iterable.set_iterator_detail("retry_blobs", retry_blobs)
                print("done.")
def _migrator_with_worker_pool ( migrator , reindexer , iterable , max_retry , num_workers ) : def work_on ( doc , key , retry_count ) : try : ok = migrator . migrate ( doc ) assert ok , "run_with_worker_pool expects success!" except Exception : err = traceback . format_exc ( ) . strip ( ) print ( "Error processing blob:\n{}" . format ( err ) ) if retry_count < max_retry : print ( "will retry {}" . format ( key ) ) retry_blobs [ key ] += 1 queue . put ( doc ) return migrator . save_backup ( doc , "too many retries" ) print ( "too many retries {}" . format ( key ) ) retry_blobs . pop ( key , None ) def retry_loop ( ) : for doc in queue : enqueue_doc ( doc ) def enqueue_doc ( doc ) : key = reindexer . get_key ( doc ) retry_count = retry_blobs . setdefault ( key , 0 ) # pool.spawn will block until a worker is available pool . spawn ( work_on , doc , key , retry_count ) # Returning True here means the underlying iterator will think # this doc has been processed successfully. Therefore we must # process this doc before the process exits or save it to be # processed on the next run. return True queue = LifoQueue ( ) loop = gevent . spawn ( retry_loop ) pool = Pool ( size = num_workers ) class gmigrator : migrate = staticmethod ( enqueue_doc ) with migrator : retry_blobs = iterable . get_iterator_detail ( "retry_blobs" ) or { } for key in list ( retry_blobs ) : queue . put ( reindexer . load ( key ) ) try : yield gmigrator finally : try : print ( "waiting for workers to stop... (Ctrl+C to abort)" ) queue . put ( StopIteration ) loop . join ( ) while not pool . join ( timeout = 10 ) : print ( "waiting for {} workers to stop..." . format ( len ( pool ) ) ) finally : iterable . set_iterator_detail ( "retry_blobs" , retry_blobs ) print ( "done." )