Explanation of how to use decentralized coordination

examples/C/Makefile

@@ -0,0 +1,3 @@
+clean:
+	rm -rf *.lft *.csv *.log src-gen fed-gen include bin
+

examples/C/README.md

@@ -5,6 +5,7 @@
 * [Browser UIs](src/browser-ui/README.md): How to create a browser-based UI for an LF program.
 * [Car Brake](src/car-brake/README.md): Sketch of ADAS system illustrating the CAL theorem.
 * [Deadlines](src/deadlines/README.md): Uses of deadlines in Lingua Franca.
+* [Decentralized coordination](src/decentralized-coordination/README.md): Tutorial on federated decentralized coordination.
 * [Distributed](src/distributed/README.md): Basic federated hello-world examples.
 * [Furuta Pendulum](src/modal_models/FurutaPendulum/README.md): A controller and simulation illustrating a modal reactor.
 * [Keyboard](src/keyboard/README.md): Responding to keyboard input using ncurses.

examples/C/src/decentralized-coordination/LiveDestination.lf

@@ -0,0 +1,35 @@
+/**
+ * A simple federated example that elaborates SimpleFeedForward so that the destination reactor has
+ * its own timer-driven activity. It has large lags that can be fixed by using STAA instead of STA.
+ * This example also explores the use of after delays vs. STA and STAA. See the README.md file for
+ * more details.
+ * @author Edward A. Lee
+ */
+target C {
+  coordination: decentralized
+}
+
+import Count from "../lib/Count.lf"
+
+reactor Destination(STA: time = 100 weeks) {
+  input in: int
+  timer t(0, 500 ms)
+
+  reaction(t) {=
+    interval_t lag = lf_time_physical() - lf_time_logical();
+    lf_print("**** Local reaction lag is " PRINTF_TIME "us at logical time " PRINTF_TIME "us.",
+        lag/1000, lf_time_logical_elapsed()/1000);
+  =}
+
+  reaction(in) {=
+    interval_t lag = lf_time_physical() - lf_time_logical();
+    lf_print("**** Reaction to network input %d lag is " PRINTF_TIME "us at logical time " PRINTF_TIME "us.",
+        in->value, lag/1000, lf_time_logical_elapsed()/1000);
+  =}
+}
+
+federated reactor {
+  c = new Count(offset=0, period = 3 s)
+  p = new Destination()
+  c.out -> p.in
+}

examples/C/src/decentralized-coordination/MonteCarloPi.lf

@@ -0,0 +1,79 @@
+/**
+ * Federated example that parallelizes a Monte Carlo estimation of Pi, as described here:
+ * https://en.wikipedia.org/wiki/Monte_Carlo_integration. This example illustrates subtleties with
+ * programs that dynamically choose a stopping time by calling `lf_request_stop()`. See the
+ * README.md file for more details.
+ */
+target C {
+  coordination: decentralized
+}
+
+import Random from "../lib/Random.lf"
+
+// Pi needs an STA because its event queue is usually empty and otherwise it will advance to the stop time.
+reactor Pi(n: int = 10000000, bank_index: int = 0, STA: time = 100 weeks) extends Random {
+  input trigger: bool
+  output out: double
+
+  method in_circle(x: double, y: double): bool {=
+    return x * x + y * y <= 1.0;
+  =}
+
+  reaction(startup) {=
+    self->seed = self->bank_index + 1;
+  =}
+
+  reaction(trigger) -> out {=
+    tag_t now = lf_tag();
+    int count = 0;
+    for (int i = 0; i < self->n; i++) {
+      double x = uniform_double(-1.0, 1.0);
+      double y = uniform_double(-1.0, 1.0);
+      if (in_circle(x, y)) count++;
+    }
+    lf_set(out, (4.0 * count) / self->n);
+  =}
+}
+
+reactor Gather(parallelism: int = 4, desired_precision: double = 0.000001) {
+  preamble {=
+    #include <math.h>
+  =}
+  input[parallelism] in: double
+  output next: bool
+  logical action a
+  state sum: double = 0.0
+  state count: int = 0
+
+  reaction(startup, a) -> next {=
+    lf_set(next, true);
+    tag_t now = lf_tag();
+  =}
+
+  reaction(in) -> a {=
+    for (size_t i = 0; i < self->parallelism; i++) {
+      self->sum += in[i]->value;
+      self->count++;
+    }
+    double estimate = self->sum / self->count;
+    tag_t now = lf_tag();
+    lf_print("**** %d: estimated pi is %f at tag " PRINTF_TAG,
+        self->count, estimate, now.time - lf_time_start(), now.microstep);
+    if (fabs(estimate - M_PI) <= self->desired_precision) {
+      lf_request_stop();
+    }
+    // Note: Even if we are requesting stopping, we need to execute at the stop tag because
+    // we are using a dataflow style and the downstream reactor will block until they receive our data.
+    lf_schedule(a, 0);
+  =} STP(100 weeks) {=
+    tag_t now = lf_tag();
+    lf_print_error("STP violation at Gather at tag " PRINTF_TAG, now.time - lf_time_start(), now.microstep);
+  =}
+}
+
+federated reactor(parallelism: int = 4) {
+  pi = new[parallelism] Pi()
+  g = new Gather(parallelism=parallelism)
+  pi.out -> g.in
+  (g.next)+ -> pi.trigger
+}