s3: piddir creation fix part 2.
[ira/wip.git] / lib / tsocket / tsocket_helpers.c
1 /*
2    Unix SMB/CIFS implementation.
3
4    Copyright (C) Stefan Metzmacher 2009
5
6      ** NOTE! The following LGPL license applies to the tsocket
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/filesys.h"
26 #include "tsocket.h"
27 #include "tsocket_internal.h"
28
29 struct tdgram_sendto_queue_state {
30         /* this structs are owned by the caller */
31         struct {
32                 struct tevent_context *ev;
33                 struct tdgram_context *dgram;
34                 const uint8_t *buf;
35                 size_t len;
36                 const struct tsocket_address *dst;
37         } caller;
38         ssize_t ret;
39 };
40
41 static void tdgram_sendto_queue_trigger(struct tevent_req *req,
42                                          void *private_data);
43 static void tdgram_sendto_queue_done(struct tevent_req *subreq);
44
45 struct tevent_req *tdgram_sendto_queue_send(TALLOC_CTX *mem_ctx,
46                                             struct tevent_context *ev,
47                                             struct tdgram_context *dgram,
48                                             struct tevent_queue *queue,
49                                             const uint8_t *buf,
50                                             size_t len,
51                                             struct tsocket_address *dst)
52 {
53         struct tevent_req *req;
54         struct tdgram_sendto_queue_state *state;
55         struct tevent_queue_entry *e;
56
57         req = tevent_req_create(mem_ctx, &state,
58                                 struct tdgram_sendto_queue_state);
59         if (!req) {
60                 return NULL;
61         }
62
63         state->caller.ev        = ev;
64         state->caller.dgram     = dgram;
65         state->caller.buf       = buf;
66         state->caller.len       = len;
67         state->caller.dst       = dst;
68         state->ret              = -1;
69
70         /*
71          * we use tevent_queue_add_optimize_empty() with allow_direct
72          * in order to optimize for the empty queue case.
73          */
74         e = tevent_queue_add_optimize_empty(
75                                 queue,
76                                 ev,
77                                 req,
78                                 tdgram_sendto_queue_trigger,
79                                 NULL);
80         if (tevent_req_nomem(e, req)) {
81                 return tevent_req_post(req, ev);
82         }
83         if (!tevent_req_is_in_progress(req)) {
84                 return tevent_req_post(req, ev);
85         }
86
87         return req;
88 }
89
90 static void tdgram_sendto_queue_trigger(struct tevent_req *req,
91                                          void *private_data)
92 {
93         struct tdgram_sendto_queue_state *state = tevent_req_data(req,
94                                         struct tdgram_sendto_queue_state);
95         struct tevent_req *subreq;
96
97         subreq = tdgram_sendto_send(state,
98                                     state->caller.ev,
99                                     state->caller.dgram,
100                                     state->caller.buf,
101                                     state->caller.len,
102                                     state->caller.dst);
103         if (tevent_req_nomem(subreq, req)) {
104                 return;
105         }
106         tevent_req_set_callback(subreq, tdgram_sendto_queue_done, req);
107 }
108
109 static void tdgram_sendto_queue_done(struct tevent_req *subreq)
110 {
111         struct tevent_req *req = tevent_req_callback_data(subreq,
112                                  struct tevent_req);
113         struct tdgram_sendto_queue_state *state = tevent_req_data(req,
114                                         struct tdgram_sendto_queue_state);
115         ssize_t ret;
116         int sys_errno;
117
118         ret = tdgram_sendto_recv(subreq, &sys_errno);
119         talloc_free(subreq);
120         if (ret == -1) {
121                 tevent_req_error(req, sys_errno);
122                 return;
123         }
124         state->ret = ret;
125
126         tevent_req_done(req);
127 }
128
129 ssize_t tdgram_sendto_queue_recv(struct tevent_req *req, int *perrno)
130 {
131         struct tdgram_sendto_queue_state *state = tevent_req_data(req,
132                                         struct tdgram_sendto_queue_state);
133         ssize_t ret;
134
135         ret = tsocket_simple_int_recv(req, perrno);
136         if (ret == 0) {
137                 ret = state->ret;
138         }
139
140         tevent_req_received(req);
141         return ret;
142 }
143
144 struct tstream_readv_pdu_state {
145         /* this structs are owned by the caller */
146         struct {
147                 struct tevent_context *ev;
148                 struct tstream_context *stream;
149                 tstream_readv_pdu_next_vector_t next_vector_fn;
150                 void *next_vector_private;
151         } caller;
152
153         /*
154          * Each call to the callback resets iov and count
155          * the callback allocated the iov as child of our state,
156          * that means we are allowed to modify and free it.
157          *
158          * we should call the callback every time we filled the given
159          * vector and ask for a new vector. We return if the callback
160          * ask for 0 bytes.
161          */
162         struct iovec *vector;
163         size_t count;
164
165         /*
166          * the total number of bytes we read,
167          * the return value of the _recv function
168          */
169         int total_read;
170 };
171
172 static void tstream_readv_pdu_ask_for_next_vector(struct tevent_req *req);
173 static void tstream_readv_pdu_readv_done(struct tevent_req *subreq);
174
175 struct tevent_req *tstream_readv_pdu_send(TALLOC_CTX *mem_ctx,
176                                 struct tevent_context *ev,
177                                 struct tstream_context *stream,
178                                 tstream_readv_pdu_next_vector_t next_vector_fn,
179                                 void *next_vector_private)
180 {
181         struct tevent_req *req;
182         struct tstream_readv_pdu_state *state;
183
184         req = tevent_req_create(mem_ctx, &state,
185                                 struct tstream_readv_pdu_state);
186         if (!req) {
187                 return NULL;
188         }
189
190         state->caller.ev                        = ev;
191         state->caller.stream                    = stream;
192         state->caller.next_vector_fn            = next_vector_fn;
193         state->caller.next_vector_private       = next_vector_private;
194
195         state->vector           = NULL;
196         state->count            = 0;
197         state->total_read       = 0;
198
199         tstream_readv_pdu_ask_for_next_vector(req);
200         if (!tevent_req_is_in_progress(req)) {
201                 goto post;
202         }
203
204         return req;
205
206  post:
207         return tevent_req_post(req, ev);
208 }
209
210 static void tstream_readv_pdu_ask_for_next_vector(struct tevent_req *req)
211 {
212         struct tstream_readv_pdu_state *state = tevent_req_data(req,
213                                             struct tstream_readv_pdu_state);
214         int ret;
215         size_t to_read = 0;
216         size_t i;
217         struct tevent_req *subreq;
218
219         TALLOC_FREE(state->vector);
220         state->count = 0;
221
222         ret = state->caller.next_vector_fn(state->caller.stream,
223                                            state->caller.next_vector_private,
224                                            state, &state->vector, &state->count);
225         if (ret == -1) {
226                 tevent_req_error(req, errno);
227                 return;
228         }
229
230         if (state->count == 0) {
231                 tevent_req_done(req);
232                 return;
233         }
234
235         for (i=0; i < state->count; i++) {
236                 size_t tmp = to_read;
237                 tmp += state->vector[i].iov_len;
238
239                 if (tmp < to_read) {
240                         tevent_req_error(req, EMSGSIZE);
241                         return;
242                 }
243
244                 to_read = tmp;
245         }
246
247         /*
248          * this is invalid the next vector function should have
249          * reported count == 0.
250          */
251         if (to_read == 0) {
252                 tevent_req_error(req, EINVAL);
253                 return;
254         }
255
256         if (state->total_read + to_read < state->total_read) {
257                 tevent_req_error(req, EMSGSIZE);
258                 return;
259         }
260
261         subreq = tstream_readv_send(state,
262                                     state->caller.ev,
263                                     state->caller.stream,
264                                     state->vector,
265                                     state->count);
266         if (tevent_req_nomem(subreq, req)) {
267                 return;
268         }
269         tevent_req_set_callback(subreq, tstream_readv_pdu_readv_done, req);
270 }
271
272 static void tstream_readv_pdu_readv_done(struct tevent_req *subreq)
273 {
274         struct tevent_req *req = tevent_req_callback_data(subreq,
275                                  struct tevent_req);
276         struct tstream_readv_pdu_state *state = tevent_req_data(req,
277                                             struct tstream_readv_pdu_state);
278         int ret;
279         int sys_errno;
280
281         ret = tstream_readv_recv(subreq, &sys_errno);
282         if (ret == -1) {
283                 tevent_req_error(req, sys_errno);
284                 return;
285         }
286
287         state->total_read += ret;
288
289         /* ask the callback for a new vector we should fill */
290         tstream_readv_pdu_ask_for_next_vector(req);
291 }
292
293 int tstream_readv_pdu_recv(struct tevent_req *req, int *perrno)
294 {
295         struct tstream_readv_pdu_state *state = tevent_req_data(req,
296                                             struct tstream_readv_pdu_state);
297         int ret;
298
299         ret = tsocket_simple_int_recv(req, perrno);
300         if (ret == 0) {
301                 ret = state->total_read;
302         }
303
304         tevent_req_received(req);
305         return ret;
306 }
307
308 struct tstream_readv_pdu_queue_state {
309         /* this structs are owned by the caller */
310         struct {
311                 struct tevent_context *ev;
312                 struct tstream_context *stream;
313                 tstream_readv_pdu_next_vector_t next_vector_fn;
314                 void *next_vector_private;
315         } caller;
316         int ret;
317 };
318
319 static void tstream_readv_pdu_queue_trigger(struct tevent_req *req,
320                                          void *private_data);
321 static void tstream_readv_pdu_queue_done(struct tevent_req *subreq);
322
323 struct tevent_req *tstream_readv_pdu_queue_send(TALLOC_CTX *mem_ctx,
324                                 struct tevent_context *ev,
325                                 struct tstream_context *stream,
326                                 struct tevent_queue *queue,
327                                 tstream_readv_pdu_next_vector_t next_vector_fn,
328                                 void *next_vector_private)
329 {
330         struct tevent_req *req;
331         struct tstream_readv_pdu_queue_state *state;
332         struct tevent_queue_entry *e;
333
334         req = tevent_req_create(mem_ctx, &state,
335                                 struct tstream_readv_pdu_queue_state);
336         if (!req) {
337                 return NULL;
338         }
339
340         state->caller.ev                        = ev;
341         state->caller.stream                    = stream;
342         state->caller.next_vector_fn            = next_vector_fn;
343         state->caller.next_vector_private       = next_vector_private;
344         state->ret                              = -1;
345
346         /*
347          * we use tevent_queue_add_optimize_empty() with allow_direct
348          * in order to optimize for the empty queue case.
349          */
350         e = tevent_queue_add_optimize_empty(
351                                 queue,
352                                 ev,
353                                 req,
354                                 tstream_readv_pdu_queue_trigger,
355                                 NULL);
356         if (tevent_req_nomem(e, req)) {
357                 return tevent_req_post(req, ev);
358         }
359         if (!tevent_req_is_in_progress(req)) {
360                 return tevent_req_post(req, ev);
361         }
362
363         return req;
364 }
365
366 static void tstream_readv_pdu_queue_trigger(struct tevent_req *req,
367                                          void *private_data)
368 {
369         struct tstream_readv_pdu_queue_state *state = tevent_req_data(req,
370                                         struct tstream_readv_pdu_queue_state);
371         struct tevent_req *subreq;
372
373         subreq = tstream_readv_pdu_send(state,
374                                         state->caller.ev,
375                                         state->caller.stream,
376                                         state->caller.next_vector_fn,
377                                         state->caller.next_vector_private);
378         if (tevent_req_nomem(subreq, req)) {
379                 return;
380         }
381         tevent_req_set_callback(subreq, tstream_readv_pdu_queue_done ,req);
382 }
383
384 static void tstream_readv_pdu_queue_done(struct tevent_req *subreq)
385 {
386         struct tevent_req *req = tevent_req_callback_data(subreq,
387                                  struct tevent_req);
388         struct tstream_readv_pdu_queue_state *state = tevent_req_data(req,
389                                         struct tstream_readv_pdu_queue_state);
390         int ret;
391         int sys_errno;
392
393         ret = tstream_readv_pdu_recv(subreq, &sys_errno);
394         talloc_free(subreq);
395         if (ret == -1) {
396                 tevent_req_error(req, sys_errno);
397                 return;
398         }
399         state->ret = ret;
400
401         tevent_req_done(req);
402 }
403
404 int tstream_readv_pdu_queue_recv(struct tevent_req *req, int *perrno)
405 {
406         struct tstream_readv_pdu_queue_state *state = tevent_req_data(req,
407                                         struct tstream_readv_pdu_queue_state);
408         int ret;
409
410         ret = tsocket_simple_int_recv(req, perrno);
411         if (ret == 0) {
412                 ret = state->ret;
413         }
414
415         tevent_req_received(req);
416         return ret;
417 }
418
419 struct tstream_writev_queue_state {
420         /* this structs are owned by the caller */
421         struct {
422                 struct tevent_context *ev;
423                 struct tstream_context *stream;
424                 const struct iovec *vector;
425                 size_t count;
426         } caller;
427         int ret;
428 };
429
430 static void tstream_writev_queue_trigger(struct tevent_req *req,
431                                          void *private_data);
432 static void tstream_writev_queue_done(struct tevent_req *subreq);
433
434 struct tevent_req *tstream_writev_queue_send(TALLOC_CTX *mem_ctx,
435                                              struct tevent_context *ev,
436                                              struct tstream_context *stream,
437                                              struct tevent_queue *queue,
438                                              const struct iovec *vector,
439                                              size_t count)
440 {
441         struct tevent_req *req;
442         struct tstream_writev_queue_state *state;
443         struct tevent_queue_entry *e;
444
445         req = tevent_req_create(mem_ctx, &state,
446                                 struct tstream_writev_queue_state);
447         if (!req) {
448                 return NULL;
449         }
450
451         state->caller.ev        = ev;
452         state->caller.stream    = stream;
453         state->caller.vector    = vector;
454         state->caller.count     = count;
455         state->ret              = -1;
456
457         /*
458          * we use tevent_queue_add_optimize_empty() with allow_direct
459          * in order to optimize for the empty queue case.
460          */
461         e = tevent_queue_add_optimize_empty(
462                                 queue,
463                                 ev,
464                                 req,
465                                 tstream_writev_queue_trigger,
466                                 NULL);
467         if (tevent_req_nomem(e, req)) {
468                 return tevent_req_post(req, ev);
469         }
470         if (!tevent_req_is_in_progress(req)) {
471                 return tevent_req_post(req, ev);
472         }
473
474         return req;
475 }
476
477 static void tstream_writev_queue_trigger(struct tevent_req *req,
478                                          void *private_data)
479 {
480         struct tstream_writev_queue_state *state = tevent_req_data(req,
481                                         struct tstream_writev_queue_state);
482         struct tevent_req *subreq;
483
484         subreq = tstream_writev_send(state,
485                                      state->caller.ev,
486                                      state->caller.stream,
487                                      state->caller.vector,
488                                      state->caller.count);
489         if (tevent_req_nomem(subreq, req)) {
490                 return;
491         }
492         tevent_req_set_callback(subreq, tstream_writev_queue_done ,req);
493 }
494
495 static void tstream_writev_queue_done(struct tevent_req *subreq)
496 {
497         struct tevent_req *req = tevent_req_callback_data(subreq,
498                                  struct tevent_req);
499         struct tstream_writev_queue_state *state = tevent_req_data(req,
500                                         struct tstream_writev_queue_state);
501         int ret;
502         int sys_errno;
503
504         ret = tstream_writev_recv(subreq, &sys_errno);
505         talloc_free(subreq);
506         if (ret == -1) {
507                 tevent_req_error(req, sys_errno);
508                 return;
509         }
510         state->ret = ret;
511
512         tevent_req_done(req);
513 }
514
515 int tstream_writev_queue_recv(struct tevent_req *req, int *perrno)
516 {
517         struct tstream_writev_queue_state *state = tevent_req_data(req,
518                                         struct tstream_writev_queue_state);
519         int ret;
520
521         ret = tsocket_simple_int_recv(req, perrno);
522         if (ret == 0) {
523                 ret = state->ret;
524         }
525
526         tevent_req_received(req);
527         return ret;
528 }
529