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 };
 331
 332 static void writev_trigger(struct tevent_req *req, void *private_data);
 333 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 334                            uint16_t flags, void *private_data);
 335
 336 struct tevent_req *writev_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 337                                struct tevent_queue *queue, int fd,
 338                                struct iovec *iov, int count)
 339 {
 340         struct tevent_req *req;
 341         struct writev_state *state;
 342
 343         req = tevent_req_create(mem_ctx, &state, struct writev_state);
 344         if (req == NULL) {
 345                 return NULL;
 346         }
 347         state->ev = ev;
 348         state->fd = fd;
 349         state->total_size = 0;
 350         state->count = count;
 351         state->iov = (struct iovec *)talloc_memdup(
 352                 state, iov, sizeof(struct iovec) * count);
 353         if (state->iov == NULL) {
 354                 goto fail;
 355         }
 356
 357         if (queue == NULL) {
 358                 struct tevent_fd *fde;
 359                 fde = tevent_add_fd(state->ev, state, state->fd,
 360                                     TEVENT_FD_WRITE, writev_handler, req);
 361                 if (tevent_req_nomem(fde, req)) {
 362                         return tevent_req_post(req, ev);
 363                 }
 364                 return req;
 365         }
 366
 367         if (!tevent_queue_add(queue, ev, req, writev_trigger, NULL)) {
 368                 goto fail;
 369         }
 370         return req;
 371  fail:
 372         TALLOC_FREE(req);
 373         return NULL;
 374 }
 375
 376 static void writev_trigger(struct tevent_req *req, void *private_data)
 377 {
 378         struct writev_state *state = tevent_req_data(req, struct writev_state);
 379         struct tevent_fd *fde;
 380
 381         fde = tevent_add_fd(state->ev, state, state->fd, TEVENT_FD_WRITE,
 382                             writev_handler, req);
 383         if (fde == NULL) {
 384                 tevent_req_error(req, ENOMEM);
 385         }
 386 }
 387
 388 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 389                            uint16_t flags, void *private_data)
 390 {
 391         struct tevent_req *req = talloc_get_type_abort(
 392                 private_data, struct tevent_req);
 393         struct writev_state *state =
 394                 tevent_req_data(req, struct writev_state);
 395         size_t to_write, written;
 396         int i;
 397
 398         to_write = 0;
 399
 400         for (i=0; i<state->count; i++) {
 401                 to_write += state->iov[i].iov_len;
 402         }
 403
 404         written = sys_writev(state->fd, state->iov, state->count);
 405         if (written == -1) {
 406                 tevent_req_error(req, errno);
 407                 return;
 408         }
 409         if (written == 0) {
 410                 tevent_req_error(req, EPIPE);
 411                 return;
 412         }
 413         state->total_size += written;
 414
 415         if (written == to_write) {
 416                 tevent_req_done(req);
 417                 return;
 418         }
 419
 420         /*
 421          * We've written less than we were asked to, drop stuff from
 422          * state->iov.
 423          */
 424
 425         while (written > 0) {
 426                 if (written < state->iov[0].iov_len) {
 427                         state->iov[0].iov_base =
 428                                 (char *)state->iov[0].iov_base + written;
 429                         state->iov[0].iov_len -= written;
 430                         break;
 431                 }
 432                 written -= state->iov[0].iov_len;
 433                 state->iov += 1;
 434                 state->count -= 1;
 435         }
 436 }
 437
 438 ssize_t writev_recv(struct tevent_req *req, int *perrno)
 439 {
 440         struct writev_state *state =
 441                 tevent_req_data(req, struct writev_state);
 442
 443         if (tevent_req_is_unix_error(req, perrno)) {
 444                 return -1;
 445         }
 446         return state->total_size;
 447 }
 448
 449 struct read_packet_state {
 450         int fd;
 451         uint8_t *buf;
 452         size_t nread;
 453         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
 454         void *private_data;
 455 };
 456
 457 static void read_packet_handler(struct tevent_context *ev,
 458                                 struct tevent_fd *fde,
 459                                 uint16_t flags, void *private_data);
 460
 461 struct tevent_req *read_packet_send(TALLOC_CTX *mem_ctx,
 462                                     struct tevent_context *ev,
 463                                     int fd, size_t initial,
 464                                     ssize_t (*more)(uint8_t *buf,
 465                                                     size_t buflen,
 466                                                     void *private_data),
 467                                     void *private_data)
 468 {
 469         struct tevent_req *result;
 470         struct read_packet_state *state;
 471         struct tevent_fd *fde;
 472
 473         result = tevent_req_create(mem_ctx, &state, struct read_packet_state);
 474         if (result == NULL) {
 475                 return NULL;
 476         }
 477         state->fd = fd;
 478         state->nread = 0;
 479         state->more = more;
 480         state->private_data = private_data;
 481
 482         state->buf = talloc_array(state, uint8_t, initial);
 483         if (state->buf == NULL) {
 484                 goto fail;
 485         }
 486
 487         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, read_packet_handler,
 488                             result);
 489         if (fde == NULL) {
 490                 goto fail;
 491         }
 492         return result;
 493  fail:
 494         TALLOC_FREE(result);
 495         return NULL;
 496 }
 497
 498 static void read_packet_handler(struct tevent_context *ev,
 499                                 struct tevent_fd *fde,
 500                                 uint16_t flags, void *private_data)
 501 {
 502         struct tevent_req *req = talloc_get_type_abort(
 503                 private_data, struct tevent_req);
 504         struct read_packet_state *state =
 505                 tevent_req_data(req, struct read_packet_state);
 506         size_t total = talloc_get_size(state->buf);
 507         ssize_t nread, more;
 508         uint8_t *tmp;
 509
 510         nread = recv(state->fd, state->buf+state->nread, total-state->nread,
 511                      0);
 512         if (nread == -1) {
 513                 tevent_req_error(req, errno);
 514                 return;
 515         }
 516         if (nread == 0) {
 517                 tevent_req_error(req, EPIPE);
 518                 return;
 519         }
 520
 521         state->nread += nread;
 522         if (state->nread < total) {
 523                 /* Come back later */
 524                 return;
 525         }
 526
 527         /*
 528          * We got what was initially requested. See if "more" asks for -- more.
 529          */
 530         if (state->more == NULL) {
 531                 /* Nobody to ask, this is a async read_data */
 532                 tevent_req_done(req);
 533                 return;
 534         }
 535
 536         more = state->more(state->buf, total, state->private_data);
 537         if (more == -1) {
 538                 /* We got an invalid packet, tell the caller */
 539                 tevent_req_error(req, EIO);
 540                 return;
 541         }
 542         if (more == 0) {
 543                 /* We're done, full packet received */
 544                 tevent_req_done(req);
 545                 return;
 546         }
 547
 548         tmp = TALLOC_REALLOC_ARRAY(state, state->buf, uint8_t, total+more);
 549         if (tevent_req_nomem(tmp, req)) {
 550                 return;
 551         }
 552         state->buf = tmp;
 553 }
 554
 555 ssize_t read_packet_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 556                          uint8_t **pbuf, int *perrno)
 557 {
 558         struct read_packet_state *state =
 559                 tevent_req_data(req, struct read_packet_state);
 560
 561         if (tevent_req_is_unix_error(req, perrno)) {
 562                 return -1;
 563         }
 564         *pbuf = talloc_move(mem_ctx, &state->buf);
 565         return talloc_get_size(*pbuf);
 566 }