{
  "markdown": "# Ariadne\n\nAriadne is an Erlang implementation of timely dataflow, following the model\nformalized by Martín Abadi and Michael Isard in\n[*Timely Dataflow: A Model*](okf/references/papers/timely-dataflow-a-model.md)\n(the paper is also kept as a [PDF](okf/references/papers/timely-dataflow-a-model.pdf)).\n\n## What it is\n\nPicture a streaming computation as a graph: boxes that keep some state,\narrows that carry messages between them. Data flows in, gets transformed\nbox by box, and flows out. Plain dataflow, nothing new.\n\nNow add the question every stateful stream processor eventually has to\nanswer: *\"I have been summing up the events of the last hour — is the hour\nover yet?\"* Batch systems know because the input ends. Streaming systems\nusually guess: wait a bit longer, hope nothing is late, or bolt on\nwatermarks by hand.\n\nTimely dataflow answers the question exactly. Every message carries a\nlogical time — the epoch it belongs to, plus one extra coordinate for each\nloop it is travelling around. A node can say to the runtime: *\"tell me when\ntime 5 is done.\"* The runtime tracks what is still in flight and where it\ncan go, and delivers that notification the moment nothing in the graph could\never produce another message for time 5 at that node. No timeouts, no\nguessing, and it works inside loops too: a node in a loop learns when an\niteration is complete, so iterative algorithms run on streams without\nleaving the graph.\n\nAriadne turns this model into an Erlang library. You describe the graph in\na small DSL, write the nodes as callback modules, feed the input and run it.\nThe current runtime, `ari_local_runtime`, executes the whole graph inside\none Erlang process, one message or one notification per atomic step, until\nthe graph goes quiet or a step budget runs out. It is meant as the\nreference: distributing nodes across processes is the next step and must\nnot change what the graph observes — the order on an edge, the results of\nthe nodes and the moment a notification becomes admissible.\n\n## Example\n\nA node that counts words per epoch and emits the totals of an epoch once no\nmore words of that epoch can arrive:\n\n```erlang\n-module(tally).\n-behaviour(ariadne_node).\n-export([input/0, output/0, init/1, handle_message/4, handle_notification/2]).\n\ninput() -> [words].\noutput() -> [totals].\n\ninit(_Args) ->\n    {#{}, []}.\n\n%% Count the word and ask to be notified when its epoch is complete.\nhandle_message(words, Word, Time, Counts) ->\n    Tally = maps:get(Time, Counts, #{}),\n    Counts1 = Counts#{Time => maps:update_with(Word, fun(N) -> N + 1 end, 1, Tally)},\n    {Counts1, [Time], []}.\n\n%% No message with this time can arrive any more: emit the totals.\nhandle_notification(Time, Counts) ->\n    {Tally, Counts1} = maps:take(Time, Counts),\n    {Counts1, [], [{totals, Tally, Time}]}.\n```\n\nA graph with this node, an input half-edge feeding it and an output\nhalf-edge collecting the totals:\n\n```erlang\nGraph = ari_graph:graph([\n    ari_graph:node(tally, tally, #{}),\n    ari_graph:in(words, {tally, words}),\n    ari_graph:out(totals, {tally, totals})\n]),\n\n{ok, Program} = ari_local_runtime:compile(Graph),\nT1 = ari_vtime:new(1),\nT2 = ari_vtime:new(2),\n{ok, Execution} = ari_local_runtime:new(Program, [\n    {words, [{cat, T1}, {dog, T1}, {cat, T1}, {dog, T2}]}\n]),\n{done, Done} = ari_local_runtime:advance(Execution, infinity),\nari_local_runtime:outputs(Done).\n%% #{totals => [{#{cat => 2, dog => 1}, {1, []}}, {#{dog => 1}, {2, []}}]}\n```\n\nThe epoch 1 totals appear only after all three epoch 1 words have been\ncounted, and the epoch 2 total after the last word.\n\n## Documentation\n\nDesign decisions live in the [OKF knowledge base](okf/index.md):\nthe [graph DSL](okf/design/graph-dsl.md), the [node contract](okf/design/node.md),\n[logical time](okf/design/time.md), the [local runtime](okf/design/local-runtime.md)\nand its [performance measurements](okf/design/local-runtime-performance.md).\n\n## Build\n\n```console\n./silent_rebar3 compile\n./silent_rebar3 test        # eunit and coverage\n./silent_rebar3 dialyzer\n```\n",
  "bytes": 4037,
  "sha": "c5f379d3454e594ffa6a32f86a05de6fc61d27d46810529ee4b2e2a5ecaf05d2",
  "repo_slug": "regikul/ariadne",
  "fonte": "repo",
  "truncated": false,
  "api": "https://agentalog.com/api/listings/okf_regikul_ariadne_okf_index_md_ba6a059d/readme"
}