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 struct sockaddr_un *addr, 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         size_t pathlen;
 148         int ret;
 149
 150         if (addr != NULL) {
 151                 pathlen = strlen(addr->sun_path)+1;
 152         } else {
 153                 pathlen = 1;
 154         }
 155
 156         ctx = malloc(offsetof(struct unix_dgram_ctx, path) + pathlen);
 157         if (ctx == NULL) {
 158                 return ENOMEM;
 159         }
 160         if (addr != NULL) {
 161                 memcpy(ctx->path, addr->sun_path, pathlen);
 162         } else {
 163                 ctx->path[0] = '\0';
 164         }
 165
 166         ctx->recv_buf = malloc(max_msg);
 167         if (ctx->recv_buf == NULL) {
 168                 free(ctx);
 169                 return ENOMEM;
 170         }
 171         ctx->max_msg = max_msg;
 172         ctx->ev_funcs = ev_funcs;
 173         ctx->recv_callback = recv_callback;
 174         ctx->private_data = private_data;
 175         ctx->sock_read_watch = NULL;
 176         ctx->send_pool = NULL;
 177         ctx->pool_read_watch = NULL;
 178         ctx->send_queues = NULL;
 179         ctx->created_pid = (pid_t)-1;
 180
 181         ctx->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 182         if (ctx->sock == -1) {
 183                 ret = errno;
 184                 goto fail_free;
 185         }
 186
 187         /* Set non-blocking and close-on-exec. */
 188         ret = prepare_socket(ctx->sock);
 189         if (ret != 0) {
 190                 goto fail_close;
 191         }
 192
 193         if (addr != NULL) {
 194                 ret = bind(ctx->sock,
 195                            (const struct sockaddr *)(const void *)addr,
 196                            sizeof(*addr));
 197                 if (ret == -1) {
 198                         ret = errno;
 199                         goto fail_close;
 200                 }
 201
 202                 ctx->created_pid = getpid();
 203
 204                 ctx->sock_read_watch = ctx->ev_funcs->watch_new(
 205                         ctx->ev_funcs, ctx->sock, POLLIN,
 206                         unix_dgram_recv_handler, ctx);
 207
 208                 if (ctx->sock_read_watch == NULL) {
 209                         ret = ENOMEM;
 210                         goto fail_close;
 211                 }
 212         }
 213
 214         *result = ctx;
 215         return 0;
 216
 217 fail_close:
 218         close(ctx->sock);
 219 fail_free:
 220         free(ctx->recv_buf);
 221         free(ctx);
 222         return ret;
 223 }
 224
 225 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
 226                                     void *private_data)
 227 {
 228         struct unix_dgram_ctx *ctx = (struct unix_dgram_ctx *)private_data;
 229         ssize_t received;
 230         struct msghdr msg;
 231         struct iovec iov;
 232
 233         iov = (struct iovec) {
 234                 .iov_base = (void *)ctx->recv_buf,
 235                 .iov_len = ctx->max_msg,
 236         };
 237
 238         msg = (struct msghdr) {
 239                 .msg_iov = &iov,
 240                 .msg_iovlen = 1,
 241 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
 242                 .msg_control = NULL,
 243                 .msg_controllen = 0,
 244 #endif
 245         };
 246
 247         received = recvmsg(fd, &msg, 0);
 248         if (received == -1) {
 249                 if ((errno == EAGAIN) ||
 250 #ifdef EWOULDBLOCK
 251                     (errno == EWOULDBLOCK) ||
 252 #endif
 253                     (errno == EINTR) || (errno == ENOMEM)) {
 254                         /* Not really an error - just try again. */
 255                         return;
 256                 }
 257                 /* Problem with the socket. Set it unreadable. */
 258                 ctx->ev_funcs->watch_update(w, 0);
 259                 return;
 260         }
 261         if (received > ctx->max_msg) {
 262                 /* More than we expected, not for us */
 263                 return;
 264         }
 265         ctx->recv_callback(ctx, ctx->recv_buf, received, ctx->private_data);
 266 }
 267
 268 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 269                                     void *private_data);
 270
 271 static int unix_dgram_init_pthreadpool(struct unix_dgram_ctx *ctx)
 272 {
 273         int ret, signalfd;
 274
 275         if (ctx->send_pool != NULL) {
 276                 return 0;
 277         }
 278
 279         ret = pthreadpool_init(0, &ctx->send_pool);
 280         if (ret != 0) {
 281                 return ret;
 282         }
 283
 284         signalfd = pthreadpool_signal_fd(ctx->send_pool);
 285
 286         ctx->pool_read_watch = ctx->ev_funcs->watch_new(
 287                 ctx->ev_funcs, signalfd, POLLIN,
 288                 unix_dgram_job_finished, ctx);
 289         if (ctx->pool_read_watch == NULL) {
 290                 pthreadpool_destroy(ctx->send_pool);
 291                 ctx->send_pool = NULL;
 292                 return ENOMEM;
 293         }
 294
 295         return 0;
 296 }
 297
 298 static int unix_dgram_send_queue_init(
 299         struct unix_dgram_ctx *ctx, const char *path,
 300         struct unix_dgram_send_queue **result)
 301 {
 302         struct unix_dgram_send_queue *q;
 303         struct sockaddr_un addr = { 0, };
 304         size_t pathlen;
 305         int ret, err;
 306
 307         pathlen = strlen(path)+1;
 308
 309         if (pathlen > sizeof(addr.sun_path)) {
 310                 return ENAMETOOLONG;
 311         }
 312
 313         q = malloc(offsetof(struct unix_dgram_send_queue, path) + pathlen);
 314         if (q == NULL) {
 315                 return ENOMEM;
 316         }
 317         q->ctx = ctx;
 318         q->msgs = NULL;
 319         memcpy(q->path, path, pathlen);
 320
 321         q->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 322         if (q->sock == -1) {
 323                 err = errno;
 324                 goto fail_free;
 325         }
 326
 327         err = prepare_socket_cloexec(q->sock);
 328         if (err != 0) {
 329                 goto fail_close;
 330         }
 331
 332         addr.sun_family = AF_UNIX;
 333         memcpy(addr.sun_path, path, pathlen+1);
 334
 335         do {
 336                 ret = connect(q->sock, (struct sockaddr *)&addr, sizeof(addr));
 337         } while ((ret == -1) && (errno == EINTR));
 338
 339         if (ret == -1) {
 340                 err = errno;
 341                 goto fail_close;
 342         }
 343
 344         err = unix_dgram_init_pthreadpool(ctx);
 345         if (err != 0) {
 346                 goto fail_close;
 347         }
 348
 349         DLIST_ADD(ctx->send_queues, q);
 350
 351         *result = q;
 352         return 0;
 353
 354 fail_close:
 355         close(q->sock);
 356 fail_free:
 357         free(q);
 358         return err;
 359 }
 360
 361 static void unix_dgram_send_queue_free(struct unix_dgram_send_queue *q)
 362 {
 363         struct unix_dgram_ctx *ctx = q->ctx;
 364
 365         while (q->msgs != NULL) {
 366                 struct unix_dgram_msg *msg;
 367                 msg = q->msgs;
 368                 DLIST_REMOVE(q->msgs, msg);
 369                 free(msg);
 370         }
 371         close(q->sock);
 372         DLIST_REMOVE(ctx->send_queues, q);
 373         free(q);
 374 }
 375
 376 static struct unix_dgram_send_queue *find_send_queue(
 377         struct unix_dgram_ctx *ctx, const char *dst_sock)
 378 {
 379         struct unix_dgram_send_queue *s;
 380
 381         for (s = ctx->send_queues; s != NULL; s = s->next) {
 382                 if (strcmp(s->path, dst_sock) == 0) {
 383                         return s;
 384                 }
 385         }
 386         return NULL;
 387 }
 388
 389 static int queue_msg(struct unix_dgram_send_queue *q,
 390                      const struct iovec *iov, int iovlen)
 391 {
 392         struct unix_dgram_msg *msg;
 393         ssize_t buflen;
 394         size_t msglen;
 395         int i;
 396
 397         buflen = iov_buflen(iov, iovlen);
 398         if (buflen == -1) {
 399                 return EINVAL;
 400         }
 401
 402         msglen = offsetof(struct unix_dgram_msg, buf) + buflen;
 403         if ((msglen < buflen) ||
 404             (msglen < offsetof(struct unix_dgram_msg, buf))) {
 405                 /* overflow */
 406                 return EINVAL;
 407         }
 408
 409         msg = malloc(msglen);
 410         if (msg == NULL) {
 411                 return ENOMEM;
 412         }
 413         msg->buflen = buflen;
 414         msg->sock = q->sock;
 415
 416         buflen = 0;
 417         for (i=0; i<iovlen; i++) {
 418                 memcpy(&msg->buf[buflen], iov[i].iov_base, iov[i].iov_len);
 419                 buflen += iov[i].iov_len;
 420         }
 421
 422         DLIST_ADD_END(q->msgs, msg, struct unix_dgram_msg);
 423         return 0;
 424 }
 425
 426 static void unix_dgram_send_job(void *private_data)
 427 {
 428         struct unix_dgram_msg *msg = private_data;
 429
 430         do {
 431                 msg->sent = send(msg->sock, msg->buf, msg->buflen, 0);
 432         } while ((msg->sent == -1) && (errno == EINTR));
 433 }
 434
 435 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 436                                     void *private_data)
 437 {
 438         struct unix_dgram_ctx *ctx = private_data;
 439         struct unix_dgram_send_queue *q;
 440         struct unix_dgram_msg *msg;
 441         int ret, job;
 442
 443         ret = pthreadpool_finished_jobs(ctx->send_pool, &job, 1);
 444         if (ret != 1) {
 445                 return;
 446         }
 447
 448         for (q = ctx->send_queues; q != NULL; q = q->next) {
 449                 if (job == q->sock) {
 450                         break;
 451                 }
 452         }
 453
 454         if (q == NULL) {
 455                 /* Huh? Should not happen */
 456                 return;
 457         }
 458
 459         msg = q->msgs;
 460         DLIST_REMOVE(q->msgs, msg);
 461         free(msg);
 462
 463         if (q->msgs != NULL) {
 464                 ret = pthreadpool_add_job(ctx->send_pool, q->sock,
 465                                           unix_dgram_send_job, q->msgs);
 466                 if (ret == 0) {
 467                         return;
 468                 }
 469         }
 470
 471         unix_dgram_send_queue_free(q);
 472 }
 473
 474 static int unix_dgram_send(struct unix_dgram_ctx *ctx, const char *dst_sock,
 475                            const struct iovec *iov, int iovlen)
 476 {
 477         struct unix_dgram_send_queue *q;
 478         struct sockaddr_un addr = { 0, };
 479         struct msghdr msg;
 480         size_t dst_len;
 481         int ret;
 482
 483         dst_len = strlen(dst_sock);
 484         if (dst_len >= sizeof(addr.sun_path)) {
 485                 return ENAMETOOLONG;
 486         }
 487
 488         /*
 489          * To preserve message ordering, we have to queue a message when
 490          * others are waiting in line already.
 491          */
 492         q = find_send_queue(ctx, dst_sock);
 493         if (q != NULL) {
 494                 return queue_msg(q, iov, iovlen);
 495         }
 496
 497         /*
 498          * Try a cheap nonblocking send
 499          */
 500
 501         addr.sun_family = AF_UNIX;
 502         memcpy(addr.sun_path, dst_sock, dst_len);
 503
 504         msg.msg_name = &addr;
 505         msg.msg_namelen = sizeof(addr);
 506         msg.msg_iov = discard_const_p(struct iovec, iov);
 507         msg.msg_iovlen = iovlen;
 508 #ifdef HAVE_STRUCT_MSGHDR_MSG_CONTROL
 509         msg.msg_control = NULL;
 510         msg.msg_controllen = 0;
 511 #endif
 512         msg.msg_flags = 0;
 513
 514         ret = sendmsg(ctx->sock, &msg, 0);
 515         if (ret >= 0) {
 516                 return 0;
 517         }
 518 #ifdef EWOULDBLOCK
 519         if ((errno != EWOULDBLOCK) && (errno != EAGAIN) && (errno != EINTR)) {
 520 #else
 521         if ((errno != EAGAIN) && (errno != EINTR)) {
 522 #endif
 523                 return errno;
 524         }
 525
 526         ret = unix_dgram_send_queue_init(ctx, dst_sock, &q);
 527         if (ret != 0) {
 528                 return ret;
 529         }
 530         ret = queue_msg(q, iov, iovlen);
 531         if (ret != 0) {
 532                 unix_dgram_send_queue_free(q);
 533                 return ret;
 534         }
 535         ret = pthreadpool_add_job(ctx->send_pool, q->sock,
 536                                   unix_dgram_send_job, q->msgs);
 537         if (ret != 0) {
 538                 unix_dgram_send_queue_free(q);
 539                 return ret;
 540         }
 541         return 0;
 542 }
 543
 544 static int unix_dgram_sock(struct unix_dgram_ctx *ctx)
 545 {
 546         return ctx->sock;
 547 }
 548
 549 static int unix_dgram_free(struct unix_dgram_ctx *ctx)
 550 {
 551         if (ctx->send_queues != NULL) {
 552                 return EBUSY;
 553         }
 554
 555         if (ctx->send_pool != NULL) {
 556                 int ret = pthreadpool_destroy(ctx->send_pool);
 557                 if (ret != 0) {
 558                         return ret;
 559                 }
 560                 ctx->ev_funcs->watch_free(ctx->pool_read_watch);
 561         }
 562
 563         ctx->ev_funcs->watch_free(ctx->sock_read_watch);
 564
 565         if (getpid() == ctx->created_pid) {
 566                 /* If we created it, unlink. Otherwise someone else might
 567                  * still have it open */
 568                 unlink(ctx->path);
 569         }
 570
 571         close(ctx->sock);
 572         free(ctx->recv_buf);
 573         free(ctx);
 574         return 0;
 575 }
 576
 577 /*
 578  * Every message starts with a uint64_t cookie.
 579  *
 580  * A value of 0 indicates a single-fragment message which is complete in
 581  * itself. The data immediately follows the cookie.
 582  *
 583  * Every multi-fragment message has a cookie != 0 and starts with a cookie
 584  * followed by a struct unix_msg_header and then the data. The pid and sock
 585  * fields are used to assure uniqueness on the receiver side.
 586  */
 587
 588 struct unix_msg_hdr {
 589         size_t msglen;
 590         pid_t pid;
 591         int sock;
 592 };
 593
 594 struct unix_msg {
 595         struct unix_msg *prev, *next;
 596         size_t msglen;
 597         size_t received;
 598         pid_t sender_pid;
 599         int sender_sock;
 600         uint64_t cookie;
 601         uint8_t buf[1];
 602 };
 603
 604 struct unix_msg_ctx {
 605         struct unix_dgram_ctx *dgram;
 606         size_t fragment_len;
 607         uint64_t cookie;
 608
 609         void (*recv_callback)(struct unix_msg_ctx *ctx,
 610                               uint8_t *msg, size_t msg_len,
 611                               void *private_data);
 612         void *private_data;
 613
 614         struct unix_msg *msgs;
 615 };
 616
 617 static void unix_msg_recv(struct unix_dgram_ctx *ctx,
 618                           uint8_t *msg, size_t msg_len,
 619                           void *private_data);
 620
 621 int unix_msg_init(const char *path, const struct poll_funcs *ev_funcs,
 622                   size_t fragment_len, uint64_t cookie,
 623                   void (*recv_callback)(struct unix_msg_ctx *ctx,
 624                                         uint8_t *msg, size_t msg_len,
 625                                         void *private_data),
 626                   void *private_data,
 627                   struct unix_msg_ctx **result)
 628 {
 629         struct unix_msg_ctx *ctx;
 630         struct sockaddr_un addr;
 631         struct sockaddr_un *paddr = NULL;
 632         int ret;
 633
 634         ctx = malloc(sizeof(*ctx));
 635         if (ctx == NULL) {
 636                 return ENOMEM;
 637         }
 638
 639         if (path != NULL) {
 640                 size_t pathlen = strlen(path)+1;
 641
 642                 if (pathlen > sizeof(addr.sun_path)) {
 643                         return ENAMETOOLONG;
 644                 }
 645                 addr = (struct sockaddr_un) { .sun_family = AF_UNIX };
 646                 memcpy(addr.sun_path, path, pathlen);
 647                 paddr = &addr;
 648         }
 649
 650         ret = unix_dgram_init(paddr, fragment_len, ev_funcs,
 651                               unix_msg_recv, ctx, &ctx->dgram);
 652         if (ret != 0) {
 653                 free(ctx);
 654                 return ret;
 655         }
 656
 657         ctx->fragment_len = fragment_len;
 658         ctx->cookie = cookie;
 659         ctx->recv_callback = recv_callback;
 660         ctx->private_data = private_data;
 661         ctx->msgs = NULL;
 662
 663         *result = ctx;
 664         return 0;
 665 }
 666
 667 int unix_msg_send(struct unix_msg_ctx *ctx, const char *dst_sock,
 668                   const struct iovec *iov, int iovlen)
 669 {
 670         ssize_t msglen;
 671         size_t sent;
 672         int ret = 0;
 673         struct iovec *iov_copy;
 674         struct unix_msg_hdr hdr;
 675         struct iovec src_iov;
 676
 677         if (iovlen < 0) {
 678                 return EINVAL;
 679         }
 680
 681         msglen = iov_buflen(iov, iovlen);
 682         if (msglen == -1) {
 683                 return EINVAL;
 684         }
 685
 686         if (msglen <= (ctx->fragment_len - sizeof(uint64_t))) {
 687                 struct iovec tmp_iov[iovlen+1];
 688                 uint64_t cookie = 0;
 689
 690                 tmp_iov[0].iov_base = &cookie;
 691                 tmp_iov[0].iov_len = sizeof(cookie);
 692                 if (iovlen > 0) {
 693                         memcpy(&tmp_iov[1], iov,
 694                                sizeof(struct iovec) * iovlen);
 695                 }
 696
 697                 return unix_dgram_send(ctx->dgram, dst_sock, tmp_iov,
 698                                        iovlen+1);
 699         }
 700
 701         hdr.msglen = msglen;
 702         hdr.pid = getpid();
 703         hdr.sock = unix_dgram_sock(ctx->dgram);
 704
 705         iov_copy = malloc(sizeof(struct iovec) * (iovlen + 2));
 706         if (iov_copy == NULL) {
 707                 return ENOMEM;
 708         }
 709         iov_copy[0].iov_base = &ctx->cookie;
 710         iov_copy[0].iov_len = sizeof(ctx->cookie);
 711         iov_copy[1].iov_base = &hdr;
 712         iov_copy[1].iov_len = sizeof(hdr);
 713
 714         sent = 0;
 715         src_iov = iov[0];
 716
 717         /*
 718          * The following write loop sends the user message in pieces. We have
 719          * filled the first two iovecs above with "cookie" and "hdr". In the
 720          * following loops we pull message chunks from the user iov array and
 721          * fill iov_copy piece by piece, possibly truncating chunks from the
 722          * caller's iov array. Ugly, but hopefully efficient.
 723          */
 724
 725         while (sent < msglen) {
 726                 size_t fragment_len;
 727                 size_t iov_index = 2;
 728
 729                 fragment_len = sizeof(ctx->cookie) + sizeof(hdr);
 730
 731                 while (fragment_len < ctx->fragment_len) {
 732                         size_t space, chunk;
 733
 734                         space = ctx->fragment_len - fragment_len;
 735                         chunk = MIN(space, src_iov.iov_len);
 736
 737                         iov_copy[iov_index].iov_base = src_iov.iov_base;
 738                         iov_copy[iov_index].iov_len = chunk;
 739                         iov_index += 1;
 740
 741                         src_iov.iov_base = (char *)src_iov.iov_base + chunk;
 742                         src_iov.iov_len -= chunk;
 743                         fragment_len += chunk;
 744
 745                         if (src_iov.iov_len == 0) {
 746                                 iov += 1;
 747                                 iovlen -= 1;
 748                                 if (iovlen == 0) {
 749                                         break;
 750                                 }
 751                                 src_iov = iov[0];
 752                         }
 753                 }
 754                 sent += (fragment_len - sizeof(ctx->cookie) - sizeof(hdr));
 755
 756                 ret = unix_dgram_send(ctx->dgram, dst_sock,
 757                                       iov_copy, iov_index);
 758                 if (ret != 0) {
 759                         break;
 760                 }
 761         }
 762
 763         free(iov_copy);
 764
 765         ctx->cookie += 1;
 766         if (ctx->cookie == 0) {
 767                 ctx->cookie += 1;
 768         }
 769
 770         return ret;
 771 }
 772
 773 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
 774                           uint8_t *buf, size_t buflen,
 775                           void *private_data)
 776 {
 777         struct unix_msg_ctx *ctx = (struct unix_msg_ctx *)private_data;
 778         struct unix_msg_hdr hdr;
 779         struct unix_msg *msg;
 780         size_t space;
 781         uint64_t cookie;
 782
 783         if (buflen < sizeof(cookie)) {
 784                 return;
 785         }
 786         memcpy(&cookie, buf, sizeof(cookie));
 787
 788         buf += sizeof(cookie);
 789         buflen -= sizeof(cookie);
 790
 791         if (cookie == 0) {
 792                 ctx->recv_callback(ctx, buf, buflen, ctx->private_data);
 793                 return;
 794         }
 795
 796         if (buflen < sizeof(hdr)) {
 797                 return;
 798         }
 799         memcpy(&hdr, buf, sizeof(hdr));
 800
 801         buf += sizeof(hdr);
 802         buflen -= sizeof(hdr);
 803
 804         for (msg = ctx->msgs; msg != NULL; msg = msg->next) {
 805                 if ((msg->sender_pid == hdr.pid) &&
 806                     (msg->sender_sock == hdr.sock)) {
 807                         break;
 808                 }
 809         }
 810
 811         if ((msg != NULL) && (msg->cookie != cookie)) {
 812                 DLIST_REMOVE(ctx->msgs, msg);
 813                 free(msg);
 814                 msg = NULL;
 815         }
 816
 817         if (msg == NULL) {
 818                 msg = malloc(offsetof(struct unix_msg, buf) + hdr.msglen);
 819                 if (msg == NULL) {
 820                         return;
 821                 }
 822                 msg->msglen = hdr.msglen;
 823                 msg->received = 0;
 824                 msg->sender_pid = hdr.pid;
 825                 msg->sender_sock = hdr.sock;
 826                 msg->cookie = cookie;
 827                 DLIST_ADD(ctx->msgs, msg);
 828         }
 829
 830         space = msg->msglen - msg->received;
 831         if (buflen > space) {
 832                 return;
 833         }
 834
 835         memcpy(msg->buf + msg->received, buf, buflen);
 836         msg->received += buflen;
 837
 838         if (msg->received < msg->msglen) {
 839                 return;
 840         }
 841
 842         DLIST_REMOVE(ctx->msgs, msg);
 843         ctx->recv_callback(ctx, msg->buf, msg->msglen, ctx->private_data);
 844         free(msg);
 845 }
 846
 847 int unix_msg_free(struct unix_msg_ctx *ctx)
 848 {
 849         int ret;
 850
 851         ret = unix_dgram_free(ctx->dgram);
 852         if (ret != 0) {
 853                 return ret;
 854         }
 855
 856         while (ctx->msgs != NULL) {
 857                 struct unix_msg *msg = ctx->msgs;
 858                 DLIST_REMOVE(ctx->msgs, msg);
 859                 free(msg);
 860         }
 861
 862         free(ctx);
 863         return 0;
 864 }
 865
 866 static ssize_t iov_buflen(const struct iovec *iov, int iovlen)
 867 {
 868         size_t buflen = 0;
 869         int i;
 870
 871         for (i=0; i<iovlen; i++) {
 872                 size_t thislen = iov[i].iov_len;
 873                 size_t tmp = buflen + thislen;
 874
 875                 if ((tmp < buflen) || (tmp < thislen)) {
 876                         /* overflow */
 877                         return -1;
 878                 }
 879                 buflen = tmp;
 880         }
 881         return buflen;
 882 }