What's it all about, Alfie?

Today I messed around with the boost::variant problem I'd been dealing with lately trying to get the code to compile and work properly. Finally, after about five hours, I gave up. It's simply too hard to get working, and even if I did, the maintenance costs of dealing with these kinds of compiler errors would be far too high for a junior developer.

So I took a very different track: I simply wrote my own. Honestly, it wasn't all that hard. I took out the definition of the boost::variant, and in it's place I put a simple union:

  private:
    tVariantType      mType;
    union {
      std::map<std::string, variant>    *mMapValue;
      std::list<variant>                *mListValue;
      std::string                       *mStringValue;
      int64_t                           mIntValue;
      double                            mDoubleValue;
      uuid_t                            *mUUIDValue;
      bool                              mBoolValue;
      error_t                           *mErrorValue;
    };

and then all the setters cleared out the old value and replaced it with the new. It's something I've written before, and so I knew a lot of the pitfalls to avoid. But it's not as nice as a stack-based template version. When I change values I'm hitting the heap for new space. While this isn't horrible for a lot of applications, it kills performance when you're trying to do something really fast.

Still... this is far easier to understand, and once we have it all buttoned-up, there's no real chance of a leak, and it's a solid way to handle the variant problem.

Once I got the main part written, I was able to attack the serialization and de-serialization schemes for this guy - based on the work of another group who has defined the scheme we'll be using. It's decently flexible, and should be really nice to use across the board.

Still lots of testing to do, but I'll get to that tomorrow.

Once more unto the boost... this time to try and get the boost::variant working. In my particular application, I've got a self-defined data stream that can include:

• Map - all keys are up to 256-character strings, values are variants
• List - all elements are variants
• Integer
• Double
• String
• Boolean
• NULL
• UUID
• Date
• Error - which is a boost::tuple of a UUID, an integer, and a variant

and with the recursion in the definition with the map and list, I wanted to try the boost make_recursive_variant capabilities.

I sure do wish boost had better docs, because even getting to the point that the code compiles was an all-day affair. Primarily due to two lines of code:

  void variant::set( std::map<std::string, variant> & aMap )
  {
    mValue = aMap;
  }

and:

  void variant::set( std::list<variant> & aList )
  {
    mValue = aList;
  }

In theory, and in practice, this should set the value of the ivar mValue, the boost::variant, to the map and list, respectively. But I got the most insane compiler errors I've ever seen. Oddly enough, when I do:

  void variant::set( std::string & aValue )
  {
    mValue = aValue;
  }

everything works just fine. So it's clearly something about the recursive definition in the code. Possibly in the const-ness of one of something, but I tried all possible permutations I could think of. Nothing worked.

Finally, I tried this:

  void variant::set( std::map<std::string, variant> & aMap )
  {
    mValue.get< std::map<std::string, variant> > = aMap;
  }

and it compiled, but when I called this code and the value in the variant was not already a map, this threw a "bad cast" boost exception. Very understandable... I'm saying give me the existing map, and then set this guy there, but there's no existing map.

Exceptionally frustrating, but I'll have to hit it again on Monday.

I ran across this today and I simply could not believe what I was seeing. It's right up there on Daily WTF - or should be, anyway. First, a little set-up...

This code is part of an incoming exchange data decoder. The Exchange will send messages on udp multicast, and it's up to us to grab them, decode them, place them in out message formats, and pass them on to all waiting listeners. What's important to realize is that these decoders are supposed to be efficient and fast. After all, they are decoding hundreds of thousands of messages a second. It's a lot of data.

So... the exchange dictates it's message format, and as in the olden days of the mainframe, most all the data is in fixed-length ASCII records. Specifically, the integer for the size of an order might be 8 characters and look like this:

  120.....

where the '.'s are spaces. Eight characters total, in ASCII format for the number 120. Simple. Not very efficient, but simple.

Since these fixed-length records will be end-to-end, there's no terminating NULL characters to make it easier to parse - you have to know what you're looking for. Well, the code I saw started with this method:

  /*
   * Convert unsafe char array string to int.
   */
  inline int uatoi(const char *nptr, size_t l)
  {
    int rv = 0;
    char *nullTermStr = cmalloc2(l + 1);
    if (nullTermStr == NULL) {
      errno = ENOMEM;
      return INT_MAX;
    }
 
    memcpy(nullTermStr, nptr, l);
    rv = atoi(nullTermStr);
    free(nullTermStr);
 
    return rv;
  }

where:

  /*
   * Malloc2 that returns a char pointer.
   */
  inline char *cmalloc2(size_t size)
  {
    return (char *) malloc2(size);
  }

and:

  /*
   * Malloc memory and initialize it to zero
   */
  inline void *malloc2(size_t size)
  {
    if (size < 0)
      return NULL;
 
    void *ptr = calloc(size, 1);
 
    return ptr;
  }

OK... this is really quite stunning. You want to parse a (char *) and so you duplicate it, by calling a useless method and then calloc with a repeat count of 1, parse it and then free the memory. And this is fast? For upwards of 10 fields a message, hundreds of thousands of times a second?

When I re-wrote the functionality I was decidedly simpler:

  /*
   * Convert unsafe char array string to int.
   */
  inline int uatoi( char *nptr, size_t width )
  {
    char    hold = nptr[width];
    nptr[width] = '\0';
    int     retval = atoi(nptr);
    nptr[width] = hold;
    return retval;
  }

Sure, I had to "loose" the const in the signature because I was modifying the data as I parsed it, but hey - it's a message from some data source - that's OK. It's also the same "logic" of using atoi() in the decoding. But now I'm not calling something to create some memory and then copying it, and destroying it. I can't believe they didn't look at all this before. It's incredible!

I know there comes a time when people don't look at the code anymore and just think the whole thing is too complicated... but really... guys... let's try a little harder. This is a horrible performance penalty for parsing. It should have been looked at long ago.

I guess I'm the one that decided to really look at it.

Today I spent most of the day trying to flesh out the infrastructure for a few concrete message classes of exchange data in my new ticker plant. The initial work I'd been doing was focused on simple message types, and just about anything would work there - so I used my simple 'Hello' and 'GoodBye' TCP handshaking messages, and they worked fine. But now, I've got people wanting to test components that require the sequence number from exchange messages, and that means I needed a lot more infrastructure than I had built up.

The first thing was the idea of the conflation, or 'compression', key on the messages. This is something that I know I'm going to need in the client code as the current system has historically had a serious problem of not being able to handle a slow consumer very well at all. I settled on a simple uint64_t as the conflation key type and added a very nice little string hash method to get decently diverse integer values from strings:

  static const size_t    __initialFNV = 2166136261U;
  static const size_t    __fnvMultiple = 16777619;
  size_t Message::hash( const std::string & aString ) const
  {
    size_t   hash = __initialFNV;
    size_t   len = aString.length();
    const char *p = aString.data();
    for (size_t i = 0; i < len; ++i) {
      hash = hash ^ (*(p++));
      hash *= __fnvMultiple;
    }
    return hash;
  }

and from what I've read, this Fowler, Noll, Vo algorithm is pretty decent at creating a diverse mapping of strings into the integer space. It's not something I'm going to count on as unique, but it's decent enough for some message types.

Now that I had something to put into the getConflationKey() for my existing messages, I started with the exchange messages. I needed to make a base message for exchange data, and then a faux price message that I could use for testing. This is going to be something that is very close to a "real" message, but will be strictly used for testing.

Why? Because I can control everything about it and it'll never change. This is the kind of test framework we need for the performance testing and comparison runs. True, we'll also need real-world exchange data for some additional testing, but that doesn't negate the value of this kind of test framework.

Then I realized that I needed to make a final determination on how to integrate the "parsing" of the external data into msg::Message instances. I decided to go with a very simplistic API. The MessageFactory has three basic methods - the create() method to take external data into our messages, the extractSequenceNumber() to extract just the sequence number from said external data, and the masquerade() method to be the inverse of the create() method and attempt to make external data from the message.

I then had to put this all in place and write the faux price message and it's data conversion class. It was a lot of coding and I didn't quite get it all done today, but should be able to finish it up tomorrow.

I was able to spend another 20 mins on the code conversion to a single ZeroMQ socket for my ZMQ receiver this morning. I then ran the tests, and BINGO! it worked like a charm! Excellent. I still need to do a lot more tests, but this is an amazing step in the right direction.

Today I need to test multiple multicast channels sending to a single receiver, and then I need to get a little deeper into the actual data messages that we'll be sending as I need to do some load tests to see what it's all capable of.

Very exciting times.

The design we have for my ticker plant is to have the different products and exchanges spread out over a large number of reliable multicast channels, address and port combinations. In order to do this with ZeroMQ, I was under the impression that I needed to have a zmq::socket_t for each channel - which in ZeroMQ terms is really a URL like: epgm://eth0;225.1.1.1:55555. After all, the basic code for a multicast receiver looks like:

  #include <zmq.hpp>
 
  // make a ZeroMQ context to handle all this - use one thread for I/O
  zmq::context_t   context(1);
  // make a simple socket, and connect it to the multicast channel
  zmq::socket_t    socket(context, ZMQ_SUB);
  socket.connect("epgm://eth0;225.1.1.1:55555");
  // now set it up for all subscriptions
  socket.setsockopt(ZMQ_SUBSCRIBE, "", 0);
  // receive a single message
  zmq::message_t   msg;
  socket.recv(&msg);

So in many ways, the ZeroMQ socket looks and acts like a regular socket. But I've read in many of the mailing list posts, and even talking to a guy here at The Shop that has dug into the code, that these aren't really sockets - they are just the logical way the ZeroMQ guys made their code appear to the users of their stuff.

This is never more obvious than the mailing list post I read today about having multiple connections to a single zmq::socket_t. The maintainer of the library said that there can be multiple connections to a single socket:

  #include <zmq.hpp>
 
  // make a ZeroMQ context to handle all this - use one thread for I/O
  zmq::context_t   context(1);
  // make a simple socket, and connect it to the multicast channel
  zmq::socket_t    socket(context, ZMQ_SUB);
  socket.connect("epgm://eth0;225.1.1.1:55555");
  socket.connect("epgm://eth0;225.1.1.1:55666");
  socket.connect("epgm://eth0;225.1.1.1:77777");
  socket.connect("epgm://eth0;225.1.1.1:88888");
  // now set it up for all subscriptions
  socket.setsockopt(ZMQ_SUBSCRIBE, "", 0);
  // receive a single message
  zmq::message_t   msg;
  socket.recv(&msg);

and the recv() will then pick off the first available message on any one of the connected sockets. That's great news! As opposed to needing n sockets for n channels, I can have just one, and that simplifies my code a huge amount.

So that's what I'm working on. I'm nearly done, and then I can test. Can't wait to see how this works.

I finally got ZeroMQ integrated into my application library and wrote a few test clients - one to send and the other to receive. When I tried to run them I got the following error:

  $ zmqReceiver
  ..(startup)...
  (process:8444): Pgm-WARNING **: DSCP setting requires CAP_NET_ADMIN of ADMIN
  capability.

I seemed to remember something about this in the mailing list, but try as I might, I wasn't able to find that bloody reference again, and so I was stuck trying to figure this out from scratch.

Basically, because of the way OpenPGM runs, it essentially requires "privileged" access on the network drivers. OK, I can believe that. So how to I accomplish that? My limited google fu yields that the command setcap should do the trick, but that command is not on the CentOS 5 install I have, and yet the command execcap is.

It seems that the execcap runs the single command with the specified capabilities, so if I do:

  $ sudo /usr/sbin/execcap 'cap_net_admin=eip' ./zmqReceiver

then all runs well.

Sort of.

We still run into the problem of LD_LIBRARY_PATH not being transferred, so I ended up making a simple script that set the LD_LIBRARY_PATH and then called the command and that script I put in the execcap command to get things working.

And they worked wonderfully.

Now I need to find a way to make it "stick" without running it as root.

[7/26] UPDATE: when I came in this morning, the linux server I code on had been rebooted. Odd... so I decided to check and see what the state of the CAP_NET_ADMIN capabilities was. Lo and behold... it was fixed! Seems the admin guy had to reboot the box to make the change, and made it he did. I can now launch these processes without the grief of the launch script or execcap. Sweet!

I was looking at several different reliable multicast systems today and came across one that appears to be promising. It's called ZeroMQ, and I'm not sure if it's going to work out, but I wanted to give it a try as it is planning to be used by another group in The Shop, and they have really gotten deep into the guts of the code. So it's probably going to be OK. I just won't know until I really get into it.

The first thing I need to do is to build it, and I might as well build an RPM for it for CentOS 5. There's no such RPM in existence, and yet they provide a nice zeromq.spec file for building an RPM, so I decided to go with that.

For the most part, it works OK... but there's a problem that is a killer if you don't get past it. It won't even compile with OpenPGM activated (--with-pgm). The solution was on the mailing list, and there was no easy way for me to put the fix into the zeromq.spec file, so I had to un-tar the ZeroMQ tarball, edit the src/Makefile.in file and then tar it back up and then build the RPM.

The critical line is:

  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DCONFIG_HAVE_HPET \
  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DPGM_GNUC_INTERNAL=G_GNUC_INTERNAL \
  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DGETTEXT_PACKAGE='"pgm"' \

to:

  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DCONFIG_HAVE_HPET \
  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DPGM_GNUC_INTERNAL= \
  @BUILD_PGM_TRUE@@ON_MINGW_FALSE@   -DGETTEXT_PACKAGE='"pgm"' \

and then the rpmbuild -bb zeromq.spec will work just fine.

I'm interested to see if I can get reliable multicast working with this guy.

Today has been a long and tiring day, but I think finally I have the answers I've been looking for. Specifically, I've been trying to complete my next source/sink pair for the ticker plant that I'm working on, and this pair is meant to use 29West. Their LBM (Latency Buster Messaging) product, specifically. In the past, I've used 29West in what I consider a fairly "standard" configuration - there's a resolver for a large series of services - but separated out for production, UAT, development and possibly even multiple sites. But the point is to make these resolvers handle as much as possible so they actually do the job their name implies: resolve.

Then you have a series of topics like all messaging products, and you start sending and listening for packets on those topics. It's not the most elegant messaging system I've ever used - I think it doesn't encapsulate enough for the user, but that's a corporate design decision, and while I don't agree with theirs, it is theirs to make.

When I started looking at the code from the existing system using 29West, I noticed that it was doing a lot of very specific multicast address stuff, and different resolvers, and I got the feeling that they weren't using it like this at all. In fact, I got the distinct impression that they were using it to bridge multiple multicast networks together.

When we got together and I asked a few questions I realized that they were using it as basically nothing more than a reliable multicast system. The topics were really overly simplistic, but they had independent 29West networks for each of the different message types, and then each of the different first letters in the symbol names. It was really just a reliable multicast library.

When I realized this, I knew the real solution was to get rid of 29West and go towards a more simplistic, yet solid, reliable multicast library. I just had to find it.

I was doing work today with boost asio and the buffering of async socket I/O and was stunned at the realization that the size() was being used in the buffering of data in a std::string or a std::vector<char>. But let me explain.

The standard call to async_receive_from() - or, in fact, any boost async reader method, is a boost::asio::buffer. This is a standard buffer class for boost that takes an underlying data container - like a char [] or a std::string oe a std::vector<char> and then populates that underlying data container with the data coming from the source - be that UDP or TCP, and it works very nicely.

But what I expected was to do something like the following:

  std::vector<char>   buff;
  buff.reserve(16384);
  ...
  mSocket.async_receive_from(boost::asio::buffer(buff),
      boost::bind(...);

such that I had created (allocated) 16kB of space in the std::vector<char> and then when I got the callback, I'd look at size() and see what was really in there. But that's not how it works.

If you do that, then nothing will be read in to the buffer because boost uses the size() to know how big the buffer is. This means that you really have to "pre-fill" the container with junk, and then let the boost asio methods overwrite the junk with good data. It's then mandatory that you get the number of bytes read from boost, as there's no way for you to know what's in the container that's from the I/O and what was put there by you prior to the call.

The code looks deceptively simple:

  std::vector<char>   buff(16384);
  ...
  mSocket.async_receive_from(boost::asio::buffer(buff),
      boost::bind(...);

and in many cases, the value in the constructor might appear to be a capacity. But it's really a size of junk data. Very odd.

Given that boost chose to use size() versus capacity(), I have to wonder why. It makes no sense to me at all. But hey... it's not the first, and it sure won't be the last.