s3-tevent: only include ../lib/util/tevent wrappers where needed.
[bbaumbach/samba-autobuild/.git] / source3 / lib / fncall.c
1 /*
2  * Unix SMB/CIFS implementation.
3  * Async fn calls
4  * Copyright (C) Volker Lendecke 2009
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/util/tevent_unix.h"
22
23 #if WITH_PTHREADPOOL
24
25 #include "lib/pthreadpool/pthreadpool.h"
26
27 struct fncall_state {
28         struct fncall_context *ctx;
29         int job_id;
30         bool done;
31
32         void *private_parent;
33         void *job_private;
34 };
35
36 struct fncall_context {
37         struct pthreadpool *pool;
38         int next_job_id;
39         int sig_fd;
40         struct tevent_req **pending;
41
42         struct fncall_state **orphaned;
43         int num_orphaned;
44
45         struct fd_event *fde;
46 };
47
48 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
49                            uint16_t flags, void *private_data);
50
51 static int fncall_context_destructor(struct fncall_context *ctx)
52 {
53         while (talloc_array_length(ctx->pending) != 0) {
54                 /* No TALLOC_FREE here */
55                 talloc_free(ctx->pending[0]);
56         }
57
58         while (ctx->num_orphaned != 0) {
59                 /*
60                  * We've got jobs in the queue for which the tevent_req has
61                  * been finished already. Wait for all of them to finish.
62                  */
63                 fncall_handler(NULL, NULL, TEVENT_FD_READ, ctx);
64         }
65
66         pthreadpool_destroy(ctx->pool);
67         ctx->pool = NULL;
68
69         return 0;
70 }
71
72 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
73                                            int max_threads)
74 {
75         struct fncall_context *ctx;
76         int ret;
77
78         ctx = talloc_zero(mem_ctx, struct fncall_context);
79         if (ctx == NULL) {
80                 return NULL;
81         }
82
83         ret = pthreadpool_init(max_threads, &ctx->pool);
84         if (ret != 0) {
85                 TALLOC_FREE(ctx);
86                 return NULL;
87         }
88         talloc_set_destructor(ctx, fncall_context_destructor);
89
90         ctx->sig_fd = pthreadpool_signal_fd(ctx->pool);
91         if (ctx->sig_fd == -1) {
92                 TALLOC_FREE(ctx);
93                 return NULL;
94         }
95
96         return ctx;
97 }
98
99 static int fncall_next_job_id(struct fncall_context *ctx)
100 {
101         int num_pending = talloc_array_length(ctx->pending);
102         int result;
103
104         while (true) {
105                 int i;
106
107                 result = ctx->next_job_id++;
108                 if (result == 0) {
109                         continue;
110                 }
111
112                 for (i=0; i<num_pending; i++) {
113                         struct fncall_state *state = tevent_req_data(
114                                 ctx->pending[i], struct fncall_state);
115
116                         if (result == state->job_id) {
117                                 break;
118                         }
119                 }
120                 if (i == num_pending) {
121                         return result;
122                 }
123         }
124 }
125
126 static void fncall_unset_pending(struct tevent_req *req);
127 static int fncall_destructor(struct tevent_req *req);
128
129 static bool fncall_set_pending(struct tevent_req *req,
130                                struct fncall_context *ctx,
131                                struct tevent_context *ev)
132 {
133         struct tevent_req **pending;
134         int num_pending, orphaned_array_length;
135
136         num_pending = talloc_array_length(ctx->pending);
137
138         pending = talloc_realloc(ctx, ctx->pending, struct tevent_req *,
139                                  num_pending+1);
140         if (pending == NULL) {
141                 return false;
142         }
143         pending[num_pending] = req;
144         num_pending += 1;
145         ctx->pending = pending;
146         talloc_set_destructor(req, fncall_destructor);
147
148         /*
149          * Make sure that the orphaned array of fncall_state structs has
150          * enough space. A job can change from pending to orphaned in
151          * fncall_destructor, and to fail in a talloc destructor should be
152          * avoided if possible.
153          */
154
155         orphaned_array_length = talloc_array_length(ctx->orphaned);
156         if (num_pending > orphaned_array_length) {
157                 struct fncall_state **orphaned;
158
159                 orphaned = talloc_realloc(ctx, ctx->orphaned,
160                                           struct fncall_state *,
161                                           orphaned_array_length + 1);
162                 if (orphaned == NULL) {
163                         fncall_unset_pending(req);
164                         return false;
165                 }
166                 ctx->orphaned = orphaned;
167         }
168
169         if (ctx->fde != NULL) {
170                 return true;
171         }
172
173         ctx->fde = tevent_add_fd(ev, ctx->pending, ctx->sig_fd, TEVENT_FD_READ,
174                                  fncall_handler, ctx);
175         if (ctx->fde == NULL) {
176                 fncall_unset_pending(req);
177                 return false;
178         }
179         return true;
180 }
181
182 static void fncall_unset_pending(struct tevent_req *req)
183 {
184         struct fncall_state *state = tevent_req_data(req, struct fncall_state);
185         struct fncall_context *ctx = state->ctx;
186         int num_pending = talloc_array_length(ctx->pending);
187         int i;
188
189         if (num_pending == 1) {
190                 TALLOC_FREE(ctx->fde);
191                 TALLOC_FREE(ctx->pending);
192                 return;
193         }
194
195         for (i=0; i<num_pending; i++) {
196                 if (req == ctx->pending[i]) {
197                         break;
198                 }
199         }
200         if (i == num_pending) {
201                 return;
202         }
203         if (num_pending > 1) {
204                 ctx->pending[i] = ctx->pending[num_pending-1];
205         }
206         ctx->pending = talloc_realloc(NULL, ctx->pending, struct tevent_req *,
207                                       num_pending - 1);
208 }
209
210 static int fncall_destructor(struct tevent_req *req)
211 {
212         struct fncall_state *state = tevent_req_data(
213                 req, struct fncall_state);
214         struct fncall_context *ctx = state->ctx;
215
216         fncall_unset_pending(req);
217
218         if (state->done) {
219                 return 0;
220         }
221
222         /*
223          * Keep around the state of the deleted request until the request has
224          * finished in the helper thread. fncall_handler will destroy it.
225          */
226         ctx->orphaned[ctx->num_orphaned] = talloc_move(ctx->orphaned, &state);
227         ctx->num_orphaned += 1;
228
229         return 0;
230 }
231
232 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
233                                struct fncall_context *ctx,
234                                void (*fn)(void *private_data),
235                                void *private_data)
236 {
237         struct tevent_req *req;
238         struct fncall_state *state;
239         int ret;
240
241         req = tevent_req_create(mem_ctx, &state, struct fncall_state);
242         if (req == NULL) {
243                 return NULL;
244         }
245         state->ctx = ctx;
246         state->job_id = fncall_next_job_id(state->ctx);
247         state->done = false;
248
249         /*
250          * We need to keep the private data we handed out to the thread around
251          * as long as the job is not finished. This is a bit of an abstraction
252          * violation, because the "req->state1->subreq->state2" (we're
253          * "subreq", "req" is the request our caller creates) is broken to
254          * "ctx->state2->state1", but we are right now in the destructor for
255          * "subreq2", so what can we do. We need to keep state1 around,
256          * otherwise the helper thread will have no place to put its results.
257          */
258
259         state->private_parent = talloc_parent(private_data);
260         state->job_private = talloc_move(state, &private_data);
261
262         ret = pthreadpool_add_job(state->ctx->pool, state->job_id, fn,
263                                   state->job_private);
264         if (ret == -1) {
265                 tevent_req_error(req, errno);
266                 return tevent_req_post(req, ev);
267         }
268         if (!fncall_set_pending(req, state->ctx, ev)) {
269                 tevent_req_nomem(NULL, req);
270                 return tevent_req_post(req, ev);
271         }
272         return req;
273 }
274
275 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
276                            uint16_t flags, void *private_data)
277 {
278         struct fncall_context *ctx = talloc_get_type_abort(
279                 private_data, struct fncall_context);
280         int i, num_pending;
281         int job_id;
282
283         job_id = pthreadpool_finished_job(ctx->pool);
284         if (job_id <= 0) {
285                 return;
286         }
287
288         num_pending = talloc_array_length(ctx->pending);
289
290         for (i=0; i<num_pending; i++) {
291                 struct fncall_state *state = tevent_req_data(
292                         ctx->pending[i], struct fncall_state);
293
294                 if (job_id == state->job_id) {
295                         state->done = true;
296                         talloc_move(state->private_parent,
297                                     &state->job_private);
298                         tevent_req_done(ctx->pending[i]);
299                         return;
300                 }
301         }
302
303         for (i=0; i<ctx->num_orphaned; i++) {
304                 if (job_id == ctx->orphaned[i]->job_id) {
305                         break;
306                 }
307         }
308         if (i == ctx->num_orphaned) {
309                 return;
310         }
311
312         TALLOC_FREE(ctx->orphaned[i]);
313
314         if (i < ctx->num_orphaned-1) {
315                 ctx->orphaned[i] = ctx->orphaned[ctx->num_orphaned-1];
316         }
317         ctx->num_orphaned -= 1;
318 }
319
320 int fncall_recv(struct tevent_req *req, int *perr)
321 {
322         if (tevent_req_is_unix_error(req, perr)) {
323                 return -1;
324         }
325         return 0;
326 }
327
328 #else  /* WITH_PTHREADPOOL */
329
330 struct fncall_context {
331         uint8_t dummy;
332 };
333
334 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
335                                            int max_threads)
336 {
337         return talloc(mem_ctx, struct fncall_context);
338 }
339
340 struct fncall_state {
341         uint8_t dummy;
342 };
343
344 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
345                                struct fncall_context *ctx,
346                                void (*fn)(void *private_data),
347                                void *private_data)
348 {
349         struct tevent_req *req;
350         struct fncall_state *state;
351
352         req = tevent_req_create(mem_ctx, &state, struct fncall_state);
353         if (req == NULL) {
354                 return NULL;
355         }
356         fn(private_data);
357         tevent_req_post(req, ev);
358         return req;
359 }
360
361 int fncall_recv(struct tevent_req *req, int *perr)
362 {
363         return 0;
364 }
365
366 #endif