lib/async_req/async_sock.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    async socket syscalls
   4    Copyright (C) Volker Lendecke 2008
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "includes.h"
  21 #include "lib/talloc/talloc.h"
  22 #include "lib/tevent/tevent.h"
  23 #include "lib/async_req/async_sock.h"
  24 #include "lib/util/tevent_unix.h"
  25 #include <fcntl.h>
  26
  27 #ifndef TALLOC_FREE
  28 #define TALLOC_FREE(ctx) do { talloc_free(ctx); ctx=NULL; } while(0)
  29 #endif
  30
  31 struct async_send_state {
  32         int fd;
  33         const void *buf;
  34         size_t len;
  35         int flags;
  36         ssize_t sent;
  37 };
  38
  39 static void async_send_handler(struct tevent_context *ev,
  40                                struct tevent_fd *fde,
  41                                uint16_t flags, void *private_data);
  42
  43 struct tevent_req *async_send_send(TALLOC_CTX *mem_ctx,
  44                                    struct tevent_context *ev,
  45                                    int fd, const void *buf, size_t len,
  46                                    int flags)
  47 {
  48         struct tevent_req *result;
  49         struct async_send_state *state;
  50         struct tevent_fd *fde;
  51
  52         result = tevent_req_create(mem_ctx, &state, struct async_send_state);
  53         if (result == NULL) {
  54                 return result;
  55         }
  56         state->fd = fd;
  57         state->buf = buf;
  58         state->len = len;
  59         state->flags = flags;
  60
  61         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_WRITE, async_send_handler,
  62                             result);
  63         if (fde == NULL) {
  64                 TALLOC_FREE(result);
  65                 return NULL;
  66         }
  67         return result;
  68 }
  69
  70 static void async_send_handler(struct tevent_context *ev,
  71                                struct tevent_fd *fde,
  72                                uint16_t flags, void *private_data)
  73 {
  74         struct tevent_req *req = talloc_get_type_abort(
  75                 private_data, struct tevent_req);
  76         struct async_send_state *state =
  77                 tevent_req_data(req, struct async_send_state);
  78
  79         state->sent = send(state->fd, state->buf, state->len, state->flags);
  80         if (state->sent == -1) {
  81                 tevent_req_error(req, errno);
  82                 return;
  83         }
  84         tevent_req_done(req);
  85 }
  86
  87 ssize_t async_send_recv(struct tevent_req *req, int *perrno)
  88 {
  89         struct async_send_state *state =
  90                 tevent_req_data(req, struct async_send_state);
  91
  92         if (tevent_req_is_unix_error(req, perrno)) {
  93                 return -1;
  94         }
  95         return state->sent;
  96 }
  97
  98 struct async_recv_state {
  99         int fd;
 100         void *buf;
 101         size_t len;
 102         int flags;
 103         ssize_t received;
 104 };
 105
 106 static void async_recv_handler(struct tevent_context *ev,
 107                                struct tevent_fd *fde,
 108                                uint16_t flags, void *private_data);
 109
 110 struct tevent_req *async_recv_send(TALLOC_CTX *mem_ctx,
 111                                    struct tevent_context *ev,
 112                                    int fd, void *buf, size_t len, int flags)
 113 {
 114         struct tevent_req *result;
 115         struct async_recv_state *state;
 116         struct tevent_fd *fde;
 117
 118         result = tevent_req_create(mem_ctx, &state, struct async_recv_state);
 119         if (result == NULL) {
 120                 return result;
 121         }
 122         state->fd = fd;
 123         state->buf = buf;
 124         state->len = len;
 125         state->flags = flags;
 126
 127         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, async_recv_handler,
 128                             result);
 129         if (fde == NULL) {
 130                 TALLOC_FREE(result);
 131                 return NULL;
 132         }
 133         return result;
 134 }
 135
 136 static void async_recv_handler(struct tevent_context *ev,
 137                                struct tevent_fd *fde,
 138                                uint16_t flags, void *private_data)
 139 {
 140         struct tevent_req *req = talloc_get_type_abort(
 141                 private_data, struct tevent_req);
 142         struct async_recv_state *state =
 143                 tevent_req_data(req, struct async_recv_state);
 144
 145         state->received = recv(state->fd, state->buf, state->len,
 146                                state->flags);
 147         if (state->received == -1) {
 148                 tevent_req_error(req, errno);
 149                 return;
 150         }
 151         tevent_req_done(req);
 152 }
 153
 154 ssize_t async_recv_recv(struct tevent_req *req, int *perrno)
 155 {
 156         struct async_recv_state *state =
 157                 tevent_req_data(req, struct async_recv_state);
 158
 159         if (tevent_req_is_unix_error(req, perrno)) {
 160                 return -1;
 161         }
 162         return state->received;
 163 }
 164
 165 struct async_connect_state {
 166         int fd;
 167         int result;
 168         int sys_errno;
 169         long old_sockflags;
 170 };
 171
 172 static void async_connect_connected(struct tevent_context *ev,
 173                                     struct tevent_fd *fde, uint16_t flags,
 174                                     void *priv);
 175
 176 /**
 177  * @brief async version of connect(2)
 178  * @param[in] mem_ctx   The memory context to hang the result off
 179  * @param[in] ev        The event context to work from
 180  * @param[in] fd        The socket to recv from
 181  * @param[in] address   Where to connect?
 182  * @param[in] address_len Length of *address
 183  * @retval The async request
 184  *
 185  * This function sets the socket into non-blocking state to be able to call
 186  * connect in an async state. This will be reset when the request is finished.
 187  */
 188
 189 struct tevent_req *async_connect_send(TALLOC_CTX *mem_ctx,
 190                                       struct tevent_context *ev,
 191                                       int fd, const struct sockaddr *address,
 192                                       socklen_t address_len)
 193 {
 194         struct tevent_req *result;
 195         struct async_connect_state *state;
 196         struct tevent_fd *fde;
 197
 198         result = tevent_req_create(
 199                 mem_ctx, &state, struct async_connect_state);
 200         if (result == NULL) {
 201                 return NULL;
 202         }
 203
 204         /**
 205          * We have to set the socket to nonblocking for async connect(2). Keep
 206          * the old sockflags around.
 207          */
 208
 209         state->fd = fd;
 210         state->sys_errno = 0;
 211
 212         state->old_sockflags = fcntl(fd, F_GETFL, 0);
 213         if (state->old_sockflags == -1) {
 214                 goto post_errno;
 215         }
 216
 217         set_blocking(fd, false);
 218
 219         state->result = connect(fd, address, address_len);
 220         if (state->result == 0) {
 221                 tevent_req_done(result);
 222                 goto done;
 223         }
 224
 225         /**
 226          * A number of error messages show that something good is progressing
 227          * and that we have to wait for readability.
 228          *
 229          * If none of them are present, bail out.
 230          */
 231
 232         if (!(errno == EINPROGRESS || errno == EALREADY ||
 233 #ifdef EISCONN
 234               errno == EISCONN ||
 235 #endif
 236               errno == EAGAIN || errno == EINTR)) {
 237                 state->sys_errno = errno;
 238                 goto post_errno;
 239         }
 240
 241         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ | TEVENT_FD_WRITE,
 242                            async_connect_connected, result);
 243         if (fde == NULL) {
 244                 state->sys_errno = ENOMEM;
 245                 goto post_errno;
 246         }
 247         return result;
 248
 249  post_errno:
 250         tevent_req_error(result, state->sys_errno);
 251  done:
 252         fcntl(fd, F_SETFL, state->old_sockflags);
 253         return tevent_req_post(result, ev);
 254 }
 255
 256 /**
 257  * fde event handler for connect(2)
 258  * @param[in] ev        The event context that sent us here
 259  * @param[in] fde       The file descriptor event associated with the connect
 260  * @param[in] flags     Indicate read/writeability of the socket
 261  * @param[in] priv      private data, "struct async_req *" in this case
 262  */
 263
 264 static void async_connect_connected(struct tevent_context *ev,
 265                                     struct tevent_fd *fde, uint16_t flags,
 266                                     void *priv)
 267 {
 268         struct tevent_req *req = talloc_get_type_abort(
 269                 priv, struct tevent_req);
 270         struct async_connect_state *state =
 271                 tevent_req_data(req, struct async_connect_state);
 272
 273         TALLOC_FREE(fde);
 274
 275         /*
 276          * Stevens, Network Programming says that if there's a
 277          * successful connect, the socket is only writable. Upon an
 278          * error, it's both readable and writable.
 279          */
 280         if ((flags & (TEVENT_FD_READ|TEVENT_FD_WRITE))
 281             == (TEVENT_FD_READ|TEVENT_FD_WRITE)) {
 282                 int sockerr;
 283                 socklen_t err_len = sizeof(sockerr);
 284
 285                 if (getsockopt(state->fd, SOL_SOCKET, SO_ERROR,
 286                                (void *)&sockerr, &err_len) == 0) {
 287                         errno = sockerr;
 288                 }
 289
 290                 state->sys_errno = errno;
 291
 292                 DEBUG(10, ("connect returned %s\n", strerror(errno)));
 293
 294                 fcntl(state->fd, F_SETFL, state->old_sockflags);
 295                 tevent_req_error(req, state->sys_errno);
 296                 return;
 297         }
 298
 299         state->sys_errno = 0;
 300         tevent_req_done(req);
 301 }
 302
 303 int async_connect_recv(struct tevent_req *req, int *perrno)
 304 {
 305         struct async_connect_state *state =
 306                 tevent_req_data(req, struct async_connect_state);
 307         int err;
 308
 309         fcntl(state->fd, F_SETFL, state->old_sockflags);
 310
 311         if (tevent_req_is_unix_error(req, &err)) {
 312                 *perrno = err;
 313                 return -1;
 314         }
 315
 316         if (state->sys_errno == 0) {
 317                 return 0;
 318         }
 319
 320         *perrno = state->sys_errno;
 321         return -1;
 322 }
 323
 324 struct writev_state {
 325         struct tevent_context *ev;
 326         int fd;
 327         struct iovec *iov;
 328         int count;
 329         size_t total_size;
 330         uint16_t flags;
 331 };
 332
 333 static void writev_trigger(struct tevent_req *req, void *private_data);
 334 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 335                            uint16_t flags, void *private_data);
 336
 337 struct tevent_req *writev_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 338                                struct tevent_queue *queue, int fd,
 339                                bool err_on_readability,
 340                                struct iovec *iov, int count)
 341 {
 342         struct tevent_req *req;
 343         struct writev_state *state;
 344
 345         req = tevent_req_create(mem_ctx, &state, struct writev_state);
 346         if (req == NULL) {
 347                 return NULL;
 348         }
 349         state->ev = ev;
 350         state->fd = fd;
 351         state->total_size = 0;
 352         state->count = count;
 353         state->iov = (struct iovec *)talloc_memdup(
 354                 state, iov, sizeof(struct iovec) * count);
 355         if (state->iov == NULL) {
 356                 goto fail;
 357         }
 358         state->flags = TEVENT_FD_WRITE;
 359         if (err_on_readability) {
 360                 state->flags |= TEVENT_FD_READ;
 361         }
 362
 363         if (queue == NULL) {
 364                 struct tevent_fd *fde;
 365                 fde = tevent_add_fd(state->ev, state, state->fd,
 366                                     state->flags, writev_handler, req);
 367                 if (tevent_req_nomem(fde, req)) {
 368                         return tevent_req_post(req, ev);
 369                 }
 370                 return req;
 371         }
 372
 373         if (!tevent_queue_add(queue, ev, req, writev_trigger, NULL)) {
 374                 goto fail;
 375         }
 376         return req;
 377  fail:
 378         TALLOC_FREE(req);
 379         return NULL;
 380 }
 381
 382 static void writev_trigger(struct tevent_req *req, void *private_data)
 383 {
 384         struct writev_state *state = tevent_req_data(req, struct writev_state);
 385         struct tevent_fd *fde;
 386
 387         fde = tevent_add_fd(state->ev, state, state->fd, state->flags,
 388                             writev_handler, req);
 389         if (fde == NULL) {
 390                 tevent_req_error(req, ENOMEM);
 391         }
 392 }
 393
 394 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 395                            uint16_t flags, void *private_data)
 396 {
 397         struct tevent_req *req = talloc_get_type_abort(
 398                 private_data, struct tevent_req);
 399         struct writev_state *state =
 400                 tevent_req_data(req, struct writev_state);
 401         size_t to_write, written;
 402         int i;
 403
 404         to_write = 0;
 405
 406         if (flags & TEVENT_FD_READ) {
 407                 tevent_req_error(req, EPIPE);
 408                 return;
 409         }
 410
 411         for (i=0; i<state->count; i++) {
 412                 to_write += state->iov[i].iov_len;
 413         }
 414
 415         written = sys_writev(state->fd, state->iov, state->count);
 416         if (written == -1) {
 417                 tevent_req_error(req, errno);
 418                 return;
 419         }
 420         if (written == 0) {
 421                 tevent_req_error(req, EPIPE);
 422                 return;
 423         }
 424         state->total_size += written;
 425
 426         if (written == to_write) {
 427                 tevent_req_done(req);
 428                 return;
 429         }
 430
 431         /*
 432          * We've written less than we were asked to, drop stuff from
 433          * state->iov.
 434          */
 435
 436         while (written > 0) {
 437                 if (written < state->iov[0].iov_len) {
 438                         state->iov[0].iov_base =
 439                                 (char *)state->iov[0].iov_base + written;
 440                         state->iov[0].iov_len -= written;
 441                         break;
 442                 }
 443                 written -= state->iov[0].iov_len;
 444                 state->iov += 1;
 445                 state->count -= 1;
 446         }
 447 }
 448
 449 ssize_t writev_recv(struct tevent_req *req, int *perrno)
 450 {
 451         struct writev_state *state =
 452                 tevent_req_data(req, struct writev_state);
 453
 454         if (tevent_req_is_unix_error(req, perrno)) {
 455                 return -1;
 456         }
 457         return state->total_size;
 458 }
 459
 460 struct read_packet_state {
 461         int fd;
 462         uint8_t *buf;
 463         size_t nread;
 464         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
 465         void *private_data;
 466 };
 467
 468 static void read_packet_handler(struct tevent_context *ev,
 469                                 struct tevent_fd *fde,
 470                                 uint16_t flags, void *private_data);
 471
 472 struct tevent_req *read_packet_send(TALLOC_CTX *mem_ctx,
 473                                     struct tevent_context *ev,
 474                                     int fd, size_t initial,
 475                                     ssize_t (*more)(uint8_t *buf,
 476                                                     size_t buflen,
 477                                                     void *private_data),
 478                                     void *private_data)
 479 {
 480         struct tevent_req *result;
 481         struct read_packet_state *state;
 482         struct tevent_fd *fde;
 483
 484         result = tevent_req_create(mem_ctx, &state, struct read_packet_state);
 485         if (result == NULL) {
 486                 return NULL;
 487         }
 488         state->fd = fd;
 489         state->nread = 0;
 490         state->more = more;
 491         state->private_data = private_data;
 492
 493         state->buf = talloc_array(state, uint8_t, initial);
 494         if (state->buf == NULL) {
 495                 goto fail;
 496         }
 497
 498         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, read_packet_handler,
 499                             result);
 500         if (fde == NULL) {
 501                 goto fail;
 502         }
 503         return result;
 504  fail:
 505         TALLOC_FREE(result);
 506         return NULL;
 507 }
 508
 509 static void read_packet_handler(struct tevent_context *ev,
 510                                 struct tevent_fd *fde,
 511                                 uint16_t flags, void *private_data)
 512 {
 513         struct tevent_req *req = talloc_get_type_abort(
 514                 private_data, struct tevent_req);
 515         struct read_packet_state *state =
 516                 tevent_req_data(req, struct read_packet_state);
 517         size_t total = talloc_get_size(state->buf);
 518         ssize_t nread, more;
 519         uint8_t *tmp;
 520
 521         nread = recv(state->fd, state->buf+state->nread, total-state->nread,
 522                      0);
 523         if (nread == -1) {
 524                 tevent_req_error(req, errno);
 525                 return;
 526         }
 527         if (nread == 0) {
 528                 tevent_req_error(req, EPIPE);
 529                 return;
 530         }
 531
 532         state->nread += nread;
 533         if (state->nread < total) {
 534                 /* Come back later */
 535                 return;
 536         }
 537
 538         /*
 539          * We got what was initially requested. See if "more" asks for -- more.
 540          */
 541         if (state->more == NULL) {
 542                 /* Nobody to ask, this is a async read_data */
 543                 tevent_req_done(req);
 544                 return;
 545         }
 546
 547         more = state->more(state->buf, total, state->private_data);
 548         if (more == -1) {
 549                 /* We got an invalid packet, tell the caller */
 550                 tevent_req_error(req, EIO);
 551                 return;
 552         }
 553         if (more == 0) {
 554                 /* We're done, full packet received */
 555                 tevent_req_done(req);
 556                 return;
 557         }
 558
 559         tmp = TALLOC_REALLOC_ARRAY(state, state->buf, uint8_t, total+more);
 560         if (tevent_req_nomem(tmp, req)) {
 561                 return;
 562         }
 563         state->buf = tmp;
 564 }
 565
 566 ssize_t read_packet_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 567                          uint8_t **pbuf, int *perrno)
 568 {
 569         struct read_packet_state *state =
 570                 tevent_req_data(req, struct read_packet_state);
 571
 572         if (tevent_req_is_unix_error(req, perrno)) {
 573                 return -1;
 574         }
 575         *pbuf = talloc_move(mem_ctx, &state->buf);
 576         return talloc_get_size(*pbuf);
 577 }