lib/tsocket/tsocket_writev.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    Copyright (C) Stefan Metzmacher 2009
   5
   6      ** NOTE! The following LGPL license applies to the tevent
   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    Lesser General Public License for more details.
  19
  20    You should have received a copy of the GNU Lesser General Public
  21    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  22 */
  23
  24 #include "replace.h"
  25 #include "system/network.h"
  26 #include "tsocket.h"
  27 #include "tsocket_internal.h"
  28
  29 struct tsocket_writev_state {
  30         /* this structs are owned by the caller */
  31         struct {
  32                 struct tsocket_context *sock;
  33                 const struct iovec *vector;
  34                 size_t count;
  35         } caller;
  36
  37         struct iovec *iov;
  38         size_t count;
  39         int total_written;
  40 };
  41
  42 static int tsocket_writev_state_destructor(struct tsocket_writev_state *state)
  43 {
  44         if (state->caller.sock) {
  45                 tsocket_set_writeable_handler(state->caller.sock, NULL, NULL);
  46         }
  47         ZERO_STRUCT(state->caller);
  48
  49         return 0;
  50 }
  51
  52 static void tsocket_writev_handler(struct tsocket_context *sock,
  53                                    void *private_data);
  54
  55 struct tevent_req *tsocket_writev_send(struct tsocket_context *sock,
  56                                        TALLOC_CTX *mem_ctx,
  57                                        const struct iovec *vector,
  58                                        size_t count)
  59 {
  60         struct tevent_req *req;
  61         struct tsocket_writev_state *state;
  62         int ret;
  63         int err;
  64         bool dummy;
  65         int to_write = 0;
  66         size_t i;
  67
  68         req = tevent_req_create(mem_ctx, &state,
  69                                 struct tsocket_writev_state);
  70         if (!req) {
  71                 return NULL;
  72         }
  73
  74         state->caller.sock      = sock;
  75         state->caller.vector    = vector;
  76         state->caller.count     = count;
  77
  78         state->iov              = NULL;
  79         state->count            = count;
  80         state->total_written    = 0;
  81
  82         state->iov = talloc_array(state, struct iovec, count);
  83         if (tevent_req_nomem(state->iov, req)) {
  84                 goto post;
  85         }
  86         memcpy(state->iov, vector, sizeof(struct iovec) * count);
  87
  88         for (i=0; i < count; i++) {
  89                 int tmp = to_write;
  90
  91                 tmp += state->iov[i].iov_len;
  92
  93                 if (tmp < to_write) {
  94                         tevent_req_error(req, EMSGSIZE);
  95                         goto post;
  96                 }
  97
  98                 to_write = tmp;
  99         }
 100
 101         if (to_write == 0) {
 102                 tevent_req_done(req);
 103                 goto post;
 104         }
 105
 106         /*
 107          * this is a fast path, not waiting for the
 108          * socket to become explicit writeable gains
 109          * about 10%-20% performance in benchmark tests.
 110          */
 111         tsocket_writev_handler(sock, req);
 112         if (!tevent_req_is_in_progress(req)) {
 113                 goto post;
 114         }
 115
 116         talloc_set_destructor(state, tsocket_writev_state_destructor);
 117
 118         ret = tsocket_set_writeable_handler(sock,
 119                                             tsocket_writev_handler,
 120                                             req);
 121         err = tsocket_error_from_errno(ret, errno, &dummy);
 122         if (tevent_req_error(req, err)) {
 123                 goto post;
 124         }
 125
 126         return req;
 127
 128  post:
 129         return tevent_req_post(req, sock->event.ctx);
 130 }
 131
 132 static void tsocket_writev_handler(struct tsocket_context *sock,
 133                                    void *private_data)
 134 {
 135         struct tevent_req *req = talloc_get_type(private_data,
 136                                  struct tevent_req);
 137         struct tsocket_writev_state *state = tevent_req_data(req,
 138                                              struct tsocket_writev_state);
 139         int ret;
 140         int err;
 141         bool retry;
 142
 143         ret = tsocket_writev(state->caller.sock,
 144                              state->iov,
 145                              state->count);
 146         err = tsocket_error_from_errno(ret, errno, &retry);
 147         if (retry) {
 148                 /* retry later */
 149                 return;
 150         }
 151         if (tevent_req_error(req, err)) {
 152                 return;
 153         }
 154
 155         state->total_written += ret;
 156
 157         /*
 158          * we have not written everything yet, so we need to truncate
 159          * the already written bytes from our iov copy
 160          */
 161         while (ret > 0) {
 162                 if (ret < state->iov[0].iov_len) {
 163                         uint8_t *base;
 164                         base = (uint8_t *)state->iov[0].iov_base;
 165                         base += ret;
 166                         state->iov[0].iov_base = base;
 167                         state->iov[0].iov_len -= ret;
 168                         break;
 169                 }
 170                 ret -= state->iov[0].iov_len;
 171                 state->iov += 1;
 172                 state->count -= 1;
 173         }
 174
 175         if (state->count > 0) {
 176                 /* more to write */
 177                 return;
 178         }
 179
 180         tevent_req_done(req);
 181 }
 182
 183 int tsocket_writev_recv(struct tevent_req *req, int *perrno)
 184 {
 185         struct tsocket_writev_state *state = tevent_req_data(req,
 186                                              struct tsocket_writev_state);
 187         int ret;
 188
 189         ret = tsocket_simple_int_recv(req, perrno);
 190         if (ret == 0) {
 191                 ret = state->total_written;
 192         }
 193
 194         tevent_req_received(req);
 195         return ret;
 196 }
 197
 198 struct tsocket_writev_queue_state {
 199         /* this structs are owned by the caller */
 200         struct {
 201                 struct tsocket_context *sock;
 202                 const struct iovec *vector;
 203                 size_t count;
 204         } caller;
 205         int ret;
 206 };
 207
 208 static void tsocket_writev_queue_trigger(struct tevent_req *req,
 209                                          void *private_data);
 210 static void tsocket_writev_queue_done(struct tevent_req *subreq);
 211
 212 /**
 213  * @brief Queue a dgram blob for sending through the socket
 214  * @param[in] mem_ctx   The memory context for the result
 215  * @param[in] sock      The socket to send data through
 216  * @param[in] queue     The existing send queue
 217  * @param[in] vector    The iovec vector so write
 218  * @param[in] count     The size of the vector
 219  * @retval              The async request handle
 220  *
 221  * This function queues a blob for sending to destination through an existing
 222  * dgram socket. The async callback is triggered when the whole blob is
 223  * delivered to the underlying system socket.
 224  *
 225  * The caller needs to make sure that all non-scalar input parameters hang
 226  * arround for the whole lifetime of the request.
 227  */
 228 struct tevent_req *tsocket_writev_queue_send(TALLOC_CTX *mem_ctx,
 229                                              struct tsocket_context *sock,
 230                                              struct tevent_queue *queue,
 231                                              const struct iovec *vector,
 232                                              size_t count)
 233 {
 234         struct tevent_req *req;
 235         struct tsocket_writev_queue_state *state;
 236         bool ok;
 237
 238         req = tevent_req_create(mem_ctx, &state,
 239                                 struct tsocket_writev_queue_state);
 240         if (!req) {
 241                 return NULL;
 242         }
 243
 244         state->caller.sock      = sock;
 245         state->caller.vector    = vector;
 246         state->caller.count     = count;
 247         state->ret              = -1;
 248
 249         ok = tevent_queue_add(queue,
 250                               sock->event.ctx,
 251                               req,
 252                               tsocket_writev_queue_trigger,
 253                               NULL);
 254         if (!ok) {
 255                 tevent_req_nomem(NULL, req);
 256                 goto post;
 257         }
 258
 259         return req;
 260
 261  post:
 262         return tevent_req_post(req, sock->event.ctx);
 263 }
 264
 265 static void tsocket_writev_queue_trigger(struct tevent_req *req,
 266                                          void *private_data)
 267 {
 268         struct tsocket_writev_queue_state *state = tevent_req_data(req,
 269                                         struct tsocket_writev_queue_state);
 270         struct tevent_req *subreq;
 271
 272         subreq = tsocket_writev_send(state->caller.sock,
 273                                      state,
 274                                      state->caller.vector,
 275                                      state->caller.count);
 276         if (tevent_req_nomem(subreq, req)) {
 277                 return;
 278         }
 279         tevent_req_set_callback(subreq, tsocket_writev_queue_done ,req);
 280 }
 281
 282 static void tsocket_writev_queue_done(struct tevent_req *subreq)
 283 {
 284         struct tevent_req *req = tevent_req_callback_data(subreq,
 285                                  struct tevent_req);
 286         struct tsocket_writev_queue_state *state = tevent_req_data(req,
 287                                         struct tsocket_writev_queue_state);
 288         int ret;
 289         int sys_errno;
 290
 291         ret = tsocket_writev_recv(subreq, &sys_errno);
 292         talloc_free(subreq);
 293         if (ret == -1) {
 294                 tevent_req_error(req, sys_errno);
 295                 return;
 296         }
 297         state->ret = ret;
 298
 299         tevent_req_done(req);
 300 }
 301
 302 int tsocket_writev_queue_recv(struct tevent_req *req, int *perrno)
 303 {
 304         struct tsocket_writev_queue_state *state = tevent_req_data(req,
 305                                         struct tsocket_writev_queue_state);
 306         int ret;
 307
 308         ret = tsocket_simple_int_recv(req, perrno);
 309         if (ret == 0) {
 310                 ret = state->ret;
 311         }
 312
 313         tevent_req_received(req);
 314         return ret;
 315 }
 316