|
|
@ -101,7 +101,7 @@ hash_key = get_hash_key()
|
|
|
|
@asyncio.coroutine
|
|
|
|
@asyncio.coroutine
|
|
|
|
async def spawn_processing_threads(data):
|
|
|
|
async def spawn_processing_threads(data):
|
|
|
|
len_data = len(data)
|
|
|
|
len_data = len(data)
|
|
|
|
log.debug(f"Spawning processing threads for batch of {len_data} messages")
|
|
|
|
# log.debug(f"Spawning processing threads for batch of {len_data} messages")
|
|
|
|
|
|
|
|
|
|
|
|
loop = asyncio.get_event_loop()
|
|
|
|
loop = asyncio.get_event_loop()
|
|
|
|
tasks = []
|
|
|
|
tasks = []
|
|
|
@ -112,18 +112,23 @@ async def spawn_processing_threads(data):
|
|
|
|
msg_per_core = int(len(data) / CPU_THREADS)
|
|
|
|
msg_per_core = int(len(data) / CPU_THREADS)
|
|
|
|
split_data = array_split(data, ceil(len(data) / msg_per_core))
|
|
|
|
split_data = array_split(data, ceil(len(data) / msg_per_core))
|
|
|
|
for index, split in enumerate(split_data):
|
|
|
|
for index, split in enumerate(split_data):
|
|
|
|
log.debug(f"Delegating processing of {len(split)} messages to thread {index}")
|
|
|
|
# log.debug(f"Delegating processing of {len(split)} messages to thread {index}")
|
|
|
|
task = loop.run_in_executor(p, process_data, split)
|
|
|
|
task = loop.run_in_executor(p, process_data, split)
|
|
|
|
tasks.append(task)
|
|
|
|
tasks.append(task)
|
|
|
|
|
|
|
|
|
|
|
|
results = [await task for task in tasks]
|
|
|
|
results = [await task for task in tasks]
|
|
|
|
log.debug(f"Results from processing of {len_data} messages: {len(results)}")
|
|
|
|
log.debug(
|
|
|
|
|
|
|
|
(
|
|
|
|
|
|
|
|
f"Results from processing of {len_data} messages in "
|
|
|
|
|
|
|
|
f"{len(split_data)} threads: {len(results)}"
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
# Join the results back from the split list
|
|
|
|
# Join the results back from the split list
|
|
|
|
flat_list = [item for sublist in results for item in sublist]
|
|
|
|
flat_list = [item for sublist in results for item in sublist]
|
|
|
|
await db.store_kafka_batch(flat_list)
|
|
|
|
await db.store_kafka_batch(flat_list)
|
|
|
|
|
|
|
|
|
|
|
|
log.debug(f"Finished processing {len_data} messages")
|
|
|
|
# log.debug(f"Finished processing {len_data} messages")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def process_data(data):
|
|
|
|
def process_data(data):
|
|
|
|