Runtimeerror Cannot Enter Into Task

[Randell Magtoto]

Iam using FastAPI, and strangely, only on the first API call I get the error "RuntimeError: Cannot enter into task while another task is being executed." I have supplied some sample code that should reproduce this error.

The links provide so much detail that I cannot figure out what the problem is. I am not doubting there is a problem, I just need help understanding it. Could someone give a scenario that shows the need for back-propagation without going into too much detail of FastAPI? (I am not familiar with it so the linked issues are difficult for me to follow.)

I take it the middleware does not run in the same worker thread as the endpoint? But it would like to see the user (which is apparently parsed out of the request by the endpoint code) when logging the response.

Then whatever is using it can manually pass the context around (e.g. passing the object into a worker thread) thus giving it control of how the execution looks from the outside by composing current_context() with asyncio.to_thread() and asyncio.create_task()

A synchronous function acts as an entrypoint to the engine. The engine executes on a dedicated "global loop" thread. For asynchronous flow calls, we return a coroutine from the entrypoint so the user can enter the engine without blocking their event loop.

The flow run is orchestrated through states, calling the user's function as necessary. Generally the user's function is sent for execution on the user thread. If the flow function cannot be safely executed on the user thread, e.g. it is a synchronous child in an asynchronous parent it will be scheduled on a worker thread instead.

A future is created for the task call. Creation of the task run and submission to the task runner is scheduled as a background task so submission of many tasks can occur concurrently.

The engine branches depending on if a future, state, or result is requested. If a future is requested, it is returned immediately to the user thread. Otherwise, the engine will wait for the task run to complete and return the final state or result.

An engine function is submitted to the task runner. The task runner will schedule this function for execution on a worker. When executed, it will prepare for orchestration and wait for completion of the run.

Note that the flow_run contains a parameters attribute which is the serializedparameters sent to the backend while the parameters argument here should be thedeserialized and validated dictionary of python objects.

This method may be called from a worker so we ensure the settings context has beenentered. For example, with a runner that is executing tasks in the same event loop,we will likely not enter the context again because the current context alreadymatches:

main thread:--> Flow called with settings A--> begin_task_run is scheduled on a remote worker, settings A is serializedremote worker:--> Remote worker imports Prefect (may not occur)--> Global settings is loaded with default settings--> begin_task_run executes on a different event loop than the flow--> Current settings is not set or does not match, settings A is entered

Subflows differ from parent flows in that they- Resolve futures in passed parameters into values- Create a dummy task for representation in the parent flow- Retrieve default result storage from the parent flow rather than the server

Differs from enter_flow_run_engine_from_flow_call in that we have a flow run idbut not a flow object. The flow must be retrieved before execution can begin.Additionally, this assumes that the caller is always in a context without an eventloop as this should be called from a fresh process.

Since async flows are run directly in the main event loop, timeout behavior willmatch that described by anyio. If the flow is awaiting something, it willimmediately return; otherwise, the next time it awaits it will exit. Sync flowsare being task runner in a worker thread, which cannot be interrupted. The workerthread will exit at the next task call. The worker thread also has access to thestatus of the cancellation scope at FlowRunContext.timeout_scope.cancel_calledwhich allows it to raise a TimeoutError to respect the timeout.

When the user function is run, the result will be used to determine a final state- if an exception is encountered, it is trapped and stored in a FAILED state- otherwise, return_value_to_state is used to determine the state

The final state is then proposed- if accepted, this is the final state and will be returned- if rejected and a new final state is provided, it will be returned- if rejected and a non-final state is provided, we will attempt to enter a RUNNING state again

When called within a flow run, execution will block and no downstream tasks willrun until the flow is resumed. Task runs that have already started will continuerunning. A timeout parameter can be passed that will fail the flow run if it has notbeen resumed within the specified time.

a flow run id. If supplied, this function will attempt to pausethe specified flow run outside of the flow run process. When paused, theflow run will continue execution until the NEXT task is orchestrated, atwhich point the flow will exit. Any tasks that have already started willrun until completion. When resumed, the flow run will be rescheduled tofinish execution. In order pause a flow run in this way, the flow needs tohave an associated deployment and results need to be configured with thepersist_results option.

the number of seconds to wait for the flow to be resumed beforefailing. Defaults to 1 hour (3600 seconds). If the pause timeout exceedsany configured flow-level timeout, the flow might fail even after resuming.

Flag that will reschedule the flow run if resumed. Instead ofblocking execution, the flow will gracefully exit (with no result returned)instead. To use this flag, a flow needs to have an associated deployment andresults need to be configured with the persist_results option.

An optional key to prevent calling pauses more than once. This defaults tothe number of pauses observed by the flow so far, and prevents pauses thatuse the "reschedule" option from running the same pause twice. A custom keycan be supplied for custom pausing behavior.

a subclass of RunInput or any type supported byPydantic. If provided when the flow pauses, the flow will wait for theinput to be provided before resuming. If the flow is resumed withoutproviding the input, the flow will fail. If the flow is resumed with theinput, the flow will resume and the input will be loaded and returnedfrom this function.

When suspended, the flow run will continue execution until the NEXT task isorchestrated, at which point the flow will exit. Any tasks that havealready started will run until completion. When resumed, the flow run willbe rescheduled to finish execution. In order suspend a flow run in thisway, the flow needs to have an associated deployment and results need to beconfigured with the persist_results option.

the number of seconds to wait for the flow to be resumed beforefailing. Defaults to 1 hour (3600 seconds). If the pause timeoutexceeds any configured flow-level timeout, the flow might fail evenafter resuming.

An optional key to prevent calling suspend more than once. Thisdefaults to a random string and prevents suspends from running thesame suspend twice. A custom key can be supplied for customsuspending behavior.

a subclass of RunInput or any type supported byPydantic. If provided when the flow suspends, the flow will remainsuspended until receiving the input before resuming. If the flow isresumed without providing the input, the flow will fail. If the flow isresumed with the input, the flow will resume and the input will beloaded and returned from this function.

Try adding Microsoft Excel into the task list, I bet it is missing and that string literal cannot access the specific task because it does not exist. Copy the users task list on the machine that this works on and copy it over to any other machines that you wish to run this on.

My problems were very similar to the write access errors mentioned here: [4.2.1] whoosh search not working, initiate reindex search yields to errors in log (#1096) Issues Mayan EDMS / Mayan EDMS GitLab

The idea was, only 1 object should be entered into the whoosh database at the same time. Because according to the explanation in issue #1096 currently not more than one process can write to the whoosh database at the same time.

I suspect that a WRITELOCK of another process was not cleared away before. After that all further write attempts to the whoosh database fail, because the lock is never cleared again. Before the LockErrors the following is in the log:

This setting allows indexing many things in a single request by building a data structure of many objects. It is only used for bulk indexing tasks, specially to avoid creating a task entry for each object in the system, such as if you have 1M document this avoid 1M task entries in RabbitMQ and the associated memory usage spike. Bulk indexing is not used for search refresh during normal use.

Raising a LockError is not fatal and is the intended protection. This is similar to an Ethernet collision (Collision domain - Wikipedia). It just means the lock system worked and the later task is going to be re-schuled and retried later.

This is an interesting hypothesis. Research will be needed to determine if a new Whoosh process is able to invalidate or take control of an existing orphan Whoosh WRITELOCK. That should be the case given that Whoosh works as a single process per index update and this would be a common scenario expected by the library.

thank you very much for your explanations. I now understand some things better than before. I will try the debugging and then give feedback if I could find something interesting. I use direct deployment and not a docker environment. So I need to slightly modify your commands for debugging. But I know what I need to do.

3a8082e126