irmd: Use buffer_t for piggybacked data

Instead of passing a const void * and len, it now passes buffer_t to
operations that send piggybacked data (flow_req_arr and flow_reply)
and a buffer_t * for operations that send and receive piggybacked data
(flow_alloc and flow_accept).

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

6 files changed, 138 insertions, 157 deletions

diff --git a/include/ouroboros/utils.h b/include/ouroboros/utils.h
index bd174149..f5f8610d 100644
--- a/include/ouroboros/utils.h
+++ b/include/ouroboros/utils.h
@@ -25,16 +25,20 @@
 
 #include <stdint.h>
 #include <unistd.h>
+#include <string.h>
 
 #define MIN(a,b) (((a) < (b)) ? (a) : (b))
 #define MAX(a,b) (((a) > (b)) ? (a) : (b))
 #define ABS(a)   ((a) > 0 ? (a) : -(a))
+#define clrbuf(buf) do { memset(&(buf), 0, sizeof(buf)); } while (0);
+#define freebuf(buf) do { free((buf).data); clrbuf(buf); } while (0);
 
 typedef struct {
         uint8_t * data;
         size_t    len;
 } buffer_t;
 
+
 /*
  * Returns the number of characters a uint would
  * need when represented as a string
diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index c062809c..360934b0 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -99,6 +99,7 @@ ipcp_msg_t * send_recv_ipcp_msg(pid_t        pid,
        default:
                tv.tv_sec  = SOCKET_TIMEOUT / 1000;
                tv.tv_usec = (SOCKET_TIMEOUT % 1000) * 1000;
+
                break;
        }
 
@@ -176,7 +177,7 @@ pid_t ipcp_create(const char *   name,
 
         argv[4] = NULL;
 
-        if (posix_spawn(&pid, argv[0], NULL, NULL, argv, NULL)) {
+        if (posix_spawn(&pid, argv[0], NULL, NULL, argv, NULL) != 0) {
                 log_err("Failed to spawn new process");
                 return -1;
         }
@@ -434,8 +435,7 @@ static int __ipcp_flow_alloc(pid_t           pid,
                              size_t          len,
                              qosspec_t       qs,
                              bool            join,
-                             const void *    data,
-                             size_t          dlen)
+                             const buffer_t  data)
 {
         ipcp_msg_t    msg      = IPCP_MSG__INIT;
         ipcp_msg_t *  recv_msg = NULL;
@@ -454,8 +454,8 @@ static int __ipcp_flow_alloc(pid_t           pid,
         msg.hash.data    = (uint8_t *) dst;
         msg.qosspec      = qos_spec_s_to_msg(&qs);;
         msg.has_pk       = true;
-        msg.pk.data      = (uint8_t *) data;
-        msg.pk.len       = (uint32_t) dlen;
+        msg.pk.data      = data.data;
+        msg.pk.len       = data.len;
 
         recv_msg = send_recv_ipcp_msg(pid, &msg);
         free(msg.qosspec);
@@ -479,11 +479,10 @@ int ipcp_flow_alloc(pid_t           pid,
                     const uint8_t * dst,
                     size_t          len,
                     qosspec_t       qs,
-                    const void *    data,
-                    size_t          dlen)
+                    const buffer_t  data)
 {
-        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst, len, qs, false,
-                                 data, dlen);
+        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst,
+                                 len, qs, false, data);
 }
 
 int ipcp_flow_join(pid_t           pid,
@@ -493,16 +492,16 @@ int ipcp_flow_join(pid_t           pid,
                    size_t          len,
                    qosspec_t       qs)
 {
-        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst, len, qs, true,
-                                 NULL, 0);
+        buffer_t data = {NULL, 0};
+        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst,
+                                 len, qs, true, data);
 }
 
-int ipcp_flow_alloc_resp(pid_t        pid,
-                         int          flow_id,
-                         pid_t        n_pid,
-                         int          response,
-                         const void * data,
-                         size_t       len)
+int ipcp_flow_alloc_resp(pid_t          pid,
+                         int            flow_id,
+                         pid_t          n_pid,
+                         int            response,
+                         const buffer_t data)
 {
         ipcp_msg_t   msg      = IPCP_MSG__INIT;
         ipcp_msg_t * recv_msg = NULL;
@@ -516,8 +515,8 @@ int ipcp_flow_alloc_resp(pid_t        pid,
         msg.has_response = true;
         msg.response     = response;
         msg.has_pk       = true;
-        msg.pk.data      = (uint8_t *) data;
-        msg.pk.len       = (uint32_t) len;
+        msg.pk.data      = data.data;
+        msg.pk.len       = data.len;
 
         recv_msg = send_recv_ipcp_msg(pid, &msg);
         if (recv_msg == NULL)
diff --git a/src/irmd/ipcp.h b/src/irmd/ipcp.h
index ad96ad1c..bedbfe01 100644
--- a/src/irmd/ipcp.h
+++ b/src/irmd/ipcp.h
@@ -69,8 +69,7 @@ int   ipcp_flow_alloc(pid_t           pid,
                       const uint8_t * dst,
                       size_t          len,
                       qosspec_t       qs,
-                      const void *    data,
-                      size_t          dlen);
+                      const buffer_t  data);
 
 int   ipcp_flow_join(pid_t           pid,
                      int             flow_id,
@@ -79,12 +78,11 @@ int   ipcp_flow_join(pid_t           pid,
                      size_t          len,
                      qosspec_t       qs);
 
-int   ipcp_flow_alloc_resp(pid_t        pid,
-                           int          flow_id,
-                           pid_t        n_pid,
-                           int          response,
-                           const void * data,
-                           size_t       len);
+int   ipcp_flow_alloc_resp(pid_t          pid,
+                           int            flow_id,
+                           pid_t          n_pid,
+                           int            response,
+                           const buffer_t data);
 
 int   ipcp_flow_dealloc(pid_t  pid,
                         int    flow_id,
diff --git a/src/irmd/irm_flow.c b/src/irmd/irm_flow.c
index 44d3fb7e..6d22cbc8 100644
--- a/src/irmd/irm_flow.c
+++ b/src/irmd/irm_flow.c
@@ -33,9 +33,10 @@
 
 #include "irm_flow.h"
 
-#include <stdlib.h>
-#include <stdbool.h>
 #include <assert.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
 
 struct irm_flow * irm_flow_create(pid_t     n_pid,
                                   pid_t     n_1_pid,
@@ -43,10 +44,14 @@ struct irm_flow * irm_flow_create(pid_t     n_pid,
                                   qosspec_t qs)
 {
         pthread_condattr_t cattr;
-        struct irm_flow *  f = malloc(sizeof(*f));
+        struct irm_flow *  f;
+
+        f = malloc(sizeof(*f));
         if (f == NULL)
                 goto fail_malloc;
 
+        memset(f, 0, sizeof(*f));
+
         if (pthread_condattr_init(&cattr))
                 goto fail_cattr;
 
@@ -59,14 +64,6 @@ struct irm_flow * irm_flow_create(pid_t     n_pid,
         if (pthread_mutex_init(&f->state_lock, NULL))
                 goto fail_mutex;
 
-        f->n_pid   = n_pid;
-        f->n_1_pid = n_1_pid;
-        f->flow_id = flow_id;
-        f->mpl     = 0;
-        f->qs      = qs;
-        f->data    = NULL;
-        f->len     = 0;
-
         f->n_rb = shm_rbuff_create(n_pid, flow_id);
         if (f->n_rb == NULL) {
                 log_err("Could not create ringbuffer for process %d.", n_pid);
@@ -79,11 +76,16 @@ struct irm_flow * irm_flow_create(pid_t     n_pid,
                 goto fail_n_1_rbuff;
         }
 
-        f->state = FLOW_ALLOC_PENDING;
-
         if (clock_gettime(CLOCK_MONOTONIC, &f->t0) < 0)
                 log_warn("Failed to set timestamp.");
 
+        f->n_pid   = n_pid;
+        f->n_1_pid = n_1_pid;
+        f->flow_id = flow_id;
+        f->qs      = qs;
+
+        f->state = FLOW_ALLOC_PENDING;
+
         pthread_condattr_destroy(&cattr);
 
         return f;
@@ -123,7 +125,7 @@ void irm_flow_destroy(struct irm_flow * f)
 
         pthread_mutex_lock(&f->state_lock);
 
-        assert(f->len == 0);
+        assert(f->data.len == 0);
 
         if (f->state == FLOW_DESTROY) {
                 pthread_mutex_unlock(&f->state_lock);
diff --git a/src/irmd/irm_flow.h b/src/irmd/irm_flow.h
index af613d36..a1794f5f 100644
--- a/src/irmd/irm_flow.h
+++ b/src/irmd/irm_flow.h
@@ -26,6 +26,7 @@
 #include <ouroboros/list.h>
 #include <ouroboros/qos.h>
 #include <ouroboros/shm_rbuff.h>
+#include <ouroboros/utils.h>
 
 #include <sys/types.h>
 #include <pthread.h>
@@ -50,8 +51,7 @@ struct irm_flow {
 
         qosspec_t          qs;
         time_t             mpl;
-        void *             data;
-        size_t             len;
+        buffer_t           data;
 
         struct shm_rbuff * n_rb;
         struct shm_rbuff * n_1_rb;
diff --git a/src/irmd/main.c b/src/irmd/main.c
index e8a73aa0..75e6c2c7 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -173,9 +173,9 @@ static void clear_irm_flow(struct irm_flow * f) {
 
         assert(f);
 
-        if (f->len != 0) {
-                free(f->data);
-                f->len = 0;
+        if (f->data.len != 0) {
+                free(f->data.data);
+                f->data.len = 0;
         }
 
         while ((idx = shm_rbuff_read(f->n_rb)) >= 0)
@@ -1277,18 +1277,17 @@ static int proc_announce(pid_t  pid,
 static int flow_accept(pid_t             pid,
                        struct timespec * dl,
                        struct irm_flow * f_out,
-                       const void *      data,
-                       size_t            len)
+                       buffer_t *        data)
 {
-        struct irm_flow  *  f  = NULL;
-        struct proc_entry * pe = NULL;
-        struct reg_entry *  re = NULL;
-        struct list_head *  p  = NULL;
-
-        pid_t pid_n;
-        pid_t pid_n_1;
-        int   flow_id;
-        int   ret;
+        struct irm_flow  *  f;
+        struct proc_entry * pe;
+        struct reg_entry *  re;
+        struct list_head *  p;
+        pid_t               pid_n;
+        pid_t               pid_n_1;
+        int                 flow_id;
+        int                 ret;
+        buffer_t            tmp = {NULL, 0};
 
         pthread_rwlock_wrlock(&irmd.reg_lock);
 
@@ -1345,7 +1344,7 @@ static int flow_accept(pid_t             pid,
                 list_del(&f->next);
                 bmp_release(irmd.flow_ids, f->flow_id);
                 pthread_rwlock_unlock(&irmd.flows_lock);
-                ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, -1, NULL, 0);
+                ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, -1, tmp);
                 clear_irm_flow(f);
                 irm_flow_set_state(f, FLOW_NULL);
                 irm_flow_destroy(f);
@@ -1365,7 +1364,7 @@ static int flow_accept(pid_t             pid,
                 list_del(&f->next);
                 bmp_release(irmd.flow_ids, f->flow_id);
                 pthread_rwlock_unlock(&irmd.flows_lock);
-                ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, -1, NULL, 0);
+                ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, -1, tmp);
                 clear_irm_flow(f);
                 irm_flow_set_state(f, FLOW_NULL);
                 irm_flow_destroy(f);
@@ -1384,20 +1383,16 @@ static int flow_accept(pid_t             pid,
         f_out->n_1_pid = f->n_1_pid;
         f_out->qs      = f->qs;
         f_out->mpl     = f->mpl;
-        f_out->data    = f->data; /* pass owner */
-        f_out->len     = f->len;
 
-        f->data = NULL;
-        f->len  = 0;
+        if (f->qs.cypher_s != 0) /* crypto requested, send pubkey */
+                tmp = *data;
 
-        pthread_rwlock_unlock(&irmd.flows_lock);
+        *data = f->data; /* pass owner */
+        clrbuf (f->data);
 
-        if (f->qs.cypher_s == 0) { /* no crypto requested, don't send pubkey */
-                data = NULL;
-                len = 0;
-        }
+        pthread_rwlock_unlock(&irmd.flows_lock);
 
-        if (ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, 0, data, len)) {
+        if (ipcp_flow_alloc_resp(pid_n_1, flow_id, pid_n, 0, tmp)) {
                 pthread_rwlock_wrlock(&irmd.flows_lock);
                 list_del(&f->next);
                 pthread_rwlock_unlock(&irmd.flows_lock);
@@ -1494,11 +1489,10 @@ static int flow_join(pid_t              pid,
         f_out->flow_id = f->flow_id;
         f_out->n_pid   = f->n_pid;
         f_out->n_1_pid = f->n_1_pid;
-        f_out->data    = f->data; /* pass owner */
-        f_out->len     = f->len;
         f_out->mpl     = f->mpl;
-        f->data        = NULL;
-        f->len         = 0;
+
+        assert(f->data.data == NULL);
+        assert(f->data.len  == 0);
 
         pthread_rwlock_unlock(&irmd.flows_lock);
 
@@ -1512,8 +1506,7 @@ static int flow_alloc(pid_t              pid,
                       qosspec_t          qs,
                       struct timespec *  dl,
                       struct irm_flow *  f_out,
-                      const void *       data,
-                      size_t             len)
+                      buffer_t *         data)
 {
         struct irm_flow *   f;
         struct ipcp_entry * ipcp;
@@ -1560,7 +1553,7 @@ static int flow_alloc(pid_t              pid,
         str_hash(ipcp->dir_hash_algo, hash, dst);
 
         if (ipcp_flow_alloc(ipcp->pid, flow_id, pid, hash,
-                            IPCP_HASH_LEN(ipcp), qs, data, len)) {
+                            IPCP_HASH_LEN(ipcp), qs, *data)) {
                 irm_flow_set_state(f, FLOW_NULL);
                 /* sanitizer cleans this */
                 log_info("Flow_allocation failed.");
@@ -1588,11 +1581,9 @@ static int flow_alloc(pid_t              pid,
         f_out->flow_id = f->flow_id;
         f_out->n_pid   = f->n_pid;
         f_out->n_1_pid = f->n_1_pid;
-        f_out->data    = f->data; /* pass owner */
-        f_out->len     = f->len;
         f_out->mpl     = f->mpl;
-        f->data        = NULL;
-        f->len         = 0;
+        *data          = f->data; /* pass owner */
+        clrbuf(f->data);
 
         pthread_rwlock_unlock(&irmd.flows_lock);
 
@@ -1687,13 +1678,12 @@ static int flow_req_arr(pid_t             pid,
                         const uint8_t *   hash,
                         time_t            mpl,
                         qosspec_t         qs,
-                        const void *      data,
-                        size_t            len)
+                        buffer_t          data)
 {
-        struct reg_entry *  re = NULL;
-        struct prog_entry * a  = NULL;
-        struct proc_entry * pe = NULL;
-        struct irm_flow *   f  = NULL;
+        struct reg_entry *  re;
+        struct prog_entry * a;
+        struct proc_entry * pe;
+        struct irm_flow *   f;
 
         struct pid_el *     c_pid;
         struct ipcp_entry * ipcp;
@@ -1801,22 +1791,8 @@ static int flow_req_arr(pid_t             pid,
         }
 
         f->state = FLOW_ALLOC_REQ_PENDING;
-        f->mpl = mpl;
-
-        if (len != 0) {
-                assert(data);
-                f->data = malloc(len);
-                if (f->data == NULL) {
-                        bmp_release(irmd.flow_ids, flow_id);
-                        pthread_rwlock_unlock(&irmd.flows_lock);
-                        log_err("Could not piggyback data.");
-                        return -1;
-                }
-
-                f->len = len;
-                if (len > 0)
-                        memcpy(f->data, data, len);
-        }
+        f->mpl   = mpl;
+        f->data  = data;
 
         list_add(&f->next, &irmd.irm_flows);
 
@@ -1834,8 +1810,7 @@ static int flow_req_arr(pid_t             pid,
                 list_del(&f->next);
                 pthread_rwlock_unlock(&irmd.flows_lock);
                 log_err("Could not get process table entry for %d.", h_pid);
-                free(f->data);
-                f->len = 0;
+                freebuf(f->data);
                 irm_flow_destroy(f);
                 return -1;
         }
@@ -1852,11 +1827,10 @@ static int flow_req_arr(pid_t             pid,
         return 0;
 }
 
-static int flow_alloc_reply(int          flow_id,
-                            int          response,
-                            time_t       mpl,
-                            const void * data,
-                            size_t       len)
+static int flow_alloc_reply(int      flow_id,
+                            int      response,
+                            time_t   mpl,
+                            buffer_t data)
 {
         struct irm_flow * f;
 
@@ -1874,16 +1848,7 @@ static int flow_alloc_reply(int          flow_id,
         else
                 irm_flow_set_state(f, FLOW_NULL);
 
-        f->data = malloc(len);
-        if (f->data == NULL) {
-                pthread_rwlock_unlock(&irmd.flows_lock);
-                return -1;
-        }
-
-        if (len > 0)
-                memcpy(f->data, data, len);
-
-        f->len = len;
+        f->data = data;
 
         pthread_rwlock_unlock(&irmd.flows_lock);
 
@@ -2073,14 +2038,15 @@ static void * mainloop(void * o)
 
         while (true) {
                 irm_msg_t *        ret_msg;
-                struct irm_flow    e;
+                struct irm_flow    f;
                 struct ipcp_config conf;
                 struct timespec *  dl      = NULL;
                 struct timespec    ts      = {0, 0};
                 struct cmd *       cmd;
                 int                result;
+                buffer_t           data;
 
-                memset(&e, 0, sizeof(e));
+                memset(&f, 0, sizeof(f));
 
                 ret_msg = malloc(sizeof(*ret_msg));
                 if (ret_msg == NULL)
@@ -2198,54 +2164,58 @@ static void * mainloop(void * o)
                         result = name_unreg(msg->name, msg->pid);
                         break;
                 case IRM_MSG_CODE__IRM_FLOW_ACCEPT:
-                        assert(msg->pk.len > 0 ? msg->pk.data != NULL
-                               : msg->pk.data == NULL);
-                        result = flow_accept(msg->pid, dl, &e,
-                                             msg->pk.data, msg->pk.len);
+                        data.len  = msg->pk.len;
+                        data.data = msg->pk.data;
+                        assert(data.len > 0 ? data.data != NULL
+                                            : data.data == NULL);
+                        result = flow_accept(msg->pid, dl, &f, &data);
                         if (result == 0) {
+                                qosspec_msg_t * qs_msg;
+                                qs_msg = qos_spec_s_to_msg(&f.qs);
                                 ret_msg->has_flow_id = true;
-                                ret_msg->flow_id     = e.flow_id;
+                                ret_msg->flow_id     = f.flow_id;
                                 ret_msg->has_pid     = true;
-                                ret_msg->pid         = e.n_1_pid;
-                                ret_msg->qosspec     = qos_spec_s_to_msg(&e.qs);
-                                ret_msg->has_pk      = true;
-                                ret_msg->pk.data     = e.data;
-                                ret_msg->pk.len      = e.len;
+                                ret_msg->pid         = f.n_1_pid;
+                                ret_msg->qosspec     = qs_msg;
                                 ret_msg->has_mpl     = true;
-                                ret_msg->mpl         = e.mpl;
+                                ret_msg->mpl         = f.mpl;
+                                ret_msg->has_pk      = true;
+                                ret_msg->pk.data     = data.data;
+                                ret_msg->pk.len      = data.len;
                         }
                         break;
                 case IRM_MSG_CODE__IRM_FLOW_ALLOC:
-                        assert(msg->pk.len > 0 ? msg->pk.data != NULL
-                                               : msg->pk.data == NULL);
+                        data.len  = msg->pk.len;
+                        data.data = msg->pk.data;
+                        assert(data.len > 0 ? data.data != NULL
+                                            : data.data == NULL);
                         result = flow_alloc(msg->pid, msg->dst,
                                             qos_spec_msg_to_s(msg->qosspec),
-                                            dl, &e, msg->pk.data,
-                                            msg->pk.len);
+                                            dl, &f, &data);
                         if (result == 0) {
                                 ret_msg->has_flow_id = true;
-                                ret_msg->flow_id     = e.flow_id;
+                                ret_msg->flow_id     = f.flow_id;
                                 ret_msg->has_pid     = true;
-                                ret_msg->pid         = e.n_1_pid;
-                                ret_msg->has_pk      = true;
-                                ret_msg->pk.data     = e.data;
-                                ret_msg->pk.len      = e.len;
+                                ret_msg->pid         = f.n_1_pid;
                                 ret_msg->has_mpl     = true;
-                                ret_msg->mpl         = e.mpl;
+                                ret_msg->mpl         = f.mpl;
+                                ret_msg->has_pk      = true;
+                                ret_msg->pk.data     = data.data;
+                                ret_msg->pk.len      = data.len;
                         }
                         break;
                 case IRM_MSG_CODE__IRM_FLOW_JOIN:
                         assert(msg->pk.len == 0 && msg->pk.data == NULL);
                         result = flow_join(msg->pid, msg->dst,
                                            qos_spec_msg_to_s(msg->qosspec),
-                                           dl, &e);
+                                           dl, &f);
                         if (result == 0) {
                                 ret_msg->has_flow_id = true;
-                                ret_msg->flow_id     = e.flow_id;
+                                ret_msg->flow_id     = f.flow_id;
                                 ret_msg->has_pid     = true;
-                                ret_msg->pid         = e.n_1_pid;
+                                ret_msg->pid         = f.n_1_pid;
                                 ret_msg->has_mpl     = true;
-                                ret_msg->mpl         = e.mpl;
+                                ret_msg->mpl         = f.mpl;
                         }
                         break;
                 case IRM_MSG_CODE__IRM_FLOW_DEALLOC:
@@ -2254,30 +2224,38 @@ static void * mainloop(void * o)
                                               msg->timeo_sec);
                         break;
                 case IRM_MSG_CODE__IPCP_FLOW_REQ_ARR:
-                        assert(msg->pk.len > 0 ? msg->pk.data != NULL
-                                               : msg->pk.data == NULL);
+                        data.len  = msg->pk.len;
+                        data.data = msg->pk.data;
+                        msg->has_pk  = false; /* pass data */
+                        msg->pk.data = NULL;
+                        msg->pk.len  = 0;
+                        assert(data.len > 0 ? data.data != NULL
+                                            : data.data == NULL);
                         result = flow_req_arr(msg->pid,
-                                              &e,
+                                              &f,
                                               msg->hash.data,
                                               msg->mpl,
                                               qos_spec_msg_to_s(msg->qosspec),
-                                              msg->pk.data,
-                                              msg->pk.len);
+                                              data);
                         if (result == 0) {
                                 ret_msg->has_flow_id = true;
-                                ret_msg->flow_id     = e.flow_id;
+                                ret_msg->flow_id     = f.flow_id;
                                 ret_msg->has_pid     = true;
-                                ret_msg->pid         = e.n_pid;
+                                ret_msg->pid         = f.n_pid;
                         }
                         break;
                 case IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY:
-                        assert(msg->pk.len > 0 ? msg->pk.data != NULL
-                                               : msg->pk.data == NULL);
+                        data.len  = msg->pk.len;
+                        data.data = msg->pk.data;
+                        msg->has_pk  = false; /* pass data */
+                        msg->pk.data = NULL;
+                        msg->pk.len  = 0;
+                        assert(data.len > 0 ? data.data != NULL
+                                            : data.data == NULL);
                         result = flow_alloc_reply(msg->flow_id,
                                                   msg->response,
                                                   msg->mpl,
-                                                  msg->pk.data,
-                                                  msg->pk.len);
+                                                  data);
                         break;
                 default:
                         log_err("Don't know that message code.");