Executors, task parallelism, channels APIs #347
Comments
|
|
Good point we might want another overload for array-based channels. List-based channels are necessary for:
That doesn't mean we need to provide them in the standard library, but this would make channels interface consistent across all future libraries. |
|
For array-based channels we can have the following proc trySendBatch[T](chan: var Chan, src: var openArray[Isolated[T]]): bool =
## Send an array of items to the back of the channel
## Returns true if successful (channel had enough free slots)
##
## This destroys `src` on success
proc tryRecvBatch[T](chan: var Chan, dst: var openArray[Isolated[T]]): int =
## Try receiving all items buffered in the channel
##
## The channel may not return all items due to in-progress competition
## between the consumer and producer(s)
##
## Returns the number of items receivedNow I realized an issue with func trySend*[T](chan: var Chan, src: sink Isolated[T]): boolis incorrect because in case of failure the sender will likely want to resend later so func trySend*[T](chan: var Chan, src: var Isolated[T]): boolthoughts? |
In the new channel implementation (based on yours) I use a bounded channel with a condition variable so that there is a blocking |
|
If the return type is not simply an int, but instead an opaque type that can be queried about the number of items, via accessor of some variety. Then could said opaque type also return the non-consumed part of the array or the entire array wholesale? If so would that not allow for a "free" sink? Apologies, still rather green with ARC/destructors. |
In the future it makes sense to allow returning iterators so that you can parallelize infinite streams. Requiring the number of items is probably too restrictive. |
It was late, I muddled the received count of
So beyond this particular call, I meant it as a stylistic thing for the API design for the calls as they return status. This by no means will mitigate all circumstances. |
Any reason to support If
proc isReady(fv: Flowvar): bool
This is mostly to test my own understanding of the intention here, am I on the right track with: This reads to me as if the spawner is slow to consume, then the executor cannot free resources until that happens. This is also preferred as the fork-join model is typically used in cases where the issuer/consumer are one in the same and the issuer should be "in charge". Overall API Thoughts: I really like the clarity of this, it's awesome. A potential issue with the API overall, especially with only having Can we disallow |
|
Heads up, the definition of trySend plus the isolated concept, has flaws that are independed of an implementation. proc producer =
for i in 0 ..< numIters:
var p = isolate(Foo(id: $(i + seed))) # crashes
while not chan.trySend(move p): cpuRelax()Here is just the while (1) {
tyObject_Isolated__rEIebJMJoptZEKusFoG28w T7_;
NIM_BOOL T8_;
nimZeroMem((void*)(&T7_), sizeof(tyObject_Isolated__rEIebJMJoptZEKusFoG28w));
T7_.value = p.value;
nimZeroMem((void*)(&p), sizeof(tyObject_Isolated__rEIebJMJoptZEKusFoG28w));
T8_ = (NIM_BOOL)0;
T8_ = push_tring_236((&rng_tring_12), T7_);
if (!!(T8_)) goto LA6;
cpuRelax_system_2802();
} LA6: ;As you can see if it fails the first time, p would be moved and empty, causing it send empty data. while not chan.trySend(isolate(Foo(id: $(i + seed)))): cpuRelax()But here is another fault, inside the implementation of trySend, pseudocode with expandarc: if channelBufferIsFull:
result = false
else:
`=sink`(this.data[head], extract value)
result = true
`=destroy`(value)As you can see value is destroyed inside trySend. It can also be worked-around with On the other hand with these workarounds if you only trySend once, your data may be leaked. Upd: So the correct function signature is |
|
...also the templates that hide Isolated[T] need warnings about multiple evaluations when used in a loop. |
This comment was marked as outdated.
This comment was marked as outdated.
This is an API specification for:
The RFC is also available in Weave: https://github.com/mratsim/weave/blob/ba99dce/rfcs/multithreading_apis.md, it might be easier to discuss it in a PR.
I can split the RFC in 3.
Executors, task parallelism, channels APIs
Abstract
This document:
Channels libraries SHOULD implement this API.
Table of Contents
spawn: scheduling left to the executor [REQUIRED]submit: scheduling MUST NOT block the submitter thread [Experimental]Introduction
The Nim programming language has significantly evolved since its threadpool and channels modules were introduced in the standard library.
inboxes.nimlikely with the idea of making first-class actors in Nim.spawnandparallel.Since then, few projects actually used channels or threadpools in part due to:
||.With the progress of core low-level primitives and compiler guarantees, Nim foundations are becoming solid enough to build an ergonomic and safe multithreading ecosystem.
This document specifies the multithreading interfaces related to channels and threadpools so that they can be evolved or replaced with principled foundations.
This document is interested in the public user API and does not specify the underlying implementation.
This document does not require compiler support. Primitives can be implemented as a library.
This document also defines related multithreaded concepts that may be implemented in some libraries without specifying them. The interface is left open until more feedback is gathered.
This document is written under the assumptions that there is no one-size-fits-all and that projects may want to use multiple threadpools, executors or schedulers within the same application or library. Furthermore, libraries may offer to support multiple parallelism backends and as a project evolves dedicated specialized threadpools may be written that are tuned to specific workloads.
The core primitives mentioned that facilitate multithreading are:
Isolated[T]which allow transferring ownership of GC typesThis blog post explains why it's important to embrace multiple schedylers:
Multithreading is a very large subject that cannot be covered in a single specification. Nonetheless to facilitate future RFCs, this document will define terms that are likely to come up in future specifications in a non-goals section.
While channels and task parallelism API are presented in the specification document, they MAY be implemented in different libraries.
Requirements notation
The key words "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED", "NOT RECOMMENDED", "MAY", and "OPTIONAL" in this document are to be interpreted as described in the Internet Engineering Task Force (IETF) BCP14, RFC2119, RFC8174 when, and only when, they appear in all capitals, as shown here.
Definitions
Definitions are ordered in lexicographical order. Definitions may include specification or implementation recommendation.
Channels
A channel is a cross-thread communication queue or ring-buffer.
A channel can come in many flavors:
Note: not using locks does not imply that a channel is lockfree.
lockfreeis a progress guarantee, even without locks an implementation can block.We call such channels
lockless. It is worth noting that usually the stronger the progress guarantees the less throughput your channel has. Strong progress guarantees are needed for low-latency or real-time processing.For reference types, a channel transfers ownership of the reference.
A channel MUST NOT copy, this is required for both correctness and performance.
In terms of distributed systems, we require that shared-memory channels achieve exactly-once delivery.
This ensures that producer and consumer can rely on the message being sent and received.
It also prevents side-effect being executed twice if a message is duplicated for example sending "launch missiles!".
A channel is shared by at least 2 threads. Shared ownership requires ensuring that the shared object is not freed twice and not freed before its last users.
ptr objectand ensuring that the thread that will free the shared object awaits all child tasks.Closures
A closure is a data structure representing procedure and its parameters (environment).
For multithreading purpose, it MUST be using a threadsafe memory allocator and/or garbage collection. Nim closures with Nim default refc GC is not compatible with multithreading.
Closures SHOULD be non-copyable, only movable. A closure might free a temporary resource at the end of a computation (CPU or GPU memory, a socket, a database handle), that resource might be manually managed or managed by destructors.
Concurrency
Concurrency is the ability to make progress on more than one task at a time.
Parallelism imply concurrency but concurrency does not imply parallelism.
A program may interleave execution of task A, then task B, then task A on a single thread and be concurrent and not parallel.
Since IO workloads requires waiting to make progress and CPU workloads require working to make progress, concurrency is often associated with IO-bound tasks and parallelism with CPU-bound tasks.
Continuations
A continuation represents the state of execution of a program at a certain point.
Running a continuation would restore that state and run the program from that point.
Concretely, a continuation is a data structure with a procedure that represents the following statements (instructions) to run and all the parameters required to run them (environment).
At a data structure level, a closure and a continuation have the same fields.
Executor & execution context
An executor is a handle to an execution context. An execution context represents:
Parallelism
Parallelism is the ability to physically run part of a program on multiple hardware resources at the same time.
Fork-join model
The fork-join model allows execution of a program to branch off ('fork') on two threads at designated points in the programs. The forks can be 'joined' at designated points.
Fork points are often called
spawnincluding in Nim threadpool. Join points are often calledsync. They are called^in Nim threadpool.The fork-join model can be implemented on top of:
fork().At a fork point, multiple dispatching strategies can be used, for example on parallel merge sort:
At the
spawnpoints:mergeSort(a, b, left, middle)can be packaged in a closure and sent to an executor, becoming atask. The current threads then executes the next linemergeSort.This is called help-first scheduling because the current thread helps creating new tasks first. It is also called child-stealing because the executor steals the child task.
mergeSort(a, b, middle, right); sync(fut); merge(a, b, left, middle, right)can be packaged and sent to an executor. The current thread then suspends the current procedure and enters the spawnedmergeSort.This is called work-first scheduling because the current thread work immediately on new tasks. It is also called continuation-stealing and parent-stealing because the executor steals the continuation of the current (parent) procedure.
A high-level overview of continuation-stealing vs child-stealing is given in
Task parallelism
Task parallelism dispatches heterogeneous (procedure, data) pairs on execution resources.
The fork-join model is the usual way to split a program into tasks.
Task
A task is a data structure representing a procedure, its parameters (environment) and its execution context.
Closures MAY be packaged in a task.
The task data structure is left unspecified.
Threadpool
A threadpool is an executor with a naive scheduler that has no advanced load balancing technique.
Threads & Fibers
A thread is a collection of execution resources:
With OS (or kernel) threads, the stack is managed by the kernel.
Userspace threads where the library allocates, switches and manages its own stack are called fibers (and also stackful coroutines).
std/corois an implementation of fibers.GPUs also have threads, their stack is managed by the GPU driver.
Task Parallelism API
Task parallelism is implemented via the fork join model. This is the model used in:
The API has a REQUIRED and OPTIONAL interface. Some APIs are unspecified.
The API intentionally mirrors async/await without reusing the names to differentiate between concurrency (async/await) and parallelism (spawn/sync).
Creating a task
spawn: scheduling left to the executor [REQUIRED]A task-parallel executor MUST implement either the local executor or the global executor API.
It MAY implement both.
spawneagerly adds the task to the executor for execution as soon as an execution resource is available. Scheduling is not delayed until a graph of tasks is built (see dataflow parallelism in Non-goals).Local executor API
A new task MUST be created on a specific executor with a template or macro with the following signature
For example
Global executor API
A new task MUST be created on a global executor with a template or macro with the following signature
For example
Disambiguation between global executors can be done by prefixing the module name.
Global executors MAY use a dummy type parameter to refer to their global executor.
Future handle
spawnshould return a future handle under the typeFlowVar[T]orvoid.Tis the type returned by the procedure being forked. If it returnsvoid,spawnreturnsvoid.If the future of a void procedure is needed, the end user SHOULD wrap that function in a function with a return value, for example a function that returns
boolor refactor their code.Scheduling
spawnis a hint to the executor that processing MAY happen in parallel.The executor is free to schedule the task on a different thread or not depending on hardware resources, current load and other factors.
At a
spawnstatement, the threadpool implementation may choose to have the current thread execute the child task (continuation-stealing) or the continuation (child-stealing).Scheduling is done eagerly, there is no abstract computation graph being built that is launched at a later point in time.
Awaiting a future [REQUIRED]
The operation to await a task-parallel future is called
sync.This leaves
awaitopen for async libraries and framework.It is also the usual name used in multithreading framework, going back to Cilk (1995).
Awaiting a single future [REQUIRED]
Program execution can be suspended until a
Flowvaris completed by callingsyncon aFlowvar.A
Flowvarcan only by synced once, hence the return value within theFlowVar[T]can be moved.Assuming the task completes this ensure exactly-once delivery for Flowvars.
A
FlowvarMUST NOT be copied, only moved. This SHOULD be enforced by overloading=copy.If a
Flowvaris created and awaited within the same procedure, theFlowvarMAY use heap allocation elision as an optimization to reduce heap allocation.At a
syncstatement, the current thread SHOULD participate in running the executor tasks. The current thread MAY choose to only process tasks that the awaitedFlowvardepends on to optimize latency.Awaiting multiple futures [UNSPECIFIED]
Awaiting multiple futures ("await until any" or "await until all") is unspecified.
Structured parallelism [OPTIONAL]
Structured parallelism ensures that all tasks and their descendants created within a scope are completed before exiting that scope.
References:
More recently, concurrency frameworks also converged to similar "structured concurrency" (https://en.wikipedia.org/wiki/Structured_concurrency).
Check if a task was spawned [REQUIRED]
And user can call
isSpawned()to check if aFlowvaris associated with a spawned task.This allows users to build speculative algorithms that may or may not spawn tasks (unbalanced task tree), for example nqueens with backtracking.
Check if a result has been computed [REQUIRED]
An user can call
isReadyon aFlowvarto check if the result is present.The
FlowvarMUST be associated with a spawned task.This allows users to know if the current thread would block or not when calling
syncon A Flowvar.Error handling
Procs that are spawned MUST NOT throw an exception. They MAY throw
Defectwhich would end the program.Exceptions MUST be trapped by
try/exceptblocks and converted to another form of error handling such as status codes, error codes orResult[T, E]type.Threadpool and task parallelism libraries SHOULD document that constraint to end users and SHOULD enforce that constraint with the effect system.
Note: even assuming C++ exceptions, or exceptions that don't use the heap or a GC without thread-local heap, exceptions work by unwinding the stack. As each thread has its own stack, you cannot catch exceptions thrown in a thread in another.
Thread-local variables
Procs that are spawned MUST NOT use thread-local storage unless they are internal to the executor library.
Executors make no guarantees about the thread of execution.
Threadpool and task parallelism libraries SHOULD document that constraint to end users.
Cancellation
Caller -> Callee
This RFC does not require cancellation primitives so that the caller can cancel the callee.
Rationales:
Alternative:
Callee -> Caller
The callee can notify the caller that it was cancelled with boolean, enums or Result[T, E]
Buffered channels
Channels are a basic inter-thread communication primitive.
This specifies buffered channels, i.e. channels that can hold at least one item.
Unbuffered channels, also called rendez-vous channels, are unspecified.
The channel flavors should be communicated clearly in-particular:
whether it's single or multi producer and consumer.
There are 2 kinds of SPMC queues:
For channels we only specify the "racing" type, a broadcasting channel can be implemented on top of SPSC channels.
whether its bounded or unbounded
the progress guarantees
the item ordering guarantees:
whether the channel is intrusive (requires a
next: Atomic[T]with T aptr object) or notNon-blocking send [REQUIRED]
Non-blocking receive [REQUIRED]
Blocking send [OPTIONAL]
Blocking send still returns a bool for backpressure management.
If blocking or overwriting the oldest is chosen, sending is always successful if the function returns.
Blocking receive [OPTIONAL]
Blocking receive still returns a bool for backpressure management.
Batched operations [OPTIONAL]
We define batch operations for list-based channels.
Working with integers in a synchronization primitive like channels MUST NOT throw an exception.
Elements count [OPTIONAL]
Channels MAY keep track of the elements enqueued or dequeued.
In that case they MAY provide an approximate count of items with
peek.Working with integers in a synchronization primitive like channel MUST NOT throw an exception.
peekMUST NOT block the caller.Due to the non-deterministic nature of multithreading, even if a channel is locked to get the exact count, it would become an approximation as soon as the channel is unlocked.
If called on a channel with a single consumer, from the consumer thread, the approximation is a lower bound as producers can enqueue items concurrently.
If called on a channel with a single producer, from the producer thread, the approximation is a lower bound as consumers can dequeue items concurrently.
If called on a channel with multiple producers, from a producer thread, no conclusion is possible as other producers enqueues items and the consumer(s) thread dequeue(s) them concurrently.
Similarly on a channel with multiple consumers, from a consumer thread.
API + documentation on a MPSC channel.
Non-goals
Experimental non-blocking Task Parallelism API
In the proposed API, the scheduler may have the spawning thread participate in clearing the work queue(s)
at
syncstatements.If the spawning thread is handling network or UI event this is would block network or UI handling,
in that case it is desirable to ensure that the work cannot happen in the spawning thread.
This proposes
submit,Job,Pendingthat mirrorspawn,Task,Flowvar.It is left unspecified for now as given the competing latency (IO schedulers) vs throughput (CPU schedulers) requirements, there might be no use case for
submitthat isn't better covered by allowing multiple specialized schedulers in a program.Alternatively, the application could be separated in an UI/networking thread and a heavy processing thread with the heavy processing thread managing a threadpool.
Creating a task
submit: scheduling MUST NOT block the submitter thread [Experimental]submitis a tentative alternative tospawnthat guarantees execution on a different thread.Similar to
spawnthe API would be:submit,Job,runInBackground(Weave)andsetupSubmitterThread(Weave)were added to Weave followingto improve interoperability with async event loops by allowing a thread to submit
Jobto an executorwhile guaranteeing that execution always happens in a different thread from the async thread.
Definitions
This section gives a definition of other terms related to multithreading and async so that there is common vocabulary within the Nim community to talk about those concepts.
However it does not specify them.
resumable functions
Traditional functions have two effects on control flow:
Resumable functions adds the following 2 key capabilities:
3. The callee can suspend itself, returning control to "something".
4. The current owner (not always the initial caller) of the function can suspend itself and resume the continuation (aka the unexecuted "rest of the function").
coroutines
Resumable functions are called coroutines if the continuation (aka the unexecuted "rest of the function") can only be called once and it is delimited in scope (we exit the function at the end and return control to the caller). "Coroutines are one-shot delimited continuations".
Coroutines that uses the stack of their caller (like a normal function) are called stackless coroutines.
As a reminder fibers (stackful coroutines or green threads) allocate some memory and move the stack pointer to it (with assembly) just like hardware thread.
Closure iterators are stackless coroutines.
Note: stackful vs stackless is about the function call stack (stacktraces) nor about stack vs heap allocation of the coroutine state.
CPU-bound vs IO-bound tasks:
See https://nim-lang.org/blog/2021/02/26/multithreading-flavors.html
Latency:
The time required to wait to receive the result of 1 unit of work.
Latency is often important for IO and most important for real-time.
A single client/consumer of a service only cares about its latency
Throughput
The time required to expedite all units of work.
Throughput is often important for CPU-bound tasks where all work need to be expedited,
and the order it is done is inconsequential, for example for batch transforming 1000 images,
it doesn't matter whether we start from the first or the last as long as all are done as fast as possible.
Data parallelism
While task parallelism is the ability to run different tasks in parallel,
data parallelism is the ability to run the same task, but parallelizing at the data level,
for exemple dividing an array of size N so that each core receives an equal share of work.
This is often presented as a parallel for loop.
Dataflow parallelism
Dataflow parallelism is the ability to specify task or data dependencies and schedule the resulting computation graph in parallel.
Dataflow parallelism APIs present themselves under either:
Dataflow parallelism is used to model complex pipelined computations, for example video processing
where certain areas of the video might be less complex and so, in that area, later steps of processing can start earlier.
Dataflow parallelism is also called:
Communicating Sequential Processes
CSP is a model of computation based on message passing via rendez-vous channels (both sender and receiver must be available at the same time). CSP is a model that can be formally verified to be free of deadlocks and livelocks and so is suitable for safety-critical systems.
CSP uses named explicit channels for message passing that trigger anonymous behaviours.
Actors
Actors
is a model of computation based on the actor primitive, a process with a set of behaviours and an inbox.
The actor model uses named behaviors that are passing messages via anonymous channels.
Non-goals not covered in this specification
The text was updated successfully, but these errors were encountered: