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 "lib/util/dlinklist.h"
  25 #include "pthreadpool/pthreadpool_pipe.h"
  26 #include "lib/util/iov_buf.h"
  27 #include "lib/util/msghdr.h"
  28 #include <fcntl.h>
  29
  30 /*
  31  * This file implements two abstractions: The "unix_dgram" functions implement
  32  * queueing for unix domain datagram sockets. You can send to a destination
  33  * socket, and if that has no free space available, it will fall back to an
  34  * anonymous socket that will poll for writability. "unix_dgram" expects the
  35  * data size not to exceed the system limit.
  36  *
  37  * The "unix_msg" functions implement the fragmentation of large messages on
  38  * top of "unix_dgram". This is what is exposed to the user of this API.
  39  */
  40
  41 struct unix_dgram_msg {
  42         struct unix_dgram_msg *prev, *next;
  43
  44         int sock;
  45         ssize_t sent;
  46         int sys_errno;
  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[];
  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                               int *fds, size_t num_fds,
  66                               void *private_data);
  67         void *private_data;
  68
  69         struct poll_watch *sock_read_watch;
  70         struct unix_dgram_send_queue *send_queues;
  71
  72         struct pthreadpool_pipe *send_pool;
  73         struct poll_watch *pool_read_watch;
  74
  75         uint8_t *recv_buf;
  76         char path[];
  77 };
  78
  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, bool nonblock)
  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         if (nonblock) {
 101                 flags |= FLAG_TO_SET;
 102         } else {
 103                 flags &= ~FLAG_TO_SET;
 104         }
 105         if (fcntl(sock, F_SETFL, flags) == -1) {
 106                 return errno;
 107         }
 108
 109 #undef FLAG_TO_SET
 110         return 0;
 111 }
 112
 113 /* Set socket close on exec. */
 114 static int prepare_socket_cloexec(int sock)
 115 {
 116 #ifdef FD_CLOEXEC
 117         int flags;
 118
 119         flags = fcntl(sock, F_GETFD, 0);
 120         if (flags == -1) {
 121                 return errno;
 122         }
 123         flags |= FD_CLOEXEC;
 124         if (fcntl(sock, F_SETFD, flags) == -1) {
 125                 return errno;
 126         }
 127 #endif
 128         return 0;
 129 }
 130
 131 /* Set socket non blocking and close on exec. */
 132 static int prepare_socket(int sock)
 133 {
 134         int ret = prepare_socket_nonblock(sock, true);
 135
 136         if (ret) {
 137                 return ret;
 138         }
 139         return prepare_socket_cloexec(sock);
 140 }
 141
 142 static size_t unix_dgram_msg_size(void)
 143 {
 144         size_t msgsize = sizeof(struct unix_dgram_msg);
 145         msgsize = (msgsize + 15) & ~15; /* align to 16 */
 146         return msgsize;
 147 }
 148
 149 static struct msghdr_buf *unix_dgram_msghdr(struct unix_dgram_msg *msg)
 150 {
 151         /*
 152          * Not portable in C99, but "msg" is aligned and so is
 153          * unix_dgram_msg_size()
 154          */
 155         return (struct msghdr_buf *)(((char *)msg) + unix_dgram_msg_size());
 156 }
 157
 158 static void close_fd_array(int *fds, size_t num_fds)
 159 {
 160         size_t i;
 161
 162         for (i = 0; i < num_fds; i++) {
 163                 if (fds[i] == -1) {
 164                         continue;
 165                 }
 166
 167                 close(fds[i]);
 168                 fds[i] = -1;
 169         }
 170 }
 171
 172 static void close_fd_array_dgram_msg(struct unix_dgram_msg *dmsg)
 173 {
 174         struct msghdr_buf *hdr = unix_dgram_msghdr(dmsg);
 175         struct msghdr *msg = msghdr_buf_msghdr(hdr);
 176         size_t num_fds = msghdr_extract_fds(msg, NULL, 0);
 177         int fds[num_fds];
 178
 179         msghdr_extract_fds(msg, fds, num_fds);
 180
 181         close_fd_array(fds, num_fds);
 182 }
 183
 184 static int unix_dgram_init(const struct sockaddr_un *addr, size_t max_msg,
 185                            const struct poll_funcs *ev_funcs,
 186                            void (*recv_callback)(struct unix_dgram_ctx *ctx,
 187                                                  uint8_t *msg, size_t msg_len,
 188                                                  int *fds, size_t num_fds,
 189                                                  void *private_data),
 190                            void *private_data,
 191                            struct unix_dgram_ctx **result)
 192 {
 193         struct unix_dgram_ctx *ctx;
 194         size_t pathlen;
 195         int ret;
 196
 197         if (addr != NULL) {
 198                 pathlen = strlen(addr->sun_path)+1;
 199         } else {
 200                 pathlen = 1;
 201         }
 202
 203         ctx = malloc(offsetof(struct unix_dgram_ctx, path) + pathlen);
 204         if (ctx == NULL) {
 205                 return ENOMEM;
 206         }
 207         if (addr != NULL) {
 208                 memcpy(ctx->path, addr->sun_path, pathlen);
 209         } else {
 210                 ctx->path[0] = '\0';
 211         }
 212
 213         *ctx = (struct unix_dgram_ctx) {
 214                 .max_msg = max_msg,
 215                 .ev_funcs = ev_funcs,
 216                 .recv_callback = recv_callback,
 217                 .private_data = private_data,
 218                 .created_pid = (pid_t)-1
 219         };
 220
 221         ctx->recv_buf = malloc(max_msg);
 222         if (ctx->recv_buf == NULL) {
 223                 free(ctx);
 224                 return ENOMEM;
 225         }
 226
 227         ctx->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 228         if (ctx->sock == -1) {
 229                 ret = errno;
 230                 goto fail_free;
 231         }
 232
 233         /* Set non-blocking and close-on-exec. */
 234         ret = prepare_socket(ctx->sock);
 235         if (ret != 0) {
 236                 goto fail_close;
 237         }
 238
 239         if (addr != NULL) {
 240                 ret = bind(ctx->sock,
 241                            (const struct sockaddr *)(const void *)addr,
 242                            sizeof(*addr));
 243                 if (ret == -1) {
 244                         ret = errno;
 245                         goto fail_close;
 246                 }
 247
 248                 ctx->created_pid = getpid();
 249
 250                 ctx->sock_read_watch = ctx->ev_funcs->watch_new(
 251                         ctx->ev_funcs, ctx->sock, POLLIN,
 252                         unix_dgram_recv_handler, ctx);
 253
 254                 if (ctx->sock_read_watch == NULL) {
 255                         ret = ENOMEM;
 256                         goto fail_close;
 257                 }
 258         }
 259
 260         *result = ctx;
 261         return 0;
 262
 263 fail_close:
 264         close(ctx->sock);
 265 fail_free:
 266         free(ctx->recv_buf);
 267         free(ctx);
 268         return ret;
 269 }
 270
 271 static void unix_dgram_recv_handler(struct poll_watch *w, int fd, short events,
 272                                     void *private_data)
 273 {
 274         struct unix_dgram_ctx *ctx = (struct unix_dgram_ctx *)private_data;
 275         ssize_t received;
 276         int flags = 0;
 277         struct msghdr msg;
 278         struct iovec iov;
 279         size_t bufsize = msghdr_prep_recv_fds(NULL, NULL, 0, INT8_MAX);
 280         uint8_t buf[bufsize];
 281
 282         iov = (struct iovec) {
 283                 .iov_base = (void *)ctx->recv_buf,
 284                 .iov_len = ctx->max_msg,
 285         };
 286
 287         msg = (struct msghdr) {
 288                 .msg_iov = &iov,
 289                 .msg_iovlen = 1,
 290         };
 291
 292         msghdr_prep_recv_fds(&msg, buf, bufsize, INT8_MAX);
 293
 294 #ifdef MSG_CMSG_CLOEXEC
 295         flags |= MSG_CMSG_CLOEXEC;
 296 #endif
 297
 298         received = recvmsg(fd, &msg, flags);
 299         if (received == -1) {
 300                 if ((errno == EAGAIN) ||
 301                     (errno == EWOULDBLOCK) ||
 302                     (errno == EINTR) || (errno == ENOMEM)) {
 303                         /* Not really an error - just try again. */
 304                         return;
 305                 }
 306                 /* Problem with the socket. Set it unreadable. */
 307                 ctx->ev_funcs->watch_update(w, 0);
 308                 return;
 309         }
 310         if (received > ctx->max_msg) {
 311                 /* More than we expected, not for us */
 312                 return;
 313         }
 314
 315         {
 316                 size_t num_fds = msghdr_extract_fds(&msg, NULL, 0);
 317                 int fds[num_fds];
 318                 int i;
 319
 320                 msghdr_extract_fds(&msg, fds, num_fds);
 321
 322                 for (i = 0; i < num_fds; i++) {
 323                         int err;
 324
 325                         err = prepare_socket_cloexec(fds[i]);
 326                         if (err != 0) {
 327                                 close_fd_array(fds, num_fds);
 328                                 num_fds = 0;
 329                         }
 330                 }
 331
 332                 ctx->recv_callback(ctx, ctx->recv_buf, received,
 333                                    fds, num_fds, ctx->private_data);
 334         }
 335 }
 336
 337 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 338                                     void *private_data);
 339
 340 static int unix_dgram_init_pthreadpool(struct unix_dgram_ctx *ctx)
 341 {
 342         int ret, signalfd;
 343
 344         if (ctx->send_pool != NULL) {
 345                 return 0;
 346         }
 347
 348         ret = pthreadpool_pipe_init(0, &ctx->send_pool);
 349         if (ret != 0) {
 350                 return ret;
 351         }
 352
 353         signalfd = pthreadpool_pipe_signal_fd(ctx->send_pool);
 354
 355         ctx->pool_read_watch = ctx->ev_funcs->watch_new(
 356                 ctx->ev_funcs, signalfd, POLLIN,
 357                 unix_dgram_job_finished, ctx);
 358         if (ctx->pool_read_watch == NULL) {
 359                 pthreadpool_pipe_destroy(ctx->send_pool);
 360                 ctx->send_pool = NULL;
 361                 return ENOMEM;
 362         }
 363
 364         return 0;
 365 }
 366
 367 static int unix_dgram_send_queue_init(
 368         struct unix_dgram_ctx *ctx, const struct sockaddr_un *dst,
 369         struct unix_dgram_send_queue **result)
 370 {
 371         struct unix_dgram_send_queue *q;
 372         size_t pathlen;
 373         int ret, err;
 374
 375         pathlen = strlen(dst->sun_path)+1;
 376
 377         q = malloc(offsetof(struct unix_dgram_send_queue, path) + pathlen);
 378         if (q == NULL) {
 379                 return ENOMEM;
 380         }
 381         q->ctx = ctx;
 382         q->msgs = NULL;
 383         memcpy(q->path, dst->sun_path, pathlen);
 384
 385         q->sock = socket(AF_UNIX, SOCK_DGRAM, 0);
 386         if (q->sock == -1) {
 387                 err = errno;
 388                 goto fail_free;
 389         }
 390
 391         err = prepare_socket_cloexec(q->sock);
 392         if (err != 0) {
 393                 goto fail_close;
 394         }
 395
 396         do {
 397                 ret = connect(q->sock,
 398                               (const struct sockaddr *)(const void *)dst,
 399                               sizeof(*dst));
 400         } while ((ret == -1) && (errno == EINTR));
 401
 402         if (ret == -1) {
 403                 err = errno;
 404                 goto fail_close;
 405         }
 406
 407         err = unix_dgram_init_pthreadpool(ctx);
 408         if (err != 0) {
 409                 goto fail_close;
 410         }
 411
 412         DLIST_ADD(ctx->send_queues, q);
 413
 414         *result = q;
 415         return 0;
 416
 417 fail_close:
 418         close(q->sock);
 419 fail_free:
 420         free(q);
 421         return err;
 422 }
 423
 424 static void unix_dgram_send_queue_free(struct unix_dgram_send_queue *q)
 425 {
 426         struct unix_dgram_ctx *ctx = q->ctx;
 427
 428         while (q->msgs != NULL) {
 429                 struct unix_dgram_msg *msg;
 430                 msg = q->msgs;
 431                 DLIST_REMOVE(q->msgs, msg);
 432                 close_fd_array_dgram_msg(msg);
 433                 free(msg);
 434         }
 435         close(q->sock);
 436         DLIST_REMOVE(ctx->send_queues, q);
 437         free(q);
 438 }
 439
 440 static int find_send_queue(struct unix_dgram_ctx *ctx,
 441                            const struct sockaddr_un *dst,
 442                            struct unix_dgram_send_queue **ps)
 443 {
 444         struct unix_dgram_send_queue *s;
 445
 446         for (s = ctx->send_queues; s != NULL; s = s->next) {
 447                 if (strcmp(s->path, dst->sun_path) == 0) {
 448                         *ps = s;
 449                         return 0;
 450                 }
 451         }
 452         return ENOENT;
 453 }
 454
 455 static int queue_msg(struct unix_dgram_send_queue *q,
 456                      const struct iovec *iov, int iovcnt,
 457                      const int *fds, size_t num_fds)
 458 {
 459         struct unix_dgram_msg *msg;
 460         struct msghdr_buf *hdr;
 461         size_t msglen, needed;
 462         ssize_t msghdrlen;
 463         int fds_copy[MIN(num_fds, INT8_MAX)];
 464         int i, ret;
 465
 466         for (i=0; i<num_fds; i++) {
 467                 fds_copy[i] = -1;
 468         }
 469
 470         for (i = 0; i < num_fds; i++) {
 471                 fds_copy[i] = dup(fds[i]);
 472                 if (fds_copy[i] == -1) {
 473                         ret = errno;
 474                         goto fail;
 475                 }
 476         }
 477
 478         msglen = unix_dgram_msg_size();
 479
 480         msghdrlen = msghdr_copy(NULL, 0, NULL, 0, iov, iovcnt,
 481                                 fds_copy, num_fds);
 482         if (msghdrlen == -1) {
 483                 ret = EMSGSIZE;
 484                 goto fail;
 485         }
 486
 487         needed = msglen + msghdrlen;
 488         if (needed < msglen) {
 489                 ret = EMSGSIZE;
 490                 goto fail;
 491         }
 492
 493         msg = malloc(needed);
 494         if (msg == NULL) {
 495                 ret = ENOMEM;
 496                 goto fail;
 497         }
 498         hdr = unix_dgram_msghdr(msg);
 499
 500         msg->sock = q->sock;
 501         msghdr_copy(hdr, msghdrlen, NULL, 0, iov, iovcnt,
 502                     fds_copy, num_fds);
 503
 504         DLIST_ADD_END(q->msgs, msg);
 505         return 0;
 506 fail:
 507         close_fd_array(fds_copy, num_fds);
 508         return ret;
 509 }
 510
 511 static void unix_dgram_send_job(void *private_data)
 512 {
 513         struct unix_dgram_msg *dmsg = private_data;
 514
 515         do {
 516                 struct msghdr_buf *hdr = unix_dgram_msghdr(dmsg);
 517                 struct msghdr *msg = msghdr_buf_msghdr(hdr);
 518                 dmsg->sent = sendmsg(dmsg->sock, msg, 0);
 519         } while ((dmsg->sent == -1) && (errno == EINTR));
 520
 521         if (dmsg->sent == -1) {
 522                 dmsg->sys_errno = errno;
 523         }
 524 }
 525
 526 static void unix_dgram_job_finished(struct poll_watch *w, int fd, short events,
 527                                     void *private_data)
 528 {
 529         struct unix_dgram_ctx *ctx = private_data;
 530         struct unix_dgram_send_queue *q;
 531         struct unix_dgram_msg *msg;
 532         int ret, job;
 533
 534         ret = pthreadpool_pipe_finished_jobs(ctx->send_pool, &job, 1);
 535         if (ret != 1) {
 536                 return;
 537         }
 538
 539         for (q = ctx->send_queues; q != NULL; q = q->next) {
 540                 if (job == q->sock) {
 541                         break;
 542                 }
 543         }
 544
 545         if (q == NULL) {
 546                 /* Huh? Should not happen */
 547                 return;
 548         }
 549
 550         msg = q->msgs;
 551         DLIST_REMOVE(q->msgs, msg);
 552         close_fd_array_dgram_msg(msg);
 553         free(msg);
 554
 555         if (q->msgs != NULL) {
 556                 ret = pthreadpool_pipe_add_job(ctx->send_pool, q->sock,
 557                                                unix_dgram_send_job, q->msgs);
 558                 if (ret == 0) {
 559                         return;
 560                 }
 561         }
 562
 563         unix_dgram_send_queue_free(q);
 564 }
 565
 566 static int unix_dgram_send(struct unix_dgram_ctx *ctx,
 567                            const struct sockaddr_un *dst,
 568                            const struct iovec *iov, int iovlen,
 569                            const int *fds, size_t num_fds)
 570 {
 571         struct unix_dgram_send_queue *q;
 572         struct msghdr msg;
 573         ssize_t fdlen;
 574         int ret;
 575         int i;
 576
 577         if (num_fds > INT8_MAX) {
 578                 return EINVAL;
 579         }
 580
 581 #if !defined(HAVE_STRUCT_MSGHDR_MSG_CONTROL) && !defined(HAVE_STRUCT_MSGHDR_MSG_ACCRIGHTS)
 582         if (num_fds > 0) {
 583                 return ENOSYS;
 584         }
 585 #endif
 586
 587         for (i = 0; i < num_fds; i++) {
 588                 /*
 589                  * Make sure we only allow fd passing
 590                  * for communication channels,
 591                  * e.g. sockets, pipes, fifos, ...
 592                  */
 593                 ret = lseek(fds[i], 0, SEEK_CUR);
 594                 if (ret == -1 && errno == ESPIPE) {
 595                         /* ok */
 596                         continue;
 597                 }
 598
 599                 /*
 600                  * Reject the message as we may need to call dup(),
 601                  * if we queue the message.
 602                  *
 603                  * That might result in unexpected behavior for the caller
 604                  * for files and broken posix locking.
 605                  */
 606                 return EINVAL;
 607         }
 608
 609         /*
 610          * To preserve message ordering, we have to queue a message when
 611          * others are waiting in line already.
 612          */
 613         ret = find_send_queue(ctx, dst, &q);
 614         if (ret == 0) {
 615                 return queue_msg(q, iov, iovlen, fds, num_fds);
 616         }
 617
 618         /*
 619          * Try a cheap nonblocking send
 620          */
 621
 622         msg = (struct msghdr) {
 623                 .msg_name = discard_const_p(struct sockaddr_un, dst),
 624                 .msg_namelen = sizeof(*dst),
 625                 .msg_iov = discard_const_p(struct iovec, iov),
 626                 .msg_iovlen = iovlen
 627         };
 628
 629         fdlen = msghdr_prep_fds(&msg, NULL, 0, fds, num_fds);
 630         if (fdlen == -1) {
 631                 return EINVAL;
 632         }
 633
 634         {
 635                 uint8_t buf[fdlen];
 636                 msghdr_prep_fds(&msg, buf, fdlen, fds, num_fds);
 637
 638                 ret = sendmsg(ctx->sock, &msg, 0);
 639         }
 640
 641         if (ret >= 0) {
 642                 return 0;
 643         }
 644         if ((errno != EWOULDBLOCK) &&
 645             (errno != EAGAIN) &&
 646 #ifdef ENOBUFS
 647             /* FreeBSD can give this for large messages */
 648             (errno != ENOBUFS) &&
 649 #endif
 650             (errno != EINTR)) {
 651                 return errno;
 652         }
 653
 654         ret = unix_dgram_send_queue_init(ctx, dst, &q);
 655         if (ret != 0) {
 656                 return ret;
 657         }
 658         ret = queue_msg(q, iov, iovlen, fds, num_fds);
 659         if (ret != 0) {
 660                 unix_dgram_send_queue_free(q);
 661                 return ret;
 662         }
 663         ret = pthreadpool_pipe_add_job(ctx->send_pool, q->sock,
 664                                        unix_dgram_send_job, q->msgs);
 665         if (ret != 0) {
 666                 unix_dgram_send_queue_free(q);
 667                 return ret;
 668         }
 669         return 0;
 670 }
 671
 672 static int unix_dgram_sock(struct unix_dgram_ctx *ctx)
 673 {
 674         return ctx->sock;
 675 }
 676
 677 static int unix_dgram_free(struct unix_dgram_ctx *ctx)
 678 {
 679         if (ctx->send_queues != NULL) {
 680                 return EBUSY;
 681         }
 682
 683         if (ctx->send_pool != NULL) {
 684                 int ret = pthreadpool_pipe_destroy(ctx->send_pool);
 685                 if (ret != 0) {
 686                         return ret;
 687                 }
 688                 ctx->ev_funcs->watch_free(ctx->pool_read_watch);
 689         }
 690
 691         ctx->ev_funcs->watch_free(ctx->sock_read_watch);
 692
 693         close(ctx->sock);
 694         if (getpid() == ctx->created_pid) {
 695                 /* If we created it, unlink. Otherwise someone else might
 696                  * still have it open */
 697                 unlink(ctx->path);
 698         }
 699
 700         free(ctx->recv_buf);
 701         free(ctx);
 702         return 0;
 703 }
 704
 705 /*
 706  * Every message starts with a uint64_t cookie.
 707  *
 708  * A value of 0 indicates a single-fragment message which is complete in
 709  * itself. The data immediately follows the cookie.
 710  *
 711  * Every multi-fragment message has a cookie != 0 and starts with a cookie
 712  * followed by a struct unix_msg_header and then the data. The pid and sock
 713  * fields are used to assure uniqueness on the receiver side.
 714  */
 715
 716 struct unix_msg_hdr {
 717         size_t msglen;
 718         pid_t pid;
 719         int sock;
 720 };
 721
 722 struct unix_msg {
 723         struct unix_msg *prev, *next;
 724         size_t msglen;
 725         size_t received;
 726         pid_t sender_pid;
 727         int sender_sock;
 728         uint64_t cookie;
 729         uint8_t buf[1];
 730 };
 731
 732 struct unix_msg_ctx {
 733         struct unix_dgram_ctx *dgram;
 734         size_t fragment_len;
 735         uint64_t cookie;
 736
 737         void (*recv_callback)(struct unix_msg_ctx *ctx,
 738                               uint8_t *msg, size_t msg_len,
 739                               int *fds, size_t num_fds,
 740                               void *private_data);
 741         void *private_data;
 742
 743         struct unix_msg *msgs;
 744 };
 745
 746 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
 747                           uint8_t *buf, size_t buflen,
 748                           int *fds, size_t num_fds,
 749                           void *private_data);
 750
 751 int unix_msg_init(const struct sockaddr_un *addr,
 752                   const struct poll_funcs *ev_funcs,
 753                   size_t fragment_len,
 754                   void (*recv_callback)(struct unix_msg_ctx *ctx,
 755                                         uint8_t *msg, size_t msg_len,
 756                                         int *fds, size_t num_fds,
 757                                         void *private_data),
 758                   void *private_data,
 759                   struct unix_msg_ctx **result)
 760 {
 761         struct unix_msg_ctx *ctx;
 762         int ret;
 763
 764         ctx = malloc(sizeof(*ctx));
 765         if (ctx == NULL) {
 766                 return ENOMEM;
 767         }
 768
 769         *ctx = (struct unix_msg_ctx) {
 770                 .fragment_len = fragment_len,
 771                 .cookie = 1,
 772                 .recv_callback = recv_callback,
 773                 .private_data = private_data
 774         };
 775
 776         ret = unix_dgram_init(addr, fragment_len, ev_funcs,
 777                               unix_msg_recv, ctx, &ctx->dgram);
 778         if (ret != 0) {
 779                 free(ctx);
 780                 return ret;
 781         }
 782
 783         *result = ctx;
 784         return 0;
 785 }
 786
 787 int unix_msg_send(struct unix_msg_ctx *ctx, const struct sockaddr_un *dst,
 788                   const struct iovec *iov, int iovlen,
 789                   const int *fds, size_t num_fds)
 790 {
 791         ssize_t msglen;
 792         size_t sent;
 793         int ret = 0;
 794         struct iovec iov_copy[iovlen+2];
 795         struct unix_msg_hdr hdr;
 796         struct iovec src_iov;
 797
 798         if (iovlen < 0) {
 799                 return EINVAL;
 800         }
 801
 802         msglen = iov_buflen(iov, iovlen);
 803         if (msglen == -1) {
 804                 return EINVAL;
 805         }
 806
 807         if (num_fds > INT8_MAX) {
 808                 return EINVAL;
 809         }
 810
 811         if (msglen <= (ctx->fragment_len - sizeof(uint64_t))) {
 812                 uint64_t cookie = 0;
 813
 814                 iov_copy[0].iov_base = &cookie;
 815                 iov_copy[0].iov_len = sizeof(cookie);
 816                 if (iovlen > 0) {
 817                         memcpy(&iov_copy[1], iov,
 818                                sizeof(struct iovec) * iovlen);
 819                 }
 820
 821                 return unix_dgram_send(ctx->dgram, dst, iov_copy, iovlen+1,
 822                                        fds, num_fds);
 823         }
 824
 825         hdr = (struct unix_msg_hdr) {
 826                 .msglen = msglen,
 827                 .pid = getpid(),
 828                 .sock = unix_dgram_sock(ctx->dgram)
 829         };
 830
 831         iov_copy[0].iov_base = &ctx->cookie;
 832         iov_copy[0].iov_len = sizeof(ctx->cookie);
 833         iov_copy[1].iov_base = &hdr;
 834         iov_copy[1].iov_len = sizeof(hdr);
 835
 836         sent = 0;
 837         src_iov = iov[0];
 838
 839         /*
 840          * The following write loop sends the user message in pieces. We have
 841          * filled the first two iovecs above with "cookie" and "hdr". In the
 842          * following loops we pull message chunks from the user iov array and
 843          * fill iov_copy piece by piece, possibly truncating chunks from the
 844          * caller's iov array. Ugly, but hopefully efficient.
 845          */
 846
 847         while (sent < msglen) {
 848                 size_t fragment_len;
 849                 size_t iov_index = 2;
 850
 851                 fragment_len = sizeof(ctx->cookie) + sizeof(hdr);
 852
 853                 while (fragment_len < ctx->fragment_len) {
 854                         size_t space, chunk;
 855
 856                         space = ctx->fragment_len - fragment_len;
 857                         chunk = MIN(space, src_iov.iov_len);
 858
 859                         iov_copy[iov_index].iov_base = src_iov.iov_base;
 860                         iov_copy[iov_index].iov_len = chunk;
 861                         iov_index += 1;
 862
 863                         src_iov.iov_base = (char *)src_iov.iov_base + chunk;
 864                         src_iov.iov_len -= chunk;
 865                         fragment_len += chunk;
 866
 867                         if (src_iov.iov_len == 0) {
 868                                 iov += 1;
 869                                 iovlen -= 1;
 870                                 if (iovlen == 0) {
 871                                         break;
 872                                 }
 873                                 src_iov = iov[0];
 874                         }
 875                 }
 876                 sent += (fragment_len - sizeof(ctx->cookie) - sizeof(hdr));
 877
 878                 /*
 879                  * only the last fragment should pass the fd array.
 880                  * That simplifies the receiver a lot.
 881                  */
 882                 if (sent < msglen) {
 883                         ret = unix_dgram_send(ctx->dgram, dst,
 884                                               iov_copy, iov_index,
 885                                               NULL, 0);
 886                 } else {
 887                         ret = unix_dgram_send(ctx->dgram, dst,
 888                                               iov_copy, iov_index,
 889                                               fds, num_fds);
 890                 }
 891                 if (ret != 0) {
 892                         break;
 893                 }
 894         }
 895
 896         ctx->cookie += 1;
 897         if (ctx->cookie == 0) {
 898                 ctx->cookie += 1;
 899         }
 900
 901         return ret;
 902 }
 903
 904 static void unix_msg_recv(struct unix_dgram_ctx *dgram_ctx,
 905                           uint8_t *buf, size_t buflen,
 906                           int *fds, size_t num_fds,
 907                           void *private_data)
 908 {
 909         struct unix_msg_ctx *ctx = (struct unix_msg_ctx *)private_data;
 910         struct unix_msg_hdr hdr;
 911         struct unix_msg *msg;
 912         size_t space;
 913         uint64_t cookie;
 914
 915         if (buflen < sizeof(cookie)) {
 916                 goto close_fds;
 917         }
 918
 919         memcpy(&cookie, buf, sizeof(cookie));
 920
 921         buf += sizeof(cookie);
 922         buflen -= sizeof(cookie);
 923
 924         if (cookie == 0) {
 925                 ctx->recv_callback(ctx, buf, buflen, fds, num_fds,
 926                                    ctx->private_data);
 927                 return;
 928         }
 929
 930         if (buflen < sizeof(hdr)) {
 931                 goto close_fds;
 932         }
 933         memcpy(&hdr, buf, sizeof(hdr));
 934
 935         buf += sizeof(hdr);
 936         buflen -= sizeof(hdr);
 937
 938         for (msg = ctx->msgs; msg != NULL; msg = msg->next) {
 939                 if ((msg->sender_pid == hdr.pid) &&
 940                     (msg->sender_sock == hdr.sock)) {
 941                         break;
 942                 }
 943         }
 944
 945         if ((msg != NULL) && (msg->cookie != cookie)) {
 946                 DLIST_REMOVE(ctx->msgs, msg);
 947                 free(msg);
 948                 msg = NULL;
 949         }
 950
 951         if (msg == NULL) {
 952                 msg = malloc(offsetof(struct unix_msg, buf) + hdr.msglen);
 953                 if (msg == NULL) {
 954                         goto close_fds;
 955                 }
 956                 *msg = (struct unix_msg) {
 957                         .msglen = hdr.msglen,
 958                         .sender_pid = hdr.pid,
 959                         .sender_sock = hdr.sock,
 960                         .cookie = cookie
 961                 };
 962                 DLIST_ADD(ctx->msgs, msg);
 963         }
 964
 965         space = msg->msglen - msg->received;
 966         if (buflen > space) {
 967                 goto close_fds;
 968         }
 969
 970         memcpy(msg->buf + msg->received, buf, buflen);
 971         msg->received += buflen;
 972
 973         if (msg->received < msg->msglen) {
 974                 goto close_fds;
 975         }
 976
 977         DLIST_REMOVE(ctx->msgs, msg);
 978         ctx->recv_callback(ctx, msg->buf, msg->msglen, fds, num_fds,
 979                            ctx->private_data);
 980         free(msg);
 981         return;
 982
 983 close_fds:
 984         close_fd_array(fds, num_fds);
 985 }
 986
 987 int unix_msg_free(struct unix_msg_ctx *ctx)
 988 {
 989         int ret;
 990
 991         ret = unix_dgram_free(ctx->dgram);
 992         if (ret != 0) {
 993                 return ret;
 994         }
 995
 996         while (ctx->msgs != NULL) {
 997                 struct unix_msg *msg = ctx->msgs;
 998                 DLIST_REMOVE(ctx->msgs, msg);
 999                 free(msg);
1000         }
1001
1002         free(ctx);
1003         return 0;
1004 }