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