Introduction
EVpath is designed to be an event transport middleware layer. Specifically, it is designed to allow for the easy implementation of overlay networks, with active data processing, routing and management at all points in the overlay.
EVpath specifically does not encompass global overlay creation, management or destruction functions. Rather it focusses on providing efficient environment for routine transport, while providing interfaces necessary for external management layers.
The package is built around the concept of "stones" (like stepping stones) which are linked to build end-to-end "paths". While the "path" is not an explicitly supported construct in EVpath, the goal is to provide all the local, stone-level, support necessary to accomplish their setup, monitoring, management, modification and tear-down.
Stones in EVpath are lightweight entities that roughly correspond to processing points in dataflow diagrams. Stones of different types perform data filtering, data transformation, mux and demux of data as well as transmission of data between processes over network links.
Stone Types
Connected collections of stones (or paths) specify a particular flow path of data, typically a directed acyclic graph, across a complete distributed system. Before detailing the various types of stones, we'll examine some sample applications that drove the development of EVPath. For example, one might use stones to build an overlay network to distribute data from the data source to the sinks, as is shown below:
We can also imagine that each of those sinks might actually want a slightly customized version of the event stream (with sink i customizing it's stream with function i) and that an efficient implementation of event delivery would place those filter functions as close to the source as possible to avoid transmitting data that is only to be discarded later.
A yet more efficient implementation, armed with knowledge of the nature of functions 1 through 4 might use that knowledge to introduce combined filter functions, for example using filter FC to eliminate data that would be rejected by both filters F3 anf F4 before it is sent.
Examples
Single Process Example
The simplest EVpath program, which involves submitting and receiving an event inside a single program (address space), is shown below.
Include evpath.h and define the data structure we're using:Here, FMField and FMStructDescRec declarations describe to EVPath the data structure (simple_rec) that is to be transmitted. EVPath deals with structured data and that it provides the the safe marshalling of complex structures, even between architectures with different machine representations of data. These will be described in more detail in Section Types and FFS. For the moment it is merely necessary to be aware that all data in EVPath is typed and types play a role in determining actions.
Next, define a function which will be called to consume data from a stone as a passive event receiver. Note that the event is delivered as a void* and cast into the appropriate data type.
In the main program, we declare some values we'll need later and create a CManager to handle the network (even though we're not doing network data transmission in this simple example, EVPath is network-focussed and won't work without Connection Manager).
For the sink (event receiver) side of the application, we create a stone and associate a terminal action with it, specifying the simple_handler() function from above and the simple_format_list which describes the data type that simple_handler expects to receive.
For the source (event sender) side of the application, we create a source handle (EVsource) with simple_format_list describing the type of data that we will submit with this source handle, and the stone to which it will be submitted. (Here we are using the same data specification for both the source and sink. In less tightly-coupled applications, these data types can vary, causing special rules to be invoked. More in Section Types and FFS. It is also possible to create many sources, each of which submit different types of data to the same stone.)
Simple Multi-Process Example
The trival program in the previous example isn't very useful. The next step is to split the sender and receiver and introduce network data transmission using bridge stones.
The Receiving Side
This example will use the same data definitions and handler that were defined in the example above. However, the main program is somewhat different. In addition to the CManager_create(), we call CMlisten() so that Connection Manager will listen for incoming connections. (Strictly speaking, this is not necessary for process that only make and don't receive connections, but we'll do it for all the examples from here on.)Contact list "0:AQIAAENJUEGCzwVFQ0lQUAAA0mY="
and then sit quietly waiting for events. When an event is received it will print the value of the "integer_field" in the event data.
The Sending Side
The sending side of our multi-process example uses the same data definitions from the trivial example above (but without the simple_handler() routine, which is not necessary). The main program begins similarly, but we use the first command line arguments to specify the contact information for the receiver, extracted to the variablesremote_stone and contact_list. string_list we extracted from the argument. To run this program pair, first run net_recv in one window. After it prints out the contact information, run net_send in another window, specifying that contact information as the only argument to net_send. When net_send runs, net_recv should print out "I got 318". The logical architecture of these two programs is shown below:
Sending to multiple receivers
Net_send and net_recv above constitute a basic mechanism for sending data across the net, but what if we want to expand it to something more like a more capable publish-subscribe communication system? The first obvious limitation is net_send only supports sending to one receiver. Our next example is a modification of the sending side to support replication of events to multiple receivers. Again, we use the same data definitions as before, but we modify the program body to introduce a split action:
Some further notes. The text above uses the term "split stone" as a shorthand for "a stone with a split action associated with it". "bridge stone" is used similarly. Actions which are not associated with a specific data format are installed as the default action on stones and are performed when there is no other action that more specifically matches the data format of the triggering event. There is only one default action per stone, so registering a split action on top of an bridge action will overwrite the former (and likely leak some memory). Also, it's technically possible to register, for example, a terminal action and an bridge action on the same stone. In this case, the terminal action would be applied to incoming data of compatible formats, but the bridge action would be applied to other data (more about data compatability in Section Types and FFS). At this point, I'm not sure if there are circumstances where this sort of flexibility is more useful than confusing. But the implementation makes it easy, so for the moment the EVPath API will remain capable of these things. Don't shoot yourself in the foot.
Specializing the event stream
The receiving side
With bridge stones and split stones we can build the sorts of overlay networks described at the beginning of this description. So we'll focus next on how clients can specialize the event stream that they receive. In order to keep the example realistic, we'll assume that only the receiver starts out with the knowledge of how he wants to specialize his data stream. To this end, we'll modify the receiver in a simple way to encode it's desired specialization in the contact information that it prints out. This is done by using create_filter_action_spec() to create a char* value that encodes both the data type that the filter function expects and the body of the function, in this case a simple statement that returns the modulus of the field integer_field in the input. If the function returns false (0), the event is to be discarded. For non-zero returns the event will be passed unchanged through the filter. The resulting filter specification string is added to the printed contact list after being base64 encoded. (The filter spec string contains shell newlines and special characters such as quotes, so it is not easily specified on the command line. By using a base64 encoding (as attributes are encoded), we map the string to characters that are shell-safe. The function used here for encoding is exported from the ATtribute List package and is defined in atl.h.)The sending side
So, the actual architecture of the receiver is unchanged, we just have it export information about its filtering needs. The actual specialization is to occur on the sending side. We'll start with multi_send and change it so that it supports both the original net_recv clients as well as our new derived_recv clients. The program preamble is the same as multi_send, so we'll skip forward to the body of the for loop. In this modified code, we parse out the base64-encoded filter specification by looking for the ':' separator if present (base64 encoding uses the upper- and lower-case alphabet, the decimal digits, '+' and '/', so ':' will never be represented.).
Both cases require the creation of an appropriately-targetted bridge stone:
The EVassoc_intermediate_action() can be used to install handlers which take only a single event as input and can therefore run and "consume" their data immediately. In particular, they are distinct from actions which may leave their input data enqueued for some time (typically handlers which might require more than one event to act). The current EVPath implementation supports only immediate actions with one input and one output, but multiple output actions will be implemented soon.
Finally, we modify the submit action of the program to submit multiple events (so that some will be passed by the filters and others discarded). If you run a mix of net_recv and derived_recv, then run multi_send with all the contact strings on the command line, each of the net_recv programs should receive all 10 events submitted by derived_send, but the derived_recv programs will receive only the odd events. The logical architecture of these programs is shown below :
Further details
EVPath action specifications of the sort created by create_filter_action_spec() are designed to encompass exactly the details of the system's data processing that are not captured by diagrams like the one above, to wit, the details about data types and the nature of the processing that occurs. The idea being that the more structural aspects of the system are best managed separately. This also allows us to extend the functionality of EVpath without disrupting the structural aspects of the API.
So far, we have only introduced the create_filter_action_spec() call as a mechanism for creating action specifications. The semantics associated with the filter action are relatively simple. The data format list specified in the create_filter_action_spec specifies the data type for a parameter named "input" through which the incoming event is made available to the filter function. There is a fair amount of "magic" surrounding the EVpath type handling in any circumstance and to some extent, this magic is customized for the class of processing that is occurring. In the handling of actions created by create_filter_action_spec(), EVpath performs very flexible type matching and applies the filter function to any incoming event which contains a superset of the data fields that are required by the filter. In particular, EVpath is careful that all matching events that satisfy the predicate are passed through unchanged. This is a fairly obvious and natural semantic for a pure filter and is possible because of the type flexibility of the dynamic code generation package that underlies EVpath.
Transforming streaming data
EVpath actions are also capable of processing that modifies or transforms the data, potentially modifying its data type in the process. This is slightly more complex than the simple filter because we have to specify an output data type, so in comparison to create_filter_action_spec(), create_transform_action_spec() has an CMFormatList parameter called out_format_list. Semantically, the out_format_list specifies the data type of an "output" parameter to the specified function. The output record is allocated, but set to zero when provided to the function. It should be initialized properly before the function exits. The following example shows the stream transformation to add an "average" data value to the event. We start with derived_recv and add a declaration of the transform's output data, adding the new "average" field, but also keeping the basic data field for good measure :
We also define a new handler function that prints out the new field as well:
Because both filter and transform actions are immediate, and because the architecture in both examples is identical, we can use derived_send to install the transform action. When we run transform_recv with derived_send, the output should be:
I got 322, average is now 320 I got 327, average is now 322.5
Multiple actions on a stone
To this point, only a single EVpath action has been associated with any stone. However, because actions are type-specialized and multiple data types can flow along a single path, a stone can hold multiple actions. In order to demonstrate this, we'll modify the transform_recv2 program to create transform_recv2. Transform_recv2 will receive the same data as before, but we'll introduce a second data type that we'll use in a second transformation action:.
On the sending side, we have to modify our derived_send program to create derived_send2 which submits two data types and can associate multiple actions with filter stones. We'll use the second_rec data type that we specified above and modify the body of the main program to potentially parse multiple filter specs and use them to associate multiple immediate actions to the created filter stones.
Once the output and filter stones are established, all that remains is to submit events. In order to submit two different data types to our path, we need two EVsource handles, one associated with each data type:
When transform_recv2 and derived_send2 are run, the following output should appear:
I got 322, average is now 320 I got 22, average is now 20.8 I got 327, average is now 322.5 I got 27, average is now 23.3
In the output above, notice that the average values don't really look "right". Rather than consistent values near 
(I.E. around the average of our starting values, 
and 
), we have two lines that are unchanged from the output of the previous example and two lines that seem to represent the average of the floating point values. This is because the "static" declarations in the two filter functions are disjoint. That is, the "static double sum" in trans_func does not reference the same data value that "static double sum" in trans_func2 references. This may or may not be intuitive to you. What is probably not intuitive is that even if there were just one filter function involved, it's possible for there to be multiple, disjoint values for "sum" and "count". This can occur because a single action can match more than one incoming data type. For example, the first function is designed to match incoming data that looks like:
typedef struct _simple_rec {
int integer_field;
} simple_rec, *simple_rec_ptr;
but it will also match any record that has a top-level field named "integer field". Such as
typedef struct _simple_rec {
double new_stuff; /* extra field */
int integer_field;
} simple_rec, *simple_rec_ptr;
Or even
typedef struct _simple_rec {
double integer_field; /* different type for integer_field */
int more_stuff[5]; /* more fields */
} simple_rec, *simple_rec_ptr;
The details of type matching will be described more fully in Section Types and FFS, but for our current purposes the important point is that EVpath generates a new instance of the filter function for each distinct incoming type. This helps to minimize recurring overhead (eliminating the need to map each new type into some single common representation allows us to minimize data transformation), but at the cost of making "static" variables a somewhat trickier tool to use.
Communicating between actions
EVpath does afford a more reliable mechanism for communicating data between different incarnations of an action using attribute lists. In particular, evpath defines a variable, stone_attrs, that is accessible from within each action and which represents a set of attributes (name/value pairs) that is associated with the stone. All actions on a particular stone share the same attribute list. This attribute list can be operated on with a set of standard ECL functions to set and retrieve values. We can modify the transformation functions in transform_recv2.c to utilize this attribute list to get results that represent the real average of data flowing through the stone:
I got 320, average is now 199.12 I got 22, average is now 170.4 I got 325, average is now 181.64 I got 27, average is now 172.9
Router stones
We have previously created 'split' actions to replicate events on multiple output paths, but sometimes what ones wants to do is to select a single output path down which an event should propogate. EVPath allows this to be done with a 'router' stone, and with the previously used EVaction_set_output() routine. EVPath stones can have multiple numbered output 'ports'. In split stones, the output ports are identical and largely anonymous. In filter and transform stones, we only use output '0' as they can only support a single output. However in router stones we can have many outputs, each linked with a specific target stone via EVaction_set_output(). In particular, the code associated with a router action is somewhat like that of a filter stone, except that where the filter would return 0 or 1 indicating that the event should be dropped or passed forward, the code in the router function should return an positive integer indicating which of the output ports to which the even should be forwarded. That is, if one has associated target stones with three output ports, something like:
then the router function should return a value between 0 and 2 inclusive, and the event will be forwarded to the corresponding output stone. (No sample programs included here. Hopefully you've gotten to the point where you can write your own.)
Multi-stones
Multi-stones (sometimes called multi-queue or multi-type stones), are the most complex stones in EVPath. All of the previously-described stone actions are 'immediate' stones, in which the action of the stone is applied to a single event and generally evpath will try to perform those actions immediately upon submission. In particular, immediate processing of the events, with the result of either further submission or dropping the event, is automatic. Situations which require aggregation of information from multiple events, possibly of different data types, are more directly supported by multi-stones.
Two main challenges in creating an event-aggregation functionality in the context of a system like EVPath are 1) specifying when a multi-event action is to run, and 2) in a situation where multiple different event data types might be in the queue at once making the event data visible in a type-safe way. WRT 1), while some systems take the approach of creating a language in which action trigger preconditions can be specified, EVPath takes a somewhat simpler approach. In EVPath multi-stones, the specified action is run every time a new event is placed in the stones event queue. However, there is no automatic dequeue of events, so the action has an opportunity to examine the contents of the event queue and make a decision as to whether or not its pre-conditions have been met. If they have not, the action simply exits without consuming anything from its event queue. This essentially assigns to the action the responsibility for evaluating its own preconditions, however because the action code is dynamically generated native code, the costs are minimized.
In order to allow evaluation of pre-conditions, the action must have the ability to inspect the stone's event queue. The built-in CoD functions that allow this present multiple views of the event queue, allowing events to be referenced by their absolute locations in the queue (with zero being the oldest, first-entered event), or by relative locations in type-based queues (I.E. the 2nd event in the queue that matches type 'A'). The full set of functions provided are described in Section Multi-stone Functions, but the example below demonstrates a typical action function for a multi-stone. In this case, one with a pre-condition that one event of each of its two data types ("a_rec" and "b_rec") to be present before it can operate. It tests for the presence of each type of data. If both types are not found it exits without further operation. But if they are found, it accesses the data, generates a new record and submits it to output port 0. Note that events must be explicitly discarded or they are not removed from the stone's event queue. It is an error to use the 'data' segment of an event after the event is discarded.
For completeness, we include below the types of the records referenced in the multi-stone action above and the code lines used to create and install the action:
Note that the parameter to create_multityped_action_spec() is a null-terminated list of FMStructDescList. In this case, only "a_rec" and "b_rec" are expected incoming event types. "c_rec" is not expected as an incoming event, but including it in the queue_list declares the "c_rec" type so that it can be used as the output type. The complete list of functions available in multi-stone action routines is available below in Section Multi-stone Functions.
Types and FFS
The prior examples have employed FMField and FMStructDescRec declarations describe to EVPath the data structures that are to be transmitted, as well as describing the structures into which EVPath is to place data for terminal handlers. The software package used for these specifications is FFS. FFS was designed and implemented for marshalling/unmarshalling and data filtering/transformation in high performance applications. A description of the motivating applications for FFS can be found in :
A Type System for High Performance Communication and Computation, Greg Eisenhauer, Matthew Wolf, Hasan Abasi, Scott Klasky, Karsten Schwan. in Proceedings of the Workshop on D3Science, associated with e-Science11, Dec 5, 2011, Stockholm, Sweden PDF Format
Additional resources on FFS, including a manual, can be found on the FFS web page.
CoD
The language used in the filter and transformation functions above is CoD (C-on-Demand). CoD is part of the FFS package and is a dynamically-generated subset of C. In general, CoD code is a text string representing the body of a subroutine. The parameters of the subroutine are defined not by the "user provided" text, but by the generating context. In the case of EVPath, the parameters are generally structures whose fields and data types are created from the specified input and output data types. The FFS manual will eventually be extended to describe the syntax and semantics of FFS, but if you write code that would work in C, generally it will work in CoD. In some cases, CoD has some magic that supports FFS in some way, and in other cases CoD leaves out some non-bits of C, mostly that can be replicated in some other way. For example, CoD currently doesn't support switch, do-while, bit-wise complement (~) or conditional expression (?:). The principal magic bit off CoD relates to the fact that FFS allows dynamically sized arrays with the size given by integer-typed field in the same structure. In order to maintain the relationship between the value of this "control field" and the array whose size it specifies, CoD will automatically reallocate the arrays whenever there is assignment done on the control fields. This allows some form of dynamic memory allocation in CoD without the use of malloc/realloc.
However, all of these are really properties of CoD and not specific to its use in EVPath. The unique aspects of CoD code in EVPath are related to the special functions and variables that EVPath makes visible to CoD code in filter and transformation functions (in addition to the input and output parameters that have been noted previously.)
In particular, within an EVPath CoD function, the following global entries are visible and available for use:
- int printf(string format, ...);
- void *malloc(int size);
- void free(void *pointer);
- void sleep(int seconds);
- long lrand48();
- double drand48();
- attr_list stone_attrs;
- attr_list event_attrs;
Aside from the last two attr_list entries, all of these are just making the routines from the standard C library visible. (Note that other entries are not, by default, visible or available, but mechanisms exist to add things to the set of available items.) With respect to the attr_list entries, there are attribute lists associated with each event in evpath and associated with each stone. These data values make those attributes available to CoD functions. The event_attrs is actually a parameter, while the stone_attrs is an external variable, but the difference is immaterial inside the function. WRT the stone_attrs list, it may find an example use as a communication mechanism between CoD functions associated with the same stone, or it may simply be used to query other EVPath-supplied stone attributes. event_attrs may be used to communicate between CoD functions on different stones on the event path, or simply carry event attributes from the source to the final destination.
In order to operate on these attribute lists, the following routines are also made available:
- attr_list create_attr_list();
- create a new attribute list
- void free_attr_list(attr_list l);
- free an attribute list (actually, reduce reference count by 1)
- int attr_set(attr_list l, string name);
- returns boolean true if the named attribute is set, false otherwise
- void set_long_attr(attr_list l, string name, long value);
- set the named attribute to the value
- void set_float_attr(attr_list l, string name, double value);
- set the named attribute to the value
- void set_double_attr(attr_list l, string name, double value);
- set the named attribute to the value
- void set_int_attr(attr_list l, string name, int value);
- set the named attribute to the value
- void set_string_attr(attr_list l, string name, string value);
- set the named attribute to the value
- int attr_ivalue(attr_list l, string name);
- return the int value of the named attribute
- long attr_lvalue(attr_list l, string name);
- return the long value of the named attribute
- double attr_dvalue(attr_list l, string name);
- return the double value of the named attribute
- double attr_fvalue(attr_list l, string name);
- return the float value of the named attribute
- char* attr_svalue(attr_list l, string name);
- return the string value of the named attribute
Multi-stone Functions
In addition to the CoD functions listed above that are available in all CoD-based EVPath actions, multi-stones have the following extra functions that allow for inspection and operation upon the stone's event queue.
Generally speaking, there are multiple forms each basic action. For example, the function EVcount_full() returns the total count of events in the stone queue, but for each of the datatypes specified for the multi-stone, there are also EVcount_<name>() functions available (where <name> is the string name of each datatype in the queue list). These functions return the count of events in the queue that match that particular data type. EVcount(int queue) takes a parameter which is the integer index (zero-based) of a datatype. So in the example above, EVcount(0) is equivalent to EVcount_a_rec(). There is also an integer constant <name>_ID created for each of the queue types, so EVcount(0), EVcount(a_rec_ID) and EVcount_a_rec() are all equivalent forms. In addition, because some actions may wish to manipulate events whose types are unknown, the queue_list may include a special FMStructDescList with the top-level name "anonymous" and a NULL field list. If this FMStructList entry is present, EVcount_anonymous() will return the number of events in the stone's event queue that do not match any of the other data types in the queue list. (If the "anonymous" entry is not present in the queue list, events with unmatched types will be rejected by the multi-stone.)
In addition, functions which operate on entries in the event queue generally have an index parameter that specifies which queue entry they apply to. Index is zero-based and lower numbers are the oldest (first submitted) events. For routines like EVdiscard_full(int index), index is interpreted as indicating the Nth element in the queue, regardless of type. However, EVdiscard_a_rec(int index) applies to the Nth "a_rec" in the queue, ignoring events which are of other types. (While this essentially this treats the "a_rec" elements as if they are in their own queue, there is only ever one, unified, event queue.) Note that all modifications to the event queue take effect immediately and affect the results of subsequent calls, so that, for example, the code segment
will discard the oldest extant event in the stone queue until there are none left.
The full list of available functions are:
If anonymous event availability has been enabled, the type-specific routines above are also available for the "anonymous" meta-type, except that the corresponding EVdata() routines are not present.
