Related to the above, I turned on debug and added some debug statements around the end of the task. Here's the start of the logs as the flow began, and the second log is the point at which the task should have returned (but didn't).

This is the task in question:

@task(name='Pull Oracle Data',cache_for=timedelta(hours=10),timeout=(3*60*60),max_retries=3, retry_delay = timedelta(minutes=randint(5,25)),cache_validator=prefect.engine.cache_validators.all_parameters)
def pull_oracle_data_via(oracle_query_sql, server='INTP',cache_flag = False) -> DataFrame:
    if cache_flag:
        <http://logger.info|logger.info>('Task flagged to invalidate/ignore cache.')
    <http://logger.info|logger.info>(f"Initiating Oracle Pull using {oracle_query_sql}...")
    with Oracle_DB(server=server) as o_db:
        df = o_db.pull_data_from_oracle(oracle_query_sql)
    data_amount=len(df.index)
    <http://logger.info|logger.info>(f"Rows returned in oracle data query: {data_amount:,}...")
    assert data_amount!=0 # Cause task failure if 0 is returned    
    logger.debug(f"Oracle pull completed; returning...")
    return df

And here's the flow:

with Flow(FLOW_NAME,result=LocalResult(),schedule=schedule) as flow:
    cache_invalid = Parameter('cache_invalid',default=False)
    <http://logger.info|logger.info>(f"{FLOW_NAME} Flow initiated, running in {file_path}")
    df = pull_oracle_data_via(oracle_query_sql=ORACLE_QUERY_SQL_PATH,server=ORACLE_SERVER,cache_flag = cache_invalid)
    logger.debug('Oracle pull returned.')
    df = set_data_types(df)
    create_data_summary_artifacts(df)
    u=upload_to_table(df, destination_table = DATA_DESTINATION_TABLE_NAME,append=False)    
    add_indices_to_table(upstream_tasks=[u],destination_table=DATA_DESTINATION_TABLE_NAME,column_list=INDEX_COLUMNS)

flow.run_config = LocalRun(env={'PREFECT__LOGGING__FORMAT':'[%(asctime)s-%(levelname)s - %(name)s]-[%(filename)s:%(funcName)s]-Line %(lineno)d: %(message)s',
                                'PREFECT__LOGGING__LEVEL':'DEBUG',
                            },
        working_dir=file_path, labels=["normal-process"])

flow.executor = LocalDaskExecutor(scheduler='threads',num_workers=8)

flow.register(project_name=PROJECT_NAME,parameters={'cache_invalid':False},idempotency_key=(FLOW_NAME+datetime.now().strftime("%Y%m%d-%H%M%S")))

I'm on prefect 0.15.13

Is that wrong? If I install via

pip install -U prefect

the above is what I get as the most current one

That PR does indeed look like the one that fixed this issue last time (or at least one with similar behavior) but I thought that was folded in here since 0.15.11 or so; I haven't seen the issue recently until I added this new flow today and it suddenly appeared again.

Should I prefer the rc one over the current one? Given the name I assumed rc would be less stable; not so?

One sec looking at tracebacks

And I have tried to see what's different about this flow vs. others that are very similar and I don't see any difference that might cause this

It's a bit concerning to me that it occurred here though; most of my other flows are currently set to use threads; is it possible they'll start having this issue again?

I haven't had issues with it recently, but makes me nervous when I can't explain why I did here and not in the others

We’ll need the DEBUG level logs from the timeout function to diagnose where the deadlock occurs.

In this case I turned on debug in the flow (visible in the....third post here I think it was?)

Correction; fourth post.

Sound like maybe there's a debug error here too. 😏

I don’t know how the download/pagination works in the UI off the top of my head.

I'll see if I can download more to make sure I get them all; hold on...

We’re interested in the ones about the timeout handler and pickling results

The end shows the cancellation which I triggered after I reached out here

I think we had some version of this issue last time too; you had me switch to trigger the flow manually with some manual command that turned on debug more to address it I think; maybe we need to do that again?

And if so, maybe there's another issue here that debug turned on at the flow level doesn't seem to propagate fully

The task in question here is imported from another module, so maybe that breaks the log level setting somehow?

We are reaching this line https://github.com/PrefectHQ/prefect/blob/master/src/prefect/utilities/executors.py#L299

Is it possible your run is running into memory constraints or something?

In one earlier version of this test the timeout did get hit and trigger a retry

It's a long timeout though; that's what flagged me originally to look into the task to discover this error. In the one I sent you above, I cancelled before the timeout was hit

This is my available memory over the last 24 hours on this machine, so definitely not hitting memory limits

I haven't dropped below 20GB free memory at any point in the last day

Where name/project are replaced with your flow name/project

Maybe somehow getting the return_val value causes an error, causing both line 317 and the except to fail?

You seem to be an unlucky one, I thought I’d fixed this up

hahaha, you did! I hadn't seen the issue again until today!

I'm pretty excited for orion. I have to say, it looks like I'll be able to restructure all my flows to have a LOT more code reuse; hoping that comes sooner. Every one of these flows I add now is another I'll have to restructure then.

That flow is still running, btw, just waiting for it to return. Query has been taking 15-20 minutes today

logs.txt

The pickling size line did work

And then nothing...

That would suggest it's the dumps line where we're stopping

It seems like it should be a fine size

New task:

@task(name='Pull Oracle Data',cache_for=timedelta(hours=10),timeout=(3*60*60),max_retries=3, retry_delay = timedelta(minutes=randint(5,25)),cache_validator=prefect.engine.cache_validators.all_parameters)
def pull_oracle_data_via_with_pickle(oracle_query_sql, server='INTP',cache_flag = False) -> DataFrame:
    if cache_flag:
        <http://logger.info|logger.info>('Task flagged to invalidate/ignore cache.')
    <http://logger.info|logger.info>(f"Initiating Oracle Pull using {oracle_query_sql}...")
    with Oracle_DB(server=server) as o_db:
        df = o_db.pull_data_from_oracle(oracle_query_sql)
    data_amount=len(df.index)
    <http://logger.info|logger.info>(f"Rows returned in oracle data query: {data_amount:,}...")
    assert data_amount!=0 # Cause task failure if 0 is returned    
    logger.debug(f"Oracle pull completed; returning...")
    pickled_val = cloudpickle.dumps(df)
    logger.debug(f"Pickling successful")
    return df

Thinking through what else might have caused this, I do think this particular query returns fields that are likely longer than any other I've had to date.

i.e. in the pandas dataframe, a particular column will have elements with longer values than any other I've had; some could be very long

The test task just returned; same thing happens. It gets to the dumps line and then just stops

Further testing; I removed the columns that could be very long and the error went away. 😞 That basically means there's some unknown issue, likely a length limit, to the data I can get back from a database, and encountering the issue doesn't even return an error apart from a timeout. Thoughts on anything I could do to be able to keep that kind of data in here?

Or at the least to hammer down what the limit is?

I think we’re encountering this https://stackoverflow.com/questions/11854519/python-multiprocessing-some-functions-do-not-return-when-they-are-complete-que

The data is too large to place into a multiprocessing queue and it is causing a deadlock

This data regularly works fine in csv or excel format though.

Given the above, are there options here? In the meantime I'm stuck basically excluding some data from my process.

I’d report this as a bug to cloudpickle if you can make a minimal reproducible example, I don’t think pickle should hang like this.

You could attempt to serialize it another way. Perhaps by returning

<http://df.to|df.to>_json()

from your task instead of the raw data.