feat(taskworker) Add concurrent worker

[markstory]

Move the taskworker process to be a multiprocess concurrent worker. This will help enable higher CPU usage in worker pods, as we can pack more concurrent CPU operations into each pod (at the cost of memory).

The main process is responsible for:

• Spawning children
• Making RPC requests to fill child queues and submit results.

Each child process handles:

• Resolving task names
• Checking at_most_once keys
• Enforcing processing deadlines
• Executing task functions

Instead of using more child processes to enforce timeouts, I've used SIGALRM. I've verified that tasks like

@exampletasks.register(name="examples.infinite", retry=Retry(times=2))
def infinite_task() -> None:
    try:
        while True:
            pass
    except Exception as e:
        print("haha caught exception", e)

Do not paralyze workers with infinite loops.

When a worker is terminated, it uses an Event to have children exit, and then drains any results. If there are tasks in the _child_tasks queue will not be completed, and instead will sent to another worker when the processing_deadline on the activations expires.

[codecov]

Codecov Report

Attention: Patch coverage is 87.78626% with 32 lines in your changes missing coverage. Please review.

✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
src/sentry/taskworker/worker.py	80.25%	31 Missing ⚠️
tests/sentry/taskworker/test_worker.py	99.04%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #83254      +/-   ##
==========================================
+ Coverage   87.49%   87.56%   +0.06%     
==========================================
  Files        9404     9393      -11     
  Lines      537180   536973     -207     
  Branches    21133    21048      -85     
==========================================
+ Hits       470024   470203     +179     
+ Misses      66798    66414     -384     
+ Partials      358      356       -2     

[evanh]

I don't think get_nowait is necessarily correct here. I understand this is ensuring that the process doesn't block while waiting for a task before checking for the shutdown, but I think some kind of timeout/delay would good to here to avoid spiking the CPU. Maybe like 100ms or something like that?

[markstory]

Good point about the potential CPU burn on an empty queue. I'll put a blocking get with a timeout in.

[sentry-io]

Suspect Issues

This pull request was deployed and Sentry observed the following issues:

• ‼️ AssertionError: expected call not found. pytest.runtest.protocol tests/sentry/taskworker... View Issue

_{Did you find this useful? React with a 👍 or 👎}