lib/tsocket: add a fast path to tsocket_writev_send/recv()
[ira/wip.git] / 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