https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am looking for performance improvem...
# marvin-ai
l
@Marvin I am looking for performance improvements, i have an event based flow which requests data from an API and processes every response separately. We get multiple searches at once and they are handled in chunks, one search however is comprised of many results which are processed using map. It looks like everything starts parallel but, takes almost the same amount of time, performance is not great, what could the reason be? Here's part of the flow for reference.
Copy code
@task(cache_policy=NO_CACHE)
def processing_search_results(
    search_results, distance, rabbitmq_secret_block, pg_secret_block, config, search_parameters
):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Start processing: {datetime.now()}")

    proc_pg_utils = build_pg_utils_from_block(pg_secret_block)

    processed_batch = process_single.map(unmapped(config), search_results, distance).result(raise_on_failure=True)
    save_requests_pg(processed_batch, logger, proc_pg_utils)
    <http://logger.info|logger.info>("Requests data persistence completed.")
    <http://logger.info|logger.info>(f"Publish message: {datetime.now()}")
    publish_refresh_message(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
    )


@flow(task_runner=ThreadPoolTaskRunner())
def timocom_search_flow(
    search_parameters: dict,
    pg_secret_block: str,
    timocom_secret_block: str,
    rabbitmq_secret_block: str,
    ors_server_variable: str,
    price_calculation_server_variable: str,
    predictive_scoring_server_variable: str,
    here_maps_api_key_variable: str,
):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Starting Timocom search with parameters: {search_parameters} at {datetime.now()}")

    try:
        sanitized_search_parameters = sanitize_and_validate_search_parameters(search_parameters)
        pg_utils = build_pg_utils_from_block(pg_secret_block)

        # Timocom credentials are stored as "tenant_id": {"user": "username", "pass": "password"}
        creds = Secret.load(timocom_secret_block).get()
        tenant_ref = search_parameters["tenant_ref"]
        <http://logger.info|logger.info>(f"Using Timocom credentials for tenant: {tenant_ref}")
        tenant_creds = creds.get(tenant_ref)
        if not tenant_creds:
            raise ValueError(f"No Timocom credentials found for tenant {tenant_ref}")

        user = tenant_creds["user"]
        password = tenant_creds["pass"]

        # Initialize Timocom instance
        timocom_instance = Timocom(logger, pg_utils, user, password)
        # Create search tasks
        search_tasks = timocom_instance.create_search_tasks(sanitized_search_parameters)
        search_futures = timocom_instance.search.map(search_tasks)
        <http://logger.info|logger.info>("Timocom search process completed.")

        config = get_config(
            ors_server_variable,
            price_calculation_server_variable,
            predictive_scoring_server_variable,
            here_maps_api_key_variable,
        )

        # Process searches in batches of 5
        for batch in chunked(search_futures, 5):
            <http://logger.info|logger.info>(f"Processing batch of {len(batch)} results...")

            distance_mapping = get_distance_mapping(batch, config)

            processing_search_results.map(
                batch,
                distance_mapping,
                unmapped(rabbitmq_secret_block),
                unmapped(pg_secret_block),
                unmapped(config),
                unmapped(search_parameters),
            ).result(raise_on_failure=True)

        <http://logger.info|logger.info>("Post-processing of freight requests completed.")

    except Exception as e:
        logger.error(f"Error during Timocom search initiation: {e}")
        publish_refresh_message(
            rabbitmq_secret_block, search_parameters["tab_id"], search_parameters["job_id"], "FAILED"
        )
        raise
m
✅ thought for 220.8 seconds
Thanks for sharing the snippet — a few patterns here can quietly erase your expected parallelism. Most likely causes (and how to confirm/fix): 1) Nested mapping + a small thread pool = thread starvation - What’s happening: you run up to 5 concurrent parent tasks (
processing_search_results
) and each parent then calls 
process_single.map(...).result(...)
. While those parents wait on their children, they continue occupying worker threads. If your 
ThreadPoolTaskRunner
doesn’t have enough threads, the child tasks won’t get scheduled and everything “looks parallel” but effectively runs serially. - How to confirm: temporarily set a very low 
max_workers
(e.g., 5) and watch everything stall. Then increase it a lot and see throughput improve. - Fixes: - Don’t map inside a task when you’ll also run many of those tasks in parallel. Move the inner 
.map
up to the flow level or use a subflow to isolate a separate runner. - If you keep the nested map, raise 
max_workers
so there are enough threads for both the “outer” tasks and their “inner” mapped tasks:
Copy code
@flow(task_runner=ThreadPoolTaskRunner(max_workers=32))
    def timocom_search_flow(...):
        ...
- Avoid calling 
.result()
on large collections inside a task; return the futures and resolve them in the flow, or break the follow‑up (save/publish) into separate tasks so the parent isn’t blocking a thread. 2) Heavy synchronous work done serially in the flow thread - In your loop you call 
get_distance_mapping(batch, config)
directly in the flow. If it does API calls or heavy computation, you’re blocking the flow thread and gating every batch. - Fix: make it a task (and parallelize it if possible):
Copy code
@task
  def get_distance_mapping_task(batch, config):
      return get_distance_mapping(batch, config)

  distance_mapping = get_distance_mapping_task.submit(batch, config)
Then pass 
unmapped(distance_mapping)
(a single future) into downstream mapping. 3) CPU-bound work under ThreadPoolTaskRunner - If 
process_single
or distance/scoring logic is CPU heavy, threads won’t help due to the GIL. - Fix: move those CPU-bound pieces to 
ProcessPoolTaskRunner
(ensure arguments/returns are picklable) or split them into separate CPU-focused tasks/work pool:
Copy code
@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
  def cpu_heavy_flow(...):
      ...
4) External bottlenecks (API rate limits, DB contention) - If the upstream API or your database is rate limited, more concurrency won’t help and may worsen tail latency. - Fixes: - Tune batch size (your 
5
) to the service’s limits. - Add client-side rate limiting/backoff. - Bulk-insert to the DB and move 
save_requests_pg
into its own task so it doesn’t block the parent thread. If you’re doing per-row inserts, switch to batched operations. 5) Large data marshaling and blocking waits - Calling 
.result(raise_on_failure=True)
inside 
processing_search_results
forces all results to be materialized in that task thread. If the payloads are large, you’re paying serialization/memory costs and holding the worker. - Fix: return futures to the flow and aggregate there, or do a bulk-save task that accepts futures and resolves inside it. Concrete refactor sketch to avoid thread starvation: - Make the pre-processing and post-processing separate tasks - Keep mapping at the flow level (or isolate with a subflow) ``` @task def get_distance_mapping_task(batch, config): return get_distance_mapping(batch, config) @task def save_requests_pg_task(processed_batch, pg_secret_block): proc_pg_utils = build_pg_utils_from_block(pg_secret_block) save_requests_pg(processed_batch, get_run_logger(), proc_pg_utils) @task def publish_refresh_task(rabbitmq_secret_block, tab_id, job_id): publish_refresh_message(rabbitmq_secret_block, tab_id, job_id, "REFRESH") @flow(task_runner=ThreadPoolTaskRunner(max_workers=32)) def timocom_search_flow(...): ...
search_futures = timocom_instance.search.map(search_tasks) for batch in chunked(search_futures, 5): distance_mapping_fut = get_distance_mapping_task.submit(batch, config) # Map the inner work at the flow level processed = process_single.map( unmapped(config), batch, unmapped(distance_mapping_fut), ) # Kick off post-processing as separate tasks so we don’t block threads save_requests_pg_task.submit(processed, pg_secret_block) publish_refresh_task.submit(rabbitmq_secret_block, search_parameters["tab_id"], search_parameters["job_id"]) ... ``` A few quick questions to pinpoint the right path: - Roughly how many items does 
process_single.map(...)
handle per search? And what does 
process_single
actually do (API calls vs heavy compute)? - Have you set 
max_workers
explicitly yet? - Does 
get_distance_mapping
do network calls? If so, are they parallelizable? - Are you seeing API rate limits or DB locks/contention? If you share those details, I can suggest an exact sizing for 
max_workers
and the best split between threads vs processes. Related docs: - Task runners overview: Task runners - Concurrency and mapping: Concurrency and parallelism and Mapping (see notes on nested mapping)
l
@Tanya Hristova ⬆️
🙌 1
2 Views
Open in Slack
PreviousNext
Powered by