Nodus

(WARNING: EXPERIMENTAL)

Also, note that most of this README is out-of-date and inconsistent at the moment...

Framework for parallel data-flow based applications.

It is influenced by, inspired by, and in many cases similar to:

• Functional Reactive Programming (also here)
• Kahn Process Networks
• Algorithmic Skeletons (see also here)
• Iteratee IO
• Railway Oriented Programming
• Erlang
• And of course the hundreds of other reinventions with names like /(((data|signal|packet)?(stream|flow)|pipe(line)?|(flow|event|signal|reactive)-(processing|programming|architecture|computing|language)/ combined with the multitude of parallelization, concurrency, and clustering paradigms, etc. etc...

• Guiding philosophies
  • Steady-state
  • Composable
  • Proportionate
  • Real-time & simulation-time
  • Resilient (failure modes as first class features, exception/failure paths, bulkheads, circuitbreakers, hibernation, ...)
• Simple single-stream sequential
  • URI-like notation
  • Token state accumulation
• Single-stream sequential with explicit sink
• Single-stream parallel
  • Branches
  • Forks
  • Waypoints
  • Pattern Branching
• Multi-stream
• Getting away with duck-typing

This library and associated commandline tools are most appropriate for these types of (overlapping and somewhat redundant) problems & constraints:

It is additionally tuned for (but doesn't assume) problems with the following properties:

Use Little's law for queue bounds

+---+     +------+     +------+
| G | --> | f(x) | --> | g(x) | -->
+---+     +------+     +------+

• No sink specified at the end, so it simply outputs everything to STDOUT
• Generator types, from simplest to most complex:
  1. Scalar? Not very useful; does it make sense to allow it?
  2. Simpler proc/block/lambda than an enumerable? - instantiate via implicit counter generator with an offset which uses the simple proc as the next block (so you could have a generator such as ->(x){Math.sin(x/100.0)})
  3. Any enumerable or even just an object with an each method (will be made lazy if it isn't already)
  4. Simple DSL
  5. State-machine class
  6. Decoupler
  7. Synchronized Decoupler

Defined by:

• The eventual property distribution of its constituent tokens (possibly given as a shorthand name/"token-type"), which, because the token's internal state ends up mirroring the operations that have occurred, means it is also a way of describing the graph of process nodes that act on tokens within the stream (explicitly or implicitly- don't know yet) (usually all or a subset of a nodus application specification);
• An origination node for that specific stream (not necessarily a standalone origination node like for an app)
• A unique session identifier (uuid) identifying this run-instance, possibly shared with other streams (or Nexus' / Decouplers)
• Creates the initial instances of tokens, with a monotonic order indexed from the first token of the session. (Hence, streams can be theoretically infinite into the future, but always have a finite well-defined past, at least in the context of a session).
• Sometimes a version, which can affect caching, for example, or conflict resolution when a sink is permanently saving state from a running stream, etc.
• No cycles. Cycle-like behavior is handled by creating another stream with a delay (for example).

Stream == token type == (the data accumulated in the token are related) & especially (same intervals/timing)

• A chunk of data - the ordered atoms of a stream.
• Has a "type"/name that coincides with the stream
• Analogous to a rolling stone gathering moss. The token starts out very small, and accumulates more pieces of state/data as it moves through the nodes, with it's most recent state being the most relevant for the next node.
• Has a tree-like internal view of the accumulated state that ends up mirroring the topology of the node network.

• The context within which the runtime system executes the dataflows/streams.

A node may be a generator for one or more streams AND/OR a sink for one or more streams, AND/OR an operator on one or more streams (accepting tokens from one or more upstream nodes and emitting them to one (or more?) downstream nodes. Generally though most nodes will deal with only a single stream- and most of those will be processor nodes, sandwiched between a generator at the beginning and a sink at the end.

Most nodes are intra-stream nodes...

Parameterization (applying the specified or calculated parameters) only happens once, when the first token is about to hit the node. Any state change after that needs to be handled manually.

(maybe nodes with no generators applied get default behavior if they are run standalone- integer sequence generator, for example, or default to stdin, argf, etc...)

Operate on data within a single stream

|

NOTES:

• Branch without remerge -> when first branch path finishes (or none are deemed applicable) parent node operation continues forward- meaning the other data may or may not be available by the time it reaches a sink.
• (possibly named sync-points / stream entry/exit points as syntactical sugar to break up description of graph)
• Special Wait nodes might include logic that allows a "partial" token to go ahead depending on certain system conditions etc.

• Application: Adhoc collection of process networks meant to be run by itself. Usually with a single generator and a single final output sink- although that stuff can be inferred potentially. While it's meant to be standalone, other nodes should still be allowed to use it when composing more.

• Generators: Shorthand for a node with zero input streams and one or more output streams. In a broader semantic perspective a generator is a node that creates any new streams regardless of whether it has input streams (maybe we should call those Originators or something instead?)

  • sequences/counters
    • deterministic
      • integer
        • monotonic
        • sequence ... (e.g., from http://oeis.org)
      • real
        • ... (monotonic with offsets, scaling, ...)
      • ... (e.g., complex?, digits of irrationals like pi, ...)
    • stochastic (or deterministic with seed in some cases)
      • rng ... (prng, high quality, ...)
      • quasi-random ...
      • distributions ...
  • clocks/timers ... (including simulated timers? e.g., meaningful timing sequence without waiting the actual intervals between tokens...)
  • system (although usually used as normal internal functions for the nodes instead of streams)
    • state changes (e.g., network connectivity, service status, ...)
    • ... (e.g., resource utilization, uptime, network load, ...)
  • interprocess
    • signals
    • argv
    • file
      • simple/read
      • followed (i.e., tail -f-like or even -F)
    • database ... (via query or change-monitoring etc.)
    • pipe
      • stdin
      • named
      • unix pipe (via file descriptor)
    • argf (combo stdin and/or file(s))
    • message queue
    • semaphore
    • shared memory
    • mmap
    • unix socket
    • net
      • socket
      • secure socket
      • HTTP
      • HTTPS
      • ... (e.g., AMQP, Erlang-node, ...)

• Projections (one input stream, one different output stream with timing change & probably data change):

  • SeqDelay: (buffer, drop(n), drop-while, fifo-queue) output stream is same as input tokens but delayed by specified number of steps (implied (n - 1) delay when a cycle is specified or detected?)
  • TimeDelay: introduce a system-time latency
  • Sink: (fold, accumulator, aggregate, reduce, inject) special ending node that only outputs a result stream with a single token that reports some reduced version of all input tokens. (can be specified with a single operator, e.g., :+ for sum).
    • Decoupler: persists records
    • Future:
      • first
      • include?
      • any?
      • count
      • ... other enumerable & functional operators that act on the whole population.
  • Reject: (filter) Filters out specified tokens
  • Select: (grep, search) Selects specified tokens
  • Reactors: Nodes that give an impulse when they encounter certain conditions on the input stream: specifically for monitoring and reacting to certain conditions like queue lengths etc. (can be thought of as being implemented via -->[tap]-->[select])
  • Mutate: Timing of new stream is the same as input, but data has been mutated in a way that incompatible with the incoming stream
  • Future:
    • pattern detection
    • sample/alias (time-based chunking)
    • sketch (in lieu of sort, for example)
    • chunk
    • collect-concat / flat-map
    • cycle
    • each-cons
    • take, take-while, drop, drop-while, skip, skip-while (?)
    • ... other enumerable-like projection functions
    • ... sink-like enumerable functions but "so far"- e.g., 'current-max,' 'current-minmax', ...
  • DIFFICULT (or impossible considering infinite streams and steady-state, but with some possible partial-solutions)
    • reverse, sort, sort-by (although that could be some kind of sketch...)

• Merge junctions:

  • Mux: when token arrives on any input stream an output token is generated with copies of the latest of all input tokens along with how long it has been since those tokens (respectively) arrived.
  • Zip: combines specified number of input tokens (at least one each) laterally and emits an output token. Note that this can cause non-steady-state behavior (including even race conditions) if one of the streams bottlenecks and the other stream builds up a huge backlog. (Need to at least throw an error when this seems like it's happening, and ideally even have mechanisms for automatically dropping, etc.)
  • AggregateZip: emits an output token when at least one of every input stream is received. Tokens received from streams multiple times while waiting for the first token on one of the other streams are aggregated together (so the final resulting token contains a vector for each incoming stream, with every vector containing at least one item and at least one vector containing only one item [the last stream to give an inbound token while the others were building up]).
  • StreamFold: has a kernel that updates state and/or outputs a token at any input receipt from any stream.
  • Future
    • concat (if it's ever actually needed- nodus isn't really meant for it...)

• Scatter junctions:

  • Switch: output streams contain subsets of the input tokens where each token travels down a path (or more than one?) determined by a case statement. (can be thought of as being implemented via -->[branch]--*>[filter])
  • StateSwitch: input tokens get fed to output stream depending on the last given token from a different input stream.

• Generic custom Junctions: Implementing a Junction node requires that it has to actively use actor-like semantics (including timeouts and possibly polling multiple streams) to request tokens on the input streams. [implemented by the fact that in reality it is waiting on all streams at all times, and then decides whether it applies to the given wait condition].

(could have just as easily been named 'Stream Processing Graph' or 'Nodus Application Specification', ...)

Technically a normal node that happens to:

• Have neither inputs nor (maybe?) outputs
• Contain one or more initial generating nodes
• (therefore) allows execution by the nodus execution engine

In practical terms, it:

• Creates a stream which can then potentially create other streams
• Ends with explicit or implicit sinks/end-points/reductions
• Is "executable" in a standalone fashion
• Is generally behave as a steady-state daemon processing a hypothetically infinite stream of tokens
• Is usually specified at least at the highest level in a ".node" file that also describes its high-level purpose
• (in other words) is the top-most specification of a nodus, an Application.
• Will leave side-effects / external actions for the sinks
• Will finish by having executed the external action, outputting a reduced/folded result (not ideal since it implies batch-like single-run behavior rather than daemon-like), or persisting the final tokens (or a form thereof) in a Decoupler (persistent store separate from caches described shortly).

• Parses/processes a nodus specification/application/standalone-process-graph
• Starts everything up for the nodus and (normally) daemonizes
• /etc/init.d-(etc.)-like daemon instance management
• Erlang-like server behavior, ideally (or maybe even actually running on a distributed OTP network...)
• Could eventually conceivably evolve into tools with higher levels of management/abstraction (for distributed processing and cluster management, for example)

• Ability for nodus to specify which applications should be running autonomously all the time (system-level daemonized), including restart policy and footprint monitoring vs. those that are experimental / dev / transient, vs. those that are something in-between.

Token: any object instance but usually openstruct with hash-like access as well (so like HashWithIndifferentAccess) but with WRITE-ONCE semantics! (throws error if the same process/node tries to write the same field a second time)

suggested behavior:

• lock-field-value-on-write
• unlock-all-field-values (for when a node starts its turn | possibly specify that this proc/node is one that can write to it)
• exception on rewrite of a field after it's locked

• Check out the latest master to make sure the feature hasn't been implemented or the bug hasn't been fixed yet.
• Check out the issue tracker to make sure someone already hasn't requested it and/or contributed it.
• Fork the project.
• Start a feature/bugfix branch.
• Commit and push until you are happy with your contribution.
• Ensure that it is accompanied with tests.
• Please try not to mess with the Rakefile, version, or history etc. If you want to have your own version, or it is otherwise necessary, then fine, but please isolate to its own commit so I can cherry-pick around it.

Copyright (c) 2014 Joseph Wecker. MIT License. See LICENSE.txt for further details.