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