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
  31 /* Note: lib/util/ is currently GPL */
  32 #include "lib/util/tevent_unix.h"
  33 #include "lib/util/util.h"
  34
  35 #ifndef TALLOC_FREE
  36 #define TALLOC_FREE(ctx) do { talloc_free(ctx); ctx=NULL; } while(0)
  37 #endif
  38
  39 struct sendto_state {
  40         int fd;
  41         const void *buf;
  42         size_t len;
  43         int flags;
  44         const struct sockaddr_storage *addr;
  45         socklen_t addr_len;
  46         ssize_t sent;
  47 };
  48
  49 static void sendto_handler(struct tevent_context *ev,
  50                                struct tevent_fd *fde,
  51                                uint16_t flags, void *private_data);
  52
  53 struct tevent_req *sendto_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  54                                int fd, const void *buf, size_t len, int flags,
  55                                const struct sockaddr_storage *addr)
  56 {
  57         struct tevent_req *result;
  58         struct sendto_state *state;
  59         struct tevent_fd *fde;
  60
  61         result = tevent_req_create(mem_ctx, &state, struct sendto_state);
  62         if (result == NULL) {
  63                 return result;
  64         }
  65         state->fd = fd;
  66         state->buf = buf;
  67         state->len = len;
  68         state->flags = flags;
  69         state->addr = addr;
  70
  71         switch (addr->ss_family) {
  72         case AF_INET:
  73                 state->addr_len = sizeof(struct sockaddr_in);
  74                 break;
  75 #if defined(HAVE_IPV6)
  76         case AF_INET6:
  77                 state->addr_len = sizeof(struct sockaddr_in6);
  78                 break;
  79 #endif
  80         case AF_UNIX:
  81                 state->addr_len = sizeof(struct sockaddr_un);
  82                 break;
  83         default:
  84                 state->addr_len = sizeof(struct sockaddr_storage);
  85                 break;
  86         }
  87
  88         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_WRITE, sendto_handler,
  89                             result);
  90         if (fde == NULL) {
  91                 TALLOC_FREE(result);
  92                 return NULL;
  93         }
  94         return result;
  95 }
  96
  97 static void sendto_handler(struct tevent_context *ev,
  98                                struct tevent_fd *fde,
  99                                uint16_t flags, void *private_data)
 100 {
 101         struct tevent_req *req = talloc_get_type_abort(
 102                 private_data, struct tevent_req);
 103         struct sendto_state *state =
 104                 tevent_req_data(req, struct sendto_state);
 105
 106         state->sent = sendto(state->fd, state->buf, state->len, state->flags,
 107                              (struct sockaddr *)state->addr, state->addr_len);
 108         if ((state->sent == -1) && (errno == EINTR)) {
 109                 /* retry */
 110                 return;
 111         }
 112         if (state->sent == -1) {
 113                 tevent_req_error(req, errno);
 114                 return;
 115         }
 116         tevent_req_done(req);
 117 }
 118
 119 ssize_t sendto_recv(struct tevent_req *req, int *perrno)
 120 {
 121         struct sendto_state *state =
 122                 tevent_req_data(req, struct sendto_state);
 123
 124         if (tevent_req_is_unix_error(req, perrno)) {
 125                 return -1;
 126         }
 127         return state->sent;
 128 }
 129
 130 struct recvfrom_state {
 131         int fd;
 132         void *buf;
 133         size_t len;
 134         int flags;
 135         struct sockaddr_storage *addr;
 136         socklen_t *addr_len;
 137         ssize_t received;
 138 };
 139
 140 static void recvfrom_handler(struct tevent_context *ev,
 141                                struct tevent_fd *fde,
 142                                uint16_t flags, void *private_data);
 143
 144 struct tevent_req *recvfrom_send(TALLOC_CTX *mem_ctx,
 145                                  struct tevent_context *ev,
 146                                  int fd, void *buf, size_t len, int flags,
 147                                  struct sockaddr_storage *addr,
 148                                  socklen_t *addr_len)
 149 {
 150         struct tevent_req *result;
 151         struct recvfrom_state *state;
 152         struct tevent_fd *fde;
 153
 154         result = tevent_req_create(mem_ctx, &state, struct recvfrom_state);
 155         if (result == NULL) {
 156                 return result;
 157         }
 158         state->fd = fd;
 159         state->buf = buf;
 160         state->len = len;
 161         state->flags = flags;
 162         state->addr = addr;
 163         state->addr_len = addr_len;
 164
 165         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, recvfrom_handler,
 166                             result);
 167         if (fde == NULL) {
 168                 TALLOC_FREE(result);
 169                 return NULL;
 170         }
 171         return result;
 172 }
 173
 174 static void recvfrom_handler(struct tevent_context *ev,
 175                                struct tevent_fd *fde,
 176                                uint16_t flags, void *private_data)
 177 {
 178         struct tevent_req *req = talloc_get_type_abort(
 179                 private_data, struct tevent_req);
 180         struct recvfrom_state *state =
 181                 tevent_req_data(req, struct recvfrom_state);
 182
 183         state->received = recvfrom(state->fd, state->buf, state->len,
 184                                    state->flags, (struct sockaddr *)state->addr,
 185                                    state->addr_len);
 186         if ((state->received == -1) && (errno == EINTR)) {
 187                 /* retry */
 188                 return;
 189         }
 190         if (state->received == 0) {
 191                 tevent_req_error(req, EPIPE);
 192                 return;
 193         }
 194         if (state->received == -1) {
 195                 tevent_req_error(req, errno);
 196                 return;
 197         }
 198         tevent_req_done(req);
 199 }
 200
 201 ssize_t recvfrom_recv(struct tevent_req *req, int *perrno)
 202 {
 203         struct recvfrom_state *state =
 204                 tevent_req_data(req, struct recvfrom_state);
 205
 206         if (tevent_req_is_unix_error(req, perrno)) {
 207                 return -1;
 208         }
 209         return state->received;
 210 }
 211
 212 struct async_connect_state {
 213         int fd;
 214         int result;
 215         int sys_errno;
 216         long old_sockflags;
 217         socklen_t address_len;
 218         struct sockaddr_storage address;
 219 };
 220
 221 static void async_connect_connected(struct tevent_context *ev,
 222                                     struct tevent_fd *fde, uint16_t flags,
 223                                     void *priv);
 224
 225 /**
 226  * @brief async version of connect(2)
 227  * @param[in] mem_ctx   The memory context to hang the result off
 228  * @param[in] ev        The event context to work from
 229  * @param[in] fd        The socket to recv from
 230  * @param[in] address   Where to connect?
 231  * @param[in] address_len Length of *address
 232  * @retval The async request
 233  *
 234  * This function sets the socket into non-blocking state to be able to call
 235  * connect in an async state. This will be reset when the request is finished.
 236  */
 237
 238 struct tevent_req *async_connect_send(TALLOC_CTX *mem_ctx,
 239                                       struct tevent_context *ev,
 240                                       int fd, const struct sockaddr *address,
 241                                       socklen_t address_len)
 242 {
 243         struct tevent_req *result;
 244         struct async_connect_state *state;
 245         struct tevent_fd *fde;
 246
 247         result = tevent_req_create(
 248                 mem_ctx, &state, struct async_connect_state);
 249         if (result == NULL) {
 250                 return NULL;
 251         }
 252
 253         /**
 254          * We have to set the socket to nonblocking for async connect(2). Keep
 255          * the old sockflags around.
 256          */
 257
 258         state->fd = fd;
 259         state->sys_errno = 0;
 260
 261         state->old_sockflags = fcntl(fd, F_GETFL, 0);
 262         if (state->old_sockflags == -1) {
 263                 goto post_errno;
 264         }
 265
 266         state->address_len = address_len;
 267         if (address_len > sizeof(state->address)) {
 268                 errno = EINVAL;
 269                 goto post_errno;
 270         }
 271         memcpy(&state->address, address, address_len);
 272
 273         set_blocking(fd, false);
 274
 275         state->result = connect(fd, address, address_len);
 276         if (state->result == 0) {
 277                 tevent_req_done(result);
 278                 goto done;
 279         }
 280
 281         /**
 282          * A number of error messages show that something good is progressing
 283          * and that we have to wait for readability.
 284          *
 285          * If none of them are present, bail out.
 286          */
 287
 288         if (!(errno == EINPROGRESS || errno == EALREADY ||
 289 #ifdef EISCONN
 290               errno == EISCONN ||
 291 #endif
 292               errno == EAGAIN || errno == EINTR)) {
 293                 state->sys_errno = errno;
 294                 goto post_errno;
 295         }
 296
 297         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ | TEVENT_FD_WRITE,
 298                            async_connect_connected, result);
 299         if (fde == NULL) {
 300                 state->sys_errno = ENOMEM;
 301                 goto post_errno;
 302         }
 303         return result;
 304
 305  post_errno:
 306         tevent_req_error(result, state->sys_errno);
 307  done:
 308         fcntl(fd, F_SETFL, state->old_sockflags);
 309         return tevent_req_post(result, ev);
 310 }
 311
 312 /**
 313  * fde event handler for connect(2)
 314  * @param[in] ev        The event context that sent us here
 315  * @param[in] fde       The file descriptor event associated with the connect
 316  * @param[in] flags     Indicate read/writeability of the socket
 317  * @param[in] priv      private data, "struct async_req *" in this case
 318  */
 319
 320 static void async_connect_connected(struct tevent_context *ev,
 321                                     struct tevent_fd *fde, uint16_t flags,
 322                                     void *priv)
 323 {
 324         struct tevent_req *req = talloc_get_type_abort(
 325                 priv, struct tevent_req);
 326         struct async_connect_state *state =
 327                 tevent_req_data(req, struct async_connect_state);
 328
 329         /*
 330          * Stevens, Network Programming says that if there's a
 331          * successful connect, the socket is only writable. Upon an
 332          * error, it's both readable and writable.
 333          */
 334         if ((flags & (TEVENT_FD_READ|TEVENT_FD_WRITE))
 335             == (TEVENT_FD_READ|TEVENT_FD_WRITE)) {
 336                 int ret;
 337
 338                 ret = connect(state->fd,
 339                               (struct sockaddr *)(void *)&state->address,
 340                               state->address_len);
 341                 if (ret == 0) {
 342                         TALLOC_FREE(fde);
 343                         tevent_req_done(req);
 344                         return;
 345                 }
 346
 347                 if (errno == EINPROGRESS) {
 348                         /* Try again later, leave the fde around */
 349                         return;
 350                 }
 351                 TALLOC_FREE(fde);
 352                 tevent_req_error(req, errno);
 353                 return;
 354         }
 355
 356         state->sys_errno = 0;
 357         tevent_req_done(req);
 358 }
 359
 360 int async_connect_recv(struct tevent_req *req, int *perrno)
 361 {
 362         struct async_connect_state *state =
 363                 tevent_req_data(req, struct async_connect_state);
 364         int err;
 365
 366         fcntl(state->fd, F_SETFL, state->old_sockflags);
 367
 368         if (tevent_req_is_unix_error(req, &err)) {
 369                 *perrno = err;
 370                 return -1;
 371         }
 372
 373         if (state->sys_errno == 0) {
 374                 return 0;
 375         }
 376
 377         *perrno = state->sys_errno;
 378         return -1;
 379 }
 380
 381 struct writev_state {
 382         struct tevent_context *ev;
 383         int fd;
 384         struct iovec *iov;
 385         int count;
 386         size_t total_size;
 387         uint16_t flags;
 388 };
 389
 390 static void writev_trigger(struct tevent_req *req, void *private_data);
 391 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 392                            uint16_t flags, void *private_data);
 393
 394 struct tevent_req *writev_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 395                                struct tevent_queue *queue, int fd,
 396                                bool err_on_readability,
 397                                struct iovec *iov, int count)
 398 {
 399         struct tevent_req *req;
 400         struct writev_state *state;
 401
 402         req = tevent_req_create(mem_ctx, &state, struct writev_state);
 403         if (req == NULL) {
 404                 return NULL;
 405         }
 406         state->ev = ev;
 407         state->fd = fd;
 408         state->total_size = 0;
 409         state->count = count;
 410         state->iov = (struct iovec *)talloc_memdup(
 411                 state, iov, sizeof(struct iovec) * count);
 412         if (state->iov == NULL) {
 413                 goto fail;
 414         }
 415         state->flags = TEVENT_FD_WRITE;
 416         if (err_on_readability) {
 417                 state->flags |= TEVENT_FD_READ;
 418         }
 419
 420         if (queue == NULL) {
 421                 struct tevent_fd *fde;
 422                 fde = tevent_add_fd(state->ev, state, state->fd,
 423                                     state->flags, writev_handler, req);
 424                 if (tevent_req_nomem(fde, req)) {
 425                         return tevent_req_post(req, ev);
 426                 }
 427                 return req;
 428         }
 429
 430         if (!tevent_queue_add(queue, ev, req, writev_trigger, NULL)) {
 431                 goto fail;
 432         }
 433         return req;
 434  fail:
 435         TALLOC_FREE(req);
 436         return NULL;
 437 }
 438
 439 static void writev_trigger(struct tevent_req *req, void *private_data)
 440 {
 441         struct writev_state *state = tevent_req_data(req, struct writev_state);
 442         struct tevent_fd *fde;
 443
 444         fde = tevent_add_fd(state->ev, state, state->fd, state->flags,
 445                             writev_handler, req);
 446         if (fde == NULL) {
 447                 tevent_req_error(req, ENOMEM);
 448         }
 449 }
 450
 451 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 452                            uint16_t flags, void *private_data)
 453 {
 454         struct tevent_req *req = talloc_get_type_abort(
 455                 private_data, struct tevent_req);
 456         struct writev_state *state =
 457                 tevent_req_data(req, struct writev_state);
 458         size_t to_write, written;
 459         int i;
 460
 461         to_write = 0;
 462
 463         if ((state->flags & TEVENT_FD_READ) && (flags & TEVENT_FD_READ)) {
 464                 tevent_req_error(req, EPIPE);
 465                 return;
 466         }
 467
 468         for (i=0; i<state->count; i++) {
 469                 to_write += state->iov[i].iov_len;
 470         }
 471
 472         written = writev(state->fd, state->iov, state->count);
 473         if ((written == -1) && (errno == EINTR)) {
 474                 /* retry */
 475                 return;
 476         }
 477         if (written == -1) {
 478                 tevent_req_error(req, errno);
 479                 return;
 480         }
 481         if (written == 0) {
 482                 tevent_req_error(req, EPIPE);
 483                 return;
 484         }
 485         state->total_size += written;
 486
 487         if (written == to_write) {
 488                 tevent_req_done(req);
 489                 return;
 490         }
 491
 492         /*
 493          * We've written less than we were asked to, drop stuff from
 494          * state->iov.
 495          */
 496
 497         while (written > 0) {
 498                 if (written < state->iov[0].iov_len) {
 499                         state->iov[0].iov_base =
 500                                 (char *)state->iov[0].iov_base + written;
 501                         state->iov[0].iov_len -= written;
 502                         break;
 503                 }
 504                 written -= state->iov[0].iov_len;
 505                 state->iov += 1;
 506                 state->count -= 1;
 507         }
 508 }
 509
 510 ssize_t writev_recv(struct tevent_req *req, int *perrno)
 511 {
 512         struct writev_state *state =
 513                 tevent_req_data(req, struct writev_state);
 514
 515         if (tevent_req_is_unix_error(req, perrno)) {
 516                 return -1;
 517         }
 518         return state->total_size;
 519 }
 520
 521 struct read_packet_state {
 522         int fd;
 523         uint8_t *buf;
 524         size_t nread;
 525         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
 526         void *private_data;
 527 };
 528
 529 static void read_packet_handler(struct tevent_context *ev,
 530                                 struct tevent_fd *fde,
 531                                 uint16_t flags, void *private_data);
 532
 533 struct tevent_req *read_packet_send(TALLOC_CTX *mem_ctx,
 534                                     struct tevent_context *ev,
 535                                     int fd, size_t initial,
 536                                     ssize_t (*more)(uint8_t *buf,
 537                                                     size_t buflen,
 538                                                     void *private_data),
 539                                     void *private_data)
 540 {
 541         struct tevent_req *result;
 542         struct read_packet_state *state;
 543         struct tevent_fd *fde;
 544
 545         result = tevent_req_create(mem_ctx, &state, struct read_packet_state);
 546         if (result == NULL) {
 547                 return NULL;
 548         }
 549         state->fd = fd;
 550         state->nread = 0;
 551         state->more = more;
 552         state->private_data = private_data;
 553
 554         state->buf = talloc_array(state, uint8_t, initial);
 555         if (state->buf == NULL) {
 556                 goto fail;
 557         }
 558
 559         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, read_packet_handler,
 560                             result);
 561         if (fde == NULL) {
 562                 goto fail;
 563         }
 564         return result;
 565  fail:
 566         TALLOC_FREE(result);
 567         return NULL;
 568 }
 569
 570 static void read_packet_handler(struct tevent_context *ev,
 571                                 struct tevent_fd *fde,
 572                                 uint16_t flags, void *private_data)
 573 {
 574         struct tevent_req *req = talloc_get_type_abort(
 575                 private_data, struct tevent_req);
 576         struct read_packet_state *state =
 577                 tevent_req_data(req, struct read_packet_state);
 578         size_t total = talloc_get_size(state->buf);
 579         ssize_t nread, more;
 580         uint8_t *tmp;
 581
 582         nread = recv(state->fd, state->buf+state->nread, total-state->nread,
 583                      0);
 584         if ((nread == -1) && (errno == EINTR)) {
 585                 /* retry */
 586                 return;
 587         }
 588         if (nread == -1) {
 589                 tevent_req_error(req, errno);
 590                 return;
 591         }
 592         if (nread == 0) {
 593                 tevent_req_error(req, EPIPE);
 594                 return;
 595         }
 596
 597         state->nread += nread;
 598         if (state->nread < total) {
 599                 /* Come back later */
 600                 return;
 601         }
 602
 603         /*
 604          * We got what was initially requested. See if "more" asks for -- more.
 605          */
 606         if (state->more == NULL) {
 607                 /* Nobody to ask, this is a async read_data */
 608                 tevent_req_done(req);
 609                 return;
 610         }
 611
 612         more = state->more(state->buf, total, state->private_data);
 613         if (more == -1) {
 614                 /* We got an invalid packet, tell the caller */
 615                 tevent_req_error(req, EIO);
 616                 return;
 617         }
 618         if (more == 0) {
 619                 /* We're done, full packet received */
 620                 tevent_req_done(req);
 621                 return;
 622         }
 623
 624         tmp = talloc_realloc(state, state->buf, uint8_t, total+more);
 625         if (tevent_req_nomem(tmp, req)) {
 626                 return;
 627         }
 628         state->buf = tmp;
 629 }
 630
 631 ssize_t read_packet_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 632                          uint8_t **pbuf, int *perrno)
 633 {
 634         struct read_packet_state *state =
 635                 tevent_req_data(req, struct read_packet_state);
 636
 637         if (tevent_req_is_unix_error(req, perrno)) {
 638                 return -1;
 639         }
 640         *pbuf = talloc_move(mem_ctx, &state->buf);
 641         return talloc_get_size(*pbuf);
 642 }