[PULP-218] Add timeout to immediate tasks #6155

pedro-psb · 2024-12-16T20:38:05Z

No description provided.

mdellweg · 2024-12-17T13:43:47Z

pulpcore/tasking/tasks.py

+    signal.signal(signal.SIGALRM, timeout_handler)
+    signal.alarm(IMMEDIATE_TASK_TIMEOUT)


Oh, interesting.
This is quite close to low level process handling. I wonder if this might interfere with other uses of the signal handler.

Do you think we can instead require that all immediate tasks are async functions? We could then implement the timeout on the asyncio level with much more confidence.

Do you think we can instead require that all immediate tasks are async functions? We could then implement the timeout on the asyncio level with much more confidence.

Yes. I dont have a good idea of how much friction this imposes, but that's definitely more safe.

@mdellweg Hey! Can you provide an early review for the api worker part?

mdellweg · 2025-02-13T09:08:05Z

pulpcore/tasking/tasks.py

-            loop.run_until_complete(result)
+            if immediate:
+                try:
+                    loop.run_until_complete(asyncio.wait_for(result, timeout=IMMEDIATE_TIMEOUT))


Looks like a good first cut. But we still stop the heartbeat loop for up to 5 seconds here.

We should definitely plan on rewriting the workers in async.

We should definitely plan on rewriting the workers in async.

Would you say that's a blocker (for running on foreground)?

I was trying to find a way to integrate the coroutine control yielding with the select-loop, but I'm reaching dark corners.

If that's a blocker this PR could be really about the timeout (as originally intented). The commit "wip: add timeout to supervise immediate tasks" adds a timeout to the regular task process. Maybe we could stop there.

Not necessarily. It's a certain risk. But it's nothing we cannot revert when things go bad.

Maybe after all it's wise to split the work into smaller chunks to reduce that risk.
Ensuring that immediate tasks don't run too long is still an incremental improvement.

mdellweg · 2025-02-14T10:37:27Z

We cannot require all immediate tasks to be async without properly deprecating it for plugins.
But we can use sync_to_async for the ones that aren't with a deprecation warning.

pedro-psb · 2025-02-14T17:20:48Z

No idea why this fails only on s3:
https://github.com/pulp/pulpcore/actions/runs/13333279626/job/37242548757?pr=6155#step:14:18219

An exception does get caught...

mdellweg

Not sure if you wanted a review at this stage...

mdellweg · 2025-03-18T10:43:25Z

pulpcore/constants.py

@@ -45,6 +45,8 @@
 #: Tasks in an incomplete state have not finished their work yet.
 TASK_INCOMPLETE_STATES = (TASK_STATES.WAITING, TASK_STATES.RUNNING, TASK_STATES.CANCELING)

+#: Timeout for immediate tasks


Is that seconds?

Yes, I'll add that on the description

mdellweg · 2025-03-18T10:43:46Z

pulpcore/tasking/tasks.py

@@ -50,7 +52,6 @@ def wakeup_worker():


 def execute_task(task):
-    # This extra stack is needed to isolate the current_task ContextVar


Why drop this comment?

I guess that was an accident

Absolutely not. But then it doesn't seem to be telling its story.

I mean dropping the comment was an accident.
The comment is fine. I didnt know what a ContextVar was when I first read it, but then I read about it an it made sense.

Ohhh. Thanks for the clarification...

mdellweg · 2025-03-18T10:46:13Z

pulpcore/tasking/tasks.py

-        if asyncio.iscoroutine(result):
+        immediate = task.immediate
+        is_coroutine = asyncio.iscoroutine(result)


Has this ever been true? Should we no check if func is a coroutine even before executing it?

Gerrod showed that this is valid here. But I can look up a more intuitive form to check it.

The big problem is that we run the task function first and then we check if it returned something awaitable.
(That is what coroutine functions are: They are functions returning an awaitable generator object.)

While in the old code this may be working properly (unless some task accidentally returned a generator of any sort), here we don't want to run the task function in case it is immediate and not async.

So what should happen, i think, is we should check for asyncio.iscoroutinefunction(func) and then decide whether or how to execute it.

Sadly the python documentation itself uses the term "coroutine" a bit inconsistently.
Sometimes as the function declared with async, and sometimes as the awaitable object that is the result of calling the former.

mdellweg · 2025-03-18T10:48:59Z

pulpcore/tasking/worker.py

@@ -388,6 +390,7 @@ def supervise_task(self, task):
        task.save(update_fields=["worker"])
        cancel_state = None
        cancel_reason = None
+        time_limit = time.monotonic() + IMMEDIATE_TIMEOUT


Why do we not use timezone.now here?

Also didn't we want to let the timeout tick in the async loop?
That will help us maybe run even more than one immediate tasks per worker eventually.
Also the timeout will be in effect even on the api worker.

Oh, that's true 🤦
The worker ends up calling execute_task, so we only need it there with the async loop...

mdellweg · 2025-03-18T10:51:28Z

pulpcore/tests/functional/api/test_tasking.py

+        )
+        wait_until("running", task_href)
+
+    # Case 1


Each case should be a separate test. If you want to share resources, you can group them in a class.

mdellweg · 2025-03-19T22:19:56Z

pulpcore/tests/functional/api/test_tasking.py

+        AND it takes longer than timeout
+        THEN an error is raised
+        """
+        COMMON_RESOURCE = "PPP"


I'd make the resource a random string to avoid any collisions.
Also why not mark this as parallel too?

mdellweg · 2025-03-19T22:22:47Z

pulpcore/tests/functional/api/test_tasking.py

-    # Case 3
-    dispatch_long_task(requires_resource=COMMON_RESOURCE)
-    with pytest.raises(PulpTaskError) as ctx:
+class TestImmediateTaskWithLocking:


How about TestImmediateTaskWithBlockedResource?
When removing the "test" noise, the actual (first) test will read:
"Immediate task with blocked resource executes in task worker." Exactly what we want to read here...

mdellweg · 2025-03-19T22:23:23Z

pulpcore/tests/functional/api/test_tasking.py

+        THEN the task completes with a worker
+        """
+        COMMON_RESOURCE = "MMM"
+        dispatch_long_task(required_resources=[COMMON_RESOURCE])


You could cancel the long task after dispatching the regular one in order to speed up the tests dramatically.

mdellweg · 2025-03-19T22:25:04Z

pulpcore/tests/functional/api/test_tasking.py


-    # Case 2
-    with pytest.raises(PulpTaskError) as ctx:
+class TestImmediateTaskNoLocking:


TestImmediateTaskWithNoResource?

mdellweg · 2025-03-19T22:25:33Z

pulpcore/tests/functional/api/test_tasking.py

+class TestImmediateTaskNoLocking:
+
+    @pytest.mark.parallel
+    def test_succeed_on_api_worker(self, pulpcore_bindings, dispatch_task, monitor_task):


This is really nit picky...
test_succeeds_on_api_worker

mdellweg · 2025-03-19T22:28:38Z

pulpcore/tests/functional/api/test_tasking.py

-def test_immediate_task_requires_resource(pulpcore_bindings, dispatch_task, monitor_task):
-    """
-    GIVEN an async task requiring busy resources
+        task = monitor_task(task_href)


Should we monitor that task? Should it not return as completed already from dispatch_task? Also we should be able to catch the deprecation warning here and assert on it.
Also we don't want it to bubble into the collected deprecation warnings for the ci.

mdellweg · 2025-03-19T22:33:07Z

pulpcore/tasking/tasks.py

            _logger.debug(_("Task is coroutine %s"), task.pk)
+            coro = func(*args, **kwargs)


Not sure if it's wise to call this thing coro given the confusion with that name throughout the python docs.
The only idea i can come up with is "awaitable".

The usage is a bit confusing but there is a clear definition.
Do you think we can try to be strict about those terms?

IMO "awaitable" is too broad, we really expect a coroutine object, which is returned by a coroutine function.
But I can change if you still think that would be confusing.

It's probably ok.

mdellweg · 2025-03-19T22:33:53Z

pulpcore/tasking/tasks.py

-                coro = result
-
+                coro = asyncio.wait_for(coro, timeout=IMMEDIATE_TIMEOUT)
+            assert asyncio.iscoroutine(coro), coro


Extra safety? (I'm almost certain we can be sure of that now...)

Suggested change

assert asyncio.iscoroutine(coro), coro

Let's not add tests for the interpreter to the production code.

mdellweg · 2025-03-24T09:49:37Z

pulpcore/pytest_plugin.py

+    The stderr can be useful write asserts over log messages.
+    This is required because the pytest caplog fixture can't capture log messages
+    streamed from the pulpcore-manager call.


ohhhh!

Now I'm wondering if we could just use the capsys or capfd fixture instead? (If you want to try that.)
(I'd prefer not adding another fixture to do almost the same thing.)

mdellweg

I think the implementation is sound at this stage...
But still some things came up when reading through.

mdellweg · 2025-03-24T09:51:33Z

pulpcore/tasking/tasks.py

            _logger.debug(_("Task is coroutine %s"), task.pk)
+            coro = func(*args, **kwargs)


It's probably ok.

mdellweg · 2025-03-24T09:52:36Z

pulpcore/tasking/tasks.py

-                coro = result
-
+                coro = asyncio.wait_for(coro, timeout=IMMEDIATE_TIMEOUT)
+            assert asyncio.iscoroutine(coro), coro


Suggested change

assert asyncio.iscoroutine(coro), coro

Let's not add tests for the interpreter to the production code.

mdellweg · 2025-03-24T09:56:07Z

pulpcore/tasking/tasks.py

+            except asyncio.TimeoutError:
+                raise RuntimeError(
+                    _("Immediate task timed out after {} seconds").format(IMMEDIATE_TIMEOUT)
+                )


This try block really only translates the timeout error to add more information, right?
If so, please add the task uuid also, and shove something in to the log stream of the worker.

Yes, this was supposed to add more info.
But it'll be handled by the outer try/catch-all at L107 (from changed code) and receive all the context info.
Or am I missing something?

Right. But i think we can still add something like:
logger.warn("Immediate task {task.pk} timed out after ...") So we see it in the worker and api worker logs too.

Oh, ok. I've missed that comment

mdellweg · 2025-03-24T09:57:42Z

pulpcore/tests/functional/api/test_tasking.py

+        task_href = dispatch_task(
+            "pulpcore.app.tasks.test.asleep", args=(LT_TIMEOUT,), immediate=True
+        )


Incidentally, these kinds of tasks are now a big driver to wish for asgi pulp instead of wsgi.

mdellweg · 2025-03-24T10:02:58Z

pulpcore/tests/functional/api/test_tasking.py

+        raise RuntimeError("Timeout waiting for task to transition")
+
+    @contextmanager
+    def _dispatch_long_task(required_resources: list[str], duration=5):


Let us stick to the established names exclusive_resources and shared_resources here.

mdellweg · 2025-03-24T10:03:34Z

pulpcore/tests/functional/api/test_tasking.py

+            args=(duration,),
+            exclusive_resources=required_resources,
+        )
+        wait_until("running", task_href)


I don't see a reason to wait for anything else than an unfinished task in the queue.

My reasoning was that for the long task to block the immediate, it must have been picked up first.
So it could be running or cancelling, but not waiting (as we have priority on immediate tasks, it would be picked up first). So, is ["running", "canceling"] ok?

No, that's not true. A task blocks a resource even in waiting state.

mdellweg · 2025-03-24T10:04:17Z

pulpcore/tests/functional/api/test_tasking.py

+
+
+@pytest.fixture
+def dispatch_long_task(pulpcore_bindings, dispatch_task):


Now this fixture sounds kike it should be called resource_blocker.
That we use a task for it, is merely an implementation detail.

mdellweg · 2025-03-24T10:07:17Z

pulpcore/tests/functional/api/test_tasking.py

+        task = pulpcore_bindings.TasksApi.read(task_href)
+        if task.state == "running":
+            pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"})


Suggested change

task = pulpcore_bindings.TasksApi.read(task_href)

if task.state == "running":

pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"})

pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"})

No need to be polite to that task. Just make sure it goes away.
Maybe we even require it to use a non existing plugin so it will never be picked up at all.

If the task has already finished this call will raise a 409 error on the pulpcore client.
E.g:

task_href = dispatch_task("pulpcore.app.tasks.test.sleep", args=(2,)) time.sleep(3) pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"}) # <- raises

But l can just suppress that error instead, didn't think about that before.

mdellweg · 2025-03-24T10:09:24Z

pulpcore/tests/functional/api/test_tasking.py

+        assert task.worker is not None
+
+    @pytest.mark.parallel
+    def test_throws_when_non_deferrable(


This is not a newly defined behaviour. Can you do me a favor and see whether we already have a test for this?

(Ideally we combine them or pick the nicer one...)

Looks like we don't test it elsewhere:

> git grep -I -n defer -- pulpcore/tests/ pulpcore/tests/functional/api/test_tasking.py:564: def test_throws_when_non_deferrable( pulpcore/tests/functional/api/test_tasking.py:569: WHEN dispatching as immediate and not deferrable pulpcore/tests/functional/api/test_tasking.py:578: deferred=False,

mdellweg · 2025-03-24T10:10:47Z

pulpcore/tests/functional/api/test_tasking.py

+            assert "Resources temporarily unavailable." in task.error["reason"]
+
+    @pytest.mark.parallel
+    def test_timeouts_on_task_worker(


Nitpick mode: test_times_out_on_task_worker reads more English to me.

mdellweg · 2025-03-25T09:05:55Z

pulpcore/tests/functional/api/test_tasking.py

-        assert "Support for non-coroutine immediate tasks will be dropped" in stderr
+        assert "Support for non-coroutine immediate tasks will be dropped" in stderr_content


Thank you! Not adding a new fixture for this makes me happier.

mdellweg · 2025-03-25T09:07:27Z

pulpcore/tests/functional/api/test_tasking.py

+            args=(duration,),
+            exclusive_resources=required_resources,
+        )
+        wait_until("running", task_href)


No, that's not true. A task blocks a resource even in waiting state.

mdellweg · 2025-03-25T23:41:10Z

pulpcore/tests/functional/api/test_tasking.py

+            args=(duration,),
+            exclusive_resources=exclusive_resources,
+        )
+        wait_until(TASK_INCOMPLETE_STATES, task_href)


Suggested change

wait_until(TASK_INCOMPLETE_STATES, task_href)

What do you want to wait for here?

Yeah, the wait doesnt make sense anymore. Removing...

But I wanted to share some thoughts I had on this (not wanting to change anything in this PR):

The long task will always be unblocked first because it has lower pulp_created, and the short is assured to be blocked as we intend in the test, right?

I was thinking if we shouldnt sort by immediate task there too (in the unblock_task). This is a violation of the event order, but so is our immediate task prioritization in iter_tasks. I can see there is a diffrence, and feels like it would be worse to violate that in unblocking-time than it is in picking-time, but that's not crystal clear to me.

Well the thing is the immediate task will not be unblocked before the other task is in a final state. And that's the only thing needed to assure resource usage is serialized. It does not matter whether the prior task is identified unblocked or in a waiting, running or canceling state.

pulpcore/tasking/tasks.py

* Enforce immediate tasks are run are async function * Add deprecation warning that immediate tasks must be async function starting from pulpcore 3.85 * Add tests to exercise immediate and defer dispatch options Co-authored-by: Matthias Dellweg <[email protected]> Closes: pulp#5930

github-actions bot added no-changelog no-issue labels Dec 16, 2024

pedro-psb changed the title ~~Add timeout to immediate tasks~~ [PULP-218] Add timeout to immediate tasks Dec 16, 2024

mdellweg reviewed Dec 17, 2024

View reviewed changes

pedro-psb force-pushed the add-timeout-to-immediate-task branch 2 times, most recently from 138be99 to 08a9c14 Compare February 11, 2025 22:10

pedro-psb linked an issue Feb 12, 2025 that may be closed by this pull request

Add timeout to immediate tasks #5930

Closed

github-actions bot added the multi-commit label Feb 12, 2025

pedro-psb force-pushed the add-timeout-to-immediate-task branch 4 times, most recently from 54c86f9 to 1704eb4 Compare February 12, 2025 22:08

pedro-psb marked this pull request as ready for review February 12, 2025 22:18

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 1704eb4 to 88fc96d Compare February 12, 2025 22:32

mdellweg reviewed Feb 13, 2025

View reviewed changes

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 6b2781e to 1b427eb Compare February 13, 2025 19:53

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 00a071e to e834821 Compare February 14, 2025 14:21

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 4edeef0 to a7a94b9 Compare March 17, 2025 19:46

mdellweg requested changes Mar 18, 2025

View reviewed changes

github-actions bot added the wip label Mar 19, 2025

mdellweg reviewed Mar 19, 2025

View reviewed changes

pedro-psb force-pushed the add-timeout-to-immediate-task branch 3 times, most recently from 82ff903 to f46846b Compare March 22, 2025 03:18

mdellweg reviewed Mar 24, 2025

View reviewed changes

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 6658c42 to b2fca1e Compare March 24, 2025 15:13

pedro-psb requested a review from mdellweg March 24, 2025 17:35

mdellweg reviewed Mar 25, 2025

View reviewed changes

pedro-psb force-pushed the add-timeout-to-immediate-task branch 2 times, most recently from d9b1386 to 6ab7dac Compare March 26, 2025 00:21

github-actions bot removed multi-commit no-changelog wip labels Mar 26, 2025

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 6ab7dac to 52da68c Compare March 26, 2025 00:25

mdellweg reviewed Mar 26, 2025

View reviewed changes

pulpcore/tasking/tasks.py Outdated Show resolved Hide resolved

github-actions bot added multi-commit no-changelog labels Mar 26, 2025

pedro-psb force-pushed the add-timeout-to-immediate-task branch from 799969a to d3e0269 Compare March 26, 2025 12:51

github-actions bot removed multi-commit no-changelog labels Mar 26, 2025

pedro-psb force-pushed the add-timeout-to-immediate-task branch 2 times, most recently from f3b0a48 to ad769c9 Compare March 26, 2025 13:17

pedro-psb force-pushed the add-timeout-to-immediate-task branch from ad769c9 to dbae78a Compare March 26, 2025 13:54

mdellweg approved these changes Mar 26, 2025

View reviewed changes

pedro-psb merged commit 6b12bb2 into pulp:main Mar 26, 2025
12 checks passed

		signal.signal(signal.SIGALRM, timeout_handler)
		signal.alarm(IMMEDIATE_TASK_TIMEOUT)

		@@ -50,7 +52,6 @@ def wakeup_worker():


		def execute_task(task):
		# This extra stack is needed to isolate the current_task ContextVar

		_logger.debug(_("Task is coroutine %s"), task.pk)
		coro = func(args, *kwargs)



		@pytest.fixture
		def dispatch_long_task(pulpcore_bindings, dispatch_task):

		assert "Support for non-coroutine immediate tasks will be dropped" in stderr
		assert "Support for non-coroutine immediate tasks will be dropped" in stderr_content

[PULP-218] Add timeout to immediate tasks #6155

[PULP-218] Add timeout to immediate tasks #6155

Conversation

pedro-psb commented Dec 16, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdellweg commented Feb 14, 2025

pedro-psb commented Feb 14, 2025

mdellweg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdellweg Mar 24, 2025 • edited Loading

Choose a reason for hiding this comment

mdellweg left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pedro-psb Mar 24, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mdellweg Mar 24, 2025 •

edited

Loading

pedro-psb Mar 24, 2025 •

edited

Loading