2

I'm creating a framework that needs to execute certain tasks, I'll use building a car as an example for this. So, an individual task could be something like weld metal or screw bolt. But there are also collections of tasks, I call them jobs. Each job can have X amount of tasks, so combining these two tasks could get us a job build an engine:

build an engine +-- weld metal +-- screw bolts

Additionally, jobs can have sub-jobs:

build a car +-- build an engine | +-- weld metal | +-- screw bolts +-- build a frame | +-- weld metal | +-- screw bolts | +-- paint +-- combine engine and frame

So basically each node of my tree is a job, and each leaf is a task. This is easy enough, here's what my classes would look like:

class Task: def do(self): raise NotImplementedError('abstract method') class Job(Task): def __init__(self): self.tasks = [] def do(self): for task in self.tasks: task.do()

And now I'd just subclass WeldMetal and ScrewBolts from Task.

The problem arrives when we want to parallellise this. I need to somehow tell each task (and job) whether it should block the execution, or run in a separate thread and not block it. Sometimes my jobs also need results from other jobs, while those jobs may run in parallel to each other. For example in the above car building:

  • weld metal must block, since you can't screw before the metal is welded.
  • build an engine and build a frame can be ran in their own threads (and within those threads the weld metal and screw bolts must block the particular thread).
  • combine engine and frame must wait for build an engine and build a frame to finish.

Honestly I'm not too sure where to start, initially I thought of using ThreadPoolExecutor in Job.do(), but I'm not sure how to block only some jobs and not others, and another issue is that sometimes a task must be ran "alone" without even being inside of any job, i.e. WeldMetal(block=False).do() must be valid.

I ended up using this at first:

class Task: def do(self): raise NotImplementedError('abstract method') def run(self): if self.block: self.do() else: threading.Thread(target=self.do).start()

And this works for most cases, but the issue comes when combine engine and frame needs to wait for both build an engine and build a frame to finish.

Improve this question
0

2 Answers 2

Reset to default
2

You are trying to combine two concerns into one solution and that is causing problems because that solution doesn't really cover both concerns.

On the one hand, you have blocking tasks (tasks that may not run in parallel with other tasks of the same job) and on the other hand you have dependencies between tasks (a task can't start until one or more specific other tasks have completed).

You have a solution for blocking tasks that are the first tasks in a job, but not for dependencies or even for blocking tasks that come after a non-blocking task (if paint is blocking, then it can still be executed in parallel with screw bolts as part of the build a frame job).

To handle both blocking and dependencies, you can use something like this:

class Task: def __init__(self): self._done = threading.Event() self.dependencies = [] def wait_for_completion(self): self._done.wait() def do(self): raise NotImplementedError('abstract method') def do_async(self): for depends in self.dependencies: depends.wait_for_completion() self.do() self._done.set() def run(self): threading.Thread(target=self.do_async).start() class Job(Task): def __init__(self): super(Job, self).__init__() self.tasks = [] def do(self): started_tasks = [] for task in self.tasks: if task.block: self._wait_for_tasks(started_tasks) started_tasks = [] else: started_tasks.append(task) task.run() if task.block: task.wait_for_completion() self._wait_for_tasks(self.tasks) def _wait_for_tasks(self, task_list): for task in task_list: task.wait_for_completion()

The Task class takes care of the dependency handling, by blocking the task's tread until an event has been set by the dependencies. I extended the Job class with a mechanism to ensure that blocking tasks run only when no other tasks within that job are running and making sure that the Job isn't reported as completed until all contained tasks are completed. That last part is needed to let the dependency mechanism work correctly with Jobs.

I have intentionally used an Event object for the synchronization, rather than join()ing the thread, because it makes the code more robust for the case where Task A starts to wait on Task B before Task B was started. The code is only guaranteed to be free of deadlocks if the tasks and jobs form an ordered sequence (as in, executing everything in the given order on a single thread does not violate any dependencies).

Improve this answer
0

asyncio seems to already do most of what you want. For example:

import asyncio async def build_a_car(): engine_task = asyncio.create_task(build_an_engine()) frame_task = asyncio.create_task(build_a_frame()) engine = await engine_task frame = await frame_task await combine_engine_and_frame(engine, frame) async def build_an_engine(): await weld_metal() await screw_bolts() async def build_a_frame(): await weld_metal() await screw_bolts() await paint() asyncio.run(build_a_car())

You can run a coroutine alone using asyncio.run(weld_metal()). There's nothing special about the top part of the hierarchy.

That's where I would start, and it should take you quite a way. If you're looking for something heavier weight, you might look at a distributed task queue like Celery or Kubernetes Jobs, but those are more about allocating lots of compute resources in a reliable way rather than making the sequencing of dependencies easy.

Improve this answer

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.