SAFplus 7 Feature Discussion

SAFplus Messaging

Advanced Socket Layer

The socket abstraction (MsgSocket class) presents a simple read/write API for multiple scatter gather messages, consisting of receive, send, and flush APIs. At the message transport layer, an instance of the socket object directly communicates with the Linux level socket via system calls. However additional socket objects can be created that implement additional functionality, such as traffic shaping, message segmentation and reassembly and reliable messaging. These message sockets can modify the messages sent and received and then call a lower level MsgSocket instance.

Functionality can therefore be selected at socket construction time by instantiation of a stack or tree of MsgSocket-derived classes, and a single implementation of an abstract concept (such as traffic shaping) can be applied to any message transport type.

After socket construction, application code is unaware that it is accessing layered sockets rather than a direct socket implementation, making it easy to change add or remove functionality as application requirements change.

MsgSocketLayering.svg MsgSocketLayering.svg

The diagram above shows 3 applications with different socket stacks. Application 1 is using UDP transport, traffic shaping and allows large messages (segmentation and reassembly). Application 2 does not need large messages; it is just using traffic shaping. Application 3 shows a complex configuration. Its node supports both UDP and TIPC transports, so the first layer above the transport ("splitter") uses the destination nodeId to determine which transport to route the message to. Large messages are supported so the "Segmentation And Reassembly" object is next. After that, the application may need both reliable and unreliable messages, so a "Joiner" object is added that is capable of providing 2 independent MsgSocket interfaces and combining them into one (by prefixing a "message type" character to the front of every message, for example). Finally, a "reliable message" MsgSocket object provides reliable messaging. The application code sees 2 MsgSocket interfaces; one for unreliable and one for reliable messages.

Note that is is expected that the receiving application will have the same socket stack, in cases where the MsgSocket objects modify message content.

SCTP transport plugin

SCTP stands for Stream Control Transmission Protocol. SCTP is the extension of TCP, therefore, it has some typical features:

Because of the data transport in stream, so there are 2 separate methods to open socket at server and client:

However, to implement the peer-to-peer model using SCTP, the model has to be refined:

* Node IP addresses

Network Address Mapping

There are two techniques to convert node IDs into network addresses. "Direct mapping" and "cloud mode".

SAFPLUS_BACKPLANE_INTERFACE=eth0:1

and network and node ID are optionally defined: 

SAFPLUS_BACKPLANE_NETWORK=169.254.26/9 
SAFPLUS_NODE_ID=10

If SAFPLUS_BACKPLANE_NETWORK and SAFPLUS_NODE_ID are defined, the system derives and IP address from this pair (in this case, 169.254.26.10) and assigns it to the interface overriding an existing different IP address assignment.

If these variables are not defined, the interface $SAFPLUS_BACKPLANE_INTERFACE MUST have an IP address already assigned (otherwise throw exception). That IP address and netmask is read and SAFPLUS_BACKPLANE_NETWORK and SAFPLUS_NODE_ID extracted.

So given a node ID (in this example, 10) any node in the cluster can construct an IP address for the node (in this example, 169.254.26.10)

How does the node discover its own ID? It uses the SAFPLUS_NODE_ID environment variable:

In this mode, every node is configured via environment variable with the IP address of "well known" nodes in the cluster. Best practices are to make this at least 2: SAFPLUS_CLOUD_PEERS=54.23.54.13,mycluster.mycompany.com Note that domain names are allowed. The new node will contact one of these nodes to request:

  1. A node ID (the new node can supply a preferred ID and indicate whether it MUST have this ID -- i.e return error if ID is already taken, or whether it is merely a preference)
  2. the mapping between node IDs and addresses.

Upon each new registration, the cloud mode server will notify all other nodes of the existence of this node.

This is a generic mapping service so can be implemented once for all transports. However, the service should use the underlying transport (perhaps via a special port/ socket) so that part may be tricky :-).

SAFplus Management

Management Access APIs

We need a c++ API to issue management object set and gets from any application. This will allow the management data to be generically manipulated and so we will not need to support a lot of application specific APIs like log stream creation or Dynamic HA operations.

Applications can also use these APIs to implement a new northbound protocol (like SNMP and NETCONF) if they need to.

Proposal: 

mgtSet(const string& pathSpec, const string& value, Transaction& t);

string mgtGet(const string& pathSpec);

void mgtCreate(const string& pathSpec);
void mgtDelete(const string& pathSpec);

All APIs raise an exception if they cannot complete.

For example: 

mgtSet("/Log/stream["mystream"]/filename", "/var/log/myapplog")

Management CLI Directory

The current management CLI's "directory" structure from the root looks like: 

*--- root
  |--- Application
  |--- ServiceUnit  # this is ServiceUnit["su0"] but there's no way to determine this
  |--- ServiceUnit  # this is ServiceUnit["su1"]

The corresponding YANG (see SAFplusAMF.yang) is: 

module SAFplusAMF
  {
  list ServiceUnit
    {
    }
  }

*--- root
  |--- SAFplusAMF  # This is the 'prefix' field defined in the module or the name of the module if prefix is not provided.
    |--- Application
    |--- ServiceUnit  # A "list" in Yang creates a directory and the list's elements are items in the directory
      |--- su0
      |--- su1

Standard YANG Implementations

SAFplus Management will provide reference implementations for many standard YANG modules. 

xxxInitialize();
xxxFinalize();

where xxx is the name of the YANG module.

""Note, it may make sense to make these YANG modules dynamically loadable plugins. In that case they will follow the SAFplus plugin architecture""

YANG modules

  1. https://github.com/YangModels/yang/blob/master/standard/ietf/RFC/ietf-system.yang

SAFplus Checkpoint

Checkpoint deletion

When all processes close a checkpoint wait N (configurable) seconds and then delete it from memory. This is the checkpoint "retention time".

Q&A:

[Hung] Do you mean that any process (including safplus process and user process) that is using the checkpoint, in the case all of them close the checkpoint, then the data of the checkpoint stored in the shared memory must be deleted? Can you give an example for more specific if any?

[Andrew] Yes, Let's say processes A and B have a checkpoint open which is configured for a 5 minute retention time. A exits. The checkpoint stays "alive". B exits. Now no process has the checkpoint open. The system does NOT close the checkpoint. At 3 minutes after there were no users, process C opens the checkpoint. It opens the original checkpoint data because it was retained for that time. Now process C closes the checkpoint. Again no process has it open so the timer starts. After 5 minutes the data is deleted from shared memory.

For persistent checkpoints, we should have another field "persistentRetentionTime" that configures how long the data is retained on disk.

The purpose of these retention times is to clean up unused resources, but not so quickly that a failure and restart will cause the data to be deleted.

[Hung] Thanks for the clarification. This concept basically comes from Checkpoint 6: each checkpoint is provided with retentionDuration argument when it's opened. When the last call to checkpoint close is performed, a timer is started with retentionDuration and when the timer expires, data of this checkpoint will be deleted from memory.

To resolve this issue, we must have a checkpoint server which receives checkpoint open and checkpoint close notification (We may already have checkpoint server handler class named CkptServerMsgHandler which may receive and handle incoming msg - I'm not sure, need to check). It also starts or stops the timer as needed.

So, implementation for this feature should be:

  1. Add retentionDuration to the checkpoint constructor
  2. When a checkpoint is opened, init() function is called then sends notification "CKPT_OPEN"
  3. On receiving "CKPT_OPEN" notification, server will check if the timer has been started and it is not expired, then server stops the timer and increase the count of ckpt opening
  4. When a checkpoint is closed (the parent process exits), from the ckpt destructor, sends the notification "CKPT_CLOSE"
  5. On receiving "CKPT_CLOSE" notification, server will decrease the count of ckpt opening and check if this is the last ckpt close notification, then it starts the timer. If the timer expires and there is no "CKPT_OPEN" within the retentionDuration, the checkpoint data will be deleted from the memory.

[Andrew] This strategy will not work because a process can be killed. If it is killed it will not be able to send the CKPT_CLOSE notification. And the same problem exists with CKPT_OPEN: if SAFplus is killed and restarted we need to start a retention timer for a checkpoint that may not be opened. Instead do it this way:

  1. Changes to the checkpoint clients:
    1. When a checkpoint is opened, store the "retention time" in the checkpoint shared memory header (store it as milliseconds for consistency with other stored durations). If multiple processes open the same checkpoint use the maximum retention time (compare your value with the one in shared memory and overwrite if yours is larger).
    2. Also create a uint64 in the checkpoint header called "openCount". This will track the # of times the checkpoint has been opened.

    3. Also create a SAFplus::ProcSem whose ID is based on the checkpoint handle. "unlock" this ProcSem (adds 1). If the process dies for any reason, the ProcSem will be "locked" (subtracts 1)

  2. Create a CheckpointCleanup class (which will run in safplus_amf, or any other process -- one per node).

    1. Every 30 (configurable) seconds: For all checkpoints in /dev/shm (their name follows a particular format):
      1. Read the checkpoint handle from the checkpoint

      2. Create a SAFplus::ProcSem based on that handle

      3. If the value of the ProcSem is zero, start a timer (boost has timers, IIRC) based on the RetentionTime, and remember a local copy of openCount (localOpenCount)

      b. When the timer expires:

      1. if openCount == localOpenCount (that is, nobody opened the checkpoint during the RetentionTime duration) delete the checkpoint's shared memory

This approach also has the advantage that no messaging is needed.

SAFplus: Feature Discussion (last edited 2015-07-24 14:01:34 by AndrewStone)