OFlux Modules

Code re-use is a good thing.  Developers want to have encapsulated functionality which they can easily turn into libraries so that those parts can be re-used for other programming projects.  OFlux does this using its modules system which takes its inspiration from Standard ML (or OCaml for that matter).  Standard ML extends modules a bit with an idea called functors for making modules (and their signatures -- like interfaces) generic.  OFlux does not yet support functors, but I have toyed with the idea from time to time.

Writing an OFlux module Bar involves creating the following within C++ namespace Bar:

• a Bar::ModuleConfig structure to hold its "self" object state (each instance has its own object).  Often this object encapsulates the state of the module.  Static modules do not have a ModuleConfig structure -- so they are stateless in a way.
• in a file called Bar.flux (and after including needed OFlux source) the module is declared using the syntax module Bar begin ... end.  All of the usual things go inside of this block (nodes, guards, flow, instances of other modules).  A module is not a recursive concept, so you can't instantiate it within itself.
• to instantiate the module in another OFlux source location the syntax is instance Bar barinst;
• Unless a node N (within the module) is declared with special keyword mutable, it will implicitly acquire the module instance's readwrite self guard for read access.  This provides to the C++ node function Bar::N a way to access the Bar::ModuleConfig object for that instance.  Mutable nodes acquire the self guard for write instead of read, and therefore they can change the internal instance state.

The semantics of instantiating a module is that the content of the module is notionally copied into the current scope of the program and its content (nodes, guards, etc) are just prefixed with the instance name and a "." character.  In order to allow the guards inside of a module to unify (conflating them to one guard rather than two), the syntax where guard1 = instguard1, guard2 = instguard2 is appended to the guard instantiation.  

There is no notion of inheritance between modules, since composition is done by inclusion (via instantiation).  So a developer can build modules that use other modules (e.g. we include Bar.flux in a new module which will contain an instance (or more) of Bar within itself).  Inheritance is a problematic abstraction, and the rule that inclusion should be used (indicating "has a" relationships) guided the design of this language feature.

To see an example of how an OFlux module is written, please have a look at this example in the Github repo and the top-level for that example.

xplot.org on Time Series Events

Visualizing events in a distributed system is tough.  There are good solutions for experimentation on a single machine, but once the software lives on multiple machines issues crop up.  The number one issue is timing synchronization.  Out of the box NTP based time synchronization is not accurate enough (I have found) to deliver proper time synchronization.

Precise Time Synchronization

Success (on Linux) involved installing and running ptpd2 (PTP stands for Precision Time Protocol) available on sourceforge.

On the master (fileserver machine) I keep ptpd2 running with:

 ./ptpd2 -W -b eth0

And on the slave compute nodes I just keep them running in sync with the master using:

 ./ptpd2 -c -g -B

Together, these processes keep the cluster time synchronization locked to around 1 microsecond - which is sufficient for my immediate needs.

Enter xplot.org

When looking into a TCP/IP issue, I discovered tcptrace which is a program for analyzing the output of tcpdump or snoop (on Solaris).  Several visualizations of a TCP/IP session captured by those sniffing tools are available, and (most importantly) they are viualized within a tool called xplot.org.  This tool is an X windows program which can efficiently display large time series data sets in a 2D graph.  Unfortunately, its text-based data format is only really documented by reviewing its C source code.  I intend to share what I have learned about the format so that (hopefully) others can skip reading the source.

Most of the commands are issued on a single line, with the exception of commands that take a text argument.  Commands that display text in the graph require the subsequent line for the text content.  Most drawing commands looks like:

 commandname x-coordinate1 y-coordinate2 [ x-coordinate2 y-coordinate2 ] [ color ]

Configuration of a time series graph that uses UTC timestamps (second resolution with microseconds expressed after the decimal part) is done using:

timeval double

Using three text commands we can label (using xlabel and ylabel commands) the axes and title the graph itself:

title
Results from Tue Apr 24 11:45:55 2012 - run Tue Apr 24 11:48:00 2012
xlabel
Wall clock time
ylabel
Event id

Adding some lines is easy with the line command:

line 1335282355.550356 1 1335282355.550356 2 yellow
line 1335282355.550356 2 1335282355.620854 2 yellow
line 1335282355.620854 2 1335282355.620854 3 yellow
line 1335282355.620854 3 1335282355.691357 3 yellow
line 1335282355.691357 3 1335282355.691357 4 yellow

Adding some more random things:

dtick 1335282355.623986 2.004608 blue
line 1335282355.623986 2.004608 1335282355.624049 2.004608 blue
diamond 1335282355.624049 2.004608 green
diamond 1335282355.624101 2.005530 green
line 1335282355.624049 2.004608 1335282355.624101 2.005530 green
line 1335282355.624049 2.004608 1335282355.624514 2.002304 gray20
box 1335282355.624514 2.002304 gray20

Finally a look at how xplot.org paints all of this data in its viewer:

 Zooming in on the green diamond (left button on my mouse):

Zooming in once more:

Finally we can see the event tree that I am depicting here (the Y-axis is mostly used for layout of the tree), and the X-axis shows accurate timing information:

The middle button (or wheel button) on my mouse helps to pan in one direction or another, and a single left-click is used to pop back through the zoom-ins you have done recently.

Summary

If you have lots of data to process and are comfortable generating an xplot file using your favourite tool (programs like awk and Perl are useful for transformations like this), then you may get some mileage out of xplot.org for visualizing moderate amounts of time series data (100s of megabytes) using a modest Linux computer (e.g. a netbook).  More serious hardware is definitely capable of more.  I have not found an in-browser viewer which is close to being as fast as the native xplot.org program.

Enumeration on the stack

Allocating objects in C++ on the heap is not free. Although there are many implementations of allocators that are very fast (and particularly good for multi-threaded applications), it is still preferable to avoid allocating too many things on the heap. In Java programs there is not much choice in the matter, since the use of the heap is kind of an endemic habit. Fortunately garbage collection in Java has gotten a whole lot better. In this post I re-visit the enumeration interface I described earlier and try to provide two implementations of it (one uses the program stack only, and the other relies on the heap).

Enumerators Again

In Enumerator.h I describe the same generic enumerator interface, and then I offer some convenience C macros which allow a function to be called which creates the enumerator, and enables a traversal of it:

template< typename AnyT >
struct Enumerator {
  virtual bool next(const AnyT * &) = 0;
  virtual ~Enumerator() {}
};

The stack macros within Enumerator.h are as follows (note the use of "placement new" and the explicit call to the virtual destructor):

#define stack_for_each_start(EFUN,Tp,X) \
 { /*new scope*/ \
  char _enum_buff[EFUN##__storage_size]; \
  typedef Tp _enum_Tp; \
  Enumerator<Tp> * _enum##X = EFUN(_enum_buff); \
  Enumerator<Tp> * _enum = _enum##X; /*for cleanup*/ \
  const Tp * X = 0; \
  while((_enum##X) && (_enum##X)->next(X)) {

#define stack_for_each_end \
 } if(_enum) { (_enum)->~Enumerator<_enum_Tp>(); } }  

So that is reasonably ugly, and most C macros are, but the beauty on the application side is coming -- be patient! Here are the heap allocated macros:

#define heap_for_each_start(EFUN,Tp,X) \
 { /*new scope*/ \
  Enumerator<Tp> * _enum##X = EFUN(); \
  Enumerator<Tp> * _enum = _enum##X; /*for cleanup*/ \
  const Tp * X = 0; \
  while((_enum##X) &&(_enum##X)->next(X)) {

#define heap_for_each_end \
 } delete (_enum); }

Our Library

In order to write our application library header file app.h, we want to minimize the amount of junk we expose the user application code. Here we declare a single function app::get_all_ts() which will return the enumerator and use existing storage if provided as an argument (which is our way to allocating on the stack):

#include "enumerator.h"
#include <cstdlib>

namespace app {
  struct T { int id; char name[40]; }; // T defined here
  Enumerator<T> * get_all_ts(void * storage = 0);
  extern const size_t get_all_ts__storage_size;
} // namespace app

The app::get_all_ts__storage_size symbol is provided so that we know how much storage the underlying enumerator implementation needs.  The implementation of the library code behind app::get_all_ts hides almost all the detail of the collection being enumerated, and how the enumerator implementation works:

#include "app.h"
#include <new>

namespace app {

T t1 = { 1, "sam" };
T t2 = { 2, "bill" };
T * tarray[] = { &t1, &t2, 0 };

struct AllEnumeratorImpl : public Enumerator<T> {
 AllEnumeratorImpl() : at(&(tarray[0])) {}
 virtual bool next(const T * & tp) {
  if((tp = *at)) {
   ++at;
   return true;
  }
  return false;
 }
private:
 T ** at;
};

Enumerator<T> * 
get_all_ts(void * storage)
{
 return storage
  ? new(storage) AllEnumeratorImpl()
  : new AllEnumeratorImpl();
}

const size_t get_all_ts__storage_size = 
        sizeof(AllEnumeratorImpl);

} // namespace app

My simple C array tarray of objects is just an example. A real library would have a more interesting collection with less static contents.

User Code

Finally we can use these mechanics to do a traversal in the ex.cpp user code:

#include "app.h"
#include <cstdio>

int
main()
{
  stack_for_each_start(app::get_all_ts, app::T, tptr)
    printf("%d %s\n", tptr->id, tptr->name);
  stack_for_each_end
  return 0;
}

Building and running the ex.cpp code we get:

 % g++ -g app.cpp ex.cpp -o ex
 % ./ex
 1 sam
 2 bill

Checking for Leaks

Using the heap_ version of the macros available in Enumerator.h, the same result is had. The difference in Valgrind output on the two implementations is a bit interesting (first for the stack macros):

% valgrind --tool=memcheck --leak-check=full ./ex
==29306== Memcheck, a memory error detector.
==29306== Copyright (C) 2002-2008, and GNU GPL'd, by Julian Seward et al.
==29306== Using LibVEX rev 1884, a library for dynamic binary translation.
==29306== Copyright (C) 2004-2008, and GNU GPL'd, by OpenWorks LLP.
==29306== Using valgrind-3.4.1-Debian, a dynamic binary instrumentation framework.
==29306== Copyright (C) 2000-2008, and GNU GPL'd, by Julian Seward et al.
==29306== For more details, rerun with: -v
==29306== 
1 sam
2 bill
==29306== 
==29306== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 17 from 1)
==29306== malloc/free: in use at exit: 0 bytes in 0 blocks.
==29306== malloc/free: 0 allocs, 0 frees, 0 bytes allocated.
==29306== For counts of detected errors, rerun with: -v
==29306== All heap blocks were freed -- no leaks are possible. 

Comparing this with the heap implementation which does see one alloc happen:

% valgrind --tool=memcheck --leak-check=full ./ex
==29288== Memcheck, a memory error detector.
==29288== Copyright (C) 2002-2008, and GNU GPL'd, by Julian Seward et al.
==29288== Using LibVEX rev 1884, a library for dynamic binary translation.
==29288== Copyright (C) 2004-2008, and GNU GPL'd, by OpenWorks LLP.
==29288== Using valgrind-3.4.1-Debian, a dynamic binary instrumentation framework.
==29288== Copyright (C) 2000-2008, and GNU GPL'd, by Julian Seward et al.
==29288== For more details, rerun with: -v
==29288== 
1 sam
2 bill
==29288== 
==29288== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 17 from 1)
==29288== malloc/free: in use at exit: 0 bytes in 0 blocks.
==29288== malloc/free: 1 allocs, 1 frees, 8 bytes allocated.
==29288== For counts of detected errors, rerun with: -v
==29288== All heap blocks were freed -- no leaks are possible.

At this point we also know that neither method leaks memory, or segmentation faults. So they are pretty equivalent to the user program.

Performance

To compare performance I made a new sample ex_timing.cpp with no I/O but iterated 10 million times:

#include "app.h"
#include <cstdio>

int
main()
{
  for(long i = 0; i < 10000000; ++i) {
    heap_for_each_start(app::get_all_ts, app::T, tptr)
    heap_for_each_end
  }
  return 0;
}

I ran each version (stack and heap) 3 times and picked the middle user timing to get this table (these used compiler optimization -O3):

macro prefix	user time for 10 million iterations
stack	292 ms
heap	960 ms

The stack version is over three times faster!

Summary

For local things like enumerators the method of encapsulating the functionality had us returning a heap allocated object which the user code had to delete (via out macro) the object when it was done.  This was quite functional, but it has a performance penalty relative to what can be done with the stack.  The trouble with the stack is that the memory for the enumerator object has to be be "allocated" prior to the call into the library call which returns the enumerator -- and this forced us to expose the size of the implementation enumerator somehow.  I think it is an acceptable cost (some minor symbol pollution) in order to get some benefit for performance.  Of course, I wish for a way to do this which does not involve using C macros.

OFlux: Building a Flow

Most of a .flux file content will be node declarations and connecting flow.  There are two types of nodes: abstract nodes which serve as helpful connection points and do not have C++ code associated with them, and concrete nodes which are implemented via a C++ function.  In this post, I will describe the major features of the OFlux language in some detail.  First off is something programming language people call "choice".

Routing with Conditions

Suppose we have a source node Src which blocks on some input and produces an output Foo * foo.   The source node is declared as follows (its input set is empty -- which is necessary for a source node):

 node Src () => (Foo * foo);
 source Src;

As written, the oflux compiler will complain about this input since the flow rooted at Src with only one node in it does not end with a node that has a () output set.  Adding a line which has terminate Src, will silence the complaint (in effect saying "we know what we are doing, don't complain").  What we really want to accomplish is to apply a condition isFooEnough() to every foo that comes out of Src.  This condition will -- in reality -- be implemented within our C++ code using a function with prototype bool isFooEnough(Foo *):

 condition isFooEnough(Foo *) => bool;

Suppose we want to implement separate nodes ConsumeFooEnough and ConsumerFooLacking as successors depending on the outcome of the isFooEnough() test.  Needless to say, it is assumed that isFooEnough has no side-effects on its argument or the global state of the program since multiple calls to such a conditional might occur when the logic gets more complicated.  Once consumed we will dispose of foo with a node called DisposeFoo.  The abstract node ComsumeFoo is used below only to help describe the decision being made:

 node ConsumeFooEnough (Foo * foo) => (Foo *);
 node ConsumeFooLacking (Foo * foo) => (Foo *);
 node DisposeFoo (Foo * foo) => ();
 node abstract ConsumeFoo (Foo * foo) => ...;
 ConsumeFoo: [isFooEnough] = ConsumeFooEnough -> DisposeFoo;
 ConsumeFoo: [*] = ConsumeFooLacking -> DisposeFoo;
 source Src -> ConsumeFoo; /* changing the original */

Much like pattern matching in a language like OCaml or Scala, the syntax for routing is done using the rule which first matches the input.  So if isFooEnough(foo) returns true, then the routing to ConsumeFooEnough happens, and otherwise the path to ConsumeFooLacking is chosen.  This syntax survives from the original Flux language design.  Elipsis (...)  is used on the output declaration of ConsumeFoo to indicate to the compiler that we do not care to specify the output set, and if we did only () would be acceptable since other options would not unify with the output set of DisposeFoo.  Now that a basic flow is described, the application programmer only has to implement the following concrete node functions in their C++ code to create a runnable program: Src, ConsumeFooEnough, ConsumeFooLacking, and DisposeFoo.  If it is not necessary to have DisposeFoo as a separate node, I would recommend that it just be implemented as a function which is called inside of the ConsumeFooEnough/ConsumeFooLacking C++ functions.  The only reason to have a separate node (which implies a separate event and a run-time scheduling of that event -- a non-negligible cost), is if DisposeFoo is interacting with resources in the program (please see my posting on OFlux guards).

Running oflux

To compile the content above in a file called web.flux we issue the command:

 % ./oflux web.flux
OFlux v1.00-6-gc59d671 on web.flux

This causes the following output to be created locally:

• web.dot: a description of the OFlux flow which can be turned into a pretty picture showing types, conditions and nodes (blue are abstract).  To create a picture with graphviz, run dot -Tpng web.dot -o web.png:

• web.xml: the XML description of the OFlux flow which will be loaded at run-time to make the program run (edits to this file can cause changes to the flow without having to recompile. For brevity only the Src node is shown as it has the most interesting entry.):

<flow name="web.flux" ofluxversion="v1.00-6-gc59d671">
 <node name="Src" function="Src" source="true" door="false" iserrhandler="false"
 detached="false" external="false" inputunionhash="cbd4d4a285d623ee19470f7d5d68e
1c1a765263c3c9242a8d61b222d49c7e64c" outputunionhash="6e63ce559f96ef9b1f68aa4370
2f1343c9638715464dea7d140e6b4d068d9688">
  <successorlist>
   <successor name="0">
    <case nodetarget="ConsumeFooEnough">
     <condition name="isFooEnough" argno="1" isnegated="false" unionhash="6e63ce
559f96ef9b1f68aa43702f1343c9638715464dea7d140e6b4d068d9688"/>
    </case>
    <case nodetarget="ConsumeFooLacking">
     <condition name="isFooEnough" argno="1" isnegated="true" unionhash="6e63ce5
59f96ef9b1f68aa43702f1343c9638715464dea7d140e6b4d068d9688"/>
    </case>
   </successor>
   <successor name="erste">
    <case nodetarget="Src"/>
   </successor>
  </successorlist>
 </node>
 ... 
</flow>

• OFluxGenerate.h: A header file declaring the needed node and conditional functions which includes the application's mImpl.h header file (which defines the types used).  This header should be included in the application's .cpp source.  It should not be necessary to understand all of the mechanics inside of the OFluxGenerate.h file.  Using the node declarations inside of web.flux, each node N gives rise to types N_in, N_out and N_atoms which are needed for the C++ prototype  int N(const N_in *, N_out *, N_atoms).
• OFluxGenerate.cpp: The generated C++ code needed to bind the OFlux program to the run-time.  This is where the static tables used to look-up symbols read from the XML file (web.xml) live.

Building a complete application means OFlux compiling/C++ compiling/linking/running with these outputs.  Errors can crop up at any of those stages, so it is typical to have a tool like Gnu make build the whole project -- even verifying that the XML file is loadable (properly gets its symbols from the OFluxGenerate.cpp code)

Some Limitations and Philosophy

In the above example there are plenty of things for the oflux compiler to check for us.  When a node is connected to another as a successor, the compiler does an asymmetric unification of the output set with the input set.  This means that each input should exist as an output from the predecessor.  Generally this is done using the name of the argument (so matching foo with foo in my example), but if there is a mis-match in naming an attempt is made to unify using types.

If the names of the formal parameters do not match there are consequences in the generated code -- a C union is necessary to give the field two names (this is trouble if any of the argument types is non-POD, and that happens quite a bit with C++ code).  When unification fails, the offending node and its argument is indicated by the oflux compiler (the first error causes the compilation to halt).


OFlux adds another dimension to the task of programming a server.  A new server design will have nodes and flow to go with it.  My consistent impression working with developers who are new to the tool is that many many nodes end up being defined (much like functions are used in C++).  Making a flow very long (lots of nodes from source to any sink), can be quite detrimental to performance if most of the nodes are doing less work than the overhead to execute them.  My best advice is to try an initial flow with the absolute least number of nodes possible, and then investigate refining that program into more nodes in a step-wise manner.  Keep it as simple as possible!

Routing to Many Places

Occasionally concurrency is needed in the flow since an output is consumed by multiple nodes, and its inefficient to have them each process that input (a foo perhaps) sequentially.  The OFlux run-time keeps a node event (which holds its output data) alive using reference counting, so it is possible to keep a node event around long enough to be processed by multiple successor node events.

If we wanted to modify our example above to have DisposeFoo be an abstract which aliases two nodes we want to run at the same time (ReclaimFoo and CommunicateFoo).  This could be done as follows (replacing the C++ function implementation we have for DisposeFoo with implementations for ReclaimFoo and CommunicateFoo:

 node abstract DisposeFoo (Foo *foo) => ();
 node CommunicateFoo(Foo * foo) => ();
 node ReclaimFoo(Foo * foo) => ();
 DisposeFoo = CommunicateFoo & ReclaimFoo;

Using this technique, the completion of a ConsumeFooEnough event will cause two new events to be created: one for CommunicateFoo and another for ReclaimFoo.  If these nodes were detached or had a shimmed system call in them, they could both dispatch at the same time and run concurrently.

Summary

Using the basic composition syntax within OFlux an application developer can describe the top-level flow of events in his program without having to explicitly manage a thread pool themselves.  The language offers three main types of basic composition:

1. sequential (using ->)
2. concurrent (using &)
3. choice using (using :[ ? ]matching)

OCaml: Hash Anything

Tasty

OCaml has a generic marshalling code within it, and it has MD5 Digest code.  In this post I will simply compose the two of them to achieve a hash function which works on any object (a handy magical thing to have):

 % ocaml
        Objective Caml version 3.10.2

 # Marshall.to_string;;
 - : 'a -> Marshal.extern_flags list -> string = <fun>
 # Digest.string;;
 - : string -> Digest.t = <fun>
 # Digest.to_hex;;
 - : Digest.t -> string = <fun>

Just entering the component functions you can see the types of each of them when they are entered into ocaml REPL (in fact, ocaml returns symbolname : type = value on each evaluation).  Note that ;; is needed to get it to evaluate since it takes multi-line input (OCaml does not require indentation for scoping -- a "feature" that some languages make use of):

 # let hash x =
    let m_x = Marshal.to_string x [] in
    let d_m_x = Digest.string m_x
    in  Digest.to_hex d_m_x;;
 val hash : 'a -> string = <fun>

Now we can use this hash on all kinds of things (with any type):

 # hash [1;2;3];;
 - : string = "64cb37afbe72effe97fb4f089a82f9b2"
 # hash 4.5966;;
 - : string = "5eed266efb8593218f1cb1cacf1f8d89"
 # hash "Ocaml rocks!";;
 - : string = "f078e0f37bce6c8a83112f9461bf7544"

It is almost too simple.  Writing a meta-programming monster hash function in C++ is much less enjoyable.  Some have criticized the OCaml library and its incompleteness (compared to Java perhaps) as a reason for its less wide adoption.  Now that OCaml-java is a project, that excuse has less weight. Hopefully some of the parametric polymorphism available to the OCaml bundled libraries finds its way into that project.  As usual, choosing the proper tool for any job is an important step in getting it done.

TIPC: Replication Log Example

Tasty

Losing customer requests is terrible, at least for the applications that I work on.  Some applications can survive request loss by pushing the problem to the customer -- having them sort out the consistency problem and re-submit the request if necessary.  Here, I want to describe a solution I am working on to avoid request loss and leveraging the reliability of replication within a cluster to accomplish this goal.

The service of a user's request involves interacting with (possibly changing) the state of the world (persistent data on the server).  Each request received by the server needs to immediately be persisted reliably, so that only in the event of a catastrophic failure do we lose that request.  This means that we replicate that request to multiple nodes within the cluster or attempt to preserve it on a reliable storage device/appliance.

After investigating reliable storage with low latency, I came to the conclusion that I/O devices like these are very expensive and tend to be (for cheaper ones) installed in a single computer .  If you lose that computer, asking your server facility to pull a card out and put it into another machine does not lead to a zero down-time solution.  An expensive appliance sitting on your network can accomplish the goal of persistence, but then you have already opened the door to communicating on the network to do persistence.  You might as well persist by replicating the data to multiple nodes in your cluster.

Persistence by Cluster Replication

Consider the following picture where a service which has received the user request G and replicated to several instances of another service S which just stores and acknowledges the data:

Each of the pink boxes is a separate machine/node in your cluster with a separate power supply and UPS (hopefully) so that you can be reasonably certain that a minor disaster only affects one node.  In the terms used by Greg Lindahl in his talk on replication (which goes into how it is just a way better option than investing in RAID once you start to think about clustered architecture) we are achieving R3 replication.

The strategy is to receive the customer request in G, annotate it with some immediately available state into a message (e.g. timestamps, sequence numbers -- back away from that database handle hot-shot!) and send it to S with an increasing message sequence number.  S receives this stream of incoming messages i, (i+1), (i+3), ... , and "stores" them into an sequence of memory mapped archive files which are each just N-element message arrays by placing message i into mapped file (i/N) at position (i mod N). 

By using memory mapped files we get two excellent side-effects.  First, the virtual memory system of the node running S will eventually permanently persist the mapped file to disk for us on its own.  Second, we have the speed of a direct in-memory recv() into that mapped memory (assuming that the mapping of new archive files into memory is somehow done for us using another thread.  In order to get the multicast send/recv done, I am going to use TIPC and its reliable multicast socket which I discussed in an earlier posting.

Once the message is properly received, an extra word in the S log is set to indicate "valid data", and an acknowledgement (containing the sequence number) is multicast on a separate channel back to G.  Ideally, G has a thread for processing customer requests and sending S messages, a rotating buffer to keep those "in the process of persistence" messages, and a second thread which deals with acknowledgements (implementing an "ack" policy which indicates how much persistence is enough).  Having ensured persistence by the time the policy is fulfilled in this second thread, G is free to send some further acknowledgement to the customer ("we have it!").

Losing an S

Consider what happens when we lose a node that runs an S:

In this case we degrade to R2 replication, and we start to get nervous.  Fortunately our cluster resource manager (a.k.a Pacemaker for my setup) notices this and starts up another S somewhere else:

The fresh S has empty archives however, so it needs to immediately start listening to G and asking the other Ss to fill it in on what has been going on.  This is done by implementing a second thread in S dedicated to peer-to-peer replay of existing archive data for the benefit of the newcomer.

Losing a G

If G restarts or needs to be run fresh on a new node since the node it was running on fails, then it will need to come up and ask the network of Ss what the last message sequence number was (so that it can start generating new messages which do not overlap or have gaps with that).  A fourth TIPC reliable multicast channel is dedicated to this purpose to serve that information.  G waits a specified period (I think 1 second) to receive as many answers to that question as possible and uses the maximum response sequence number to begin its new stream.

It is possible to lose a request when G is lost, but that request will not have been acknowledged.  The low latency characteristics of G, make the likelihood of losing customer traffic which has yet to be received by G on the front-side more likely.

Benefits

There are a number of benefits to this scheme:

1. With low latency we are certain to reliably persist the customer's request
2. No relational databases are disturbed which might have unpredictable latency profiles (adversely affecting performance)
3. Replication scales well in a multi-cell sense (sharding) on the cluster
4. Each node with an S running could have an API to walk the memory mapped archive files independently (without disrupting S) and (possibly) do logarithmic (in space) searches on the data stored there
5. Minimal indexing is maintained -- its just a binary log of same-sized messages we are replicating -- so it does not affect the performance.
6. Less reliance on non-volatile storage and its failure characteristics (spinning rust as they call it), in favour of RAM-based storage
7. Leveraging the efficiency of virtual memory to chose the moment and method to "sync" data to non-volatile storage.

TIPC's reliable multicast really shines for this problem, since it enables a to have services at cluster level (the detail of which node they run on is not necessary for the programmer to delve into), and the ability to reliably send message-oriented packets over multicast to just the nodes who are listening is a big win.

OFlux: Detached Nodes

Tasty

Previously, I described the anti-pattern of blocking while locking, and also how it is that the OFlux run-time escapes this pitfall.  If a node in your program tends to do two or more blocking system calls, each one will be intercepted by the shim mechanism to give up the run-time's main mutex.  The context switching in a case like that could be optimized if the node does not cause side-effects within the rest of the program (mostly modifying non-local state).  The optimization is to enter the node C++ function as if it were one big system call (releasing the main run-time mutex for the duration of the function's execution).  This saves context switching on the mutex and conceivably increases the concurrency in the program (nodes events for these detached nodes are now able to run independently of the run-time more often).  Here is how we augment the basic state diagram for each run-time thread to accommodate this idea:

For nodes that are declared detached, the ability to run in the new mode (dotted line box on the right side) is available when the run-time sees that there are enough threads to allow this.  These two dotted-boxes indicate the states where the thread is not holding the main run-time mutex.

Example: Sleeping Beauty and the Seven Dwarves

Within a working copy of the OFlux Github repo, you can create a new directory called src/examples/dwarves with the following ex-contents.mk make file:

$(info Reading ex-contents.mk $(COMPONENT_DIR))

OFLUX_PROJECT_NAME:=dwarves

include $(SRCDIR)/Mk/oflux_example.mk

$(OFLUX_PROJECT_NAME)_OFLUX_CXXFLAGS+= -DHASINIT -DHASDEINIT

The dwarves.flux file describes the flow:

node SnowWhite () => (int apple_id);
node Dwarf (int apple_id) => ();
source SnowWhite -> Dwarf;

The C++ code for these nodes is pretty simple. Every 0.10 ms SnowWhite sends out an apple, and a Dwarf picks it up and does ten 0.10 ms sleeps in order to consume it (in mImpl_dwarves.cpp):

#include "OFluxGenerate_dwarves.h"
#include "OFluxRunTimeAbstract.h"
#include <sys/time.h>
#include <unistd.h>
#include <cstdlib>

long dwarf_count = 0;
extern oflux::shared_ptr<oflux::runtimeabstract> theRT;

int
SnowWhite(const SnowWhite_in *
        , SnowWhite_out * out
        , SnowWhite_atoms *)
{
        static int apples = 0;
        out->apple_id = apples++;
        if(apples>10000) {
                theRT->hard_kill();
        }
        usleep(100);
        return 0;
}

int
Dwarf(    const Dwarf_in * in
        , Dwarf_out *
        , Dwarf_atoms *)
{
        __sync_fetch_and_add(&dwarf_count,1);
        for(size_t i = 0;i < 10; ++i) {
                usleep(100);
        }
        return 0;
}

I have also added code to produce statistics when the program exits:

struct timeval tv_start;

void
deinit()
{
        struct timeval tv_end;
        gettimeofday(&tv_end,0);
        double total_time = tv_end.tv_sec-tv_start.tv_sec
                + (tv_end.tv_usec - tv_start.tv_usec)
                   /1000000.00;
        double dps = dwarf_count / total_time;
        printf("ran %lf seconds, dispatched %lf "
               "dwarves per second\n"
                , total_time
                , dps);
}

void
init(int argc,char * argv[])
{
        atexit(deinit);
        gettimeofday(&tv_start,0);
}

As is, the dwarves.flux flow will produce the following output on my Asus 1000HE netbook (which has 2 hardware contexts and 1 core):

 # ./builds/_Linux_i686_production/run-dwarves.sh \
   2> /dev/null  | grep ran
ran 5.109480 seconds, dispatched 1957.146324 dwarves per second

Detaching Dwarves

But if we make the Dwarf node detached (which I claim will likely be of benefit since the usleep shimmed system call will be called less frequently:

node SnowWhite () => (int apple_id);
node detached Dwarf (int apple_id) => ();
source SnowWhite -> Dwarf;

Re-running the test, we can see that we are running a little faster:

 # ./builds/_Linux_i686_production/run-dwarves.sh \
   2> /dev/null  | grep ran
ran 3.468819 seconds, dispatched 2882.825538 dwarves per second

So detaching nodes can pay off handsomely if it is safe to do so, since it reduces the in and out of the main run-time mutex.  It is unsafe to do this if there is something about the node source code which makes it unsafe (e.g. mutating non-local state).  Detached nodes are also useful when making calls to 3rd party libraries which (themselves) have mutexes -- in order to avoid a deadlock with the run-time mutex.