Hi everybody, im having an issue with Ray. Its creating multiple db connections (server db) but its not closing them. I have 100 connections that stay idle. How can i clean these connections from prefect server. One solution is to have a job that cleans these connections every N minutes but im wondering if there is an existing setting that takes care of this. Thank you

Hey Tedi! Please refrain from tagging folks directly 🙇 . Are you explicitly closing the connections at any point during the task or flow? Something like this?

import psycopg2  # Or the library you're using to connect to the database

@task
def query_database():
    conn = psycopg2.connect("your_connection_string")
    try:
        # Perform database operations
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM your_table")
            results = cur.fetchall()
    finally:
        # Ensure the connection is always closed
        conn.close()

@Bianca Hoch My apologies. Im not creating the connection. Prefect and Ray are creating it.

can you explain what database connections you’re talking about and how you’re observing them?

I have a flow that runs a function in parallel. My team is using Ray. Im decorating my function like this:

@flow(
    name="some_name",
    task_runner=RayTaskRunner(init_kwargs={"num_cpus": NUM_TRANSFORM_CPUS}),
)
def main(batches, data_dir,save_dir, demo_lavel,config):
    futures =t_signal.temp_transform_batches.map(
        question_configs=batches,
        data_dir=data_dir,
        save_dir=save_dir,
        level=demo_lavel,
        config_obj=unmapped(config.CONFIG),  # type: ignore
    )
    for future in futures:
        future.wait()

hmm while trying to minimally reproduce your error I ran into what I believe is a bug in either pydantic or prefect

FAILED tests/test_task_runners.py::TestRayTaskRunner::test_can_run_many_tasks_without_crashing[default_ray_task_runner] - ray.exceptions.RaySystemError: System error: Failed to unpickle serialized exception
traceback: Traceback (most recent call last):
  File "/opt/hostedtoolcache/Python/3.9.20/x64/lib/python3.9/site-packages/ray/exceptions.py", line 51, in from_ray_exception
    return pickle.loads(ray_exception.serialized_exception)
TypeError: __init__() missing 1 required keyword-only argument: 'code'

have you seen this at all client side?

i believe i have seen this issue with joblib when i use

loky

as backend

if I'm understanding, in general with the ray and dask task runners, that is the expected behavior

I see. My team will switch to joblib with

multiprocessing

backend, and will stop creating a task for each parallel process

if you don't want to use the task runner paradigm, you're free to call arbitrary python (which may call ray) inside of tasks (at that point tasks are just tools for encapsulating work / adding retries / caching etc)

Im surprised that this declaration: @task(name="foo bar", retries=10, retry_delay_seconds=6) doesnt capture the Timeout error nor does it retry

again, can you clarify what connections you're talking about and how you're counting these?

Prefect connects to a database. It stores all the task runs/logs/flows etc? I can see in AWS RDS metrics that DatabaseConnections skyrockets when this flows runs

It stores all the task runs/logs/flows etc?

yes!

if the job crashes, the flow doesnt properly close the connection and they stay idle

if the job crashes, the flow doesnt properly close the connection and they stay idle

that sounds like a bug, I'd guess likely related to using the ray task runner

postgres can handle the

idle in transaction

connection but not

idle

are you willing to open an issue with a minimal reproduction of this?

okay