Create a wrapper in order to be able to run two plans in parallel#75
Create a wrapper in order to be able to run two plans in parallel#75RafaelLyra8 wants to merge 7 commits intomainfrom
Conversation
ericonr
left a comment
ericonr
left a comment
There was a problem hiding this comment.
I think the two commits from this PR should be joined, because the first commit doesn't have a fully working implementation and reviewing the git history could be misleading.
Furthermore, I believe the commit message (or code comments) should expand on the purpose of this code. We need to bridge non-async code to run some asynchronous tasks simultaneously, and this means blocking on multiple async functions. This is necessary because we yield the tasks to be run, but if we receive a sleep/wait, we want to run that ourselves so the runner doesn't waste time.
| except Exception as exception: | ||
| plan_has_finished_list[list_id].set() |
There was a problem hiding this comment.
Is this intended to catch StopIteration for when a plan is finished?
There was a problem hiding this comment.
Yes if the plans was successful, but I also wanted to call it if one plan failed.
| def yield_msg(msg): | ||
| yield msg | ||
| async def yield_msg(msg): | ||
| await asyncio.wait_for(RE._command_registry[msg[0]](msg), timeout=10**10) |
There was a problem hiding this comment.
- is it safe to use this RE internal member?
- why use this timeout instead of
None?
There was a problem hiding this comment.
I only used the timeout because the asyncio returned an error if it was not passed, but I didn't wan't to use it.
There was a problem hiding this comment.
And I don't think RE internal member is supposed to be used outside of the run engine class, but because running plans in parallel is not something that the bluesky framework supports natively and because I want to use the same functions of the run engine for the message, I chose to use it.
There was a problem hiding this comment.
I think you can do timeout=None for an infinite timeout, then!
RafaelLyra8
force-pushed
the
parallel_plans_wrapper
branch
from
fac0d74 to
bb80d96
Compare
RafaelLyra8
force-pushed
the
parallel_plans_wrapper
branch
from
7fb5677 to
8d556cc
Compare
flowln
left a comment
flowln
left a comment
There was a problem hiding this comment.
I'm not sure this is working as intended, and this really should have automated tests. One example of broken behavior (the test is failing, it should succeed as far as I understand what you're trying to achieve here):
import time
import numpy as np
from bluesky import RunEngine, plan_stubs as bps
from bluesky.preprocessors import run_decorator
from sophys.common.plans.parallel_wrapper import parallel_plans_wrapper
def test_two_plans():
RE = RunEngine()
@run_decorator()
def single_plan():
yield from bps.sleep(0.5)
plan = parallel_plans_wrapper(single_plan(), single_plan())
_t = time.time()
RE(plan)
assert np.isclose(time.time() - _t, 0.5, atol=0.2), "Plan should last about half a second."
| @contextmanager | ||
| def thread_loop_manager(): | ||
| loop = asyncio.new_event_loop() | ||
| thread = Thread(target=run_loop, args=(loop,), daemon=True) | ||
| thread.start() | ||
| try: | ||
| yield loop | ||
| finally: | ||
| loop.call_soon_threadsafe(loop.stop) | ||
| thread.join() |
There was a problem hiding this comment.
I didn't know a contextmanager decorator existed. That's really compact!
| def yield_msg(msg): | ||
| yield msg | ||
| async def yield_msg(msg): | ||
| await asyncio.wait_for(RE._command_registry[msg[0]](msg), timeout=10**10) |
There was a problem hiding this comment.
I think you can do timeout=None for an infinite timeout, then!
| parallel_plan_list = list(args) | ||
| plan_has_finished_list = {} | ||
| current_event = {} | ||
| current_thread = {} |
There was a problem hiding this comment.
maybe rename to blocking_task_future?
| # Monitor 'wait' thread until its finished | ||
| if not current_thread[list_id].done(): | ||
| current_thread[list_id] = None |
There was a problem hiding this comment.
Shouldn't it be if current_thread[list_id].done()? The way it's written here you're throwing out the future when it's still not done.
| except Exception as exception: | ||
| plan_has_finished_list[list_id].set() | ||
| print(exception) |
There was a problem hiding this comment.
What part of the code are we expecting to throw exceptions here? I think that should be documented in a comment. Is it just the next call? If you want to catch exceptions from the tasks passed to the async loop, you need to call result() in the if ... done() block.
|
|
||
| def parallel_plans_wrapper(*args): | ||
| parallel_plan_list = list(args) | ||
| plan_has_finished_list = {} |
There was a problem hiding this comment.
Seeing as this list is now only touched in the same thread inside of this class, could it be moved to use booleans instead of Events? It would also simplify all_plans_have_finished above.
3 participants
No description provided.