Pages

Simple Apache module

My first Apache module needs to be improved in many ways. Here I am addressing to a few basic requirements, answering only to one specific request; logging using the Apache built-in facility; and generating an HTML document as answer.

We want to set our Apache web server so that it would answer with an HTML page containing a few information on the actual received request, and we want it to give this feedback only when we ask for "hello" on it.

Request-module connection

Apache should know about how to associate a user request to the handler that our module is designed to manage. To do that we add a Location directive in the Apache httpd configuration file.

You'll find this file in the Apache conf directory, named httpd.conf, we open it and we add this section:
<Location /hello>
    SetHandler hello
</Location>
We ask Apache to set the handler "hello" so that it is invoked when is issued a request to the hello page directly under the root of my server.

Secondly, we change the C source code, so that we check the passed request, and we ensure the passed handler matches our expectation:
// ...

static const char* MOD_HANDLER = "hello"; // 1

static int handler(request_rec* r)
{
    if(r->handler == NULL || strcmp(r->handler, MOD_HANDLER)) // 2
    {
        // ...
        return DECLINED; // 3
    }

    // ...

    return OK; // 4
}
1. The handler for this module, it's value is used in httpd.conf, as shown above.
2. Check the handler as passed by Apache.
3. If our module has nothing to say here we return DECLINED, to let know to Apache that it has to look elsewhere.
4. Otherwise, after we did our job, we return OK, saying in this way that Apache could considered the request accomplished.

Apache logging

It is often a good idea to log what is going on in our code, both for debugging and administrative purpose. In this module, it would be nice to have a feedback also when a request is discarded. We get this effect adding this line before "return DECLINED":
ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, r, "handling declined: %s", r->handler);
The Apache ap_log_rerror() lets us writing to the Apache error log file, you'll find it in the logs folder, in your Apache httpd installation directory, named "error_log".
APLOG_MARK is just a way to combine the standard __FILE__ and __LINE__ defines in a single symbol, to save a few keystrokes.
After it, we specify the log level, that ranges from APLOG_EMERG down to APLOG_DEBUG (not to mention the TRACE defines, not commonly used, as far as I know). In the httpd.conf file we configure which level of logging we actually want to print to the log file, setting the log level:
LogLevel debug
As you can imagine, in production it is usually a smart move to set the configured log level higher than debug.
Next parameter, here set to 0, is not very interesting here, and it is followed by the pointer to the request, as we get it from Apache.
Finally we have a string, representing what we want to log, and that could include printf-style parameters.

Building a HTML response

Admittedly, an Apache module is not the most programmer-friendly tool to create a HTML page. But it still make sense in such a simple case:
ap_set_content_type(r, "text/html"); // 1
ap_rputs("<html><head><title>Hello Module</title></head><body>", r); // 2
ap_rprintf(r, "handler: %s<br />\n", r->handler); // 3
ap_rprintf(r, "filename: %s<br />\n", r->filename);
ap_rprintf(r, "the_request: %s<br />\n", r->the_request);
ap_rprintf(r, "header_only: %d<br />\n", r->header_only);
ap_rprintf(r, "hostname: %s<br />\n", r->hostname);
ap_rputs("</body></html>", r);
1. Firstly, set the reply content type.
2. To put a static string, as this one, ap_rputs() does an excellent job.
3. When we need to send parametrized stuff, we'd better using ap_rprintf().

As you can see, I generated the answer from the request, extracting a few (more or less) interesting values as received from Apache.

The complete C code for this module is on github. To compile it and add the generated .so to the server, you can use the apxs utility (more details in the previous post). Remember to set the Apache configuration file with the Location/SetHandler directives.

Go to the full post

A minimal Hello World Apache module

After you have set up the Apache development environment, it is time to create a first module.

I tried to create a minimal module, following the K&R's "Hello World" spirit, that I reckon is so rewarding when you are approaching new stuff.

This module is going to be very impolite, trying to answer to all the user requests to the Apache server in the same way. It would even override the standard index.html page on root.

It is made of a module declaration, logically the first thing we are interested in, but traditionally placed at the bottom of the file, and a couple of static functions (I forgot to mention it, but I am developing in plain old C language).

One of the two functions, hooks(), is passed to the module declaration, and it sets a connection between Apache and the other function, that I named handler(), that it is going to be called to handle the user's jobs.

Here is the three guys above mentioned in detail:
static int handler(request_rec* r) // 1
{
    ap_set_content_type(r, "text/plain"); // 2
    ap_rputs("Hello Apache httpd module", r); // 3
    return OK; // 4
}

static void hooks(apr_pool_t* p) // 5
{
    ap_hook_handler(handler, NULL, NULL, APR_HOOK_REALLY_FIRST); // 6
}

module AP_MODULE_DECLARE_DATA hello_module = // 7
{
    STANDARD20_MODULE_STUFF, NULL, NULL, NULL, NULL, NULL, hooks // 8
};
1. This function is called by Apache so that we can provide a reply to a client request. As we can see, as parameter we get a pointer to a structure that actually represents the user request.
2. I am not doing any check here, I always prepare a plain text reply, by calling the Apache function that sets the content type, ...
3. ... and I fill it with a puny string, with the Apache version of the well known puts() function, but reinterpreted for working on a request_rec.
4. Finally I return OK, meaning that my module has been able to fulfill the user request, and Apache could happily consider this job as done.
5. Here I set the hooks that let Apache know which are the functions in my module it can call.
6. The first parameter to ap_hook_handler() is a function that Apache could call to reply to a user request, and the last one is the priority this hook should have in the collection of hooks owned by Apache. Here I am saying that I want full priority.
7. Here is my module declaration. It is known to Apache by the suggestive name of hello_module.
8. And this are a bunch of information we are passing to Apache about our module. The STANDARD20_MODULE_STUFF define is an aggregate of constants that are saying to Apache this is a standard module version 2, and there is not much more to say about it. We'll say something more on the subsequent five NULLs, but more interesting is the last parameter, the function name Apache needs to know to perform the module initialization.

This is almost everything about it. There are a couple of header inclusions you have to perform to let the compiler knowing what the heck are all those ap_... things, namely httpd.h and http_config.h, but you can see the full code on github.

And, well, you have to compile and register this code before you can actually use it on Apache. To do that, there is a nifty Apache utility, apxs, that basically does all the job for us.

In this case, I would call it something like this:
/path/to/apache/bin/apxs -a -i -c mod_hello.c
Then I stop and start Apache, run it, submit any request whatsoever to it through a we browser, and I should always get back the same reply.

Go to the full post

Installing Apache Httpd

In my current project I am also developing some stuff on the Apache HTTP Server, also known as Apache httpd, or even just Apache, in a Debian box. The environment setup is not complicated, but it does have a couple of twists. So I reckoned it was worthy to put down a few notes about the process.

Currently, on the official Apache httpd download page, we have access to three versions (2.0, 2.2, and 2.4) and a number of different formats, ready to be grabbed.

Since I want a developer install, I should avoid the lean standard cuts provided by apt-get (for Debian) or packaged in a msi (for Windows), and I should go for the "Source" releases. So in my case, I go for a 2.2 (this is the version used at work) Unix Source download.

If you check your chosen link, you would see a different provider accordingly to your geographical position, but the archive name should end like ".../httpd/httpd-2.2.22.tar.gz" (being 2.2.22 the current 2.2 version). I downloaded it through wget, and then I have extracted it by tar xvfz (that is why I have picked up the tar.gz flavor), getting all the raw stuff in a subfolder. I changed directory to there and, before starting the real installation process, I decided where to put the thing, let's call it $APACHE2, that would usually be something like $HOME/apache2.

Firstly I have to prepare the configuration, this is usually done by calling the command
./configure --prefix=$APACHE2
In my case, I want to enable the Dynamic Shared Object (DSO) Support, so that I could install shared modules in a next step. To do that, configure has to be called passing the enable-so option too:
./configure --prefix=$APACHE2 --enable-module=so
Once configure has run, it is time to make Apache. This needs two steps, a first call to "make", and a second one, to "make install".

Almost done. Still I had to set the Apache endpoint in its configuration file: I went to $APACHE2/conf, I edited httpd.conf, setting the property Listen to localhost:8080 - a common setup.

Now I can start and stop my Apache HTTP server, going to $APACHE2/bin, and running:
./apachectl start
./apachectl stop
After starting, and before stopping, I should be able to connect from my web browser to the Apache local home page, by accessing the page on localhost:8080 - if this is not the case, that means I have some trouble.

Building Apache 1.3 on a modern environment

If you need to install a legacy Apache httpd server, you would follow more or less the same procedure, but you could bump in a couple of issues.

Firstly, running configure could lead to errors and a garbled output. This is easy to solve, just run it through a new bash shell, like this:
bash ./configure --prefix=$APACHE13 --enable-module=so
Secondly, you could get a failure in making Apache, due to the use of the symbol getline, that now is part of the C standard library.
The solution here is editing the offending files, htdigest.c htpasswd.c logresolve.c, to rename the local getline function.

Go to the full post

CDS internal client error

A glitch in VMWare, running on Windows for a Linux guest. On my machine it was not possible to install the useful tools, and so I couldn't (among the other things) use the complete screen for Linux. A real pain.

This was caused by CDS internal client error (3033), issued when I tried to download VMWare Tools for Linux - version 8.8.4.

As one could expect, the solution is download the package by hand. Problem is that was not so immediate to find the right file on the net.

If you have the same issue, here is an hint:
http://softwareupdate.vmware.com/cds/vmw-desktop/player/4.0.4/744019/windows/packages/
This is the file I was looking for:
tool-linux-8.8.4.exe.tar
I un-tarred, and then executed it. And then I was able to continue the standard procedure from my Linux box.

Go to the full post

Paranoid Pirate Heartbeating - worker

The worker described here, is the companion to the ZeroMQ router server described in the previous post. In another post there is an overview for this simple heartbeat example. Full C++ source code is available on github.

The app controller part spawns a new thread for the worker on this function:
void worker(char id, int lifespan) // 1
{
    const int PATIENCE = 3;
    HeartbeatWorker worker(id); // 2

    int patience = 0;
    int cycle = 0;
    int sent = 0;
    int iteration = 1;

    while(true)
    {
        uint8_t control = worker.recv(); // 3
        if(control == PHB_NOTHING)
        {
            if(cycle++ == PATIENCE)
            {
                if(patience++ == PATIENCE)
                    break; // 4

                int factor = static_cast<int>(std::pow(2.0, patience));
                boost::this_thread::sleep(bp::millisec(factor * BASE_INTERVAL)); // 5

                worker.reset();
                cycle = 0;
            }

            if(iteration++ == lifespan) // 6
                break;
        }
        else
            cycle = patience = 0; // 7
        worker.heartbeat(); // 8
    }
    worker.shutdown(); // 9
}
1. The worker is identified by a single character that is used to generate a unique id for each socket that is created here. The thing is that each time the worker looses contact with the server, the existing socket is closed and replaced by a new one. Any new socket should have a different id, so that the server won't be confused. As a second parameter we pass to the worker function its lifespan.
2. A HeartbeatWorker class is used to make the code more readable, it is showed below.
3. HeartbeatWorker::recv() combines a poll() and a recv() call to the worker socket. If no message was pending the PHB_NOTHING value is returned.
4. If no heartbeat is received from the server for a while, the worker shuts itself down.
5. But before committing suicide, the worker tries to reconnect to the server, closing its socket, waiting for a (growing) while, and then opening a new socket, with a new identity but pointing to the same endpoint.
6. When we reach the limit imposed by the caller, even if the server is still alive, we terminate the worker.
7. If there was actually something on the socket, we reset the cycle counting .
8. In any case, send an heartbeat to the server.
9. Gracefully terminate the worker.

This is my HeartbeatWorker implementation:
class HeartbeatWorker
{
private:
    zmq::context_t context_;
    std::unique_ptr<zmq::Socket> sk_; // 1
    std::string id_; // 2
    char idRoot_;
    bp::ptime heartbeat_; // 3
    zmq::pollitem_t items_[1];

public:
    HeartbeatWorker(char id) : context_(1), idRoot_(id),
        heartbeat_(bp::microsec_clock::local_time() +  bp::millisec(BASE_INTERVAL))
    {
        reset();

        items_[0].fd = 0;
        items_[0].events = ZMQ_POLLIN;
        items_[0].revents = 0;
    }

    void reset()
    {
        id_ += idRoot_; // 4
        sk_.reset(new zmq::Socket(context_, ZMQ_DEALER, id_)); // 5
        sk_->setLinger(0); // 6
        sk_->connect(SKA_BACKEND_CLI);
        items_[0].socket = *sk_.get(); // 7

        sk_->send(PHB_READY); // 8
    }

    void heartbeat() // 9
    {
        if(bp::microsec_clock::local_time() > heartbeat_)
        {
            heartbeat_ = bp::microsec_clock::local_time() + bp::millisec(BASE_INTERVAL);
            sk_->send(PHB_HEARTBEAT);
        }
    }

    void shutdown()
    {
        sk_->send(PHB_DOWN);
    }

    uint8_t recv() // 10
    {
        if(zmq::poll(items_, 1, BASE_INTERVAL * 1000) < 1)
            return PHB_NOTHING;

        uint8_t res = PHB_NOTHING;
        if(items_[0].revents & ZMQ_POLLIN)
            res = sk_->recvAsByte();
        items_[0].revents = 0;
        
        return res;
    }
};
1. Any time we lose the connection to the server, we kill the socket and create a new one. For this reason, I use a (smart) pointer instead of an object on the stack.
2. Socket id. It is based on the idRoot_ character (defined in the next line).
3. Expected time for sending an heartbeat to the server. The bp namespace is defined as synonym of boost::posix_time.
4. The socket id is changed any time a reset occurs.
5. The unique_ptr::reset() take cares of closing the previous socket (if set) before deleting it.
6. We don't want to wait for messages sent on a socket to be delivered when we close it. By default we have an indefinite lingering, but this is no good here, because it could happen that we send messages to a server that is not there anymore, and we want to close the socket (to create a new one) even if those messages are not delivered.
7. A tricky line. We store a pointer to socket, but the zmq::pollitem_t structure requires a socket itself about its members. So dereferencing is required.
8. Sends a "ready" message to the server.
9. The heartbeat message is sent to the server only if its time has arrived.
10. HeartbeatWorker::recv() combines zmq::poll() and an actual receiving on the socket. We expect in input a single-part message containing a single byte.

Go to the full post

Paranoid Pirate Heartbeating - server

As described in the previous post, I have developed a cut to the bone ZeroMQ C++ router-dealer heartbeat exchange-only application that should help understanding the basics of the process.

The reason to write such an example is that, as they say in the ZGuide, "as for the Paranoid Pirate queue, the heartbeating is quite tricky to get right". I hope that such a stripped down example could help to clarify the concept.

In its simplest configuration, this heartbeat example is supposed to run as a monoprocess, multithread application. The main thread spawns two new threads, one for the server and one for a worker, and they are left alone to interact till their natural termination.

We could even let run just the server (or, as we'll see in the next post, the worker) alone. It won't get any ping from a worker, and after a while it would shutdown.

Here is the server function, it makes use of a custom class, HeartbeatServer, that I comment below:
void server(int lifespan) // 1
{
    HeartbeatServer server(lifespan); // 2

    while(server.isAlive()) // 3
    {
        zmq::Frames input = server.recv(); // 4
        if(!input.empty())
        {
            uint8_t control = (input.size() != 2 || input[1].empty()) ? PHB_UNEXPECTED : input[1].at(0);

            switch(control)
            {
            case PHB_READY:
                server.pushWorker(input[0]); // 5
                break;
            case PHB_HEARTBEAT:
                server.refreshWorker(input[0]); // 6
                break;
            case PHB_DOWN:
                server.dropWorker(input[0]); // 7
                break;
            default: // 8
                break;
            }
        }
        server.heartbeat(); // 9
    }
}
1. The server function expects in input a parameter stating how long it should theoretically run, passing INT_MAX (as defined in limits.h) states that we'd like to run the server (almost) forever.
2. Create an Heartbeat Server, see below for details.
3. HeartbeatServer::isAlive() returns false when the server is ready to shutdown.
4. HeartbeatServer::recv() combines a poll and a receive on the server socket. If a (multipart) message is pending on the socket, it is received and returned as deque of strings (if you wonder why, have a look at this post). We expect a two-part message, with the worker id as first element and the payload (a single byte) in second position. Whatever else could be happily discarded.
5. If the payload is a "ready" message, the worker id is stored in the server.
6. We received an "heartbeat" from the worker, we refresh its status. Remember that a worker should be seen as active by the server only if it sends at least an heartbeat once in a while. After a longish silence from its side, we should consider it as lost in space.
7. If the worker ends its life gracefully, it sends a message to notify the server, so that it would remove it on the spot.
8. We received something weird. Let it know to the user.
9. If there is a worker pending on the server, send an heartbeat to it.

And this is the server class:
class HeartbeatServer
{
private:
    zmq::context_t context_;
    zmq::Socket backend_;
    zmq::pollitem_t items_[1];

    std::string wid_; // 1
    bp::ptime expiry_; // 2

    int lifespan_; // 3
    int busy_; // 4
    bp::ptime heartbeat_; // 5
    enum { INTERVAL = 1000, BUSY = 5 };
public:
    HeartbeatServer(int lifespan = INT_MAX) : context_(1), backend_(context_, ZMQ_ROUTER), lifespan_(lifespan), 
        heartbeat_(bp::microsec_clock::local_time() + bp::millisec(INTERVAL)), busy_(BUSY)
    {
        backend_.bind(SKA_BACKEND_SRV);
        backend_.setLinger(0); // 6

        items_[0].socket = backend_; // 7
        items_[0].fd = 0;
        items_[0].events = ZMQ_POLLIN;
        items_[0].revents = 0;
    }

    bool isAlive() // 8
    {
        return busy_ > 0 && lifespan_ > 0;
    }

    zmq::Frames recv() // 9
    {
        --lifespan_;

        zmq::Frames res;
        if(zmq::poll(items_, 1, INTERVAL * 1000) > 0)
        {
            busy_ = BUSY;

            if(items_[0].revents & ZMQ_POLLIN)
                res = backend_.blockingRecv();
            items_[0].revents = 0;

            return res;
        }
        else
            --busy_;

        return res; // no message
    }

    void heartbeat() // 10
    {
        if(wid_.empty())
            return;

        // if it's time, send heartbeat
        if(bp::microsec_clock::local_time() > heartbeat_)
        {
            std::string control(&PHB_HEARTBEAT, &PHB_HEARTBEAT + 1);
            backend_.send(wid_, control);
            heartbeat_ = bp::microsec_clock::local_time() + bp::millisec(BASE_INTERVAL);
        }

        // if the worker is expired, remove its id 
        if(expiry_ < bp::microsec_clock::local_time())
            wid_ = "";
    }

    void pushWorker(const std::string& id) // 11
    {
        wid_ = id;
    }

    void refreshWorker(const std::string& id)
    {
        if(wid_ == id)
            expiry_ = bp::microsec_clock::local_time() + bp::millisec(CHECK_FACTOR * BASE_INTERVAL);
    }

    void dropWorker(const std::string& id)
    {
        if(wid_ == id)
            wid_ = "";
    }
};
1. The worker id pending on the server. In this minimal implementation we can have only one worker. If no worker is available, the string is empty.
2. Keep track of when the worker identified by (1) is going to expire. The namespace bp is a synonym for boost::posix_time.
3. Expected server lifespan.
4. If the server is idle for a while, this counter goes to zero, and then we signal that the server should shutdown.
5. When the server should send the next heartbeat.
6. The sockets in this app have a troubled life, it often happens that one sends a message to the other but that one is already gone. Besides, the worker has a custom identity, so it is marked as persistent. To avoid the server socket hanging on termination, it is better to specify that we don't want it linger. This function calls zmq_setsockopt() for the ZMQ_LINGER option on the underlying socket.
7. Initialize the array of zmq::pollitem_t to poll on the socket.
8. To be considered alive, the server should both be in the time frame defined by the user and busy.
9. Each time recv() is called, the server life expectancy decreases. If there is actually anything on the socket, the busy counter is restored to its maximum value, otherwise it decreases too.
10. An heartbeat is sent to the pending worker only if a worker is actually there, and it is actually time to send it. After sending it (if required) the worker expiration time is checked, to see if it is time to remove it from the server.
11. In this trivial implementation, pushing, refreshing, dropping a worker on the server is very easy. We have just one (or none) of them, so we just have to ensure nothing unexpected (bad worker id) happens.

The full C++ code for the application is on github.

Go to the full post

Paranoid Pirate Heartbeating

Measuring a paranoid pirate heartbeating is not for the faint of heart. I guess that that is true in any case, but here I talking about the ZeroMQ Paranoid Pirate Protocol.

As they state in the ZGuide, talking about their implementation for that protocol based on czmq, the high level C wrapper to 0MQ, "Heartbeating is simple once it works, but quite difficult to invent." For this reason I though it was interesting seeing first how heartbeating could work, and then integrating it in a paranoid pirate implementation.

I wrote the code for ZeroMQ version 2.2, developed on Windows + MSVC2010, using the standard 0MQ C++ wrapper with my zmq::Socket class that adds some features to the original zmq::socket_t. See previous posts, like this one, for details.

Since I am interested only in heartbeating, I have designed the application to be as simple as I could think, to the point of looking quite meaningless.

We have a server (without clients) with one or no associated worker. The server has a ROUTER socket, and the worker a DEALER one. They exchange only status information, no real payload.

When a worker signals to the server that it is alive, the server stores its id, and then it sends to the worker an heartbeat to signal it that it is still alive. The same from the worker, it sends to the server a heartbeat till it shutdowns.

If the server stops getting heartbeats from the worker, it simply remove its id, and doesn't care about it anymore.

The worker is more caring. It retries a few time the check on the server heartbeat, and only when it has lost any hope, it considers the server lost, wait for a while, an then create a new socket, trying to establish a new connection.

Even with this minimal requirement, the resulting code is not immediate. For this reason I split the discussion in a few posts. After this introduction you could read about the router server and the dealer worker in the next two posts.

The main thread in this heartbeat testing application runs a couple of test cases:
boost::thread_group threads;
threads.create_thread(std::bind(server, INT_MAX)); // 1
threads.create_thread(std::bind(worker, 'A', 6)); // 2
threads.join_all();
1. The server() function, expects in input an int parameter, representing how many iteration we want to run. Here I specify the largest int available, as defined in limits.h, meaning that I want to run it (almost) forever.
2. The worker() needs to know which character should be used as seed for the worker id, the second one is the number of heartbeat that the worker is going to send to the server before shutting down.

The result of this test case should be that the server should stay up till the workers sends all its heartbeats, then its is going to wait a bit more idle, before shutting down. The worker id should be "A".

Second test case:
boost::thread_group threads;
threads.create_thread(std::bind(server, 3)); // 1
threads.create_thread(std::bind(worker, 'B', 7)); // 2

boost::this_thread::sleep(boost::posix_time::seconds(10)); // 3
threads.create_thread(std::bind(server, INT_MAX)); // 4
threads.join_all();
1. This server is going to be short lived.
2. Same worker as before.
3. Ensure enough time passes before restarting the server.
4. An almost-forever server is started.

Here we expect that worker seeing the server going offline, restarting and completing its job. The worker id should swap from "A" to "AA".

The full C++ code for the complete example is on github.

Go to the full post

Sending and receiving a single byte

After the major change of using deque as container for multipart messages, I have done a minor change in my zmq::Socket class, that extends the standard ZeroMQ C++ wrapper provided with version 2 of the package.

I was reading the Paranoid Pirate Protocol when I saw that it expects the ready and heartbeat commands to be sent as single bytes. So I decided to add a couple of methods in my class to explicitly manage this case. It is not a strict necessity, but they surely make the code more readable.

Here are the two new methods:
bool send(uint8_t value, int flags =0) // 1
{
    zmq::message_t msg(sizeof(uint8_t));
    memcpy(msg.data(), &value, sizeof(uint8_t));
    return socket_t::send(msg, flags);
}

uint8_t recvAsByte(int flags =0)
{
    zmq::message_t message;
    if(!socket_t::recv(&message, flags))
        throw error_t();

    return *(static_cast<uint8_t*>(message.data()));
}
uint8_t is defined in stdint.h as unsigned char, assuming in your environment byte has 8 bits, as almost everywhere happens. Using this typedef should make clearer that we are interested in a tiny number and not in a character.

I have pushed on github the new file version.

Go to the full post

ZeroMQ Multipart message as deque

I originally designed my zmq::Socket class extending the zmq::socket_t class, as defined in the standard C++ wrapper included in the ØMQ 2.x distribution, to provide a more intuitive multipart message management, basing it on a std::vector container.

This is fine for building up multipart messages on receive, but it gets clumsy when we want to modify an existing multipart message, typically to put an envelope around it, or to strip that envelope out.

For this kind of manipulation I think is much better using std::deque.

The previous changes to the this class, till the latest introduction of a couple of methods to send and receive ints, were not dramatic ones. But this redesign breaks the original interface, a deque does not provide a reserve() method as the vector does, and I often used it in my code when creating a multipart message. Besides, I have also changed the blockingRecv() signature, since I found out that delegating to that method the check of the actual number of frames in a multipart message didn't add much value to the resulting code. On the contrary, it made it a bit less intuitive.

For this reason, I left on github the old include file, and I created a new one, zmq2a.h, where the deque version of multipart messages is defined and used.

Again, here is the change applied:

- zmq::Frames, type used for multipart messages, now is a typedef for a deque of strings.
- blockingRecv() now does not have any input parameter, it still throws an zmq::error_t exception, but only in case of failure receiving a frame in the message.

Go to the full post

Improved LRU with basic reliable queuing

We can apply the just seen lazy pirate pattern (LPP) for the LRU Queue Broker. In this way we make it more reliable inpacting only on the client side of the application.

This approach is called in the ZGuide simple pirate pattern.

What we had, a queue of least recently used workers, managed by a ROUTER to ROUTER 0MQ broker that gets in input requests coming from clients, send them to workers (accordingly to their availability), that do the job, send the result back, through their REQ socket, to the broker, that sends it back to the original client, is matched to the previously seen LPP client.

We have alredy seen about all the code, the only issue is matching it. The full example C++ code is on github, here are just the startup lines:
boost::thread_group threads; // 1
threads.create_thread(std::bind(client, 2500,  3)); //2
threads.create_thread(lruQueue); // 3
threads.create_thread(worker);

boost::this_thread::sleep(boost::posix_time::seconds(20)); // 4
dumpId("---");
threads.create_thread(worker);
threads.create_thread(std::bind(client, 2500,  3));

threads.join_all();
1. To simplify the testing, I have put all the components in the same process, each one running in its own thread. This is not realistic, but it is not an issue to refactor the example to run as a multiprocess application.
2. The function client() is the one we have seen in the LPP, see previous post for details.
3. The lruQueue() and worker() components come from the LRU example, the worker has been modified to perform restlessly.
4. Actually, the sample worker is designed to simulate crash after a few seconds, and the client would shutdown if it won't get any feedback. On the other side, the broker is designed to run forever (but it is not protected against interrupts). Here we wait a while, to be sure that the worker crashes, and the client shuts itself down. Then we create another worker and client, and we can check how the system springs back to work.

The code shown on the ZGuide is written for the high level czmq c-binding. This porting could be interesting to you, if you want to see an example of how to do the same in C++. I have used the official C++ wrapper provided for ZeroMQ 2.x, adapted with my extension to zmq::socket_t, called zmq::Socket, that provides some better suppport to multipart and int messages.

Go to the full post

Client side reliable REQ-REP

We can't assume any software being 100% reliable - actually, we can't assume that for anything in the world - but the ZeroMQ connection are designed to be fast more than reliable, so we usually want to add some fallback mechanism to avoid the more common issues.

In the ZGuide are shown a few patterns we could implement to go in the direction of an improved reliability. The lazy pirate pattern, as they call it, is the first of them. It is very simple, but it could be enough. Here I rewrite that example for C++ and using zmq::Socket, my subclass for the zmq::socket_t, as defined in the original 0MQ version 2.x C++ binding.

The point is that the REQ client sends (along with the payload, not shown in the example) a sequence number. The server replies sending it back to the client (usually along with some feedback, again, not in this example), so that the client could check if anything worked fine. If the client gets no confirmation, we assume the message we sent was lost, so we resend it. That is a bit more complicated than saying it, since the REQ-REP protocol is strictly synchronous. Sending more than one REQ before getting a REP results in an EFSM error. An easy, brutal but effective way to overcome this issue, is closing and reopening the socket before resending the message.

To keep testing easy, I put both client and server in the same process, this is the procedure in the main thread that spawns the two threads:
boost::thread_group threads;
threads.create_thread(std::bind(client, timeout, retries)); // 1
threads.create_thread(server); // 2
threads.join_all();
1. This starts the client, passing to it a couple of ints, timeout, in millisecs, and the number of retries that the client should do before assuming that the server is down. The balance between these number has to be chosen carefully, being very sensitive for giving a good behaving system. In this example I used 2500 and 3, as suggested by the ZGuide, then I played around with them to see what happened. I'd suggest you to do the same.
2. It starts the server.

A few constants used by the code:
const char* SK_ADDR_CLI = "tcp://localhost:5555"; // 1
const char* SK_ADDR_SRV = "tcp://*:5555"; // 2

const int BASE_TIMEOUT = 1000; // 3
1. The client socket connects to this address.
2. The server socket bound address.
3. I set the standard timeout for the application to 1 second (here is in millis, as you would have correctly guessed).

Server

It is very simple. Even simpler than the original one:
void server()
{
    dumpId("Starting server"); // 1

    zmq::context_t context(1);
    zmq::Socket skServer(context, ZMQ_REP);
    skServer.bind(SK_ADDR_SRV);

    for(int i = 1; i < 10; ++i) // 2
    {
        int message = skServer.recvAsInt(); // 3

        if(i % 4 == 0) // 4
        {
            dumpId("CPU overload");
            boost::this_thread::sleep(boost::posix_time::millisec(2 * BASE_TIMEOUT));
        }
        else
            dumpId("Normal request", message);

        boost::this_thread::sleep(boost::posix_time::millisec(BASE_TIMEOUT));
        skServer.send(message);
    }

    dumpId("Terminating, as for a server crash");
}
1. I paid the simplification of running both client and server in the same process, being forced to care about concurrent issues on std::cout. The dumpId() function uses a lock on a mutex to avoid that printing gets garbled by competing threads.
2. This toy server just recv/send a bunch of times before terminate simulating a crash.
3. Socket::recvAsInt() receives on the socket and converts the result in an int, that is returned to the caller.
4. After a while, a CPU overload is simulated, sleeping for a while.

The Lazy Pirate Client

To keep the client code a bit more readable, I have decided to use a utility class tailored on the example need. Actually I should say I am not completely happy about it, but I should have modified the original zmq::socket_t to get a more satisfactory result, but I am not sure that it would be a smart move in this moment. Maybe in the future.

Anyway, here is my current result:
class LazyPirateClient
{
private:
    zmq::context_t& context_;
    std::unique_ptr<zmq::Socket> sk_; // 1

    int sent_; // 2

    void reset() // 3
    {
        sk_.reset(new zmq::Socket(context_, ZMQ_REQ));
        sk_->connect(SK_ADDR_CLI);

        int linger = 0;
        sk_->setsockopt(ZMQ_LINGER, &linger, sizeof(linger));
    }
public:
    LazyPirateClient(zmq::context_t& context) : context_(context), sent_(-1) // 4
    {
        reset();
    }

    bool checkedRecv() // 5
    {
        int value = sk_->recvAsInt();
        if(value == sent_)
            return true;

        dumpId("Unexpected reply from server", value);
        return false;
    }

    zmq::Socket* operator()() { return sk_.get(); } // 6

    void send(int value) // 7
    {
        sk_->send(value);
        sent_ = value;
    }

    void resend() // 8
    {
        reset();
        send(sent_);
    }
};
1. Given the limitation of the original zmq::socket_t class, where the underlying raw socket pointer is stored in its private section, I had to think some smart alternative solution to close and reopen the lazy pirate underlying socket.
2. Internal flag, keeping memory of the last sequence number sent to the server, so that we can compare the received result.
3. Create a new REQ socket, and establish a connection to the REP server socket. The linger flag is set to zero, meaning that we want pending messages to be immediately discarded.
4. Ctor, rely on the above defined reset().
5. We don't care much of what we get back from the server, besides checking if the sequence number matches. This method does just that. It uses the Socket::recvAsInt() utility function that extract an int from the received message.
6. Utility function to get the underlying pointer to zmq::Socket
7. It sends the value (as an int, thanks to an explicit overload of send() defined in zmq::Socket), and store it so that it could be checked by (5).
8. As explained before, we can't simply resend a message, we would get an EFSM error. So we reset the socket, deleting the old one and creating a new one, before sending again the last value.

Client

This is the function on which runs the client thread:
void client(int timeout, int retries)
{
    dumpId("Starting client");

    zmq::context_t context(1);
    LazyPirateClient skClient(context);

    for(int sequence = 1; sequence < 100; ++sequence) // 1
    {
        skClient.send(sequence); // 2
        boost::this_thread::sleep(boost::posix_time::millisec(BASE_TIMEOUT));

        bool confirmed = false;
        zmq::pollitem_t items[] = { { *skClient(), 0, ZMQ_POLLIN, 0 } }; // 3
        for(int cycle = 0; cycle < retries; ++cycle)
        {
            zmq::poll(&items[0], 1, timeout * 1000); // 4

            if(items[0].revents & ZMQ_POLLIN) // 5
            {
                if(skClient.checkedRecv()) // 6
                {
                    dumpId("Server is synchronized", sequence);
                    confirmed = true;
                    break;
                }
            }
            else // 7
            {
                dumpId("Retrying", cycle);
                skClient.resend();
                items[0].socket = *skClient(); // 8
            }
        }
        if(!confirmed) // 9
        {
            dumpId("No answer from server, abandoning.");
            break;
        }
    }
}
1. Loop for some time. Actually, when the server is down, the client would automatically shutdown, and this is what is going to happen almost immediately.
2. Send the sequence number to the server. Usually we want to send more interesting stuff, too.
3. Prepare to poll on the client socket for input messages.
4. Wait for the period specified by the user.
5. We actually have something in input.
6. The sequence check succeeded, give the user a feedback and go on.
7. Nothing received in the interval, whatever happened, we simply resend the last message.
8. Remember that we want to poll on the new socket.
9. No confirmation from the server received, the client assumes that something very bad happened out there, and shutdown.

Full C++ source code is on github.

Go to the full post

Sending/receiving ints over ZeroMQ

I often send an integer as a socket message. It is not a problem to convert them to and fro a std::string, still it makes the resulting code a bit confusing, so I decided to add a send() overload and a new recvAsInt() function to my zmq::Socket class.

This is the new send() function:
bool send(int value, int flags =0)
{
    zmq::message_t msg(sizeof(int));
    memcpy(msg.data(), &value, sizeof(int));
    return socket_t::send(msg, flags);
}
A new specialized function to receive a message and format it as an int:
int recvAsInt(int flags =0)
{
    zmq::message_t message;
    if(!socket_t::recv(&message, flags))
        throw error_t();

    return *(static_cast<int*>(message.data()));
}
Even the mild EAGAIN error leads to an exception. Admittedly, this is not good.

Here is how to use this functions:
// ...
skRequest.send(42);

// ...
int value = skReply.recvAsInt();
We can also have a multipart message containing ints:
skRequest.send(42, ZMQ_SNDMORE);
skRequest.send(12);

//    ...

int val1 = skReply.recvAsInt(ZMQ_RCVMORE);
int val2 = skReply.recvAsInt();
There is no type check on the data, so we can write weird code like this:
// ...
skRequest.send("hello");

// ...
int value = skReply.recvAsInt();
That runs fine, but probably gives a surprising result to the user.

The zmq::Socket new version is on github.

Go to the full post