https://prefect.io logo
Join Slack
Powered by
Hi folks, I’ve been trying out <prefect-multiproce...
# ask-community
j
Hi folks, I’ve been trying out prefect-multiprocess to parallelize tasks, and it’s ultimately working but internally is throwing some serialization errors, and I’d like to figure out how to avoid that. Is anyone familiar with this package or have experience working with it? Details in thread
specifically I’m getting 
TypeError: cannot pickle '_thread.lock' object
, though the error message does not bother to say what object or where it is. As near as I can tell it’s something inside the 
Flow
object that’s included in the return value from 
prefect.context.serialize_context
, and then the error happens when trying to pickle that return value. I don’t think this is something my code is setting directly, so I’m not sure what we could change to avoid it
The particularly odd thing is that it only throws this error the first time – I have 
retries=1
specified on the flow decorator, and it runs as expected when retrying
Upon further investigation, it appears that the flow’s task runner is 
MultiprocessTaskRunner
the first time, which then fails with the pickle error I described above. When the flow is retried, the task runner instead becomes 
ThreadPoolTaskRunner
. @Ryan Peden - apologies for the direct ping but I believe 
prefect-multiprocess
is your library, is this something you recognize by any chance? (See also this thread; I’ll see if I can repro this more minimally.)
r
Hey, sorry for the slow reply - I'm happy to take a look. I'm fine with direct pings. Best way to get my attention. :) I have an idea about what's happening - you're probably seeing a 
ThreadPoolTaskRunner
during the retry because that's what runs inside each of the process workers. That part seems expected at least, but I'll try to reproduce the error you're seeing so I can improve the library. I'll test it out tomorrow and see if I can push a fix to PyPI. Are you using the task runner with Prefect 2, or are you using the prerelease version that works with Prefect 3?
thank you 1
j
Awesome, thank you. (It can totally wait until Monday though!)
this is Prefect 3; specifically I have 
prefect-multiprocess = "0.2.0b3"
I was working on filing a bug report, so here’s the reproducer I was able to come up with so far:
Copy code
from time import sleep

from ddtrace import tracer
from prefect import context, flow, get_run_logger, task
from prefect_multiprocess.task_runners import MultiprocessTaskRunner

@tracer.wrap(name="prefect.task", resource="mp_test_task")
def mp_test_task(val: int) -> int:
    sleep(1)
    return val

@flow(retries=1, retry_delay_seconds=2, task_runner=MultiprocessTaskRunner)
@tracer.wrap(name="prefect.flow", resource="mp_test_flow")
def mp_test_flow(count: int) -> list:
    task_runner = context.get_run_context().task_runner
    get_run_logger().info(f"*** {task_runner=}")

    vals = list(range(count))
    task_runs = task(mp_test_task).map(vals)
    results = task_runs.result()
    return results
The initial failure is, I think, because of some lock object inside the Datadog tracer object, so the 
cannot pickle
error is probably “real” in that sense. But the task runner being a different class is not something I expected, and it failing over silently made it harder to debug, because from the outside it looked like the flow was working, just slowly
let me know if I should file a bug on the main Prefect repo (per the discussion here) or on `prefect-multiprocess`; if either would help I’m happy to do so. I also can provide logs, debug info, etc
r
Hmm - I don't think this is a bug with Prefect itself. It's more that the Datadog tracer might not be portable across the process boundary. I can think of a few potential fixes/workarounds that I'm going to try. You can open an issue on 
prefect-multiprocess
if you'd like - that'll make it easy to track this.
j
Will do, thanks
If the Datadog object really isn't pickleable, that's not Prefect's fault, but even so I feel like it shouldn't change the task runner type on me unexpectedly. At the very least maybe it should emit a log warning saying something like "Falling back to task_runner=ThreadPoolTaskRunner” just so it's visible
r
That's totally fair, and it's actually the reason I've kept this as a beta and haven't done a final release yet. I have a solution in mind now that I think will get rid of the the task runner swap. Just need to add a bit of code and some tests, then I'll probably cut a new beta. As for the Datadog object not being picklable - I'll see what I can do. We won't be able to pickle a lock but I might be able to suggest some slightly different ways of creating the tracer wrap. I'll dig into datadog's library a bit to see what it's doing. I've worked with it in the past and vaguely remember working around something like this before, but it's been a couple of years and I don't remember the fine details.
👍 1
j
okay, I filed https://github.com/rpeden/prefect-multiprocess/issues/20 with some of the context for this
4 Views
Open in Slack
PreviousNext
Powered by