Split up a bucket keyspace into smaller sets for parallel iteration.

def process_bucket_partitions(
        bid, prefix_set=('',), partition='/', strategy=None, limit=4):
    """Split up a bucket keyspace into smaller sets for parallel iteration.
    """
    if strategy is None:
        return detect_partition_strategy(bid)

    account, bucket = bid.split(':', 1)
    region = connection.hget('bucket-regions', bid)
    versioned = bool(int(connection.hget('bucket-versions', bid)))
    session = get_session(
        json.loads(connection.hget('bucket-accounts', account)))
    size = int(float(connection.hget('bucket-sizes', bid)))
    s3 = session.client('s3', region_name=region, config=s3config)

    strategy = get_partition_strategy(strategy)
    strategy.limit = limit
    strategy.partition = partition
    (contents_key,
     contents_method,
     continue_tokens) = BUCKET_OBJ_DESC[versioned]
    prefix_queue = strategy.initialize_prefixes(prefix_set)

    keyset = []
    log.info(
        "Process partition bid:%s strategy:%s delimiter:%s queue:%d limit:%d",
        bid, strategy.__class__.__name__[0], partition,
        len(prefix_queue), limit)

    def statm(prefix):
        return "keyset:%d queue:%d prefix:%s bucket:%s size:%d" % (
            len(keyset), len(prefix_queue), prefix, bid, size)

    while prefix_queue:
        connection.hincrby('bucket-partition', bid, 1)
        prefix = prefix_queue.pop()
        if strategy.is_depth_exceeded(prefix):
            log.info("Partition max depth reached, %s", statm(prefix))
            invoke(process_bucket_iterator, bid, prefix)
            continue
        method = getattr(s3, contents_method, None)
        results = method(Bucket=bucket, Prefix=prefix, Delimiter=partition)
        keyset.extend(results.get(contents_key, ()))

        # As we probe we find keys, process any found
        if len(keyset) > PARTITION_KEYSET_THRESHOLD:
            log.info("Partition, processing keyset %s", statm(prefix))
            page = page_strip({contents_key: keyset}, versioned)
            if page:
                invoke(process_keyset, bid, page)
            keyset = []

        strategy.find_partitions(prefix_queue, results)

        # Do we have more than 1k keys at this level, continue iteration
        continuation_params = {
            k: results[k] for k in continue_tokens if k in results}
        if continuation_params:
            bp = int(connection.hget('bucket-partition', bid))
            log.info("Partition has 1k keys, %s %s", statm(prefix), bp)
            if not prefix_queue and bp < 5:
                log.info("Recursive detection")
                return detect_partition_strategy(bid, prefix=prefix)

            invoke(process_bucket_iterator, bid, prefix, delimiter=partition,
                   **continuation_params)

        # If the queue get too deep, then go parallel
        if len(prefix_queue) > PARTITION_QUEUE_THRESHOLD:
            log.info("Partition add friends, %s", statm(prefix))
            for s_prefix_set in chunks(
                    prefix_queue[PARTITION_QUEUE_THRESHOLD - 1:],
                    PARTITION_QUEUE_THRESHOLD - 1):

                for s in list(s_prefix_set):
                    if strategy.is_depth_exceeded(prefix):
                        invoke(process_bucket_iterator, bid, s)
                        s_prefix_set.remove(s)

                if not s_prefix_set:
                    continue
                invoke(process_bucket_partitions,
                       bid,
                       prefix_set=s_prefix_set, partition=partition,
                       strategy=strategy, limit=limit)
            prefix_queue = prefix_queue[:PARTITION_QUEUE_THRESHOLD - 1]

    if keyset:
        page = page_strip({contents_key: keyset}, versioned)
        if page:
            invoke(process_keyset, bid, page)