Turbo version using zero-copy where possible

Not going on master since this is all kind of complicated, but it's useful
to have around as a stress test

1 files changed, 85 insertions, 72 deletions

diff --git a/zmq-camera.c b/zmq-camera.c
index 5123553..5c85f76 100644
--- a/zmq-camera.c
+++ b/zmq-camera.c
@@ -79,7 +79,7 @@ const char local_camera[] = "inproc://local-camera";
 const char local_signal[] = "inproc://local-signal";
 
 /* multicast transport options */
-const uint64_t mcast_rate = 100000; /* Rate limit for multicast, kbps */
+const uint64_t mcast_rate = 250000; /* Rate limit for multicast, kbps */
 const uint64_t recovery_ivl = 5; /* Recovery interval for multicast, seconds */
 
 /*
@@ -94,18 +94,87 @@ struct sender_args_t
     int frame_rate;
 };
 
+/* New frame callback arguments */
+struct new_frame_cb_args_t
+{
+    void *s;
+    unicap_format_t dest_format;
+    int frame_rate;
+};
+
+/* Callback function called by 0MQ when message data can be freed */
+void zmq_free_cb (void *data, void *hint)
+{
+    free (data);
+}
+
+/* Callback function called by unicap when a new frame is available */
+void new_frame_cb (unicap_event_t event, unicap_handle_t handle,
+                   unicap_data_buffer_t *src_buffer, void *usr_data)
+{
+    zmq_msg_t msg;
+    size_t msg_size;
+    uint32_t image_width, image_height;
+    unsigned char *msg_data, *data;
+    unicap_data_buffer_t dest_buffer;
+    struct new_frame_cb_args_t *args = usr_data;
+    int rc;
+    
+    /* Create 0MQ message and fill in data. */
+    msg_size = args->dest_format.buffer_size + (2 * sizeof (uint32_t));
+    msg_data = malloc(msg_size);
+    assert (msg_data);
+    data = msg_data;
+
+    /* Image width (uint32_t in network byte order) */
+    image_width = htonl (args->dest_format.size.width);
+    memcpy (data, &image_width, sizeof (uint32_t));
+    data += sizeof (uint32_t);
+
+    /* Image height (uint32_t in network byte order) */
+    image_height = htonl (args->dest_format.size.height);
+    memcpy (data, &image_height, sizeof (uint32_t));
+    data += sizeof (uint32_t);
+
+    /* data now points to where RGB24 image data will go. */
+   
+    /* Setup unicap buffer for colorspace conversion, 
+       wrapped around our message data */
+    memset (&dest_buffer, 0, sizeof (unicap_data_buffer_t));
+    dest_buffer.buffer_size = args->dest_format.buffer_size;
+    dest_buffer.format = args->dest_format;
+    dest_buffer.data = data;
+   
+    /* Convert colorspace. */
+    if (!SUCCESS (ucil_convert_buffer (&dest_buffer, src_buffer))) {
+        /* TODO: This fails sometimes for unknown reasons,
+           just skip the frame for now. */
+        /* fprintf (stderr, "Failed to convert video buffer\n"); */
+    }
+
+    /* Send message to network. */
+    /* Zero-copy: wrap 0mq message around our data */
+    rc = zmq_msg_init_data (&msg, msg_data, msg_size, zmq_free_cb, NULL);
+    assert (rc == 0);
+    rc = zmq_send (args->s, &msg, 0);
+    assert (rc == 0);
+    zmq_msg_close (&msg);
+
+    /* Limit to maximum frame rate if requested */
+    if (args->frame_rate > 1)
+        usleep (1000000 / args->frame_rate);
+}
+
 void *sender_thread (void *arg)
 {
     struct sender_args_t *sender_args;
+    struct new_frame_cb_args_t new_frame_cb_args;
     void *s;
     int rc;
     unicap_handle_t handle;
     unicap_device_t device;
     unicap_format_t src_format;
     unicap_format_t dest_format;
-    unicap_data_buffer_t src_buffer;
-    unicap_data_buffer_t dest_buffer;
-    unicap_data_buffer_t *returned_buffer;
     int conversion_found = 0;
     int index = 0;
 
@@ -164,7 +233,7 @@ void *sender_thread (void *arg)
         fprintf (stderr, "the colorspace conversion code in this example.\n");
         exit (1);
     }
-    src_format.buffer_type = UNICAP_BUFFER_TYPE_USER;
+    src_format.buffer_type = UNICAP_BUFFER_TYPE_SYSTEM;
     if (!SUCCESS (unicap_set_format (handle, &src_format))) {
         fprintf (stderr, "Failed to set video format\n");
         exit (1);
@@ -181,15 +250,13 @@ void *sender_thread (void *arg)
     dest_format.bpp = 24;
     dest_format.buffer_size = dest_format.size.width *
         dest_format.size.height * 3;
-    
-    /* Initialise image buffers. */
-    memset (&src_buffer, 0, sizeof (unicap_data_buffer_t));
-    src_buffer.data = (unsigned char*) malloc (src_format.buffer_size);
-    src_buffer.buffer_size = src_format.buffer_size;
-    memset (&dest_buffer, 0, sizeof (unicap_data_buffer_t));
-    dest_buffer.data = (unsigned char*) malloc (dest_format.buffer_size);
-    dest_buffer.buffer_size = dest_format.buffer_size;
-    dest_buffer.format = dest_format;
+ 
+    /* Register callback on every new frame available from hardware */
+    new_frame_cb_args.s = s;
+    new_frame_cb_args.dest_format = dest_format;
+    new_frame_cb_args.frame_rate = sender_args->frame_rate;
+    unicap_register_callback (handle, UNICAP_EVENT_NEW_FRAME, 
+        (unicap_callback_t) new_frame_cb, (void *) &new_frame_cb_args);
 
     /* Start video capture. */
     if (!SUCCESS (unicap_start_capture (handle))) {
@@ -198,64 +265,10 @@ void *sender_thread (void *arg)
         exit (1);
     }
 
-    /* Loop, publising one message per raw video frame. */
+    /* Sleep forever, new frames will be published by new_frame_cb */
     while (1) {
-        zmq_msg_t msg;
-        size_t msg_size;
-        uint32_t image_width, image_height;
-        unsigned char *data;
-
-        /* Queue buffer for video capture. */
-        if (!SUCCESS (unicap_queue_buffer (handle, &src_buffer))) {
-            fprintf (stderr, "Failed to queue a buffer on device: %s\n",
-                device.identifier);
-            exit (1);
-        }
-
-        /* Wait until buffer is ready. */
-        if (!SUCCESS (unicap_wait_buffer (handle, &returned_buffer))) {
-            fprintf (stderr, "Failed to wait for buffer on device: %s\n",
-                device.identifier);
-            exit (1);
-        }
-
-        /* Convert colorspace. */
-        if (!SUCCESS (ucil_convert_buffer (&dest_buffer, &src_buffer))) {
-            /* TODO: This fails sometimes for unknown reasons,
-               just skip the frame for now. */
-            /* fprintf (stderr, "Failed to convert video buffer\n"); */
-        }
-
-        /* Create 0MQ message and fill in data. */
-        msg_size = dest_format.buffer_size + (2 * sizeof (uint32_t));
-        rc = zmq_msg_init_size (&msg, msg_size);
-        assert (rc == 0);
-        data = (unsigned char *)zmq_msg_data (&msg);
-
-        /* Image width (uint32_t in network byte order) */
-        image_width = htonl (dest_format.size.width);
-        memcpy (data, &image_width, sizeof (uint32_t));
-        data += sizeof (uint32_t);
-
-        /* Image height (uint32_t in network byte order) */
-        image_height = htonl (dest_format.size.height);
-        memcpy (data, &image_height, sizeof (uint32_t));
-        data += sizeof (uint32_t);
-
-        /* RGB24 image data. */
-        memcpy (data, dest_buffer.data, dest_format.buffer_size);
-        
-        /* Send message to network. */
-        rc = zmq_send (s, &msg, 0);
-        assert (rc == 0);
-        zmq_msg_close (&msg);
-
-        /* Limit to maximum frame rate if requested */
-        if (sender_args->frame_rate > 1)
-            usleep (1000000 / sender_args->frame_rate);
+        pause ();
     }
-    
-    return NULL;
 }
 
 /*
@@ -569,8 +582,8 @@ int main (int argc, char *argv [])
                         SDL_GetError());
                     exit (1);
                 }
-                screen = SDL_SetVideoMode (image_width, image_height, 32,
-                    SDL_HWSURFACE);
+                screen = SDL_SetVideoMode (image_width, image_height, 24,
+                    SDL_SWSURFACE);
                 if (screen == NULL) {
                    fprintf (stderr, "Unable to set video mode: %s\n",
                        SDL_GetError ());