The Core API Jet Execution Model

At the heart of Jet is the TaskletExecutionService. It manages the threads that perform all the computation in a Jet job. Although this class is not formally a part of Jet's public API, understanding how it schedules code for execution is essential if you want to implement a cooperative processor.

Cooperative Multithreading

Cooperative multithreading is one of the core features of Jet and can be roughly compared to green threads. It is purely a library-level feature and does not involve any low-level system or JVM tricks; the Processor API is simply designed in such a way that the processor can do a small amount of work each time it is invoked, then yield back to the Jet engine. The engine manages a thread pool of fixed size and on each thread, the processors take their turn in a round-robin fashion.

The point of cooperative multithreading is better performance. Several factors contribute to this:

• The overhead of context switching between processors is much lower since the operating system's thread scheduler is not involved.
• The worker thread driving the processors stays on the same core for longer periods, preserving the CPU cache lines.
• The worker thread has direct knowledge of the ability of a processor to make progress (by inspecting its input/output buffers).

Tasklet

The execution service doesn't deal with processors directly; instead it deals with tasklets. Tasklet is a very simple functional interface derived from the standard Java Callable<ProgressState>. The execution service manages a pool of worker threads, each being responsible for a list of tasklets. The worker thread simply invokes the call() methods on its tasklets in a round-robin fashion. The method's return value tells whether the tasklet made progress and whether it is now done.

The most important tasklet is the one driving a processor (ProcessorTasklet); there are a few others that deal with network sending/receiving and taking snapshots.

Work Stealing

When a tasklet is done, its worker will inspect all the other workers' tasklet lists to see if any of them has a longer tasklet list than its own. If it finds such a worker, it will "steal" one of its tasklets to even out the load per thread.

Exponential Backoff

If none of the worker's tasklets report having made progress, the worker will go to a short sleep. If this happens again after it wakes up, it will sleep for twice as long. Once it reaches 1 ms sleep time, it will continue retrying once per millisecond to see if any tasklets can make progress.

ProcessorTasklet

ProcessorTasklet is the one that drives a processor. It manages its inbox, outbox, inbound/outbound concurrent queues, and tracks the current processor state so it knows which of its callback methods to call.

During each tasklet.call(), ProcessorTasklet makes one call into one of its processor's callbacks. It determines the processor's progress status and reports it to the execution service.

Non-Cooperative Processor

If a processor declares itself as non-cooperative, the execution service will start a dedicated Java thread for its tasklet to run on.

Even if it's non-cooperative, the processor's callback methods must still make sure they don't run for longer than a second or so at a time. Otherwise the tasklet will never be able to initiate a snapshot on the processor.

Running a Jet job

When you submit a Job to it, Jet replicates the DAG to the whole Jet cluster and executes a copy of it on each member.

Jet executes the job on a user-configurable number of threads which use work stealing to balance the amount of work being done on each thread. Each worker thread has a list of tasklets it is in charge of and as tasklets complete at different rates, the remaining ones are moved between workers to keep the load balanced.

Each instance of a Processor is wrapped in one tasklet which the execution service repeatedly executes until it is done. A vertex with a parallelism of 8 running on 4 members would have a total of 32 tasklets running at the same time. Each member has the same number of tasklets running.

When you make a request to execute a Job, the corresponding DAG and additional resources are deployed to the Jet cluster. Jet builds an execution plan for the DAG on each member, which creates the associated tasklets for each Vertex and connects them to their inputs and outputs.

Jet uses Single Producer/Single Consumer ringbuffers to transfer the data between processors on the same member. They are data-type agnostic, so any data type can be used to transfer the data between vertices.

Ringbuffers, being bounded queues, introduce natural backpressure into the system; if a consumer’s ringbuffer is full, the producer will have to back off until it can enqueue the next item. When data is sent to another member over the network, there is no natural backpressure, so Jet uses explicit signaling in the form of adaptive receive windows.