def process_all(action, s3_key, s3_secret, bucket_name, prefix, local_folder, queue, thread_count, max_retry):
    conn = S3Connection(s3_key, s3_secret)
    bucket = Bucket(connection=conn, name=bucket_name)

    if action == 'download':
        target_function = enqueue_s3_keys_for_download
    else:
        target_function = enqueue_s3_keys_for_upload

    enqueue_thread = threading.Thread(target=target_function, args=(bucket, prefix, local_folder, queue))
    enqueue_thread.daemon = True
    enqueue_thread.start()

    queue.queuing_started()

    consume_thread = threading.Thread(target=consume_queue, args=(queue, action, thread_count, max_retry))
    consume_thread.daemon = True
    consume_thread.start()

    while not queue.all_processed:
        # report progress every 10 secs
        logger.info('{0} keys enqueued, and {1} keys {2}ed'.format(queue.enqueued_counter, queue.de_queue_counter, action))
        time.sleep(10)

    logger.info('{0} keys enqueued, and {1} keys {2}ed'.format(queue.enqueued_counter, queue.de_queue_counter, action))
Process `substitutions` for given `mapping` (modified in place)
dataset/ETHPy150Open quid/s3concurrent/s3concurrent/s3concurrent.py/process_all--crossover with--dataset/ETHPy150Open projectatomic/atomic-reactor/atomic_reactor/util.py/process_substitutions