https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> why this crash happen when running `...
# ask-marvin
y
@Marvin why this crash happen when running 
prefect concurrency-limit ls
Copy code
E:\dev\data-nebula\.venv\Lib\site-packages\pydantic_settings\main.py:426: UserWarning: Config key `pyproject_toml_table_header` is set in model_config but will be ignored because no PyprojectTomlConfigSettingsSource source is configured. To use this config key, add a PyprojectTomlConfigSettingsSource source to the settings sources via the settings_customise_sources hook.
  self._settings_warn_unused_config_keys(sources, self.model_config)
E:\dev\data-nebula\.venv\Lib\site-packages\pydantic_settings\main.py:426: UserWarning: Config key `toml_file` is set in model_config but will be ignored because no TomlConfigSettingsSource source is configured. To use this config key, add a TomlConfigSettingsSource source to the settings sources via the settings_customise_sources hook.
  self._settings_warn_unused_config_keys(sources, self.model_config)
E:\dev\data-nebula\.venv\Lib\site-packages\pydantic\_internal\_generate_schema.py:2249: UnsupportedFieldAttributeWarning: The 'default' attribute with value 'UTC' was provided to the `Field()` function, which has no effect in the context it was used. 'default' is field-specific metadata, and can only be attached to a model field using `Annotated` metadata or by assignment. This may have happened because an `Annotated` type alias using the `type` statement was used, or if the `Field()` function was attached to a single member of a union type.
  warnings.warn(
                                       Concurrency Limits
┏━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┓
┃ Tag            ┃                                   ID ┃ Concurrency Limit ┃ Active Task Runs ┃
┡━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━┩
│ wind_api_query │ 960cac05-cdd7-4d11-9fa3-0a9733f2ae45 │ 1                 │ 0                │
└────────────────┴──────────────────────────────────────┴───────────────────┴──────────────────┘
                    inspect a concurrency limit to show active task run IDs
Traceback (most recent call last):
  File "C:\Program Files\Python311\Lib\asyncio\windows_events.py", line 439, in select
    self._poll(timeout)
RuntimeError: <_overlapped.Overlapped object at 0x000001F54452D3B0> still has pending operation at deallocation, the process may crash
Traceback (most recent call last):
  File "C:\Program Files\Python311\Lib\asyncio\windows_events.py", line 439, in select
    self._poll(timeout)
RuntimeError: <_overlapped.Overlapped object at 0x000001F54452DB30> still has pending operation at deallocation, the process may crash
m
✅ thought for 422.9 seconds
Thanks for sharing the traceback — this is a Windows + asyncio shutdown quirk, not a failure of the command itself. What’s happening - Your 
prefect concurrency-limit ls
command actually ran and printed the table; the error happens during process shutdown. - On Windows, asyncio uses OS “overlapped” I/O. If the event loop is closed while any overlapped operation is still pending, Python can emit: RuntimeError: <_overlapped.Overlapped ...> still has pending operation at deallocation, the process may crash - This typically shows up when async HTTP connections (e.g., httpx/anyio under the hood) or other OS handles are still tearing down at interpreter exit. Why here - Prefect’s CLI runs async commands via asyncio.run(). On Windows, Prefect forces the Proactor event loop policy for the CLI to avoid issues spawning subprocesses. You can see that here: Prefect CLI sets WindowsProactorEventLoopPolicy - If another library or environment hook changes the event loop policy back to the Selector loop (or otherwise interferes), you can get exactly the error you’re seeing. Your traceback points into 
asyncio/windows_events.py
, which suggests the Selector loop is active. How to fix or mitigate 1) Verify versions and loop policy - Please share: - 
prefect version
- Output of:
Copy code
python -c "import asyncio, sys; print(sys.version); print(asyncio.get_event_loop_policy())"
If you see a Selector policy on Windows, that’s the culprit. 2) Update Prefect and Python - Make sure you’re on the latest Prefect 3.x (we assume 3.x unless you say otherwise). - If possible, try Python 3.12+ — there have been Windows asyncio shutdown improvements vs 3.11. 3) Ensure nothing overrides the loop policy - Check for libraries or sitecustomize/startup code that set an event loop policy (e.g., anything trying to set Selector on Windows). - uvloop is not supported on Windows, but some stacks still try to poke the loop policy. 4) Force Proactor policy for this run (diagnostic) - As a quick test, run the CLI through a tiny wrapper that forces the policy first:
Copy code
py -c "import asyncio, sys, subprocess; asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()); sys.exit(subprocess.call(['prefect','concurrency-limit','ls']))"
If the error goes away, the issue was the loop policy being changed elsewhere. 5) Short-term noise suppression (not ideal) - If the error is only on shutdown and the command output is fine, you can suppress it:
Copy code
set PYTHONWARNINGS=ignore::RuntimeError
  prefect concurrency-limit ls
Notes - I could not find an open Prefect issue specific to this traceback. It’s a known class of Windows asyncio shutdown warnings rather than a Prefect-specific failure. - The CLI command itself is valid (verified with help). For reference:
Copy code
prefect concurrency-limit ls --help
If you can share your Prefect version, Python version, and the event loop policy output, I can help pinpoint exactly what’s toggling the loop on your machine and get you to a clean exit.
y
@Marvin 140815.844 | INFO | Task run 'get_stock_list-168' - Finished in state Completed()
Copy code
14:08:40.046 | ERROR   | Task run 'get_stock_list-004' - Crash detected! Execution was cancelled by the runtime environment.
14:08:40.049 | ERROR   | Task run 'get_stock_list-004' - Finished in state Crashed('Execution was cancelled by the runtime environment.')
14:08:40.058 | ERROR   | Task run 'fetch_wind_api_data-243' - Crash detected! Execution was cancelled by the runtime environment.
14:08:40.060 | ERROR   | Task run 'fetch_wind_api_data-243' - Finished in state Crashed('Execution was cancelled by the runtime environment.')
14:08:40.066 | ERROR   | Task run 'upload_to_nas-688' - Crash detected! Execution was cancelled by the runtime environment.
14:08:40.069 | ERROR   | Task run 'upload_to_nas-688' - Finished in state Crashed('Execution was cancelled by the runtime environment.')
14:09:10.468 | ERROR   | Task run 'fetch_wind_api_data-0ab' - Crash detected! Execution was cancelled by the runtime environment.
14:09:10.471 | ERROR   | Task run 'fetch_wind_api_data-0ab' - Finished in state Crashed('Execution was cancelled by the runtime environment.')
14:09:10.477 | ERROR   | Task run 'upload_to_nas-56a' - Crash detected! Execution was cancelled by the runtime environment.
14:09:10.479 | ERROR   | Task run 'upload_to_nas-56a' - Finished in state Crashed('Execution was cancelled by the runtime environment.')
14:09:10.482 | INFO    | Flow run 'enthusiastic-chupacabra' - Finished in state Completed()
Is this related? Below is code for the main flow
Copy code
@flow(name="source_file_download_refdata")
def source_file_download_refdata_flow(
    business_dates: str = "",
    local_root_path: str = "E:\\refdata_source_download",
    remote_root_path: str = "/data/reference_data/china/source",
):
    logger = get_logger(__name__)
    bdates_dt = get_business_dates(
        timezone="Asia/Shanghai",
        business_date_str=business_dates,
    )
    trade_calendar_list = fetch_china_stock_calendar()
    process_trade_dates = filter_trade_date(bdates_dt, trade_calendar_list)

    if not process_trade_dates:
        logger.info("No trade dates to process.")
        return

    # Prefect 3 mapping
    stock_lists = get_stock_list.map(process_trade_dates)

    download_path_lists = fetch_wind_api_data.map(
        process_trade_dates,
        stock_list=stock_lists,
        local_root_path=unmapped(local_root_path),
    )

    upload_to_mumbai_nas.map(
        local_file_paths=download_path_lists,
        remote_root_path=unmapped(remote_root_path),
        trade_date=process_trade_dates,
    )
m
✅ thought for 171.4 seconds
Short answer: It can be related, but the two messages indicate different layers. - The Windows “Overlapped … pending operation” error you saw earlier is a CLI shutdown artifact. It happens after the CLI prints the table and is usually unrelated to runtime task execution. - The “Crash detected! Execution was cancelled by the runtime environment.” in your task logs means those task runs were cancelled by the worker/runtime (e.g., cancellation scope triggered, process/pod termination, or a concurrency/timeout/cancellation event), which Prefect records as a Crashed state with that specific message. Why your flow shows Completed while some mapped tasks crashed - In Prefect 3.x, a flow’s terminal state depends on what the flow returns, not on the internal task states, unless you explicitly return the futures/states. - In your flow, you don’t return the mapped futures; you return nothing. That means the flow itself can finish Completed even if some mapped child tasks crash. Common causes of “Execution was cancelled by the runtime environment” - Worker/process termination (e.g., Ctrl+C, container/pod shutdown, OS signal). - Task cancellation from the task runner/runtime (e.g., cancellation scope closed). - Hitting a task-level timeout or a work queue/work pool shutdown. - Concurrency limits or resource starvation causing cancellation/cutoff in the environment (e.g., Windows process exits). What to check next 1) Worker/Agent lifecycle around those timestamps - Did the process get stopped/restarted? Any logs from your worker/pool around 1408–1409? - Any OS signals, container/pod events, or CI job timeouts? 2) Task timeouts - Are 
get_stock_list
, 
fetch_wind_api_data
, or 
upload_to_mumbai_nas
decorated with 
timeout_seconds
or do they call APIs that may be cancelled? 3) Concurrency limits and mapping scale - You’re mapping over trade dates and then fanning out further. If many tasks start at once, ensure the worker has enough slots/resources and that you’re not hitting limits. - You can cap parallelism using a task runner concurrency limit or concurrency limits on tags. 4) Flow intent: fail flow if any child crashes? - If you want the flow to fail when any mapped task crashes, return the futures to propagate state:
Copy code
@flow(name="source_file_download_refdata")
  def source_file_download_refdata_flow(...):
      ...
      stock_lists = get_stock_list.map(process_trade_dates)

      download_path_lists = fetch_wind_api_data.map(
          process_trade_dates,
          stock_list=stock_lists,
          local_root_path=unmapped(local_root_path),
      )

      uploads = upload_to_mumbai_nas.map(
          local_file_paths=download_path_lists,
          remote_root_path=unmapped(remote_root_path),
          trade_date=process_trade_dates,
      )

      return uploads  # Causes the flow to be FAILED if any crashed/failed
5) Add diagnostics in tasks - Log start/end per mapped item and add try/finally blocks to see if tasks are being interrupted mid-execution (typical of cancellation). - If using external clients (HTTP, file I/O, network shares like NAS), ensure proper cleanup and timeouts. If you can share: - Prefect version, Python version, OS, and how you’re running (local process, Docker, Kubernetes). - Whether a worker/agent restarted around the crash time. - Any timeouts set on the tasks. I can help pinpoint whether this is environmental cancellation vs. something in the task code and suggest the cleanest mitigation.
y
@Marvin setting concurrency to 1(each time only 1 task can login in and download) seems much slower than synchronize running, why is that?
m
✅ thought for 198.4 seconds
Short version: yes, that’s expected. Setting a concurrency limit of 1 keeps all the orchestration overhead of running many mapped task runs, but removes the throughput benefits of parallelism—so it will be slower than just running the work synchronously in-process. Why it’s slower with limit=1 - Each mapped task run still pays orchestration costs: API round-trips to create the task run, state transitions (Scheduled → Pending → Running → Completed), log streaming, acquiring/releasing the concurrency slot, etc. - With a tag limit of 1 (your 
wind_api_query
tag), all those task runs queue up at the server and execute one-by-one. You’ve kept the cost of “many small tasks” but removed parallelism. - If each task logs in separately or opens a new client/session, you’ll amplify the overhead further vs reusing a single session in a loop. - On Windows, thread/process context switching and network I/O can add a bit of extra latency per task run. Patterns to make it fast while staying single-threaded for login - Collapse to a single task that loops: - Do the login once. - Iterate over your trade dates and stocks in a plain Python loop inside one task. - Return the list of downloaded paths. Then, if uploads can be parallel, map or submit those separately. Example restructure
Copy code
from prefect import flow, task, unmapped
from prefect.concurrency.sync import concurrency

@task
def download_all(process_trade_dates, local_root_path):
    # Optional: if you need to enforce cross-flow exclusivity for this login/API
    with concurrency("wind_api_query", occupy=1):
        session = login_once()
        all_paths = []
        for dt in process_trade_dates:
            stock_list = get_stock_list_impl(dt, session)  # plain Python, not a task
            paths = fetch_wind_api_data_impl(dt, stock_list, local_root_path, session)
            all_paths.extend(paths)
        return all_paths

@task
def upload_to_mumbai_nas(local_file_paths, remote_root_path, trade_date):
    ...

@flow
def source_file_download_refdata_flow(business_dates: str = "",
                                     local_root_path: str = "E:\\refdata_source_download",
                                     remote_root_path: str = "/data/reference_data/china/source"):
    # compute process_trade_dates synchronously as you do now
    process_trade_dates = ...  # your existing logic

    # Sequential login + download with minimal orchestration overhead
    all_paths = download_all(process_trade_dates, local_root_path)

    # If uploads don’t have the login restriction, fan out here
    upload_to_mumbai_nas.map(
        local_file_paths=all_paths,
        remote_root_path=unmapped(remote_root_path),
        trade_date=unmapped(None)  # or map per-date if needed
    )
Alternative options - If you must keep mapping for observability, batch it: - Map over larger batches (e.g., per trade date) and do the inner loop inside each task to reduce the number of task runs. - If you want futures but still strictly sequential, you can use a single-worker runner: - 
@flow(task_runner=ThreadPoolTaskRunner(max_workers=1))
- Still slower than a single looping task, because you keep per-task orchestration overhead. Docs for reference - Tag-based concurrency limits: https://docs.prefect.io/v3/how-to-guides/workflows/tag-based-concurrency-limits - Global concurrency + rate limits: https://docs.prefect.io/v3/how-to-guides/workflows/global-concurrency-limits - Run work concurrently and mapping guidance: https://docs.prefect.io/v3/how-to-guides/workflows/run-work-concurrently - Task runners overview: https://docs.prefect.io/v3/concepts/task-runners If you share roughly how many mapped items you run per flow (trade dates × stocks) and whether each task logs in anew, I can suggest a concrete batching size and code sketch tailored to your workload.
y
@Marvin PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS is this causing the delay, how to customize that for this tag?
m
✅ thought for 91.2 seconds
Great question. Yes, that setting controls how long a task waits before retrying to acquire a tag-based concurrency slot, and it does contribute to perceived delay when a limit is saturated. What the setting is and scope - The setting is global on the Prefect server, not per-tag. - Supported names: - PREFECT_SERVER_TASKS_TAG_CONCURRENCY_SLOT_WAIT_SECONDS (preferred) - PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS (legacy alias) - prefect.toml key: server.tasks.tag_concurrency_slot_wait_seconds - Default: 30 seconds (the server uses a randomized delay around this value to avoid thundering herd). Impact - When your tag (e.g., wind_api_query) has limit=1 and multiple tasks compete for a slot, the server tells waiting tasks to try again after roughly this many seconds. If you have many waiting tasks, they’ll step forward in ~30s increments on average, which looks “slow.” How to customize - Change it on the server (applies to all tags): - Environment variable:
Copy code
export PREFECT_SERVER_TASKS_TAG_CONCURRENCY_SLOT_WAIT_SECONDS=10
- Or in prefect.toml on the server:
Copy code
[server.tasks]
    tag_concurrency_slot_wait_seconds = 10
- Or via CLI config:
Copy code
prefect config set PREFECT_TASK_RUN_TAG_CONCURRENCY_SLOT_WAIT_SECONDS=10
- Note: There is no per-tag slot-wait setting today. It’s global. If you need per-tag behavior - Consider using global concurrency limits with the Python 
concurrency()
context manager in code to shape different resources differently. - Or split work so that only the login/critical section is guarded by the concurrency limit, then fan out downstream work without the tag. Tip - Lowering the wait seconds increases server churn (more frequent retries). Start with something like 5–10 seconds and observe.
y
@Marvin what is the best practice for control not downloading if it is holiday?
Copy code
@flow(
    name="future_bars_download",
#     on_completion=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_SUCCESS_CHANNEL])],
#     on_failure=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
#     on_cancellation=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
#     on_crashed=[notify_slack_handler(SLACK_BOT_TOKEN, [SLACK_FAILED_CHANNEL])],
)
def future_bars_download(business_dates: str = "",
                         exchanges: List[str] = None,
                         eod_root: str = "E:\\wind_futures_eod_bars",
                         min_root: str = "E:\\wind_futures_1min_bars"):
    if exchanges is None:
        raise ValueError("Exchanges must be provided as a list of strings")
    bdates_dt = get_business_dates(business_date_str=business_dates, timezone="Asia/Shanghai")

    for exchange in exchanges:
        # step 1 - get valid trading dates
        valid_dates = get_valid_exchange_dates(exchange, bdates_dt)
        if not valid_dates:
            return states.Cancelled
        for trade_date in valid_dates:
            contracts = get_trading_contracts(exchange, trade_date)
            eod_bars_df = download_eod_bars(contracts, trade_date)
            save_data(eod_bars_df, eod_root, exchange, trade_date)

            min_bars_df = download_min_bars(contracts, trade_date)
            save_data(min_bars_df, min_root, exchange, trade_date)
m
✅ thought for 278.7 seconds
Great question. In Prefect 3.x, there are two complementary best practices for “do nothing on holidays”: 1) Don’t schedule runs on holidays (best) - Use an RRule schedule that only includes business days and excludes known holidays. - This prevents the flow from starting at all. Example: build an RRule with MO–FR plus EXDATEs you generate from your exchange calendar and attach it to your deployment.
Copy code
from prefect import flow
from prefect.schedules import RRule

@flow
def future_bars_download(...):
    ...

# Example: inject EXDATEs (UTC times) for known holidays
holiday_dates = ["20250101", "20250201"]  # YYYYMMDD
exdates = "\n".join(f"EXDATE:{d}T000000Z" for d in holiday_dates)
rule = f"FREQ=DAILY;BYDAY=MO,TU,WE,TH,FR\n{exdates}"

future_bars_download.serve(
    name="futures-download",
    schedules=[RRule(rule, timezone="Asia/Shanghai")]
)
Notes: - RRule supports exclusions via EXDATE (specific dates) and can handle complex patterns. - This is ideal if you can maintain a holiday list per exchange. 2) Guard in the flow (simple, reliable fallback) - If a run is triggered anyway (backfills, ad-hoc, or unexpected holidays), return early before doing work. - In 3.x, don’t return `states.Cancelled`; either just return (COMPLETED, no-op) or raise 
CancelledRun
to mark as cancelled. Here’s a tightened version of your flow that: - Uses the scheduled date (not “now”) so backfills behave correctly - Skips per exchange without exiting the entire flow - Cleanly no-ops when no exchanges have valid dates
Copy code
from datetime import datetime
from zoneinfo import ZoneInfo
from typing import List

from prefect import flow, get_run_logger
from prefect.exceptions import CancelledRun
from prefect.runtime import flow_run

@flow(log_prints=True)
def future_bars_download(
    business_dates: str = "",
    exchanges: List[str] = None,
    eod_root: str = "E:\\wind_futures_eod_bars",
    min_root: str = "E:\\wind_futures_1min_bars",
    cancel_on_holiday: bool = False
):
    if exchanges is None:
        raise ValueError("Exchanges must be provided as a list of strings")

    logger = get_run_logger()
    tz = ZoneInfo("Asia/Shanghai")
    scheduled = flow_run.scheduled_start_time
    run_dt_local = (scheduled.astimezone(tz) if scheduled else datetime.now(tz))
    run_date = run_dt_local.date()

    # Your own logic to resolve business dates for this run
    bdates_dt = get_business_dates(business_date_str=business_dates, timezone="Asia/Shanghai")

    any_work = False
    for exchange in exchanges:
        valid_dates = get_valid_exchange_dates(exchange, bdates_dt)
        if not valid_dates:
            <http://logger.info|logger.info>(f"{exchange}: holiday or no valid trading dates for {run_date}; skipping this exchange")
            continue

        any_work = True
        for trade_date in valid_dates:
            contracts = get_trading_contracts(exchange, trade_date)
            eod_bars_df = download_eod_bars(contracts, trade_date)
            save_data(eod_bars_df, eod_root, exchange, trade_date)

            min_bars_df = download_min_bars(contracts, trade_date)
            save_data(min_bars_df, min_root, exchange, trade_date)

    if not any_work:
        msg = f"No valid trading dates for any exchange on {run_date}; skipping run"
        <http://logger.info|logger.info>(msg)
        if cancel_on_holiday:
            # Optional: mark run as Cancelled instead of Completed
            raise CancelledRun(msg)
        return  # completes successfully as a no-op
Why this is better than `return states.Cancelled`: - In Prefect 3.x, returning a state object is not how you set the run state. - To explicitly mark the run as cancelled, raise 
CancelledRun("reason")
. Otherwise, just return early to complete cleanly with no work. Which one should you choose? - If you control the calendar ahead of time: prefer the RRule schedule exclusion so runs aren’t created on holidays.
- If holiday logic is dynamic or exchange-specific per day: keep the guard in the flow and skip per exchange, only cancelling or returning early when all exchanges are closed. If you can share: - Which exchanges/calendars you need (e.g., SHFE/DCE/CZCE) and how you derive 
get_valid_exchange_dates
- Whether you want to skip entire runs or only skip specific exchanges on mixed-holiday days …I can tailor the schedule string or the guard logic accordingly.
12 Views
Open in Slack
PreviousNext
Powered by