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         socklen_t address_len;
 171         struct sockaddr_storage address;
 172 };
 173
 174 static void async_connect_connected(struct tevent_context *ev,
 175                                     struct tevent_fd *fde, uint16_t flags,
 176                                     void *priv);
 177
 178 /**
 179  * @brief async version of connect(2)
 180  * @param[in] mem_ctx   The memory context to hang the result off
 181  * @param[in] ev        The event context to work from
 182  * @param[in] fd        The socket to recv from
 183  * @param[in] address   Where to connect?
 184  * @param[in] address_len Length of *address
 185  * @retval The async request
 186  *
 187  * This function sets the socket into non-blocking state to be able to call
 188  * connect in an async state. This will be reset when the request is finished.
 189  */
 190
 191 struct tevent_req *async_connect_send(TALLOC_CTX *mem_ctx,
 192                                       struct tevent_context *ev,
 193                                       int fd, const struct sockaddr *address,
 194                                       socklen_t address_len)
 195 {
 196         struct tevent_req *result;
 197         struct async_connect_state *state;
 198         struct tevent_fd *fde;
 199
 200         result = tevent_req_create(
 201                 mem_ctx, &state, struct async_connect_state);
 202         if (result == NULL) {
 203                 return NULL;
 204         }
 205
 206         /**
 207          * We have to set the socket to nonblocking for async connect(2). Keep
 208          * the old sockflags around.
 209          */
 210
 211         state->fd = fd;
 212         state->sys_errno = 0;
 213
 214         state->address_len = address_len;
 215         if (address_len > sizeof(state->address)) {
 216                 errno = EINVAL;
 217                 goto post_errno;
 218         }
 219         memcpy(&state->address, address, address_len);
 220
 221         state->old_sockflags = fcntl(fd, F_GETFL, 0);
 222         if (state->old_sockflags == -1) {
 223                 goto post_errno;
 224         }
 225
 226         set_blocking(fd, false);
 227
 228         state->result = connect(fd, address, address_len);
 229         if (state->result == 0) {
 230                 tevent_req_done(result);
 231                 goto done;
 232         }
 233
 234         /**
 235          * A number of error messages show that something good is progressing
 236          * and that we have to wait for readability.
 237          *
 238          * If none of them are present, bail out.
 239          */
 240
 241         if (!(errno == EINPROGRESS || errno == EALREADY ||
 242 #ifdef EISCONN
 243               errno == EISCONN ||
 244 #endif
 245               errno == EAGAIN || errno == EINTR)) {
 246                 state->sys_errno = errno;
 247                 goto post_errno;
 248         }
 249
 250         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ | TEVENT_FD_WRITE,
 251                            async_connect_connected, result);
 252         if (fde == NULL) {
 253                 state->sys_errno = ENOMEM;
 254                 goto post_errno;
 255         }
 256         return result;
 257
 258  post_errno:
 259         tevent_req_error(result, state->sys_errno);
 260  done:
 261         fcntl(fd, F_SETFL, state->old_sockflags);
 262         return tevent_req_post(result, ev);
 263 }
 264
 265 /**
 266  * fde event handler for connect(2)
 267  * @param[in] ev        The event context that sent us here
 268  * @param[in] fde       The file descriptor event associated with the connect
 269  * @param[in] flags     Indicate read/writeability of the socket
 270  * @param[in] priv      private data, "struct async_req *" in this case
 271  */
 272
 273 static void async_connect_connected(struct tevent_context *ev,
 274                                     struct tevent_fd *fde, uint16_t flags,
 275                                     void *priv)
 276 {
 277         struct tevent_req *req = talloc_get_type_abort(
 278                 priv, struct tevent_req);
 279         struct async_connect_state *state =
 280                 tevent_req_data(req, struct async_connect_state);
 281
 282         /*
 283          * Stevens, Network Programming says that if there's a
 284          * successful connect, the socket is only writable. Upon an
 285          * error, it's both readable and writable.
 286          */
 287         if ((flags & (TEVENT_FD_READ|TEVENT_FD_WRITE))
 288             == (TEVENT_FD_READ|TEVENT_FD_WRITE)) {
 289                 int ret;
 290
 291                 ret = connect(state->fd,
 292                               (struct sockaddr *)(void *)&state->address,
 293                               state->address_len);
 294                 if (ret == 0) {
 295                         TALLOC_FREE(fde);
 296                         tevent_req_done(req);
 297                         return;
 298                 }
 299
 300                 if (errno == EINPROGRESS) {
 301                         /* Try again later, leave the fde around */
 302                         return;
 303                 }
 304                 TALLOC_FREE(fde);
 305                 tevent_req_error(req, errno);
 306                 return;
 307         }
 308
 309         state->sys_errno = 0;
 310         tevent_req_done(req);
 311 }
 312
 313 int async_connect_recv(struct tevent_req *req, int *perrno)
 314 {
 315         struct async_connect_state *state =
 316                 tevent_req_data(req, struct async_connect_state);
 317         int err;
 318
 319         fcntl(state->fd, F_SETFL, state->old_sockflags);
 320
 321         if (tevent_req_is_unix_error(req, &err)) {
 322                 *perrno = err;
 323                 return -1;
 324         }
 325
 326         if (state->sys_errno == 0) {
 327                 return 0;
 328         }
 329
 330         *perrno = state->sys_errno;
 331         return -1;
 332 }
 333
 334 struct writev_state {
 335         struct tevent_context *ev;
 336         int fd;
 337         struct iovec *iov;
 338         int count;
 339         size_t total_size;
 340         uint16_t flags;
 341 };
 342
 343 static void writev_trigger(struct tevent_req *req, void *private_data);
 344 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 345                            uint16_t flags, void *private_data);
 346
 347 struct tevent_req *writev_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 348                                struct tevent_queue *queue, int fd,
 349                                bool err_on_readability,
 350                                struct iovec *iov, int count)
 351 {
 352         struct tevent_req *req;
 353         struct writev_state *state;
 354
 355         req = tevent_req_create(mem_ctx, &state, struct writev_state);
 356         if (req == NULL) {
 357                 return NULL;
 358         }
 359         state->ev = ev;
 360         state->fd = fd;
 361         state->total_size = 0;
 362         state->count = count;
 363         state->iov = (struct iovec *)talloc_memdup(
 364                 state, iov, sizeof(struct iovec) * count);
 365         if (state->iov == NULL) {
 366                 goto fail;
 367         }
 368         state->flags = TEVENT_FD_WRITE;
 369         if (err_on_readability) {
 370                 state->flags |= TEVENT_FD_READ;
 371         }
 372
 373         if (queue == NULL) {
 374                 struct tevent_fd *fde;
 375                 fde = tevent_add_fd(state->ev, state, state->fd,
 376                                     state->flags, writev_handler, req);
 377                 if (tevent_req_nomem(fde, req)) {
 378                         return tevent_req_post(req, ev);
 379                 }
 380                 return req;
 381         }
 382
 383         if (!tevent_queue_add(queue, ev, req, writev_trigger, NULL)) {
 384                 goto fail;
 385         }
 386         return req;
 387  fail:
 388         TALLOC_FREE(req);
 389         return NULL;
 390 }
 391
 392 static void writev_trigger(struct tevent_req *req, void *private_data)
 393 {
 394         struct writev_state *state = tevent_req_data(req, struct writev_state);
 395         struct tevent_fd *fde;
 396
 397         fde = tevent_add_fd(state->ev, state, state->fd, state->flags,
 398                             writev_handler, req);
 399         if (fde == NULL) {
 400                 tevent_req_error(req, ENOMEM);
 401         }
 402 }
 403
 404 static void writev_handler(struct tevent_context *ev, struct tevent_fd *fde,
 405                            uint16_t flags, void *private_data)
 406 {
 407         struct tevent_req *req = talloc_get_type_abort(
 408                 private_data, struct tevent_req);
 409         struct writev_state *state =
 410                 tevent_req_data(req, struct writev_state);
 411         size_t to_write, written;
 412         int i;
 413
 414         to_write = 0;
 415
 416         if (flags & TEVENT_FD_READ) {
 417                 tevent_req_error(req, EPIPE);
 418                 return;
 419         }
 420
 421         for (i=0; i<state->count; i++) {
 422                 to_write += state->iov[i].iov_len;
 423         }
 424
 425         written = sys_writev(state->fd, state->iov, state->count);
 426         if (written == -1) {
 427                 tevent_req_error(req, errno);
 428                 return;
 429         }
 430         if (written == 0) {
 431                 tevent_req_error(req, EPIPE);
 432                 return;
 433         }
 434         state->total_size += written;
 435
 436         if (written == to_write) {
 437                 tevent_req_done(req);
 438                 return;
 439         }
 440
 441         /*
 442          * We've written less than we were asked to, drop stuff from
 443          * state->iov.
 444          */
 445
 446         while (written > 0) {
 447                 if (written < state->iov[0].iov_len) {
 448                         state->iov[0].iov_base =
 449                                 (char *)state->iov[0].iov_base + written;
 450                         state->iov[0].iov_len -= written;
 451                         break;
 452                 }
 453                 written -= state->iov[0].iov_len;
 454                 state->iov += 1;
 455                 state->count -= 1;
 456         }
 457 }
 458
 459 ssize_t writev_recv(struct tevent_req *req, int *perrno)
 460 {
 461         struct writev_state *state =
 462                 tevent_req_data(req, struct writev_state);
 463
 464         if (tevent_req_is_unix_error(req, perrno)) {
 465                 return -1;
 466         }
 467         return state->total_size;
 468 }
 469
 470 struct read_packet_state {
 471         int fd;
 472         uint8_t *buf;
 473         size_t nread;
 474         ssize_t (*more)(uint8_t *buf, size_t buflen, void *private_data);
 475         void *private_data;
 476 };
 477
 478 static void read_packet_handler(struct tevent_context *ev,
 479                                 struct tevent_fd *fde,
 480                                 uint16_t flags, void *private_data);
 481
 482 struct tevent_req *read_packet_send(TALLOC_CTX *mem_ctx,
 483                                     struct tevent_context *ev,
 484                                     int fd, size_t initial,
 485                                     ssize_t (*more)(uint8_t *buf,
 486                                                     size_t buflen,
 487                                                     void *private_data),
 488                                     void *private_data)
 489 {
 490         struct tevent_req *result;
 491         struct read_packet_state *state;
 492         struct tevent_fd *fde;
 493
 494         result = tevent_req_create(mem_ctx, &state, struct read_packet_state);
 495         if (result == NULL) {
 496                 return NULL;
 497         }
 498         state->fd = fd;
 499         state->nread = 0;
 500         state->more = more;
 501         state->private_data = private_data;
 502
 503         state->buf = talloc_array(state, uint8_t, initial);
 504         if (state->buf == NULL) {
 505                 goto fail;
 506         }
 507
 508         fde = tevent_add_fd(ev, state, fd, TEVENT_FD_READ, read_packet_handler,
 509                             result);
 510         if (fde == NULL) {
 511                 goto fail;
 512         }
 513         return result;
 514  fail:
 515         TALLOC_FREE(result);
 516         return NULL;
 517 }
 518
 519 static void read_packet_handler(struct tevent_context *ev,
 520                                 struct tevent_fd *fde,
 521                                 uint16_t flags, void *private_data)
 522 {
 523         struct tevent_req *req = talloc_get_type_abort(
 524                 private_data, struct tevent_req);
 525         struct read_packet_state *state =
 526                 tevent_req_data(req, struct read_packet_state);
 527         size_t total = talloc_get_size(state->buf);
 528         ssize_t nread, more;
 529         uint8_t *tmp;
 530
 531         nread = recv(state->fd, state->buf+state->nread, total-state->nread,
 532                      0);
 533         if (nread == -1) {
 534                 tevent_req_error(req, errno);
 535                 return;
 536         }
 537         if (nread == 0) {
 538                 tevent_req_error(req, EPIPE);
 539                 return;
 540         }
 541
 542         state->nread += nread;
 543         if (state->nread < total) {
 544                 /* Come back later */
 545                 return;
 546         }
 547
 548         /*
 549          * We got what was initially requested. See if "more" asks for -- more.
 550          */
 551         if (state->more == NULL) {
 552                 /* Nobody to ask, this is a async read_data */
 553                 tevent_req_done(req);
 554                 return;
 555         }
 556
 557         more = state->more(state->buf, total, state->private_data);
 558         if (more == -1) {
 559                 /* We got an invalid packet, tell the caller */
 560                 tevent_req_error(req, EIO);
 561                 return;
 562         }
 563         if (more == 0) {
 564                 /* We're done, full packet received */
 565                 tevent_req_done(req);
 566                 return;
 567         }
 568
 569         tmp = TALLOC_REALLOC_ARRAY(state, state->buf, uint8_t, total+more);
 570         if (tevent_req_nomem(tmp, req)) {
 571                 return;
 572         }
 573         state->buf = tmp;
 574 }
 575
 576 ssize_t read_packet_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 577                          uint8_t **pbuf, int *perrno)
 578 {
 579         struct read_packet_state *state =
 580                 tevent_req_data(req, struct read_packet_state);
 581
 582         if (tevent_req_is_unix_error(req, perrno)) {
 583                 return -1;
 584         }
 585         *pbuf = talloc_move(mem_ctx, &state->buf);
 586         return talloc_get_size(*pbuf);
 587 }