lib/async_req/async_sock.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    async socket syscalls
   4    Copyright (C) Volker Lendecke 2008
   5
   6      ** NOTE! The following LGPL license applies to the async_sock
   7      ** library. This does NOT imply that all of Samba is released
   8      ** under the LGPL
   9
  10    This library is free software; you can redistribute it and/or
  11    modify it under the terms of the GNU Lesser General Public
  12    License as published by the Free Software Foundation; either
  13    version 3 of the License, or (at your option) any later version.
  14
  15    This library is distributed in the hope that it will be useful,
  16    but WITHOUT ANY WARRANTY; without even the implied warranty of
  17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  18    Library General Public License for more details.
  19
  20    You should have received a copy of the GNU Lesser General Public License
  21    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  22 */
  23
  24 #include "replace.h"
  25 #include "system/network.h"
  26 #include "system/filesys.h"
  27 #include <talloc.h>
  28 #include <tevent.h>
  29 #include "lib/async_req/async_sock.h"
  30 #include "lib/iov_buf.h"
  31
  32 /* Note: lib/util/ is currently GPL */
  33 #include "lib/util/tevent_unix.h"
  34 #include "lib/util/samba_util.h"
  35
  36 #ifndef TALLOC_FREE
  37 #define TALLOC_FREE(ctx) do { talloc_free(ctx); ctx=NULL; } while(0)
  38 #endif
  39
  40 struct sendto_state {
  41         int fd;
  42         const void *buf;
  43         size_t len;
  44         int flags;
  45         const struct sockaddr_storage *addr;
  46         socklen_t addr_len;
  47         ssize_t sent;
  48 };
  49
  50 static void sendto_handler(struct tevent_context *ev,
  51                                struct tevent_fd *fde,
  52                                uint16_t flags, void *private_data);
  53
  54 struct tevent_req *sendto_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  55                                int fd, const void *buf, size_t len, int flags,
  56                                const struct sockaddr_storage *addr)
  57 {
  58         struct tevent_req *result;
  59         struct sendto_state *state;
  60         struct tevent_fd *fde;
  61
  62         result = tevent_req_create(mem_ctx, &state, struct sendto_state);
  63         if (result == NULL) {
  64                 return result;
  65         }
  66         state->fd = fd;
  67         state->buf = buf;
  68         state->len = len;
  69         state->flags = flags;
  70         state->addr = addr;
  71
  72         switch (addr->ss_family) {
  73         case AF_INET:
  74                 state->addr_len = sizeof(struct sockaddr_in);
  75                 break;
  76 #if defined(HAVE_IPV6)
  77         case AF_INET6:
  78                 state->addr_len = sizeof(struct sockaddr_in6);
  79                 break;
  80 #endif
  81         case AF_UNIX:
  82                 state->addr_len = sizeof(struct sockaddr_un);
  83                 break;
  84         default:
  85                 state->addr_len = sizeof(struct sockaddr_storage);
  86                 break;
  87         }
  88
  89         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_WRITE, sendto_handler,
  90                             result);
  91         if (fde == NULL) {
  92                 TALLOC_FREE(result);
  93                 return NULL;
  94         }
  95         return result;
  96 }
  97
  98 static void sendto_handler(struct tevent_context *ev,
  99                                struct tevent_fd *fde,
 100                                uint16_t flags, void *private_data)
 101 {
 102         struct tevent_req *req = talloc_get_type_abort(
 103                 private_data, struct tevent_req);
 104         struct sendto_state *state =
 105                 tevent_req_data(req, struct sendto_state);
 106
 107         state->sent = sendto(state->fd, state->buf, state->len, state->flags,
 108                              (const struct sockaddr *)state->addr,
 109                              state->addr_len);
 110         if ((state->sent == -1) && (errno == EINTR)) {
 111                 /* retry */
 112                 return;
 113         }
 114         if (state->sent == -1) {
 115                 tevent_req_error(req, errno);
 116                 return;
 117         }
 118         tevent_req_done(req);
 119 }
 120
 121 ssize_t sendto_recv(struct tevent_req *req, int *perrno)
 122 {
 123         struct sendto_state *state =
 124                 tevent_req_data(req, struct sendto_state);
 125
 126         if (tevent_req_is_unix_error(req, perrno)) {
 127                 return -1;
 128         }
 129         return state->sent;
 130 }
 131
 132 struct recvfrom_state {
 133         int fd;
 134         void *buf;
 135         size_t len;
 136         int flags;
 137         struct sockaddr_storage *addr;
 138         socklen_t *addr_len;
 139         ssize_t received;
 140 };
 141
 142 static void recvfrom_handler(struct tevent_context *ev,
 143                                struct tevent_fd *fde,
 144                                uint16_t flags, void *private_data);
 145
 146 struct tevent_req *recvfrom_send(TALLOC_CTX *mem_ctx,
 147                                  struct tevent_context *ev,
 148                                  int fd, void *buf, size_t len, int flags,
 149                                  struct sockaddr_storage *addr,
 150                                  socklen_t *addr_len)
 151 {
 152         struct tevent_req *result;
 153         struct recvfrom_state *state;
 154         struct tevent_fd *fde;
 155
 156         result = tevent_req_create(mem_ctx, &state, struct recvfrom_state);
 157         if (result == NULL) {
 158                 return result;
 159         }
 160         state->fd = fd;
 161         state->buf = buf;
 162         state->len = len;
 163         state->flags = flags;
 164         state->addr = addr;
 165         state->addr_len = addr_len;
 166
 167         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, recvfrom_handler,
 168                             result);
 169         if (fde == NULL) {
 170                 TALLOC_FREE(result);
 171                 return NULL;
 172         }
 173         return result;
 174 }
 175
 176 static void recvfrom_handler(struct tevent_context *ev,
 177                                struct tevent_fd *fde,
 178                                uint16_t flags, void *private_data)
 179 {
 180         struct tevent_req *req = talloc_get_type_abort(
 181                 private_data, struct tevent_req);
 182         struct recvfrom_state *state =
 183                 tevent_req_data(req, struct recvfrom_state);
 184
 185         state->received = recvfrom(state->fd, state->buf, state->len,
 186                                    state->flags, (struct sockaddr *)state->addr,
 187                                    state->addr_len);
 188         if ((state->received == -1) && (errno == EINTR)) {
 189                 /* retry */
 190                 return;
 191         }
 192         if (state->received == 0) {
 193                 tevent_req_error(req, EPIPE);
 194                 return;
 195         }
 196         if (state->received == -1) {
 197                 tevent_req_error(req, errno);
 198                 return;
 199         }
 200         tevent_req_done(req);
 201 }
 202
 203 ssize_t recvfrom_recv(struct tevent_req *req, int *perrno)
 204 {
 205         struct recvfrom_state *state =
 206                 tevent_req_data(req, struct recvfrom_state);
 207
 208         if (tevent_req_is_unix_error(req, perrno)) {
 209                 return -1;
 210         }
 211         return state->received;
 212 }
 213
 214 struct async_connect_state {
 215         int fd;
 216         int result;
 217         int sys_errno;
 218         long old_sockflags;
 219         socklen_t address_len;
 220         struct sockaddr_storage address;
 221
 222         void (*before_connect)(void *private_data);
 223         void (*after_connect)(void *private_data);
 224         void *private_data;
 225 };
 226
 227 static void async_connect_connected(struct tevent_context *ev,
 228                                     struct tevent_fd *fde, uint16_t flags,
 229                                     void *priv);
 230
 231 /**
 232  * @brief async version of connect(2)
 233  * @param[in] mem_ctx   The memory context to hang the result off
 234  * @param[in] ev        The event context to work from
 235  * @param[in] fd        The socket to recv from
 236  * @param[in] address   Where to connect?
 237  * @param[in] address_len Length of *address
 238  * @retval The async request
 239  *
 240  * This function sets the socket into non-blocking state to be able to call
 241  * connect in an async state. This will be reset when the request is finished.
 242  */
 243
 244 struct tevent_req *async_connect_send(
 245         TALLOC_CTX *mem_ctx, struct tevent_context *ev, int fd,
 246         const struct sockaddr *address, socklen_t address_len,
 247         void (*before_connect)(void *private_data),
 248         void (*after_connect)(void *private_data),
 249         void *private_data)
 250 {
 251         struct tevent_req *result;
 252         struct async_connect_state *state;
 253         struct tevent_fd *fde;
 254
 255         result = tevent_req_create(
 256                 mem_ctx, &state, struct async_connect_state);
 257         if (result == NULL) {
 258                 return NULL;
 259         }
 260
 261         /**
 262          * We have to set the socket to nonblocking for async connect(2). Keep
 263          * the old sockflags around.
 264          */
 265
 266         state->fd = fd;
 267         state->sys_errno = 0;
 268         state->before_connect = before_connect;
 269         state->after_connect = after_connect;
 270         state->private_data = private_data;
 271
 272         state->old_sockflags = fcntl(fd, F_GETFL, 0);
 273         if (state->old_sockflags == -1) {
 274                 goto post_errno;
 275         }
 276
 277         state->address_len = address_len;
 278         if (address_len > sizeof(state->address)) {
 279                 errno = EINVAL;
 280                 goto post_errno;
 281         }
 282         memcpy(&state->address, address, address_len);
 283
 284         set_blocking(fd, false);
 285
 286         if (state->before_connect != NULL) {
 287                 state->before_connect(state->private_data);
 288         }
 289
 290         state->result = connect(fd, address, address_len);
 291
 292         if (state->after_connect != NULL) {
 293                 state->after_connect(state->private_data);
 294         }
 295
 296         if (state->result == 0) {
 297                 tevent_req_done(result);
 298                 goto done;
 299         }
 300
 301         /**
 302          * A number of error messages show that something good is progressing
 303          * and that we have to wait for readability.
 304          *
 305          * If none of them are present, bail out.
 306          */
 307
 308         if (!(errno == EINPROGRESS || errno == EALREADY ||
 309 #ifdef EISCONN
 310               errno == EISCONN ||
 311 #endif
 312               errno == EAGAIN || errno == EINTR)) {
 313                 state->sys_errno = errno;
 314                 goto post_errno;
 315         }
 316
 317         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ | TEVENT_FD_WRITE,
 318                            async_connect_connected, result);
 319         if (fde == NULL) {
 320                 state->sys_errno = ENOMEM;
 321                 goto post_errno;
 322         }
 323         return result;
 324
 325  post_errno:
 326         tevent_req_error(result, state->sys_errno);
 327  done:
 328         fcntl(fd, F_SETFL, state->old_sockflags);
 329         return tevent_req_post(result, ev);
 330 }
 331
 332 /**
 333  * fde event handler for connect(2)
 334  * @param[in] ev        The event context that sent us here
 335  * @param[in] fde       The file descriptor event associated with the connect
 336  * @param[in] flags     Indicate read/writeability of the socket
 337  * @param[in] priv      private data, "struct async_req *" in this case
 338  */
 339
 340 static void async_connect_connected(struct tevent_context *ev,
 341                                     struct tevent_fd *fde, uint16_t flags,
 342                                     void *priv)
 343 {
 344         struct tevent_req *req = talloc_get_type_abort(
 345                 priv, struct tevent_req);
 346         struct async_connect_state *state =
 347                 tevent_req_data(req, struct async_connect_state);
 348         int ret;
 349
 350         if (state->before_connect != NULL) {
 351                 state->before_connect(state->private_data);
 352         }
 353
 354         ret = connect(state->fd, (struct sockaddr *)(void *)&state->address,
 355                       state->address_len);
 356
 357         if (state->after_connect != NULL) {
 358                 state->after_connect(state->private_data);
 359         }
 360
 361         if (ret == 0) {
 362                 state->sys_errno = 0;
 363                 TALLOC_FREE(fde);
 364                 tevent_req_done(req);
 365                 return;
 366         }
 367         if (errno == EINPROGRESS) {
 368                 /* Try again later, leave the fde around */
 369                 return;
 370         }
 371         state->sys_errno = errno;
 372         TALLOC_FREE(fde);
 373         tevent_req_error(req, errno);
 374         return;
 375 }
 376
 377 int async_connect_recv(struct tevent_req *req, int *perrno)
 378 {
 379         struct async_connect_state *state =
 380                 tevent_req_data(req, struct async_connect_state);
 381         int err;
 382
 383         fcntl(state->fd, F_SETFL, state->old_sockflags);
 384
 385         if (tevent_req_is_unix_error(req, &err)) {
 386                 *perrno = err;
 387                 return -1;
 388         }
 389
 390         if (state->sys_errno == 0) {
 391                 return 0;
 392         }
 393
 394         *perrno = state->sys_errno;
 395         return -1;
 396 }
 397
 398 struct writev_state {
 399         struct tevent_context *ev;
 400         int fd;
 401         struct iovec *iov;
 402         int count;
 403         size_t total_size;
 404         uint16_t flags;
 405         bool err_on_readability;
 406 };
 407
 408 static void writev_trigger(struct tevent_req *req, void *private_data);
 409 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 410                            uint16_t flags, void *private_data);
 411
 412 struct tevent_req *writev_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 413                                struct tevent_queue *queue, int fd,
 414                                bool err_on_readability,
 415                                struct iovec *iov, int count)
 416 {
 417         struct tevent_req *req;
 418         struct writev_state *state;
 419
 420         req = tevent_req_create(mem_ctx, &state, struct writev_state);
 421         if (req == NULL) {
 422                 return NULL;
 423         }
 424         state->ev = ev;
 425         state->fd = fd;
 426         state->total_size = 0;
 427         state->count = count;
 428         state->iov = (struct iovec *)talloc_memdup(
 429                 state, iov, sizeof(struct iovec) * count);
 430         if (state->iov == NULL) {
 431                 goto fail;
 432         }
 433         state->flags = TEVENT_FD_WRITE|TEVENT_FD_READ;
 434         state->err_on_readability = err_on_readability;
 435
 436         if (queue == NULL) {
 437                 struct tevent_fd *fde;
 438                 fde = tevent_add_fd(state->ev, state, state->fd,
 439                                     state->flags, writev_handler, req);
 440                 if (tevent_req_nomem(fde, req)) {
 441                         return tevent_req_post(req, ev);
 442                 }
 443                 return req;
 444         }
 445
 446         if (!tevent_queue_add(queue, ev, req, writev_trigger, NULL)) {
 447                 goto fail;
 448         }
 449         return req;
 450  fail:
 451         TALLOC_FREE(req);
 452         return NULL;
 453 }
 454
 455 static void writev_trigger(struct tevent_req *req, void *private_data)
 456 {
 457         struct writev_state *state = tevent_req_data(req, struct writev_state);
 458         struct tevent_fd *fde;
 459
 460         fde = tevent_add_fd(state->ev, state, state->fd, state->flags,
 461                             writev_handler, req);
 462         if (fde == NULL) {
 463                 tevent_req_error(req, ENOMEM);
 464         }
 465 }
 466
 467 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 468                            uint16_t flags, void *private_data)
 469 {
 470         struct tevent_req *req = talloc_get_type_abort(
 471                 private_data, struct tevent_req);
 472         struct writev_state *state =
 473                 tevent_req_data(req, struct writev_state);
 474         size_t written;
 475         bool ok;
 476
 477         if ((state->flags & TEVENT_FD_READ) && (flags & TEVENT_FD_READ)) {
 478                 int ret, value;
 479
 480                 if (state->err_on_readability) {
 481                         /* Readable and the caller wants an error on read. */
 482                         tevent_req_error(req, EPIPE);
 483                         return;
 484                 }
 485
 486                 /* Might be an error. Check if there are bytes to read */
 487                 ret = ioctl(state->fd, FIONREAD, &value);
 488                 /* FIXME - should we also check
 489                    for ret == 0 and value == 0 here ? */
 490                 if (ret == -1) {
 491                         /* There's an error. */
 492                         tevent_req_error(req, EPIPE);
 493                         return;
 494                 }
 495                 /* A request for TEVENT_FD_READ will succeed from now and
 496                    forevermore until the bytes are read so if there was
 497                    an error we'll wait until we do read, then get it in
 498                    the read callback function. Until then, remove TEVENT_FD_READ
 499                    from the flags we're waiting for. */
 500                 state->flags &= ~TEVENT_FD_READ;
 501                 TEVENT_FD_NOT_READABLE(fde);
 502
 503                 /* If not writable, we're done. */
 504                 if (!(flags & TEVENT_FD_WRITE)) {
 505                         return;
 506                 }
 507         }
 508
 509         written = writev(state->fd, state->iov, state->count);
 510         if ((written == -1) && (errno == EINTR)) {
 511                 /* retry */
 512                 return;
 513         }
 514         if (written == -1) {
 515                 tevent_req_error(req, errno);
 516                 return;
 517         }
 518         if (written == 0) {
 519                 tevent_req_error(req, EPIPE);
 520                 return;
 521         }
 522         state->total_size += written;
 523
 524         ok = iov_advance(&state->iov, &state->count, written);
 525         if (!ok) {
 526                 tevent_req_error(req, EIO);
 527                 return;
 528         }
 529
 530         if (state->count == 0) {
 531                 tevent_req_done(req);
 532                 return;
 533         }
 534 }
 535
 536 ssize_t writev_recv(struct tevent_req *req, int *perrno)
 537 {
 538         struct writev_state *state =
 539                 tevent_req_data(req, struct writev_state);
 540
 541         if (tevent_req_is_unix_error(req, perrno)) {
 542                 return -1;
 543         }
 544         return state->total_size;
 545 }
 546
 547 struct read_packet_state {
 548         int fd;
 549         uint8_t *buf;
 550         size_t nread;
 551         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
 552         void *private_data;
 553 };
 554
 555 static void read_packet_handler(struct tevent_context *ev,
 556                                 struct tevent_fd *fde,
 557                                 uint16_t flags, void *private_data);
 558
 559 struct tevent_req *read_packet_send(TALLOC_CTX *mem_ctx,
 560                                     struct tevent_context *ev,
 561                                     int fd, size_t initial,
 562                                     ssize_t (*more)(uint8_t *buf,
 563                                                     size_t buflen,
 564                                                     void *private_data),
 565                                     void *private_data)
 566 {
 567         struct tevent_req *result;
 568         struct read_packet_state *state;
 569         struct tevent_fd *fde;
 570
 571         result = tevent_req_create(mem_ctx, &state, struct read_packet_state);
 572         if (result == NULL) {
 573                 return NULL;
 574         }
 575         state->fd = fd;
 576         state->nread = 0;
 577         state->more = more;
 578         state->private_data = private_data;
 579
 580         state->buf = talloc_array(state, uint8_t, initial);
 581         if (state->buf == NULL) {
 582                 goto fail;
 583         }
 584
 585         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, read_packet_handler,
 586                             result);
 587         if (fde == NULL) {
 588                 goto fail;
 589         }
 590         return result;
 591  fail:
 592         TALLOC_FREE(result);
 593         return NULL;
 594 }
 595
 596 static void read_packet_handler(struct tevent_context *ev,
 597                                 struct tevent_fd *fde,
 598                                 uint16_t flags, void *private_data)
 599 {
 600         struct tevent_req *req = talloc_get_type_abort(
 601                 private_data, struct tevent_req);
 602         struct read_packet_state *state =
 603                 tevent_req_data(req, struct read_packet_state);
 604         size_t total = talloc_get_size(state->buf);
 605         ssize_t nread, more;
 606         uint8_t *tmp;
 607
 608         nread = recv(state->fd, state->buf+state->nread, total-state->nread,
 609                      0);
 610         if ((nread == -1) && (errno == ENOTSOCK)) {
 611                 nread = read(state->fd, state->buf+state->nread,
 612                              total-state->nread);
 613         }
 614         if ((nread == -1) && (errno == EINTR)) {
 615                 /* retry */
 616                 return;
 617         }
 618         if (nread == -1) {
 619                 tevent_req_error(req, errno);
 620                 return;
 621         }
 622         if (nread == 0) {
 623                 tevent_req_error(req, EPIPE);
 624                 return;
 625         }
 626
 627         state->nread += nread;
 628         if (state->nread < total) {
 629                 /* Come back later */
 630                 return;
 631         }
 632
 633         /*
 634          * We got what was initially requested. See if "more" asks for -- more.
 635          */
 636         if (state->more == NULL) {
 637                 /* Nobody to ask, this is a async read_data */
 638                 tevent_req_done(req);
 639                 return;
 640         }
 641
 642         more = state->more(state->buf, total, state->private_data);
 643         if (more == -1) {
 644                 /* We got an invalid packet, tell the caller */
 645                 tevent_req_error(req, EIO);
 646                 return;
 647         }
 648         if (more == 0) {
 649                 /* We're done, full packet received */
 650                 tevent_req_done(req);
 651                 return;
 652         }
 653
 654         if (total + more < total) {
 655                 tevent_req_error(req, EMSGSIZE);
 656                 return;
 657         }
 658
 659         tmp = talloc_realloc(state, state->buf, uint8_t, total+more);
 660         if (tevent_req_nomem(tmp, req)) {
 661                 return;
 662         }
 663         state->buf = tmp;
 664 }
 665
 666 ssize_t read_packet_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 667                          uint8_t **pbuf, int *perrno)
 668 {
 669         struct read_packet_state *state =
 670                 tevent_req_data(req, struct read_packet_state);
 671
 672         if (tevent_req_is_unix_error(req, perrno)) {
 673                 return -1;
 674         }
 675         *pbuf = talloc_move(mem_ctx, &state->buf);
 676         return talloc_get_size(*pbuf);
 677 }
 678
 679 struct wait_for_read_state {
 680         struct tevent_req *req;
 681         struct tevent_fd *fde;
 682 };
 683
 684 static void wait_for_read_done(struct tevent_context *ev,
 685                                struct tevent_fd *fde,
 686                                uint16_t flags,
 687                                void *private_data);
 688
 689 struct tevent_req *wait_for_read_send(TALLOC_CTX *mem_ctx,
 690                                       struct tevent_context *ev,
 691                                       int fd)
 692 {
 693         struct tevent_req *req;
 694         struct wait_for_read_state *state;
 695
 696         req = tevent_req_create(mem_ctx, &state, struct wait_for_read_state);
 697         if (req == NULL) {
 698                 return NULL;
 699         }
 700         state->req = req;
 701         state->fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ,
 702                                    wait_for_read_done, state);
 703         if (tevent_req_nomem(state->fde, req)) {
 704                 return tevent_req_post(req, ev);
 705         }
 706         return req;
 707 }
 708
 709 static void wait_for_read_done(struct tevent_context *ev,
 710                                struct tevent_fd *fde,
 711                                uint16_t flags,
 712                                void *private_data)
 713 {
 714         struct wait_for_read_state *state = talloc_get_type_abort(
 715                 private_data, struct wait_for_read_state);
 716
 717         if (flags & TEVENT_FD_READ) {
 718                 TALLOC_FREE(state->fde);
 719                 tevent_req_done(state->req);
 720         }
 721 }
 722
 723 bool wait_for_read_recv(struct tevent_req *req, int *perr)
 724 {
 725         int err;
 726
 727         if (tevent_req_is_unix_error(req, &err)) {
 728                 *perr = err;
 729                 return false;
 730         }
 731         return true;
 732 }