Concurrency and Complexity

HP's Russ Daniels understands a critical issue facing the next generation of software development:

"If you think about whatever those skills are that are necessary to be able to program a computer, we don't understand what they are very well, but we know that they're not very widely distributed. So you end up with a relatively small portion of the population who can write a program successfully. Then you start introducing certain kinds of abstractions: if you introduce recursion, you lose some people; if you introduce pointers, you lose some people; if you introduce event-driven models, you lose some people; when you introduce concurrency, you lose virtually everybody.

So we create these programming languages that support concurrency and then you find that there's a lock and an unlock around every method, because otherwise people hurt themselves. It presents an interesting dilemma—we need to be able to get more stuff done, we need to figure out how to do stuff in parallel, but we're pretty bad at doing that."

One of the keys to solving this complexity problem is to make the mechanics of parallelism safe, automatic, and transparent to the programmer. Ultimately, this approach needs to extend far beyond just parallelism, as we see the same failings in the areas of security, fault tolerance, fault recovery, dynamic scaling, and resource economics.

An ADE Example

In order to see how this could work, let's take a quick look at an example of how ADE transparently provides concurrency and protection of resources that must be accessed serially:

Let's imagine that we're tasked with building a simple LAN switch, and we want to provide an IP address resolution for connected Ethernet hosts. If we want to send an IP packet to a given computer, we first need to find what the MAC address of the computer with a given IP address is. This is performed using the Address Resolution Protocol (ARP), which was originally documented in IETF RFC 826.

At a simplified level, the protocol works as follows:

1. A host which wants to talk with a specific IP address sends an ARP broadcast request asking for the MAC address that traffic should be addressed.
   
2. The switch intercepts the broadcast, and looks up the IP address in its local cache. If the IP to MAC address mapping is found, it responds to the originally requesting host with the ARP response.
3. If the mapping is not found in the local cache, the switch broadcasts the ARP request to all other ports.
4. If the switch receives a ARP response, it adds the IP to MAC mapping to its local cache, and forwards the response to the originally requesting host.

This is equivalent to a simple database insert/lookup problem, and in traditional concurrent programming models, errors can easily be introduced if care is not taken to lock the local cache while performing modifications.

In contrast, because ADE handles concurrency by forcing all component interactions to be atomic transactions, no locks are required and the system can be proven to be correct and deadlock free.

In this example, we are going to consider three actors: The ARP Lookup Process, the ARP Cache Process and the ARP Discovery Process. These actors have the following roles:

• The ARP Lookup Process receives ARP lookup requests from the network hosts, asks the ARP Cache Process if the mapping is known, asks the ARP Discovery Process for the mapping if not known, and responds to the requesting host.
• The ARP Cache Process looks up mappings in the local cache, and adds mappings when asked.
• The ARP Discovery Process sends ARP lookup requests to the network, and when a mapping is discovered, it asks the ARP Cache Process to remember them.

In this model, there would be one instance of the ARP Lookup Process per ARP lookup packet received on the network, one instance of the ARP Cache Process, and one instance ARP Discovery Process per cache miss.

In a busy network, there could be dozens of ARP requests being processed, and many concurrent requests that may require changes to be made to the ARP cache. Even if ARP discovery is adding a new mappings at the same time that a new request is looking up a mapping, there is no possibility that the cache can become corrupt, because all access to the cache is automatically serialized by virtue of the entities and their messages that are exchanged.

This is the key aspect of ADE that simplifies the creation of distributed systems. Because concurrency is automatic, when problems are decomposed into collections of interacting actors, any parallelism inherent in the problem will be automatically utilized.

XAM Active Objects

The SNIA XAM standard provides a comprehensive object model and interface for the storage of dynamic data objects. The standard provides methods by which collections of data (known as XSets) can be created and deleted, data can be stored, retrieved and manipulated inside these XSets, and queries can be performed to locate XSets that match specified charactertistics.

While the first version of the standard was under development, I worked with Jared Floyd of Permabit to ensure that the XAM Job model was architected in such a way that it would include all of the components required to support active objects, like those supported in ADE.

What are XAM Active Objects?

This is best illustrated by an example:

Let us create a hypothetical XSet that includes an XStream (binary data) that contains the bytecodes defining a Java program. When this XSet is committed, it can automatically start executing. The entity that executes these active objects can either be performed by a sidecar system that attaches to the XAM Storage System (XSS), and becomes aware of new active objects via the standard XAM query facilities, or the execution of these active objects can be an intrinsic part of the XSS.

The Java program would be executed in the security context of the XSET it is contained within, and thus it can access local data stored within it's own XSET, and optionally perform XAM operations based on its security credentials. This allows it to read other XSets, create new XSets and perform queries to discover XSets.

For example, an active XAM object could remain resident within the storage system, performing queries for specific types of XSETs (for example, PDF objects), and convert them into a newer version of the PDF format. Such an model would allow dynamic format conversion of archived content, just by loading in a new XSET. This model is also be useful for analysis, where a large number of XSETs need to be analysed or datamined in the background.

Because XAM Storage Systems will often be distributed, multiple active objects can easily be parallelized across the compute infrastructure the makes up the system, and parallel computing patterns are easy to implement, as one active object can create XSETs that act as child active objects (the equivalent of performing a fork in UNIX). As code can be bundled with data, this model will also enable the creation of data-driven MIMD and SIMD parallel data processing systems.

XSETs become process contexts, complete with inputs, code, state and outputs, with the full ability to discover and access data within the storage system, create new data, spawn child processes and report status information to external systems. Because XSETs belong to user contexts, adding processing quotas, limits on computing resources, and other policies can use the same methods as used for other XAM policies.

Part of the Standard?

All this could be implemented today as a vendor-specific extension to the standard. However, adding this capability to the XAM standard would require work to be done in the following areas:

• Active object language profiles would have to be defined, since multiple languages including Java, .net and interpreted languages such as ruby could all be supported.
• Language bindings would need to be standardized to allow these programs embedded in the active objects to be perform XAM operations on the XSS they are resident in.
• Standard job-style status reporting would be beneficial to allow standardized active object execution status monitoring.
• Policies for computing-related aspects, such as resource usage and quotas would need to be defined.
• Requirements for security isolation would need to be defined.

Given that none of these would require changes to the core XAM standard, this could easily be added as an optional part of the standard, much like Level 2 query, which provides full-text search within XStreams.

In summary, the XAM standard provides a foundation upon which a rich data-driven distributed computing system can easily be created. This opens up many intriguing possibilities, and would be relatively easy to formally add to the standard.

The ADE Process Model

To illustrate how ADE-based systems operate, it is best to use an example. In this entry, we will examine at a high level how a simple ADE program is run, from bootstrapping to shutdown.

ADE Versions

Over the last ten years, there have been three major revisions of ADE:

• ADE V1 was developed in 1998, and was a minimal system designed to facilitate experiments with message-based computing. In this version, ADE was hosted as a library that ran on Mac OS.
• ADE V2 was a result of continued improvements over the next two years, and included a pure message-based execution model and the ability to run directly on a processor without requiring a host operating system.
• ADE V3 was created specifically for the Bycast StorageGRID platform during 2000 through 2004, and has been tuned to provide high levels of performance while retaining most of the advantages of the ADE approach to distributed computing. In this version, ADE was updated to use an Operating System abstraction layer, allowing it to run on a wide variety of UNIX-like and embedded operating systems.

In order to demonstrate the pure message-based execution model, this example is based on ADE V2.

Bootstrapping

When ADE first starts, there are no processes present and no messages being sent. When in this state, there are no processes to send a message to, and there are no processes that could send a message. Thus, in order to transition the system into an executing state, we must "bootstrap" the system to the point where there is at least one process executing.

Given that ADE processes are just messages, when I refer to a message as a process, I'm referring to a message that contains executable code as part of the message contents. And since messages are just self-describing data structures (known as "containers"), a message can be serialized to disk. As messages are created, manipulated and exchanged in serialized form, one of the key design criteria of the container data format was highly efficient serialization.

We use this property to bootstrap the system. Once the ADE kernel is running, we load an (initially hand-created) message into memory. This message contains executable code for our bootstrap process, as well as the state for the process.

Figure 1: An executable message (Capability ID 100)

Loading this message into memory involves the following steps:

1. An entry is added to an in-memory database, keyed by the message capability. The message capability is a randomly assigned 128 bit identifier assigned by the system that uniquely identifies the message.
2. The kernel creates a protected memory space for the message. This memory space provides the execution context.
3. The serialized message is loaded into this memory space.

At this point, we have a process in memory, ready to do work, but no messages to trigger any work to be done. In order to kick off execution, we need to load a second message into the system. This message is far simpler than the first message, as it just triggers the executable code in the first message. This message consists of a message capability, a source CID (set to zero for bootstrapping), a destination CID, which is set to the capability identifier for the first message, and a message type, which is a string that identifies which executable code in the destination that should be invoked.

Figure 2: A message destined to CID 100

This is read as, "Message CID 200, from CID 0, to CID 100, invokes /create".

Message Processing

When this second message is loaded into memory, the presence of the destination capability identifier field triggers the message to be added to a scheduler queue for delivery to the specified destination capability. The ADE kernel sees this scheduler entry, and moves the triggering message into the memory space of the destination message, where it becomes part of the destination message. The scheduler then switches to the context of the destination message, and executes the code indicated by the triggering message. As this executable code has access to the message contents, it can now inspect the contents of the triggering message, and manipulate its own message state.

Figure 3: Message CID 100, after receiving message CID 200.

In our example, message CID 100 then executes the "create" method, and the executable code in the create method creates and sends a new message to itself (which is the only possible destination, and the only destination known to the process), invoking the "destroy" method. The final step in the "create" method is to clean up the contents of triggering message from the process message state.

Figure 4: Message CID 100, with message CID 300 about to be sent.

This newly sent message in turn gets enqueued by the scheduler, then processed, causing the "destroy" method to be invoked by the destination process. This executable code for the "destroy" method removes the entire state of the message, which then results in the message itself being removed (and the entry being removed from the in-memory database), returning us to our original pre-bootstrap state.

In Summary

This process model, of sending messages to other messages, executing code in destination messages, manipulating the state of the destination messages, and creating new messages, allows full computational capabilities, and provides a rich foundation for the creation of many advanced distributed computing models.

In our next ADE post, we'll examine some of the design patterns and capabilities that emerge as a result of this process model.

SAS for Blades

Blade-based computing form factors offer many potential advantages for constructing storage subsystems using standard hardware, but until recently, the lack of cost-effective storage connectivity has hampered blades when compared to 1 and 2U servers with directly connected SCSI or SATA/SAS shelves.

As I mentioned in an earlier post about IBM's BladeCenter S storage attachment capabilities, this is starting to change as low-cost SAS storage connectivity switching is integrated into blade solutions.

And now I am pleased to see that HP has introduced a new product, the HP StorageWorks 3Gb SAS BL Switch. This product provides eight external facing SAS 3GB ports, and if you install two of these switches in a c3000 chassis, you can attach up to 16 external SAS/SATA shelves. Assuming 12 TB of SATA storage per shelf, that allows you to have 192 TB of raw addressable storage per c3000.

This is perfect for high density storage systems, as you end up with 6U for the c3000, 32U for the storage shelves, and 4U remaining for a KVM and networking equipment. As the c3000 series blades also allows you to install "Storage Blades", you can host databases locally to the blade. This makes an excellent platform for a scalable high-density StorageGRID deployment, as each site can start with two blades providing basic services, and as storage capacity is expanded, additional shelves and blades can be added as needed. And building out to 200TB per rack is a pretty respectable density. One could easily use this hardware to build a 1 PB storage system based around five racks running StorageGRID, connected together using 10 GB network uplinks.

When comparing this to the IBM offering, there are both advantages and disadvantages. In order to allow an HP blade to access the SAS switch, you have to install a Smart Array P700m controller card, which represents an additional cost. With the IBM offering, the RAID controller in part of the SAS switch module itself. But given that the HP switch allows twice the SAS attachment density, and that having the controller card as part of the blade provides a greater degree of failure isolation, I'm inclined to prefer the HP solution.

But regardless of the minor differences, the bottom line is that both systems are now far more viable for use as lower-cost storage infrastructure due to the elimination of the need for fibre-channel based connectivity to the storage shelves. This is a huge improvement in terms of costs, and one that we won't see again until FCOE (or SAS over Ethernet) becomes widely deployed.