Manifold

Manifold separates the logic of what happens from the execution model of when and where it happens. By default, Manifold conforms to the surrounding execution model: messages are processed on the thread they were originally put! on. This means Manifold can be used safely alongside other frameworks.

Under certain conditions, Manifold lazily constructs three specialized thread pools:

• wait-pool: Used for waiting on blocking operations (e.g., java.util.BlockingQueue, Clojure seqs, java.util.concurrent.Future, or Clojure promises).
• execute-pool: Used to execute bodies within manifold.deferred/future.
• scheduler-pool: Used for delayed tasks, periodic tasks, or timeouts (e.g., manifold.time/in, manifold.time/every, manifold.stream/periodically).

Statistics for the wait-pool and execute-pool can be monitored using manifold.executor/register-wait-pool-stats-callback and manifold.executor/register-execute-pool-stats-callback respectively.

Manifold streams represent an ordered sequence of asynchronous values. They provide mechanisms for asynchronous puts, takes, timeouts, and backpressure.

Streams are dual-natured: they act as both sources (emitting messages) and sinks (consuming messages).

• Sources are interacted with using take! and try-take!, which return deferred values representing the next message.
• Sinks are interacted with using put! and try-put!, which return deferred values that yield true if the put was successful, or false otherwise.

Manifold streams are compatible with Java BlockingQueues, Clojure lazy sequences, and core.async channels via conversion utilities.

(require '[manifold.stream :as s])

;; Create a new stream
(def s (s/stream))

;; Put a value into the stream (Sink operation)
(s/put! s 1)

;; Take a value from the stream (Source operation)
@(s/take! s) ; => 1

Manifold is designed as an asynchronous lingua franca to bridge incompatible data representations (like Java's BlockingQueues, core.async channels, or Clojure's lazy sequences). It provides a generic way to coerce unrealized data into a common form and pipe it between different stream representations using backpressure.

Key mental models include:

• Pervasive Asynchrony: Manifold emulates asynchrony by wrapping threads around synchronous objects when necessary.
• Deferreds: All asynchronous values and operations are represented as deferreds.
• Streams: Streams are categorized as either sources, sinks, or both.
  • Sources are interacted with via take!, try-take!, and close!.
  • Sinks are interacted with via put!, try-put!, and close!.
• Connectivity: Messages from any "sourceable" entity can be piped into any "sinkable" entity using manifold.stream/connect. This creates an explicit topology that can be walked and queried.
• Execution Control: Both deferreds and streams can have their execution offloaded to a thread pool using their respective .onto methods.

You can create derivative streams using operators similar to Clojure sequences.

• Consumption: Use s/consume to run a function for every message in a stream.
• Sequence Conversion: Use s/stream->seq to convert a stream into a standard Clojure sequence. Note that streams are not immutable, so this is an explicit transformation.
• Mapping/Filtering: Use operators like s/map or s/filter. Calling s/map on a sequence will automatically call s/->source internally.
• Transducers: For complex transformations not covered by standard operators, use s/transform with a transducer.
• Periodic Emission: Use (periodically period f) to emit the result of (f) every period milliseconds.

When you create derived streams (e.g., (s/map inc s)), all messages put into the original source s are propagated to all downstream derivatives. If the source is closed, all downstream streams are also closed.

(require '[manifold.stream :as s])

;; Transform a sequence into a stream, map it, and convert back to sequence
(->> [1 2 3]
     (s/map inc)
     s/stream->seq)
;; => (2 3 4)

;; Using a transducer
(->> [1 2 3]
     (s/transform (map inc))
     s/stream->seq)
;; => (2 3 4)

;; Create multiple derived streams from one source
(def s (s/stream))
(def a (s/map inc s))
(def b (s/map dec s))

@(s/put! s 0)
@(s/take! a) ;; => 1
@(s/take! b) ;; => -1

By default, Manifold streams have no buffer and act as direct conduits. To handle bursts or manage flow, use buffering and throttling:

• Create a buffered stream: Call (s/stream buffer-size).
• Add a buffer downstream: Call (s/buffer size stream) on an existing stream.
• Metric-based buffering: Use (s/buffer metric limit stream) to limit based on a value other than message count (e.g., count for collections).
• Throttling: Use (s/throttle max-rate stream) to limit the rate of messages emitted from a stream.

(require '[manifold.stream :as s])

;; Create a stream with a buffer of 100 messages
(def s (s/stream 100))

;; Add a buffer downstream of an existing stream
(def buffered-s (s/buffer 50 s))

;; Buffer based on the size of collections in the stream
(def metric-s (s/buffer count 1000 s))

;; Throttle the stream
(def throttled-s (s/throttle 10 s))

Because put! on a stream (or realizing a deferred) can block until the message has propagated through the entire downstream topology, you may want to decouple execution.

Using onto moves the stream or deferred onto a specific executor. This guarantees that all subsequent actions resulting from an operation will be enqueued onto that thread pool rather than being executed immediately on the caller's thread.

Key behavior: Calling onto on a single stream in a topology is often sufficient, as everything downstream of that stream will transitively be executed on that executor.

Available executors include:

• manifold.executor/instrumented-executor
• fixed-thread-executor
• utilization-executor

(require '[manifold.deferred :as d]
         '[manifold.stream :as s])

(def executor (fixed-thread-executor 42))

;; Moving a deferred onto an executor
(-> (d/future 1)
    (d/onto executor)
    (d/chain inc inc inc))

;; Moving a stream onto an executor
(->> (s/->source (range 1e3))
     (s/onto executor)
     (s/map inc))