How do I use libcouchbase with my own event loop?

Writing asynchronous programs seems to be popular these days, so I thought I should whip up an example that shows you how you may utilize libcouchbase with your own event loop.

libcouchbase performs it's IO through an "IO handle". This is a plugin system so you should be able to use whatever mechanism you want. Adding support for a new system is nothing more than implementing a handfull of function calls and place them in a shared object. When I wrote libcouchbase I created a version for libevent and a small version that use select for Windows just to ensure that it should be possible to write such plugins. Later Mark Nunberg wrote the plugin we're using for node.js that is based on libuv. I'm not going to cover how to write your own plugin in this blog post (perhaps I'll get around to do that at a later time), but I'll show you how to let libcouchbase utilize the same event base from libevent that you're using for something else.

To keep the example as short as possible I'm only going to do the very basic error handling: print out an error message and terminate the program. This means that we can start writing our first function; our error handler:

static void error_callback(lcb_t instance, lcb_error_t error, const char *errinfo) {
    fprintf(stderr, "ERROR: %s %s\n", lcb_strerror(instance, error), errinfo);
    exit(EXIT_FAILURE);
}

I'm not very good at coming up with interesting examples, so today we'll just create a small program that connects to a Couchbase cluster, stores a key and then reads the key back out again.

So let's take a look at our main function:

int main(int argc, char** argv) {
    struct event_base *evbase = event_base_new();

    if (create_libcouchbase_handle(evbase) == -1) {
        exit(EXIT_FAILURE);
    }

    event_base_loop(evbase, 0);
    event_base_free(evbase);
    exit(EXIT_SUCCESS);
}

As you see there is nothing fancy there.. We're creating the event base that the rest of our application should use (I just don't use it for anything else in this example to make the example simple), before we create our libcouchbase handle and associate it with the newly created event base (we'll look at that shortly). We then start the event loop that will run the rest of the example through the callbacks.

So how does the create_libcouchbase_handle function look like:

static int create_libcouchbase_handle(struct event_base *evbase) {
    struct lcb_create_io_ops_st ciops;

    memset(&ciops, 0, sizeof (ciops));
    ciops.v.v0.type = LCB_IO_OPS_LIBEVENT;
    ciops.v.v0.cookie = evbase;

    lcb_io_opt_t ioops;
    lcb_error_t error = lcb_create_io_ops(&ioops, &ciops);
    if (error != LCB_SUCCESS) {
        fprintf(stderr, "Failed to create an IOOPS structure for libevent: %s\n",
                lcb_strerror(NULL, error));
        return -1;
    }

Let's stop there for a moment and look what we're doing. The first thing we're doing is that we're creating an instance of the lcb_create_io_ops_st structure. Like the rest of libcouchbase we're using a versioned struct for creating objects. The "constructor" for the IO handle on top of libevent allows us to pass the event base to use as the cookie.

We can now go ahead and create the libcouchbase instance "the normal way, with the only difference that we specify the io member in the create structure.:

    struct lcb_create_st copts;
    memset(&copts, 0, sizeof (copts));
    copts.v.v0.host = "localhost:8091";
    copts.v.v0.user = "Administrator";
    copts.v.v0.passwd = "secret";
    copts.v.v0.bucket = "default";
    copts.v.v0.io = ioops;

    lcb_t instance;
    if ((error = lcb_create(&instance, &copts)) != LCB_SUCCESS) {
        fprintf(stderr, "Failed to create a libcouchbase instance: %s\n",
                lcb_strerror(NULL, error));
        return -1;
    }

The next thing we'll do is to set up the different callbacks we're going to use in our program and call lcb_connect to initiate the connect sequence:

    lcb_set_error_callback(instance, error_callback);
    lcb_set_configuration_callback(instance, configuration_callback);
    lcb_set_get_callback(instance, get_callback);
    lcb_set_store_callback(instance, store_callback);

    if ((error = lcb_connect(instance)) != LCB_SUCCESS) {
        fprintf(stderr, "Failed to connect libcouchbase instance: %s\n",
                lcb_strerror(NULL, error));
        lcb_destroy(instance);
        return -1;
    }
}

You might be curious why I'm adding a "configuration callback"? This is actually a small trick :-) You might have tried to use libcouchbase yourself and had problems that your operations failed because you forgot to do a lcb_wait() after calling connect. The thing is that libcouchbase needs to know the topology of your cluster before it may perform any operations, and it cannot do that until it receives the first configuration from the server.

So how does this configuration callback look like in our example:

static void configuration_callback(lcb_t instance, lcb_configuration_t config) {
    if (config == LCB_CONFIGURATION_NEW) {
        // Since we've got our configuration, let's go ahead and store a value
        lcb_store_cmd_t cmd;
        const lcb_store_cmd_t * cmds[] = {&cmd};
        memset(&cmd, 0, sizeof (cmd));
        cmd.v.v0.key = "foo";
        cmd.v.v0.nkey = 3;
        cmd.v.v0.bytes = "bar";
        cmd.v.v0.nbytes = 3;
        cmd.v.v0.operation = LCB_SET;
        lcb_error_t err = lcb_store(instance, NULL, 1, cmds);
        if (err != LCB_SUCCESS) {
            fprintf(stderr, "Failed to set up store request: %s\n",
                    lcb_strerror(instance, err));
            exit(EXIT_FAILURE);
        }
    }
}

As the comment tells you, we've received the configuration from the server. This means that it's safe to start using the library. In a real world application you would probably have a more advanced logic here (please note that this callback will be called if you add/remove nodes etc, so it's not safe to assume that it will be called only once!)

When we receive the result for the store command from the server, our store_callback is called:

static void store_callback(lcb_t instance, const void *cookie, lcb_storage_t operation, lcb_error_t error, const lcb_store_resp_t *resp) {
    if (error != LCB_SUCCESS) {
        fprintf(stderr, "Failed to store key: %s\n",
                lcb_strerror(instance, error));
        exit(EXIT_FAILURE);
    }

    /* Time to read it back */
    lcb_get_cmd_t cmd;
    const lcb_get_cmd_t * cmds[] = {&cmd};
    memset(&cmd, 0, sizeof (cmd));
    cmd.v.v0.key = "foo";
    cmd.v.v0.nkey = 3;
    if ((error = lcb_get(instance, NULL, 1, cmds)) != LCB_SUCCESS) {
        fprintf(stderr, "Failed to setup get request: %s\n",
                lcb_strerror(instance, error));
        exit(EXIT_FAILURE);
    }
}

If the value was successfully stored on the server, we're issuing a single get command to the server to verify that it's there. When we receive the get result from the server, our get_callback is called and we're terminating the program:

static void get_callback(lcb_t instance, const void *cookie, lcb_error_t error, const lcb_get_resp_t *resp) {
    if (error != LCB_SUCCESS) {
        fprintf(stderr, "Failed to get key: %s\n",
                lcb_strerror(instance, error));
        exit(EXIT_FAILURE);
    }

    fprintf(stdout, "I stored and retrieved the key \"foo\". Terminate program");
    exit(EXIT_SUCCESS);
}

This wasn't the most exciting example, but it shows you the basics on how you may utilize libcouchbase with your own event loop. None of the above functions will cause your application to block waiting for socket IO. If that happens I'd be more than happy to fix it if you send me a bug report (unless you wrote your own IO plugin and forgot to enable nonblocking IO ;)

Happy hacking!