Dispatching and scheduling electrons/dispatches, how should they happen ?

[santoshkumarradha]

In the case of running a single server-ed dispatcher, there are multiple ways to run both the electrons and dispatches

Few of them are

1. Electrons
2. Dispatches

Methods
a. asyincio (concurrent)
b. Threads
c. Processes

With this, we have the following matrix

• 1a2a
• 1a2b
• 1a2c
• 1b2a
• 1b2b
• 1b2c
• 1c2a
• 1c2b
• 1c2c

Any discussion on which one to choose and why ?

Constrains :

• Since electrons interact with executers, and following the ideology of executers being a user defined plugin (where we plan to make it even more easier to define executer in the future) Ideally we do not want users to go around their way to design good executers.
• There would be a healthy mix of "local" and "cloud" executers (read, non-IO blocking dispatches and IO blocking dispatches)
• Anything else ...? Nothing I can think of now.

[Emmanuel289]

In the ideal world, we would want to compare the performance analysis across all 9 options for electrons and dispatches and account for each constraint. This would be useful in discussing pros and cons of each option.

[kessler-frost]

Agreed, we should do that soon.

[cjao]

At the cost of additional complexity on our part, what if we support both sync executors -- which are easier to write -- and async executors -- as good remote executors ought to be? For instance, we could await on coroutine executors and run sync executors it in a preexisting Thread/ProcessPool using loop.run_in_executor() and await the returned Future.

[kessler-frost]

I think that's the best of both worlds. Using this we can set the default pool to be used by the run_in_executor method as the threadpool we started with covalent start. But the biggest issue is that, you cannot use a ProcessPoolExecutor, only a ThreadPoolExecutor.
Edit: Turns out that's only the case with set_default_executor.

[santoshkumarradha]

Hey @cjao could you brief on why we might need a async change in executer if we are asynci-ing in our side ?

[santoshkumarradha]

But i really like this idea, if one can support both, we can handle ours with async wherever we need, and advanced plugin writers can indeed write async while we still support non async like plugins example- GPU ones.

[kessler-frost]

Hey @santoshkumarradha, good question. So basically there is one thing that we are sure about - doing cooperative multitasking (manually handling when a context switch happens within an eventloop) will support a lot more "tasks" happening concurrently than multithreading can because of client's hardware limitations. This is in the case where most of these tasks are io/network bound. But that will mean having to make the executors async aware.

That can only happen if there are two different methods "submission" and "getting result" inside the executor.execute function, essentially if submitting a job to say a cluster can happen without waiting for its result as that gives us time to do a context switch and start processing a separate task already scheduled to run in the event loop before coming back to get the result.

So coming back to your question, what we need to do in order to take advantage of asyncio is the ability to perform a context switch within our executor's execute method. Which means it's up to the creator of that executor to make it async-aware, but at the same time we also want non-async execute methods to be run in parallel, hence we will fall back on an already running thread pool/process pool to run those electrons using asyncio's run_in_executor method.

Edit: Added process pools too.

[cjao]

@santoshkumarradha: If we use asyncio, an event loop running in a single thread (the "main" thread) submits each task for execution. If executor.execute were to block the main thread, no other tasks can be submitted until the current task completes. A typical remote executor uploads the task to a server, wait for the server to execute the task, and download the result from the server. These network I/O operations can be slow but don't use much CPU time; but if they were simply performed sequentially like

def execute():
Upload()
WaitForServer()
return Download()

the main thread would have to wait for each operation to complete, and executor.execute would take a long time to return control to the dispatcher.

To let multiple tasks run concurrently, executor.execute should ask to be paused upon initiating each expensive I/O operation. While the task is paused, the event loop is free to handle other tasks. When the IO completes, the event loop (assuming it's not blocked by another operation) resumes the original task and pauses it at the next I/O. This allows the I/O between different tasks to occur in parallel, especially for long-running operations that don't consume much bandwidth, like waiting for a reply from the server.

The following diagrams attempt to illustrate the differences. The green blocks represent time that the event loop spends in each task.
In the first case, the executors run synchronously and the event loop must wait for all steps of a given task to complete before handling the next task:

In the second case, the executors only make non-blocking networking calls and pause themselves during I/O, thereby freeing the event loop to handle other tasks or listen for I/O:

[kessler-frost]

There were several discussions that happened over the past week about this and I'll try my best to summarize them in this comment:

• In the latest implementation we have essentially detached dask from covalent and made it into an executor, we are now using python's threading module to start a new thread for each new workflow and return the dispatch_id without waiting for the dispatch to complete. Which essentially offloads the dispatch to be run in that separate thread. Now in order to achieve concurrency on the electron level, we are using asyncio's event loop to schedule tasks for later execution.

  • What this means is basically each executor.execute method is scheduled to be run in the event loop.
  • Now, inside an event loop only 1 block of code gets executed at a time which is why asyncio allows us to to tell it when is a good time to switch "context" and start running a different block of code that was scheduled, and come back to the first one and continue later.
  • Since each execute method is supposed to always return a result, and hence is a event loop blocking call, whenever any one of the execute method has been started by asyncio, asyncio will not be able to switch context until the execute method is done.
  • Which essentially means that everything will be running in serial, even though they are independent tasks and we should be able to run them in parallel. This is the problem.

• Let's look at possible solutions:

  1. We can have a thread/process pool started at the beginning of covalent start, and handle workflow level parallelism and electron level parallelism from the same pool. This, as you might see, may very quickly lead to a deadlock, e.g, say all workers in that pool are occupied and you want to submit an electron to that pool, then that electron will wait for a workflow to finish which in turn is waiting for the electron to finish. ---- Not Viable
  2. We start a process pool for catering each workflow - one workflow = one process, which means each workflow will have its own GIL which is nice. Within that process we start a thread pool and submit each execute method to its own thread.
     • In this case, we'll be dynamically creating pools, so the number of thread pools = number of workflows dispatched.
  3. We start two thread pools, one to handle workflow level parallelism and the other to submit all the electrons to.
     • This will however limit scalability, since we can have a workflow hogging up all of the threads in a pool which will affect other workflow's execution too.
  4. Using asyncio in combination with pools, in either or both of the above mentioned viable ways, so that we can support asynchronous executors whose execute methods are async-aware as well as synchronous ones which will be run in a separate thread. This is probably the better solution as it takes the best of all of the things mentioned above and can support a lot more electrons than possible threads.

• Some things to note:

  1. Stress testing with a whole bunch of workflows and each having a lot of electrons and sublattices (using Locust will help in testing this). Using this we also saw the inability of solution 3 to scale better.
  2. One thing that still remains is if an executor is designed to run something compute intensive on that same machine as the dispatcher, like the current LocalExecutor, then it won't be the ideal to use a thread.
  3. Another issue recently found out - deadlock in sublattice case similar to the one in solution 1 but on the workflow level parallelism, since a sublattice is currently run with dispatch_sync, hence it'll occupy a slot in the workflow pool.
  4. In all of the above solutions that we tried with some basic workflows, using a dask executor with a running cluster seems to perform much better than its LocalExecutor alternative.
  5. It might also be better if the user has the ability to decide how many workflow level threads/processes and how many electron level threads/processes do they expect to be run, so that we create those many in the pool. Otherwise the defaults should be sufficient without hogging resources in their own machines.
  6. One more bottleneck if using thread/process only (without async aware) in electron execution, is say the dask cluster (or any cluster for that matter) can actually run 50 functions in parallel, the submission of those functions through the executor will be in batches of number of workers in the pool. But that can be handled if we let the user decide that number, or if we/they write the executor in an async aware manner and we use solution 4, in which case that won't even be an issue (since asyncio can handle number of tasks in the order of 10^3 or even 10^4 without any issues).

[kessler-frost]

Feel free to add in the replies if you think I missed something.

[kessler-frost]

One thing I forgot to note here, which I guess is probably well known, is that even though multithreading in python is an illusion, it still works pretty good for I/O and network bound tasks, so basically good for submitting/offloading things to some other clusters (network bound). That said, asyncio and related libraries (such as aiohttp) for this case are better. One pretty good article discussing this can be found here: https://realpython.com/python-concurrency/ .

[cjao]

One more bottleneck if using thread/process only (without async aware) in electron execution, is say the dask cluster (or any cluster for that matter) can actually run 50 functions in parallel, the submission of those functions through the executor will be in batches of number of workers in the pool. But that can be handled if we let the user decide that number, or if we/they write the executor in an async aware manner and we use solution 4,

In the case of dask, it is easy to make that executor asynchronous. We should encourage users to implement their executors as efficiently as possible -- and allow well-written executors to reach their full potential -- instead of trying to make up for executor inefficiencies by increasing the dispatcher's resource usage. The thread/process pool really only ought to be for those executors which fundamentally cannot be written asynchronously.

[santoshkumarradha]

Agree