source3/lib/unix_msg/unix_msg.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * Copyright (C) Volker Lendecke 2013
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 3 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include "replace.h"
  20 #include "unix_msg.h"
  21 #include "system/select.h"
  22 #include "system/time.h"
  23 #include "system/network.h"
  24 #include "dlinklist.h"
  25 #include "pthreadpool/pthreadpool.h"
  26 #include <fcntl.h>
  27
  28 /*
  29  * This file implements two abstractions: The "unix_dgram" functions implement
  30  * queueing for unix domain datagram sockets. You can send to a destination
  31  * socket, and if that has no free space available, it will fall back to an
  32  * anonymous socket that will poll for writability. "unix_dgram" expects the
  33  * data size not to exceed the system limit.
  34  *
  35  * The "unix_msg" functions implement the fragmentation of large messages on
  36  * top of "unix_dgram". This is what is exposed to the user of this API.
  37  */
  38
  39 struct unix_dgram_msg {
  40         struct unix_dgram_msg *prev, *next;
  41
  42         int sock;
  43         ssize_t sent;
  44         int sys_errno;
  45         size_t buflen;
  46         uint8_t buf[1];
  47 };
  48
  49 struct unix_dgram_send_queue {
  50         struct unix_dgram_send_queue *prev, *next;
  51         struct unix_dgram_ctx *ctx;
  52         int sock;
  53         struct unix_dgram_msg *msgs;
  54         char path[1];
  55 };
  56
  57 struct unix_dgram_ctx {
  58         int sock;
  59         pid_t created_pid;
  60         const struct poll_funcs *ev_funcs;
  61         size_t max_msg;
  62
  63         void (*recv_callback)(struct unix_dgram_ctx *ctx,
  64                               uint8_t *msg, size_t msg_len,
  65                               void *private_data);
  66         void *private_data;
  67
  68         struct poll_watch *sock_read_watch;
  69         struct unix_dgram_send_queue *send_queues;
  70
  71         struct pthreadpool *send_pool;
  72         struct poll_watch *pool_read_watch;
  73
  74         uint8_t *recv_buf;
  75         char path[1];
  76 };
  77
  78 static ssize_t iov_buflen(const struct iovec *iov, int iovlen);
  79 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
  80                                     void *private_data);
  81
  82 /* Set socket non blocking. */
  83 static int prepare_socket_nonblock(int sock)
  84 {
  85         int flags;
  86 #ifdef O_NONBLOCK
  87 #define FLAG_TO_SET O_NONBLOCK
  88 #else
  89 #ifdef SYSV
  90 #define FLAG_TO_SET O_NDELAY
  91 #else /* BSD */
  92 #define FLAG_TO_SET FNDELAY
  93 #endif
  94 #endif
  95
  96         flags = fcntl(sock, F_GETFL);
  97         if (flags == -1) {
  98                 return errno;
  99         }
 100         flags |= FLAG_TO_SET;
 101         if (fcntl(sock, F_SETFL, flags) == -1) {
 102                 return errno;
 103         }
 104
 105 #undef FLAG_TO_SET
 106         return 0;
 107 }
 108
 109 /* Set socket close on exec. */
 110 static int prepare_socket_cloexec(int sock)
 111 {
 112 #ifdef FD_CLOEXEC
 113         int flags;
 114
 115         flags = fcntl(sock, F_GETFD, 0);
 116         if (flags == -1) {
 117                 return errno;
 118         }
 119         flags |= FD_CLOEXEC;
 120         if (fcntl(sock, F_SETFD, flags) == -1) {
 121                 return errno;
 122         }
 123 #endif
 124         return 0;
 125 }
 126
 127 /* Set socket non blocking and close on exec. */
 128 static int prepare_socket(int sock)
 129 {
 130         int ret = prepare_socket_nonblock(sock);
 131
 132         if (ret) {
 133                 return ret;
 134         }
 135         return prepare_socket_cloexec(sock);
 136 }
 137
 138 static int unix_dgram_init(const char *path, size_t max_msg,
 139                            const struct poll_funcs *ev_funcs,
 140                            void (*recv_callback)(struct unix_dgram_ctx *ctx,
 141                                                  uint8_t *msg, size_t msg_len,
 142                                                  void *private_data),
 143                            void *private_data,
 144                            struct unix_dgram_ctx **result)
 145 {
 146         struct unix_dgram_ctx *ctx;
 147         struct sockaddr_un addr = { 0, };
 148         size_t pathlen;
 149         int ret;
 150
 151         if (path != NULL) {
 152                 pathlen = strlen(path)+1;
 153                 if (pathlen > sizeof(addr.sun_path)) {
 154                         return ENAMETOOLONG;
 155                 }
 156         } else {
 157                 pathlen = 1;
 158         }
 159
 160         ctx = malloc(offsetof(struct unix_dgram_ctx, path) + pathlen);
 161         if (ctx == NULL) {
 162                 return ENOMEM;
 163         }
 164         if (path != NULL) {
 165                 memcpy(ctx->path, path, pathlen);
 166         } else {
 167                 ctx->path[0] = '\0';
 168         }
 169
 170         ctx->recv_buf = malloc(max_msg);
 171         if (ctx->recv_buf == NULL) {
 172                 free(ctx);
 173                 return ENOMEM;
 174         }
 175         ctx->max_msg = max_msg;
 176         ctx->ev_funcs = ev_funcs;
 177         ctx->recv_callback = recv_callback;
 178         ctx->private_data = private_data;
 179         ctx->sock_read_watch = NULL;
 180         ctx->send_pool = NULL;
 181         ctx->pool_read_watch = NULL;
 182         ctx->send_queues = NULL;
 183         ctx->created_pid = (pid_t)-1;
 184
 185         ctx->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 186         if (ctx->sock == -1) {
 187                 ret = errno;
 188                 goto fail_free;
 189         }
 190
 191         /* Set non-blocking and close-on-exec. */
 192         ret = prepare_socket(ctx->sock);
 193         if (ret != 0) {
 194                 goto fail_close;
 195         }
 196
 197         if (path != NULL) {
 198                 addr.sun_family = AF_UNIX;
 199                 memcpy(addr.sun_path, path, pathlen);
 200
 201                 ret = bind(ctx->sock, (struct sockaddr *)(void *)&addr,
 202                                 sizeof(addr));
 203                 if (ret == -1) {
 204                         ret = errno;
 205                         goto fail_close;
 206                 }
 207
 208                 ctx->created_pid = getpid();
 209
 210                 ctx->sock_read_watch = ctx->ev_funcs->watch_new(
 211                         ctx->ev_funcs, ctx->sock, POLLIN,
 212                         unix_dgram_recv_handler, ctx);
 213
 214                 if (ctx->sock_read_watch == NULL) {
 215                         ret = ENOMEM;
 216                         goto fail_close;
 217                 }
 218         }
 219
 220         *result = ctx;
 221         return 0;
 222
 223 fail_close:
 224         close(ctx->sock);
 225 fail_free:
 226         free(ctx->recv_buf);
 227         free(ctx);
 228         return ret;
 229 }
 230
 231 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
 232                                     void *private_data)
 233 {
 234         struct unix_dgram_ctx *ctx = (struct unix_dgram_ctx *)private_data;
 235         ssize_t received;
 236
 237         received = recv(fd, ctx->recv_buf, ctx->max_msg, 0);
 238         if (received == -1) {
 239                 if ((errno == EAGAIN) ||
 240 #ifdef EWOULDBLOCK
 241                     (errno == EWOULDBLOCK) ||
 242 #endif
 243                     (errno == EINTR) || (errno == ENOMEM)) {
 244                         /* Not really an error - just try again. */
 245                         return;
 246                 }
 247                 /* Problem with the socket. Set it unreadable. */
 248                 ctx->ev_funcs->watch_update(w, 0);
 249                 return;
 250         }
 251         if (received > ctx->max_msg) {
 252                 /* More than we expected, not for us */
 253                 return;
 254         }
 255         ctx->recv_callback(ctx, ctx->recv_buf, received, ctx->private_data);
 256 }
 257
 258 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 259                                     void *private_data);
 260
 261 static int unix_dgram_init_pthreadpool(struct unix_dgram_ctx *ctx)
 262 {
 263         int ret, signalfd;
 264
 265         if (ctx->send_pool != NULL) {
 266                 return 0;
 267         }
 268
 269         ret = pthreadpool_init(0, &ctx->send_pool);
 270         if (ret != 0) {
 271                 return ret;
 272         }
 273
 274         signalfd = pthreadpool_signal_fd(ctx->send_pool);
 275
 276         ctx->pool_read_watch = ctx->ev_funcs->watch_new(
 277                 ctx->ev_funcs, signalfd, POLLIN,
 278                 unix_dgram_job_finished, ctx);
 279         if (ctx->pool_read_watch == NULL) {
 280                 pthreadpool_destroy(ctx->send_pool);
 281                 ctx->send_pool = NULL;
 282                 return ENOMEM;
 283         }
 284
 285         return 0;
 286 }
 287
 288 static int unix_dgram_send_queue_init(
 289         struct unix_dgram_ctx *ctx, const char *path,
 290         struct unix_dgram_send_queue **result)
 291 {
 292         struct unix_dgram_send_queue *q;
 293         struct sockaddr_un addr = { 0, };
 294         size_t pathlen;
 295         int ret, err;
 296
 297         pathlen = strlen(path)+1;
 298
 299         if (pathlen > sizeof(addr.sun_path)) {
 300                 return ENAMETOOLONG;
 301         }
 302
 303         q = malloc(offsetof(struct unix_dgram_send_queue, path) + pathlen);
 304         if (q == NULL) {
 305                 return ENOMEM;
 306         }
 307         q->ctx = ctx;
 308         q->msgs = NULL;
 309         memcpy(q->path, path, pathlen);
 310
 311         q->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 312         if (q->sock == -1) {
 313                 err = errno;
 314                 goto fail_free;
 315         }
 316
 317         err = prepare_socket_cloexec(q->sock);
 318         if (err != 0) {
 319                 goto fail_close;
 320         }
 321
 322         addr.sun_family = AF_UNIX;
 323         memcpy(addr.sun_path, path, pathlen+1);
 324
 325         do {
 326                 ret = connect(q->sock, (struct sockaddr *)&addr, sizeof(addr));
 327         } while ((ret == -1) && (errno == EINTR));
 328
 329         if (ret == -1) {
 330                 err = errno;
 331                 goto fail_close;
 332         }
 333
 334         err = unix_dgram_init_pthreadpool(ctx);
 335         if (err != 0) {
 336                 goto fail_close;
 337         }
 338
 339         DLIST_ADD(ctx->send_queues, q);
 340
 341         *result = q;
 342         return 0;
 343
 344 fail_close:
 345         close(q->sock);
 346 fail_free:
 347         free(q);
 348         return err;
 349 }
 350
 351 static void unix_dgram_send_queue_free(struct unix_dgram_send_queue *q)
 352 {
 353         struct unix_dgram_ctx *ctx = q->ctx;
 354
 355         while (q->msgs != NULL) {
 356                 struct unix_dgram_msg *msg;
 357                 msg = q->msgs;
 358                 DLIST_REMOVE(q->msgs, msg);
 359                 free(msg);
 360         }
 361         close(q->sock);
 362         DLIST_REMOVE(ctx->send_queues, q);
 363         free(q);
 364 }
 365
 366 static struct unix_dgram_send_queue *find_send_queue(
 367         struct unix_dgram_ctx *ctx, const char *dst_sock)
 368 {
 369         struct unix_dgram_send_queue *s;
 370
 371         for (s = ctx->send_queues; s != NULL; s = s->next) {
 372                 if (strcmp(s->path, dst_sock) == 0) {
 373                         return s;
 374                 }
 375         }
 376         return NULL;
 377 }
 378
 379 static int queue_msg(struct unix_dgram_send_queue *q,
 380                      const struct iovec *iov, int iovlen)
 381 {
 382         struct unix_dgram_msg *msg;
 383         ssize_t buflen;
 384         size_t msglen;
 385         int i;
 386
 387         buflen = iov_buflen(iov, iovlen);
 388         if (buflen == -1) {
 389                 return EINVAL;
 390         }
 391
 392         msglen = offsetof(struct unix_dgram_msg, buf) + buflen;
 393         if ((msglen < buflen) ||
 394             (msglen < offsetof(struct unix_dgram_msg, buf))) {
 395                 /* overflow */
 396                 return EINVAL;
 397         }
 398
 399         msg = malloc(msglen);
 400         if (msg == NULL) {
 401                 return ENOMEM;
 402         }
 403         msg->buflen = buflen;
 404         msg->sock = q->sock;
 405
 406         buflen = 0;
 407         for (i=0; i<iovlen; i++) {
 408                 memcpy(&msg->buf[buflen], iov[i].iov_base, iov[i].iov_len);
 409                 buflen += iov[i].iov_len;
 410         }
 411
 412         DLIST_ADD_END(q->msgs, msg, struct unix_dgram_msg);
 413         return 0;
 414 }
 415
 416 static void unix_dgram_send_job(void *private_data)
 417 {
 418         struct unix_dgram_msg *msg = private_data;
 419
 420         do {
 421                 msg->sent = send(msg->sock, msg->buf, msg->buflen, 0);
 422         } while ((msg->sent == -1) && (errno == EINTR));
 423 }
 424
 425 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 426                                     void *private_data)
 427 {
 428         struct unix_dgram_ctx *ctx = private_data;
 429         struct unix_dgram_send_queue *q;
 430         struct unix_dgram_msg *msg;
 431         int ret, job;
 432
 433         ret = pthreadpool_finished_jobs(ctx->send_pool, &job, 1);
 434         if (ret != 1) {
 435                 return;
 436         }
 437
 438         for (q = ctx->send_queues; q != NULL; q = q->next) {
 439                 if (job == q->sock) {
 440                         break;
 441                 }
 442         }
 443
 444         if (q == NULL) {
 445                 /* Huh? Should not happen */
 446                 return;
 447         }
 448
 449         msg = q->msgs;
 450         DLIST_REMOVE(q->msgs, msg);
 451         free(msg);
 452
 453         if (q->msgs != NULL) {
 454                 ret = pthreadpool_add_job(ctx->send_pool, q->sock,
 455                                           unix_dgram_send_job, q->msgs);
 456                 if (ret == 0) {
 457                         return;
 458                 }
 459         }
 460
 461         unix_dgram_send_queue_free(q);
 462 }
 463
 464 static int unix_dgram_send(struct unix_dgram_ctx *ctx, const char *dst_sock,
 465                            const struct iovec *iov, int iovlen)
 466 {
 467         struct unix_dgram_send_queue *q;
 468         struct sockaddr_un addr = { 0, };
 469         struct msghdr msg;
 470         size_t dst_len;
 471         int ret;
 472
 473         dst_len = strlen(dst_sock);
 474         if (dst_len >= sizeof(addr.sun_path)) {
 475                 return ENAMETOOLONG;
 476         }
 477
 478         /*
 479          * To preserve message ordering, we have to queue a message when
 480          * others are waiting in line already.
 481          */
 482         q = find_send_queue(ctx, dst_sock);
 483         if (q != NULL) {
 484                 return queue_msg(q, iov, iovlen);
 485         }
 486
 487         /*
 488          * Try a cheap nonblocking send
 489          */
 490
 491         addr.sun_family = AF_UNIX;
 492         memcpy(addr.sun_path, dst_sock, dst_len);
 493
 494         msg.msg_name = &addr;
 495         msg.msg_namelen = sizeof(addr);
 496         msg.msg_iov = discard_const_p(struct iovec, iov);
 497         msg.msg_iovlen = iovlen;
 498         msg.msg_control = NULL;
 499         msg.msg_controllen = 0;
 500         msg.msg_flags = 0;
 501
 502         ret = sendmsg(ctx->sock, &msg, 0);
 503         if (ret >= 0) {
 504                 return 0;
 505         }
 506 #ifdef EWOULDBLOCK
 507         if ((errno != EWOULDBLOCK) && (errno != EAGAIN) && (errno != EINTR)) {
 508 #else
 509         if ((errno != EAGAIN) && (errno != EINTR)) {
 510 #endif
 511                 return errno;
 512         }
 513
 514         ret = unix_dgram_send_queue_init(ctx, dst_sock, &q);
 515         if (ret != 0) {
 516                 return ret;
 517         }
 518         ret = queue_msg(q, iov, iovlen);
 519         if (ret != 0) {
 520                 unix_dgram_send_queue_free(q);
 521                 return ret;
 522         }
 523         ret = pthreadpool_add_job(ctx->send_pool, q->sock,
 524                                   unix_dgram_send_job, q->msgs);
 525         if (ret != 0) {
 526                 unix_dgram_send_queue_free(q);
 527                 return ret;
 528         }
 529         return 0;
 530 }
 531
 532 static int unix_dgram_sock(struct unix_dgram_ctx *ctx)
 533 {
 534         return ctx->sock;
 535 }
 536
 537 static int unix_dgram_free(struct unix_dgram_ctx *ctx)
 538 {
 539         if (ctx->send_queues != NULL) {
 540                 return EBUSY;
 541         }
 542
 543         if (ctx->send_pool != NULL) {
 544                 int ret = pthreadpool_destroy(ctx->send_pool);
 545                 if (ret != 0) {
 546                         return ret;
 547                 }
 548                 ctx->ev_funcs->watch_free(ctx->pool_read_watch);
 549         }
 550
 551         ctx->ev_funcs->watch_free(ctx->sock_read_watch);
 552
 553         if (getpid() == ctx->created_pid) {
 554                 /* If we created it, unlink. Otherwise someone else might
 555                  * still have it open */
 556                 unlink(ctx->path);
 557         }
 558
 559         close(ctx->sock);
 560         free(ctx->recv_buf);
 561         free(ctx);
 562         return 0;
 563 }
 564
 565 /*
 566  * Every message starts with a uint64_t cookie.
 567  *
 568  * A value of 0 indicates a single-fragment message which is complete in
 569  * itself. The data immediately follows the cookie.
 570  *
 571  * Every multi-fragment message has a cookie != 0 and starts with a cookie
 572  * followed by a struct unix_msg_header and then the data. The pid and sock
 573  * fields are used to assure uniqueness on the receiver side.
 574  */
 575
 576 struct unix_msg_hdr {
 577         size_t msglen;
 578         pid_t pid;
 579         int sock;
 580 };
 581
 582 struct unix_msg {
 583         struct unix_msg *prev, *next;
 584         size_t msglen;
 585         size_t received;
 586         pid_t sender_pid;
 587         int sender_sock;
 588         uint64_t cookie;
 589         uint8_t buf[1];
 590 };
 591
 592 struct unix_msg_ctx {
 593         struct unix_dgram_ctx *dgram;
 594         size_t fragment_len;
 595         uint64_t cookie;
 596
 597         void (*recv_callback)(struct unix_msg_ctx *ctx,
 598                               uint8_t *msg, size_t msg_len,
 599                               void *private_data);
 600         void *private_data;
 601
 602         struct unix_msg *msgs;
 603 };
 604
 605 static void unix_msg_recv(struct unix_dgram_ctx *ctx,
 606                           uint8_t *msg, size_t msg_len,
 607                           void *private_data);
 608
 609 int unix_msg_init(const char *path, const struct poll_funcs *ev_funcs,
 610                   size_t fragment_len, uint64_t cookie,
 611                   void (*recv_callback)(struct unix_msg_ctx *ctx,
 612                                         uint8_t *msg, size_t msg_len,
 613                                         void *private_data),
 614                   void *private_data,
 615                   struct unix_msg_ctx **result)
 616 {
 617         struct unix_msg_ctx *ctx;
 618         int ret;
 619
 620         ctx = malloc(sizeof(*ctx));
 621         if (ctx == NULL) {
 622                 return ENOMEM;
 623         }
 624
 625         ret = unix_dgram_init(path, fragment_len, ev_funcs,
 626                               unix_msg_recv, ctx, &ctx->dgram);
 627         if (ret != 0) {
 628                 free(ctx);
 629                 return ret;
 630         }
 631
 632         ctx->fragment_len = fragment_len;
 633         ctx->cookie = cookie;
 634         ctx->recv_callback = recv_callback;
 635         ctx->private_data = private_data;
 636         ctx->msgs = NULL;
 637
 638         *result = ctx;
 639         return 0;
 640 }
 641
 642 int unix_msg_send(struct unix_msg_ctx *ctx, const char *dst_sock,
 643                   const struct iovec *iov, int iovlen)
 644 {
 645         ssize_t msglen;
 646         size_t sent;
 647         int ret = 0;
 648         struct iovec *iov_copy;
 649         struct unix_msg_hdr hdr;
 650         struct iovec src_iov;
 651
 652         if (iovlen < 0) {
 653                 return EINVAL;
 654         }
 655
 656         msglen = iov_buflen(iov, iovlen);
 657         if (msglen == -1) {
 658                 return EINVAL;
 659         }
 660
 661         if ((iovlen < 16) &&
 662             (msglen <= (ctx->fragment_len - sizeof(uint64_t)))) {
 663                 struct iovec tmp_iov[16];
 664                 uint64_t cookie = 0;
 665
 666                 tmp_iov[0].iov_base = &cookie;
 667                 tmp_iov[0].iov_len = sizeof(cookie);
 668                 if (iovlen > 0) {
 669                         memcpy(&tmp_iov[1], iov,
 670                                sizeof(struct iovec) * iovlen);
 671                 }
 672
 673                 return unix_dgram_send(ctx->dgram, dst_sock, tmp_iov,
 674                                        iovlen+1);
 675         }
 676
 677         hdr.msglen = msglen;
 678         hdr.pid = getpid();
 679         hdr.sock = unix_dgram_sock(ctx->dgram);
 680
 681         iov_copy = malloc(sizeof(struct iovec) * (iovlen + 2));
 682         if (iov_copy == NULL) {
 683                 return ENOMEM;
 684         }
 685         iov_copy[0].iov_base = &ctx->cookie;
 686         iov_copy[0].iov_len = sizeof(ctx->cookie);
 687         iov_copy[1].iov_base = &hdr;
 688         iov_copy[1].iov_len = sizeof(hdr);
 689
 690         sent = 0;
 691         src_iov = iov[0];
 692
 693         /*
 694          * The following write loop sends the user message in pieces. We have
 695          * filled the first two iovecs above with "cookie" and "hdr". In the
 696          * following loops we pull message chunks from the user iov array and
 697          * fill iov_copy piece by piece, possibly truncating chunks from the
 698          * caller's iov array. Ugly, but hopefully efficient.
 699          */
 700
 701         while (sent < msglen) {
 702                 size_t fragment_len;
 703                 size_t iov_index = 2;
 704
 705                 fragment_len = sizeof(ctx->cookie) + sizeof(hdr);
 706
 707                 while (fragment_len < ctx->fragment_len) {
 708                         size_t space, chunk;
 709
 710                         space = ctx->fragment_len - fragment_len;
 711                         chunk = MIN(space, src_iov.iov_len);
 712
 713                         iov_copy[iov_index].iov_base = src_iov.iov_base;
 714                         iov_copy[iov_index].iov_len = chunk;
 715                         iov_index += 1;
 716
 717                         src_iov.iov_base = (char *)src_iov.iov_base + chunk;
 718                         src_iov.iov_len -= chunk;
 719                         fragment_len += chunk;
 720
 721                         if (src_iov.iov_len == 0) {
 722                                 iov += 1;
 723                                 iovlen -= 1;
 724                                 if (iovlen == 0) {
 725                                         break;
 726                                 }
 727                                 src_iov = iov[0];
 728                         }
 729                 }
 730                 sent += (fragment_len - sizeof(ctx->cookie) - sizeof(hdr));
 731
 732                 ret = unix_dgram_send(ctx->dgram, dst_sock,
 733                                       iov_copy, iov_index);
 734                 if (ret != 0) {
 735                         break;
 736                 }
 737         }
 738
 739         free(iov_copy);
 740
 741         ctx->cookie += 1;
 742         if (ctx->cookie == 0) {
 743                 ctx->cookie += 1;
 744         }
 745
 746         return ret;
 747 }
 748
 749 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
 750                           uint8_t *buf, size_t buflen,
 751                           void *private_data)
 752 {
 753         struct unix_msg_ctx *ctx = (struct unix_msg_ctx *)private_data;
 754         struct unix_msg_hdr hdr;
 755         struct unix_msg *msg;
 756         size_t space;
 757         uint64_t cookie;
 758
 759         if (buflen < sizeof(cookie)) {
 760                 return;
 761         }
 762         memcpy(&cookie, buf, sizeof(cookie));
 763
 764         buf += sizeof(cookie);
 765         buflen -= sizeof(cookie);
 766
 767         if (cookie == 0) {
 768                 ctx->recv_callback(ctx, buf, buflen, ctx->private_data);
 769                 return;
 770         }
 771
 772         if (buflen < sizeof(hdr)) {
 773                 return;
 774         }
 775         memcpy(&hdr, buf, sizeof(hdr));
 776
 777         buf += sizeof(hdr);
 778         buflen -= sizeof(hdr);
 779
 780         for (msg = ctx->msgs; msg != NULL; msg = msg->next) {
 781                 if ((msg->sender_pid == hdr.pid) &&
 782                     (msg->sender_sock == hdr.sock)) {
 783                         break;
 784                 }
 785         }
 786
 787         if ((msg != NULL) && (msg->cookie != cookie)) {
 788                 DLIST_REMOVE(ctx->msgs, msg);
 789                 free(msg);
 790                 msg = NULL;
 791         }
 792
 793         if (msg == NULL) {
 794                 msg = malloc(offsetof(struct unix_msg, buf) + hdr.msglen);
 795                 if (msg == NULL) {
 796                         return;
 797                 }
 798                 msg->msglen = hdr.msglen;
 799                 msg->received = 0;
 800                 msg->sender_pid = hdr.pid;
 801                 msg->sender_sock = hdr.sock;
 802                 msg->cookie = cookie;
 803                 DLIST_ADD(ctx->msgs, msg);
 804         }
 805
 806         space = msg->msglen - msg->received;
 807         if (buflen > space) {
 808                 return;
 809         }
 810
 811         memcpy(msg->buf + msg->received, buf, buflen);
 812         msg->received += buflen;
 813
 814         if (msg->received < msg->msglen) {
 815                 return;
 816         }
 817
 818         DLIST_REMOVE(ctx->msgs, msg);
 819         ctx->recv_callback(ctx, msg->buf, msg->msglen, ctx->private_data);
 820         free(msg);
 821 }
 822
 823 int unix_msg_free(struct unix_msg_ctx *ctx)
 824 {
 825         int ret;
 826
 827         ret = unix_dgram_free(ctx->dgram);
 828         if (ret != 0) {
 829                 return ret;
 830         }
 831
 832         while (ctx->msgs != NULL) {
 833                 struct unix_msg *msg = ctx->msgs;
 834                 DLIST_REMOVE(ctx->msgs, msg);
 835                 free(msg);
 836         }
 837
 838         free(ctx);
 839         return 0;
 840 }
 841
 842 static ssize_t iov_buflen(const struct iovec *iov, int iovlen)
 843 {
 844         size_t buflen = 0;
 845         int i;
 846
 847         for (i=0; i<iovlen; i++) {
 848                 size_t thislen = iov[i].iov_len;
 849                 size_t tmp = buflen + thislen;
 850
 851                 if ((tmp < buflen) || (tmp < thislen)) {
 852                         /* overflow */
 853                         return -1;
 854                 }
 855                 buflen = tmp;
 856         }
 857         return buflen;
 858 }