• store_ref = xump_store_create (engine, store_type, store_name);
• store_ref = xump_store_fetch (engine, store_name);
• xump_store_delete (engine, store_name);
• queue_ref = xump_queue_create (engine, store_name, queue_name);
• queue_ref = xump_queue_fetch (engine, queue_name);
• xump_queue_delete (engine, queue_name);

So that the queue refers to its store by name, rather than by reference.

For the store portals, I've used the factory feature of IPR portals, which lets us create store instances more simply. At startup we create a factory (also a portal) for each store implementation:

 // Create new engine instance engine = xump_new (); // Create portal factory for each store implementation store = xump_store_ram__xump_store_factory ("RAM"); xump__xump_store_bind (engine, store); xump_store_unlink (&store); 

We can then create new store instances by using the factory name as a 'store type', e.g:

 store = xump_store_create (engine, "RAM", "store-1"); 

The actual work of finding the right factory and getting it to create a new store instance is done in the xump_store_create method.

by pieterh

I've not made the engine API yet, still working through some design concepts. Let's look at credit-based flow control. This is needed to make MOVE work safely. As I've said previously, for COPYs I don't want to use credit-based flow control, indeed any kind of flow control at all.

I'll define a "credit" unit, so we can say that one credit corresponds to one message. In AMQP I originally designed credit based on message volume as well but this is nasty because the semantics of credit are fundamentally not about network capacity but rather about how much the upstream trusts the downstream. We cannot trust a recipient with fractions of a message. (This is also why I think COPY does not need credits, because it does not imply any kind of trust relationship.)

As messages move down the directed graph of queues, credits move up the graph. Let's look at a typical scenario:

• We have a queue with two selectors that each move messages to respective external destinations.
• Each selector starts with one credit.
• Two messages arrive, and are passed respectively to each selector.
• Each selector moves the message it receives to its destination.
• Each selector now has a credit of zero.
• A new message arrives in the queue, and waits there, since there are no available selectors.
• One of the destinations sends an acknowledgment, which is passed to its selector as a credit.
• That selector is now able to process one more message.

To some extent credits and acknowledgements are the same concept, and we might use "credit" everywhere, especially since it's easier to spell. Thus applications send messages to queues, and credits to selectors.

Providing a credit to a selector is in effect paying for a message, and credits are like an internal currency. It makes sense that we aim to balance the books. When a queue is itself fed by one or more MOVE selectors, and it receives credits from a selector, it must pass these credits to its parent MOVE selector(s) in turn, i.e. pay for the messages that it received.

Selectors can, at any point, have outstanding messages. That is, they may have one or more messages they provided to a destination but did not yet receive payment for. If the selector is then deleted, it must return its outstanding messages to its source queue. These messages can then be routed to other selectors, or bounced.

Queue size limits

It makes sense in several scenarios to set a limit on the size of a queue. In fact there are two possible limits:

• On the number of unprocessed messages;
• On the history of processed messages.

When the history gets too large, there is only one sensible strategy, namely to discard the oldest messages. We can hardcode this, there is no need to allow for alternative strategies.

When the unprocessed size gets too large, the sensible strategy is to reject the new message as an exception. Discarding the message is an acceptable default (presumably the sender will realize, eventually, and retry or complain).

Message exceptions

There are several situations where a queue may find itself with a troublesome message that it has to deal with:

• When a queue has reached its limit of unprocessed messages;
• When a queue does not have any active selector that can accept a message.

Dropping the message is a good default action but in cases where it matters, we want to send the message to someone who knows how to handle it. That is often, but not always, the original sender. In AMQP we discussed this question extensively and the best design we came up with (which is not in the current protocol draft, which has taken a left turn to the Moon) was to route the message on a "return address".

Pushing the message forwards on an alternative route rather than returning it to the original source is as far as I can tell more robust and useful. It does not require us to track the message's routing path. It does not require that the source still exists. It lets us centralize exception handling for critical queues. And it lets us handle exceptions with the same mechanisms as deliveries.

When exactly is a message "unroutable"? It depends, but these are the cases I know of:

1. When a message arrives in a queue that has no selectors. E.g. if I send a document to a print queue that has no printers attached to it.
2. When a message arrives in a queue that has no matching selectors. E.g. if I send a document that requires color printing to a queue that has only black and white printers.

In the case when a queue has selectors but they are all busy (i.e. don't have the credit to accept a new message), we can't tell if the message would be routed or not and it does not make much sense to treat it as an exception, yet.

The very simplest way to handle exceptions is to move the affected messages to a 'dead-letter' queue, that is defined on a per-queue basis. On the dead-letter queue we can put selectors that route exceptions however we like. To make this work properly, we might want to mangle the message envelope so that the return address, if specified, is used as the address, and the original address is saved somewhere. This is a familiar pattern from the real world.

So here is a sketch for exception handling on queues:

• By default, if there are no (ready) selectors, hold the message as unprocessed.
• By default, if the unprocessed size exceeds a per-queue limit, discard new messages (and log some error).
• By default, if there are selectors but none match, discard the message.
• Add a BOUNCE operation that is like MOVE, but mangles the address and return addresses.
• Allow a "default selector" that is invoked if no other selectors match.
• Allow an "overflow selector" that is invoked if the unprocessed size hits a limit.

When applications try to publish messages to non-existent queues, we can bounce those messages to some per-engine dead-letter queue. This, when used with self-destructive queues, lets us implement a service lookup test.

Self-destructive queues

A pattern that has been very useful in AMQP and elsewhere is the "self-destructive queue". This is a queue that deletes itself when its last selector is deleted. Self-destructive queues are typically used for Wolfpack scenarios.

by pieterh

 <method name = "fetch" return = "self"> <doc> This public method fetches a queue from the store. It acts as a constructor and returns a new queue object when successful. The caller must unlink this queue object when finished using it. </doc> <argument name = "store" type = "xump_store_t *">Enclosing store</argument> <argument name = "name" type = "char *">Queue name, if any</argument> <declare name = "self" type = "$(selftype) *" /> // xump_store_request_queue_fetch (store, &self, name); </method> 

It's basically just a wrapper around the portal method. Now, the problem with these basic APIs is that they don't add enough value. For example, when we fetch a queue, it would be nicer to not have to know the store in advance. Queue names need to be globally unique, so that selector destinations make sense. This uniqueness should be managed by the API, so the caller code cannot get it wrong. Thus, the API has to sit above all the stores.

We already implemented a simple queue name cache that lets us work with queues as global objects within an engine instance, and I'll use this as the basis for a new engine-level API. The general model will be:

• The create and fetch methods will be implemented at the engine level, in the xump class. This makes it possible to work at the engine global level. We'll put delete at the engine level as well, for consistency.
• Access to object properties will be via object methods.
• Other methods, such as adding credit to a selector, will be done at the object level.

Starting with stores, we want to be able to create stores by name. This should be idempotent, so that they do not depend on a precise agreed engine state. E.g. creating a database-backed store should work if the store already exists.

Next, queues. These are global within an engine instance. That is, the caller can work with queues without having to specify what store they are in. The engine must manage this. It's only when we create a queue that we need to specify a store, and it makes sense to allow a default here. Another option would be to use a naming convention (like a path prefix) but that means it is impossible to reconfigure a queue to use a different store, something that might be useful. A store name as prefix remains an option, which we may come back to later.

Then, selectors. As for queues, we already have an API that works but we'll move the methods into the xump engine class as explained. The previous implementation of the selector class referred to the source queue as an object, but it would be cleaner to refer to a queue name.

The natural way to get messages out of the network seems to be to define a selector that has no destination queue, or rather, a selector that has an external destination rather than an internal queue destination. The caller will know whether a selector is an exit node or not, when it creates it. We will call this the "terminal" property of the selector. That is, a selector is a terminal when it delivers messages to an external destination rather than an internal queue. When a selector is not a terminal, its destination name must refer to an existing queue.

Lastly, messages. It seems clear there are two sets of semantics for messages:

• Message flow through queues and selectors.
• Browsing historic data in queues.

For the first case, we can rely on the fact that the engine is a closed system. That is, all output is driven by input. If there are no inputs to the system, there will be no outputs either. In a synchronous model, the calling application thus gives an incoming message or acknowledgment to the engine, and in return gets a set (zero or more) deliverable messages, which it can process in a loop.

In an asynchronous model, the processing of messages can take an arbitrary time and the caller would not want to wait. So, we'd want to work with callbacks (e.g. asynchronous portal reply methods).

For now, we'll design both the engine and the calling application as synchronous and single-threaded, so the API will provide methods to accept incoming messages and acknowledgments, and will provide a list of outgoing messages that the caller can process. Each "publish" pushes messages through the network until all routing is complete, and zero or more "deliveries" are push to that outgoing queue.

For the case of historic data, there are two use cases I can see. One is to scroll back and forwards through the history of a queue, a page of messages at a time: "get me the 50 previous messages". It matches an end-user's browsing of a queue. The semantics are classic: oldest, latest, previous, next. The second use case is to catch up with historic data since a particular time: "give me everything since 9.00am". In both cases, it only makes sense insofar as the queue in question has historic data, which depends on how it was configured and what selectors have done to it.

In earlier sketches of Xump I used the notion of "cursors" but I'm discarding this concept, it appears to be both unnecessary and unhelpful. It seems easier to create a new selector when we want to read historic data on a queue.

The simplest form of a queue is an ordered list of messages stretching from tail (newest mesage) to head (oldest message). New published messages are attached before the tail. Messages are taken off the list from the head:

[[code]]
Tail Head
| |
[M][M][M][M]
<Unprocessed —>
[[code]]

Actual queue implementations depend on the storage layer. To implement COPY, we need to divide the queue into two sections:

[[code]]
Tail Current Head
| | |
[M][M][M][M][M][M][M][M]
<Unprocessed > < Historical —>
[[code]]

When we create a selector we can thus specify where the selector starts:

• At the current position, which is the normal and default choice;
• At the head, i.e. the oldest message
• At any position between the current and the head.

by pieterh

There has been discussion in the AMQP workgroup of exchanges, and their role. In Xump's design, there is an exchange but it is universal and invisible, and is in fact the Xump instance. While AMQP extends routing functionality by defining new exchange types, Xump extends routing functionality by defining new extensions to the single routing layer. We'll do that using a portal API as we did for the storage layer, but this will come later.

Xump uses a inverted database analogy for routing. The caller defines selectors on queues, and as data flows through a queue, the routing layer applies the selectors on that queue. This is conceptually simple but harder to make than to explain. In AMQP land, some routing entities (exchanges) have no storage, while others (shared queues) do. I don't like the distinction. It should depend on the selectors and the network they feed: if the queue can empty itself immediately, it does so. And if not, it builds up and empties when it can.

So, we are going to design selectors. I'd also thought of calling these triggers, which is a neat abuse of database jargon, but I think it would be inaccurate. Yes, they are triggered by message flow, but they perform a selection action.

Selectors specify matching and/or filtering relationships between queues. Matching and filtering are two different and complementary ways of deciding whether a message M should go to queue Q or not:

• Matching is complex: every possible address A has a list of selectors that specified it. We thus match a message's address with its correspondong set of selectors, giving a "yes" decision for each of them.
• Filtering is simple: a selector S inspects a message M and makes a yes/no decision about whether or not it wants that message.

The tradeoffs are:

• Matching is a fast O(logn) operation but requires that the set of possible addresses is finite and changes slowly.
• Filtering works on any aspect of the message, or external conditions, but is a slow O(n2) operation.

Typically, filtering is good for content-based routing, while matching is ideal for topic distribution where the topic tree is finite and slow to change (like stock indices, newsgroup trees, etc.) The ideal combination is to match first, and then filter on the much smaller set of matched messages.

When a new message arrives in a queue, we match it against all selectors, giving us an interim hit set. We then filter against each selector in the hit set, giving us a final hit set. Each selector specifies either COPY or MOVE. We do all the COPYs first, and then we do the first MOVE. When we do a MOVE, we push that selector to the end of the list of selectors, so that messages will be round-robined between all MOVE selectors, if there are several.

In order to move a message, a selector needs sufficient credit. It gets credit from the destination queue, which tells it, "ready to accept N more messages". There are two common cases: infinite credit, where messages are delivered without any response from the destination, and single-step, where messages are moved one at a time, with confirmation for each message from the destination.

So, if a queue has selectors but none have sufficient credit, messages will accumulate. When any of the selectors receive credit, they will start to move messages again.

We could use credit COPY as well but this has some unpleasant side effects. First, it causes source queues to fill up when destinations are slow in processing messages. Second, it forces us to treat selectors as independent cursors on the queue, which makes O(logn) routing impossible or at least very difficult. And it does not appear to be necessary for the use-cases we see: COPY is typically used for high-volume parrot (pub-sub) patterns where messages are never acknowledged, and where messages are pushed towards the destination as rapidly as possible, and dropped when that cannot be done, rather than backing-up source queues.

A selector specifies a source queue and a destination (presumably a queue). I'm not yet sure whether the destination is required, i.e. whether selectors are the terminators in the directed network, or whether queues are.

For our initial end-to-end proof, we can ignore filtering. The only natural identification for a selector is the selector itself, i.e. the concatenation of its main properties. This is not easy to work with. So, we'll number selectors per source queue and provide a way for applications to fetch selectors by index, exactly as we do for messages.

A selector therefore defines some or all of:

• A numeric identifier.
• A source queue.
• A destination name.
• A match type, which defines the selector algorithm. I'll list a few below.
• A match argument, which is passed to the algorithm. Its meaning depends on the match type.
• An operation, which is one of MOVE or COPY.
• A credit, which is an integer number of messages. For "infinite" we'll use -1.

Selectors exist in the storage layer, attached to their source queue. It seems to make sense to hold these in the same storage layer instance as their source queue, since their lifespans are linked. If we delete a queue, we delete all selectors attached to it.

We need some mechanism to determine whether a name refers to queue, or not, and what storage layer instance that queue is in. The simplest is probably a queue name cache in the Xump engine.

For now, we'll define only two match types, "EQ", which does a literal string comparison between the message address and the argument, and the empty match type, which means "all messages". These correspond to the AMQP "direct" and "fanout" exchange types.

From the above, we need to make:

• A queue name cache that holds the set of all known queues and that can interrogate storage layers to know whether a queue is defined or not.
• A xump_selector class that lets us work with selector objects held in the storage layer. This would work similarly to xump_message.
• New methods in the store API to handle selectors (create, fetch, delete).
• Implementation of these methods in the RAM store.

This still won't let us do message routing, but we're getting closer.

by pieterh

I'm removing the message_update method, it does not make sense. We define a message fully when we create it, and all state needed to do routing should be held in the routing layer, not the storage layer.

The message_delete method needs to be idempotent, thus safe to call no matter what the state of the store. Thus, before we delete the ram message we need to check that the queue exists. We should be robust against stupid things like fetching a message, deleting the queue, then trying to delete the message. That means we do not need to worry about the order of a teardown. The cost is an extra lookup, eventually this can be optimised away.

Message properties (fields): we do not need to manipulate these as objects; they are created and used but never (or rarely) modified for existing messages. Thus, the fastest and simplest data structure will be an env-type block (name=value<nul>…)

To implement message properties, I'm going to call these "headers", to match the SMTP/HTTP semantics. Headers are name/value pairs, where both name and value are C strings (i.e. they cannot contain null characters). We need to be able to store the headers for a message when we create a new message, and we need to be able to access (but rarely modify) the headers for an existing message.

To implement this, we'll make a new xump_headers class that handles header fields. A xump_message will contain a xump_headers object (or at least opportunistically, since there is no point creating xump_header objects we don't need). Here is the test code for xump_headers:

 xump_headers_t *headers; headers = xump_headers_new (); xump_headers_set (headers, "a", "one"); xump_headers_set (headers, "b", "two"); xump_headers_set (headers, "c", "three"); xump_headers_set (headers, "a", "value"); assert (streq (xump_headers_get (headers, "a"), "value")); assert (streq (xump_headers_get (headers, "z"), "")); xump_headers_destroy (&headers); 

I've used reference counting for the xump_headers class, so that for storage layers that keep the headers in memory, there is no copying. Fetching a message just fetches a reference to the headers. When the last reference is deleted, the headers are also destroyed.

We could eventually use the same mechanism to avoid copying message data, by holding it in buckets, which are also reference counted. The break-even point seems to be at about 2,300 bytes, where copying costs more than allocating a bucket using the fastest 'direct' allocator (the ALLOCATOR=direct runtime setting for Base2, which bypasses the memory debugging framework). So in the worst case scenario, we can use buckets for messages of 2.3K and above. But if we assume each message is read at least once, then the break-even point is lower. Using buckets will also eliminate duplicate message content for messages that are on more than one queue (at least, for RAM-based queues).

We'll come back to buckets another time, using them to reduce copying work for messages of 1K and above in size. The current API is pretty fast already. Creating and deleting 1M messages takes about 1 second if I switch to the direct allocator. When compiled in release mode, I expect this to go twice as fast.

2009-06-06

We identify messages using a numeric ID that is unique within a queue. It's the storage layer's responsibility to provide new IDs. We'll use an unsigned 32-bit ID and ignore rollover. If a single queue has more than 2^32 messages, the older ones won't be accessible. Fair enough.

Thinking: xump_message and xump_queue objects should have a void * context block which is managed by the storage layer. This requires that the constructors (at least) and destructors are called from within the storage layer. We can constrain destruction of the void * block as icl_mem_free.

2009-06-05

The API is breaking one of the rules of software design, it's confusing. I find myself looking at the code to figure out where I put the message creation code. This ain't good. So, I'm going to change this again, so that we create messages like we do queues, with a 'message_create' method. This is then a 'create' method on the xump_message class, which invokes the storage portal method 'message_create'.

So what we have is:

• xump_queue.icl: create, fetch, delete (no update possible on queues).
• xump_message.icl: create, fetch, update, delete.
• xump_store.icl: queue_create, queue_fetch, queue_delete, message_create, message_fetch, message_update, message_delete.

Sorry for the detour. Sometimes we need to try stuff in order to realize that it's silly.

2009-06-04

The storage layer must be entirely ignorant of how messages flow into queues, and out again. This is handled by the engine. What the storage layer does is store messages, provide access to them, and delete them. In our current design, the head of the queue is stable insofar as the only party removing messages is the engine itself.

Access to messages is thus synchronous, i.e. the engine polls the storage layer when it needs messages. This means that we can browse queues using an index that iterates from the head of the queue, i.e. 0 = oldest message, 1 = second oldest, etc. We need methods like this, as we previously sketched:

• message_fetch (message, index) - fetch message object by index
• message_update (message) - update message from object
• message_delete (message) - delete message

Perhaps the best identifier for a message is a queue-unique id, rather than the index. Thus the engine can manipulate messages independently. In storage layers where the real message is large (e.g. a disk file), the message object acts as a reference. In storage layers where the real message is small, the message object would hold the denormalized body data.

by pieterh

Having prototyped the queue API, it looks clumsy. We'll make these improvements:

• Make the create and fetch methods work like constructors, returning a reference to a newly created object, if successful.
• Use link counting on the xump_queue object, so we can use unlink instead of destroy. This is nice because if we don't use new, we should not use destroy either.
• Scrap the update method, since this does nothing useful. We can add it back if needed but for now it looks like queue properties are either established at creation time, or read-only.

So while the portal request API still uses the RESTful model, the xump_queue class layers a more fluid object-oriented API on top of that. This is what the old calling code looked like (from the xump.icl selftest method):

 xump_queue_t *queue; queue = xump_queue_new (store, "Test queue"); xump_queue_create (queue); xump_queue_fetch (queue); xump_queue_update (queue); xump_queue_delete (queue); xump_queue_destroy (&queue); 

It's clear, but too different from the normal way of working with iCL objects. Having new and create, and delete and destroy is confusing. iCL lets me mark the new and destroy methods as private, so these are not exported to callers. The revised API looks more elegant:

 xump_queue_t *queue = NULL; // Create acts as a constructor queue = xump_queue_create (store, "Test queue"); // Unlink is an soft destructor xump_queue_unlink (&queue); // Fetch also acts as a constructor queue = xump_queue_fetch (store, "Test queue"); // Delete acts as a hard destructor xump_queue_delete (&queue); 

Commit cee4083: Implemented queue_post method

I'm not sure that the message API I sketched yesterday is the right one. In any case, for sticking new messages on queues, it's not right. The model that feels right is:

• queue_post (queue, message) - post message to queue

Which fits the RESTful pattern and makes this a method of the storage layer's queue class, which is accurate. The queue is the class that knows how to organize its messages.

Here is the test case for posting messages to a queue:

 // Create a queue and post messages to it queue = xump_queue_create (store, NULL); message = xump_message_post (queue, "address1", "abc", 4); xump_message_unlink (&message); message = xump_message_post (queue, "address2", "def", 4); xump_message_unlink (&message); xump_queue_unlink (&queue); 

The store_ram implementation is pretty trivial, it'll be more fun making a file system storage layer. Several ways of doing this:

• One file per queue, with one message per line. For text messages, especially if messages are never deleted.
• One directory per queue, with one message per file. For large messages, especially for file-transfer type work. Meta data can be in separate files.
• One file per store, with queues and messages indexed into the file. Easy management of stores, and very fast if the indexes are properly designed.

Using a database like SQLite would also be fun: one table per queue, one row per message. This would probably be 10x slower than using a custom indexed file.

by pieterh

Let's explore the queue-message-field API further. We define this API as portal requests in the xump_store.icl class. Roughly:

• queue_create (&queue, name) - create queue in store
• queue_fetch (&queue, name) - fetch queue object
• queue_update (queue) - update queue from object
• queue_delete (queue) - delete queue
• message_create (&message, queue, address, bucket) - create message in queue
• message_fetch (&message, index) - fetch message object by index
• message_update (message) - update message from object
• message_delete (message) - delete message
• field_create (&field, message, name, value) - create field in message
• field_fetch (&field, message, name) - fetch field object
• field_update (field) - update field from object
• field_delete (field) - delete field

This mirrors a RESTful model, in which a remote resource (in this case, on the other side of the API call) is managed by doing create/retrieve/update/delete on a representation of that resource (in this case, an object). In a classic OO model, the resource and the object would be the same thing. Here, they are not.

The reasons for this not-quite-OO design:

• It lets us define resources that have different life cycles than objects, which we need to define persistent resources (that remain alive when the application stops).
• It lets us do concurrent access to shared resources in a safe fashion.
• It lets us hide the implementation of the resources from the calling application.

The word "retrieve" is annoying to type, so I'll use "fetch" instead. The more often we use a word, the shorter and simpler it should be.

Commit 7e30756: Constructing the portal queue API.

Here.

To make it simple, the engine has a single store instance which is just the 'store' property. This is not a full design. We'd want to make at least one store instance per backend, but in fact allow the caller to create an arbitrary number of store instances, referenced by name. We'll add that later.

How this works now - this is from the xump.icl selftest method:

  // Create an engine instance xump_t *xump; xump = xump_new (); // Create a queue object within single store xump_queue_t *queue; queue = xump_queue_new (xump_store (xump), "Test queue"); // Check that every methods fails properly assert (xump_queue_create (queue) == -1); assert (xump_queue_fetch (queue) == -1); assert (xump_queue_update (queue) == -1); assert (xump_queue_delete (queue) == -1); // Destroy queue object and engine instance xump_queue_destroy (&queue); xump_destroy (&xump); 

Running the create method a billion times takes around 7.5 seconds on a 3GHz core. So, about 130M requests/second, 7.5 nanoseconds per request. No bottleneck here, yet, synchronous portals are fast.

Commit 5f06b01: Engine now supports multiple store registration/lookup

Here.

Added two methods to the engine, which let the caller register and lookup stores. It's more sensible that the caller creates the store portal, since it knows what extensions it is using. The API for creating and registering a new store is short and sweet:

 xump_register_store (xump, xump_store_ram__xump_store_new (NULL, storenamegoeshere)); 

I called the methods 'register_store' and 'lookup_store' because the object_verb form I'd usually prefer ('store_register' and 'store_lookup') suggest that these methods are in the store class, which they are not.

Commit 9f9a1c2: Improved engine store method names

Here.

Actually, on second thoughts, it's even nicer to make the store list look like a property of engine, and use the same style as we use for opaque object properties, namely "set_property" and "property". This gives us a readable:

 xump_set_store (xump, xump_store_ram__xump_store_new (NULL, "RAM1")); xump_set_store (xump, xump_store_ram__xump_store_new (NULL, "RAM2")); assert (xump_store (xump, "RAM1") != NULL); assert (xump_store (xump, "RAM0") == NULL); 

by pieterh

 Engine +-- Store +-- Queue +-- Message +-- Field 

Starting on the basic framework for the store portal and plugins. We have these classes:

• xump.icl - engine class
• xump_store.icl - store portal
• xump_store_ram.icl - RAM based store stub that says hello

All single threaded, so no SMT, no multicore. That makes the classes simpler, but for any kind of performance, or when using disk-based persistence, we'll need multithreading. To be designed later.

What the stub RAM store looks like (trimming comments):

 <?xml?> <class name = "xump_store_ram" comment = "Xump RAM store back-end" script = "icl_gen" license = "gpl" opaque = "1" > <inherit class = "xump_store_back" /> <method name = "announce"> icl_console_print ("I: initializing RAM-based storage layer"); </method> </class> 

And this works:

 $ boom build boom I: [Xump]: Generating files... boom I: [Xump]: Building Xump... $ boom test boom I: [Xump]: Running regression tests... 

This is the first running code. Not much, but it registers and calls a plugin. Committed as change 014a172.

by pieterh

 +--------------------------------+ | Application layer | +--------------------------------+ | Routing layer | +--------------------------------+ | Storage layer | +----------+----------+----------+ | Back-end | Back-end | Back-end | +----------+----------+----------+ 

Making it simpler… what can we ignore for now? Our use case for Xump (which is a social messaging layer above Wikidot) calls for very high volumes of quite small text messages (80-160 characters). We would expect thousands of queues, each containing thousands of messages. We'd want these to be persistent and archived, with messages being held on their queues for as long as the queues existed.

We can ignore messages with multipart contents. This is an advanced use case and arguably it is not a good one, since it creates messages that are complex for applications to work with. If my application produces multipart messages, that makes it hard for others to talk to me. So a framework that allows this may actually be helping me to shoot myself in the foot.

For high volumes of small messages, copying is not a major concern. We avoid copying by manipulating references, but that is not free. So, we will for now copy data. Later, and for larger messages, we'll see about normalizing data so that copying is reduced when the same message is sent to many consumers.

Every single message has an address, which is used heavily, so we will make this a dedicated property of the class. A message also has content body, content length, and a table of name/value fields.

Let's build the storage system around queues. That is, a store holds a number of queues, which hold a number of messages. We can interrogate the store for queues and messages, and it will return us memory objects that we can work with.

The iPR portal system provides a virtual class interface. Our portal class will be xump_store.icl. This would have implementations (called "portal back-ends") like xump_store_fs.icl (file system), xump_store_ram.icl, xump_store_sql.icl, and so on. Each portal back-end instance is equal to a store instance, and implements the portal methods. Which we can break into:

• Creating and working with queues in a particular store.
• Creating and working with messages in a particular queue (in its store).

The queues and messages that a store manages are opaque, and invisible as such to the storage layer. To work with a queue, the caller asks the back-end to provide a queue object. It can execute methods on this object (some of which may affect the stored queue). It then destroys the queue object. Messages are handled in the same way.

We will use the create-retrieve-update-delete pattern for working with queues and messages. To some extent this mirrors the RESTful pattern for working with remote resources. In this case, the remote resources sit in a store that is opaque to the caller, which accesses it via the portal interface.

To repeat in a different way: when the caller does xump_queue_destroy(), this destroys a queue object but does not destroy the queue held in the store.

The queue and message objects that the store provides to the calling layer are private, not locked, and not shared (they do not support reference counting).

The simplest workable identifiers for queues are names. It makes sense that a store can auto-name private queues. For messages, we can use sequence numbers.

The xump_store.icl portal class needs these methods for working with queues:

• Create (named or unnamed) queue in store, with given properties.
• Retrieve queue from store by name.
• Update queue with given properties.
• Delete queue from store.

And for working with messages:

• Create new message in queue, with given content and address.
• Retrieve message from queue (there are various patterns here).
• Update a message with given properties.
• Delete message from queue.

And finally, to manipulate fields on messages:

• Create new field on message, with given name and value.
• Retrieve field from message, by name.
• Update field value.
• Delete field from message.

We then need classes to hold queue, message and field references:

• xump_queue.icl, which refers to a queue in a store.
• xump_message.icl, which refers to a message in a queue.
• xump_field.icl, which refers to a field in a message.

So our resource hierarchy looks like this:

 Store +-- Queue +-- Message +-- Field 

by pieterh

Decision: all public classes will be opaque and when we need to make lists or tables of them, we'll use the ipr_looseref_list and ipr_hash classes.

So, API classes have this general interface:

objref = myclass_new (arguments);
objcopy = myclass_link (objref);
myclass_some_method (objref);
myclass_unlink (&objcopy);

• All classes are reference counted.
• All classes are synchronous.
• All classes are threadsafe.
• All classes are opaque.

by pieterh

The design of Xump is a mix of ideas collected over the last few years. The engine manages a set of FIFO queues. Messages flow through one or more queues in a directed network of any complexity. Everything is a queue so the whole network is asynchronous. The calling application (that is, the product into which we embed the Xump engine) constructs this network dynamically at runtime, though the network may also be persistent. The lifecycle of all resources is visible to, and managed by, the calling application.

Queues pull messages from other queues via selectors, which specify static matching and filtering operations. Selectors are like stored search criteria that act on the head of the queue, eating up or otherwise operating on messages. A selector reads from one queue, and writes to another queue or set of queues. Matching is a O(logn) set operation, in which Xump compares a message address to the set of selectors that read on a queue. Filtering is a O(n2) operation in which Xump compares one message to one selector. In theory we match first, and filter second.

Selector match and filter operations are: Move, Copy, and Delete. These work as you would expect. Here is how two applications divide the flow of messages on a queue (the Wolfpack pattern for workload distribution):

 Queue1 -> Selector MOVE -> Queue2 -> Selector MOVE -> Queue3 

Here is how two applications each get the full flow of messages (the Parrot pattern for information distribution):

 Queue1 -> Selector COPY -> Queue2 -> Selector COPY -> Queue3 

A mix of selector operations is valid. Here is how two applications divide the work on a queue, while a third gets a copy of every message:

 Queue -> Selector MOVE -> Recipient -> Selector MOVE -> Recipient -> Selector COPY -> Recipient 

The implementation of selectors is extensible, via a virtual class interface, and the intention is that message processing applications that use Xump would provide their own implementations for the xump_select class.

To read messages off a queue, applications (that is, messaging servers which use Xump as their embedded engine) open and close "cursors" on queues. Xump cursors are analogous to the SQL cursor concept. The main difference is that when a cursor is open, messages are sent asynchronously to the application.

Applications use Xump by:

• Constructing a directed network of queues and selectors;
• Setting up cursors.
• Publishing messages to queues.
• Delivering messages received from cursors.

The persistence layer is extensible, via a virtual class interface, with the intention being that applications which embed Xump would provide their own storage layer implementations. Persistence happens at the queue level, so it is possible to e.g. provide persistent subscriptions for some clients, and transient ones for others.

Delivery semantics (e.g. message acknowledgement) affect selectors insofar as selectors may want acknowledgment for their operations, and effectively block until they get it. This would cause the queue to build up, typical for Wolfpack. Without this, typical for Parrot, queues will process messages at full speed and mainly remain empty (unless sheer processing costs caused backlogs).

by pieterh