https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I run the following code: ```from ti...
# ask-marvin
a
@Marvin I run the following code:
Copy code
from time import sleep


@task(tags=["test-slot"])
def test_task(x):
    sleep(1)
    return x + 1


@flow
def test_flow():
    xs = list(range(1000))
    ret = test_task.map(xs)
    ret.wait()
    return ret

test_flow()
on a local prefect server (
prefect server start
) with 
tag:test-slot
limit set to 10. After approximately 800 successful tasks, they start to fail with messages:
Copy code
15:16:54.882 | ERROR   | Task run 'test_task-962' - Finished in state Pending()
Please wait for all submitted tasks to complete before exiting your flow by calling `.wait()` on the `PrefectFuture` returned from your `.submit()` calls.

Example:

from prefect import flow, task

@task
def say_hello(name):
    print(f"Hello, {name}!")

@flow
def example_flow():
    future = say_hello.submit(name="Marvin")
    future.wait()

example_flow()
and then finally the whole flow fails with exception:
Copy code
...
File /venv/lib/python3.12/site-packages/prefect/utilities/asyncutils.py:188, in run_coro_as_sync.<locals>.coroutine_wrapper()
    186     task = create_task(coroutine)
    187     if wait_for_result:
--> 188         return await task
    189 finally:
    190     RUNNING_IN_RUN_SYNC_LOOP_FLAG.reset(token1)

File /venv/lib/python3.12/site-packages/prefect/concurrency/_asyncio.py:151, in aacquire_concurrency_slots_with_lease(names, slots, mode, timeout_seconds, max_retries, lease_duration, strict, holder, suppress_warnings)
    147     raise AcquireConcurrencySlotTimeoutError(
    148         f"Attempt to acquire concurrency slots timed out after {timeout_seconds} second(s)"
    149     ) from timeout
    150 except Exception as exc:
--> 151     raise ConcurrencySlotAcquisitionError(
    152         f"Unable to acquire concurrency slots on {names!r}"
    153     ) from exc

ConcurrencySlotAcquisitionError: Unable to acquire concurrency slots on ['tag:test-slot']
Why does it happen?
m
✅ thought for 775.3 seconds
Short answer: your tasks are not failing because you forgot to call 
.wait()
— you did that correctly. They’re ending in Pending because some mapped task runs never even started; they failed while trying to acquire the tag-based concurrency slot. That Pending end-state triggers Prefect’s generic “Please wait…” hint, which is misleading in this case. What’s happening under the hood - In Prefect 3.x, each mapped 
test_task
acquires a lease for 
tag:test-slot
before it can run. With your limit at 10, tasks run in small batches. - After ~800 acquisitions/releases, your local Prefect Server (which uses SQLite by default) likely begins returning intermittent errors (e.g., “database is locked”) while handling the acquire/release churn. - When the API call to acquire a slot fails (for reasons other than “still full, please retry”), the task engine raises:
Copy code
ConcurrencySlotAcquisitionError: Unable to acquire concurrency slots on ['tag:test-slot']
The task never transitions to Running, so it “finishes” in Pending and logs that generic “Please wait…” message. - Note: Prefect’s tag-slot acquisition in 3.x is configured to wait indefinitely when slots are full. Seeing 
ConcurrencySlotAcquisitionError
indicates an API/server error — not a normal timeout and not a missing 
.wait()
. Why after ~800? - With 1,000 very short tasks and a tag limit of 10, the server processes a lot of acquire/release writes. SQLite can hit lock contention under sustained write load. Once these failures start, affected tasks never start and end in Pending; eventually the flow reports the error you saw. How to confirm - Check your Prefect Server logs around the time failures start. On local SQLite you’ll often see 500 responses and messages like “database is locked” (SQLITE_BUSY) around concurrency endpoints. Workarounds and fixes - Use Postgres for the server instead of SQLite to avoid lock contention under load. - Reduce API churn: - Batch the map in chunks so you’re not queuing 1,000 items at once:
Copy code
@flow
    def test_flow(batch=100):
        xs = list(range(1000))
        results = []
        for i in range(0, len(xs), batch):
            futs = test_task.map(xs[i:i+batch])
            futs.wait()
            # Optionally collect results
            results.extend([f.result() for f in futs])
        return results
- Or remove the tag-based limit and use a local task runner cap instead:
Copy code
from prefect.task_runners import ThreadPoolTaskRunner

    @flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
    def test_flow():
        xs = list(range(1000))
        futs = test_task.map(xs)  # no tag limit, local pool limits concurrency
        futs.wait()
        return futs
This avoids the server-side concurrency endpoints entirely for local runs. - Make sure you’re on the latest Prefect 3.x release in case of any concurrency-related fixes. If you can share: - Your Prefect version (
prefect version
) - A snippet of your server logs when the failures begin …I can help confirm it’s the SQLite lock issue and recommend the best next step (usually switching the server to Postgres or batching).
👍 1
5 Views
Open in Slack
PreviousNext
Powered by