shmsink: Add custom allocator to allow for zero-copy shared memory use

1 files changed, 28 insertions, 13 deletions

diff --git a/sys/shm/shmpipe.c b/sys/shm/shmpipe.c
index 6c7b80be7..677113b31 100644
--- a/sys/shm/shmpipe.c
+++ b/sys/shm/shmpipe.c
@@ -118,7 +118,7 @@ struct _ShmBuffer
   int num_clients;
   int clients[0];
 
-  uint64_t tag;
+  void *tag;
 };
 
 
@@ -183,14 +183,14 @@ struct CommandBuffer
 static ShmArea *sp_open_shm (char *path, int id, mode_t perms, size_t size);
 static void sp_close_shm (ShmArea * area);
 static int sp_shmbuf_dec (ShmPipe * self, ShmBuffer * buf,
-    ShmBuffer * prev_buf, ShmClient * client);
+    ShmBuffer * prev_buf, ShmClient * client, void **tag);
 static void sp_shm_area_dec (ShmPipe * self, ShmArea * area);
 
 
 
 #define RETURN_ERROR(format, ...) do {                  \
   fprintf (stderr, format, __VA_ARGS__);                \
-  sp_close (self);                                      \
+  sp_writer_close (self, NULL, NULL);                   \
   return NULL;                                          \
   } while (0)
 
@@ -424,7 +424,8 @@ sp_dec (ShmPipe * self)
 }
 
 void
-sp_close (ShmPipe * self)
+sp_writer_close (ShmPipe * self, sp_buffer_free_callback callback,
+    void *user_data)
 {
   if (self->main_socket >= 0)
     close (self->main_socket);
@@ -435,11 +436,18 @@ sp_close (ShmPipe * self)
   }
 
   while (self->clients)
-    sp_writer_close_client (self, self->clients);
+    sp_writer_close_client (self, self->clients, callback, user_data);
 
   sp_dec (self);
 }
 
+void
+sp_client_close (ShmPipe * self)
+{
+  sp_writer_close (self, NULL, NULL);
+}
+
+
 int
 sp_writer_setperms_shm (ShmPipe * self, mode_t perms)
 {
@@ -560,7 +568,7 @@ sp_writer_free_block (ShmBlock * block)
 /* Returns the number of client this has successfully been sent to */
 
 int
-sp_writer_send_buf (ShmPipe * self, char *buf, size_t size, uint64_t tag)
+sp_writer_send_buf (ShmPipe * self, char *buf, size_t size, void *tag)
 {
   ShmArea *area = NULL;
   unsigned long offset = 0;
@@ -699,7 +707,7 @@ sp_client_recv (ShmPipe * self, char **buf)
 }
 
 int
-sp_writer_recv (ShmPipe * self, ShmClient * client)
+sp_writer_recv (ShmPipe * self, ShmClient * client, void **tag)
 {
   ShmBuffer *buf = NULL, *prev_buf = NULL;
   struct CommandBuffer cb;
@@ -713,7 +721,7 @@ sp_writer_recv (ShmPipe * self, ShmClient * client)
       for (buf = self->buffers; buf; buf = buf->next) {
         if (buf->shm_area->id == cb.area_id &&
             buf->offset == cb.payload.ack_buffer.offset) {
-          sp_shmbuf_dec (self, buf, prev_buf, client);
+          return sp_shmbuf_dec (self, buf, prev_buf, client, tag);
           break;
         }
         prev_buf = buf;
@@ -786,7 +794,7 @@ sp_client_open (const char *path)
   return self;
 
 error:
-  sp_close (self);
+  sp_client_close (self);
   return NULL;
 }
 
@@ -837,7 +845,7 @@ error:
 
 static int
 sp_shmbuf_dec (ShmPipe * self, ShmBuffer * buf, ShmBuffer * prev_buf,
-    ShmClient * client)
+    ShmClient * client, void **tag)
 {
   int i;
   int had_client = 0;
@@ -866,6 +874,8 @@ sp_shmbuf_dec (ShmPipe * self, ShmBuffer * buf, ShmBuffer * prev_buf,
     else
       self->buffers = buf->next;
 
+    if (tag)
+      *tag = buf->tag;
     shm_alloc_space_block_dec (buf->ablock);
     sp_shm_area_dec (self, buf->shm_area);
     spalloc_free1 (sizeof (ShmBuffer) + sizeof (int) * buf->num_clients, buf);
@@ -875,7 +885,8 @@ sp_shmbuf_dec (ShmPipe * self, ShmBuffer * buf, ShmBuffer * prev_buf,
 }
 
 void
-sp_writer_close_client (ShmPipe * self, ShmClient * client)
+sp_writer_close_client (ShmPipe * self, ShmClient * client,
+    sp_buffer_free_callback callback, void *user_data)
 {
   ShmBuffer *buffer = NULL, *prev_buf = NULL;
   ShmClient *item = NULL, *prev_item = NULL;
@@ -885,11 +896,15 @@ sp_writer_close_client (ShmPipe * self, ShmClient * client)
 again:
   for (buffer = self->buffers; buffer; buffer = buffer->next) {
     int i;
+    void *tag = NULL;
 
     for (i = 0; i < buffer->num_clients; i++) {
       if (buffer->clients[i] == client->fd) {
-        if (!sp_shmbuf_dec (self, buffer, prev_buf, client))
+        if (!sp_shmbuf_dec (self, buffer, prev_buf, client, &tag)) {
+          if (callback)
+            callback (tag, user_data);
           goto again;
+        }
         break;
       }
     }
@@ -949,7 +964,7 @@ sp_writer_get_next_buffer (ShmBuffer * buffer)
   return buffer->next;
 }
 
-uint64_t
+void *
 sp_writer_buf_get_tag (ShmBuffer * buffer)
 {
   return buffer->tag;