Job events performance

[benthomasson]

Improves the performance of processing job events from 200 per second to 4000 per second.

[benthomasson]

This simulates 100 concurrent jobs each with 1000 job events.

time ./job_events_cannon.py --websocket-address ws://localhost:8080/api/ws2 --job-id ca73cc33-4ede-465e-bd82-6ccaaa2915d6 -c 1000 --stdout test2 --workers 100 --verbose

Events per second: 4238.371749744774

All job events were successfully recorded:

eda_server=# select count(*) from job_instance_event;
 count  
--------
 100000

[Red-HAP]

Shouldn't this reference the queue attribute?

Suggested change

	await bulk_job_instance_events.get()
	for i in range(bulk_job_instance_events.qsize())
	await bulk_job_instance_events.queue.get()
	for i in range(bulk_job_instance_events.queue.qsize())

[benthomasson]

Yes, I'll take a look at it.

[Red-HAP]

Shouldn't this also reference the queue attribute?

Suggested change

	await bulk_job_instance_hosts.get()
	for i in range(bulk_job_instance_hosts.qsize())
	await bulk_job_instance_hosts.queue.get()
	for i in range(bulk_job_instance_hosts.queue.qsize())

[cutwater]

Global dependency. These functions are used in the scope of websocket handler. It's unclear why they have to be global and completely unnecessary.

[cutwater]

Why this call is needed?

1. The get_event_loop function is deprecated for creating a new loop. See https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_event_loop
2. If it was used to create an event loop, the event loop must be created when calling asyncio.run and not here.
3. If it's used to get an existing running loop, it doesn't make sense (And in general for this you should use asyncio.get_running_loop())

[cutwater]

Use monotonic timer time.monotonic() for calculating time intervals.

[cutwater]

Because you're consuming a fixed and already known number of queue entries, you can use queue.get_nowait() here.

[cutwater]

Suggested change

	bulk_job_instance_events.start(
	app.dependency_overrides[get_db_session_factory]
	)
	bulk_job_instance_hosts.start(
	app.dependency_overrides[get_db_session_factory]
	)
	# FIXME: This is a hack. The database initialization should be refactored
	# if the database session factory must be accessed outside of the request scope.
	bulk_job_instance_events.start(
	app.dependency_overrides[get_db_session_factory]()
	)
	bulk_job_instance_hosts.start(
	app.dependency_overrides[get_db_session_factory]()
	)

The bulk processor should depend on a factory and not on a wrapper used for FastAPI DI mechanism.
At the moment there is a limitation, that the DatabaseProvider is not accessible outside of the request context.
This should be refactored.

[cutwater]

The logic here looks not very intuitive and probably not performant as well.
It runs a loop to check the queue size with fixed intervals and then passes the queue to the function that consumes messages from it. The queue size is unlimited which may lead to the back pressure problem.

While this somehow works because queue size is known and the queue access doesn't need synchronization, I'd re-write it in a different manner.

1. Use limited queue size to avoid uncontrolled growth of a queue.
2. The processor_fn callback should expect full batch: a sequence (e.g. list) of items to be processed.
3. The Batcher consumes the queue with a timeout:

# PSEUDOCODE #

while True:
    try:
        item = await asyncio.wait(queue.get(), timeout - elapsed_time)
    except TimeoutError:
        # flush
    batch.append(item)
    if len(batch) > max_batch_size:
        # flush