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