s3: Many pthreadpool fixes
[Samba.git] / source3 / lib / fncall.c
blobbd06be2cd8c09fc720529995394fba4057fd899b
1 /*
2 * Unix SMB/CIFS implementation.
3 * Async fn calls
4 * Copyright (C) Volker Lendecke 2009
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.
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.
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/>.
20 #include "includes.h"
22 #if WITH_PTHREADPOOL
24 #include "lib/pthreadpool/pthreadpool.h"
26 struct fncall_state {
27 struct fncall_context *ctx;
28 int job_id;
29 bool done;
31 void *private_parent;
32 void *job_private;
35 struct fncall_context {
36 struct pthreadpool *pool;
37 int next_job_id;
38 int sig_fd;
39 struct tevent_req **pending;
41 struct fncall_state **orphaned;
42 int num_orphaned;
44 struct fd_event *fde;
47 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
48 uint16_t flags, void *private_data);
50 static int fncall_context_destructor(struct fncall_context *ctx)
52 while (talloc_array_length(ctx->pending) != 0) {
53 /* No TALLOC_FREE here */
54 talloc_free(ctx->pending[0]);
57 while (ctx->num_orphaned != 0) {
59 * We've got jobs in the queue for which the tevent_req has
60 * been finished already. Wait for all of them to finish.
62 fncall_handler(NULL, NULL, TEVENT_FD_READ, ctx);
65 pthreadpool_destroy(ctx->pool);
66 ctx->pool = NULL;
68 return 0;
71 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
72 int max_threads)
74 struct fncall_context *ctx;
75 int ret;
77 ctx = talloc_zero(mem_ctx, struct fncall_context);
78 if (ctx == NULL) {
79 return NULL;
82 ret = pthreadpool_init(max_threads, &ctx->pool);
83 if (ret != 0) {
84 TALLOC_FREE(ctx);
85 return NULL;
87 talloc_set_destructor(ctx, fncall_context_destructor);
89 ctx->sig_fd = pthreadpool_sig_fd(ctx->pool);
90 if (ctx->sig_fd == -1) {
91 TALLOC_FREE(ctx);
92 return NULL;
95 return ctx;
98 static int fncall_next_job_id(struct fncall_context *ctx)
100 int num_pending = talloc_array_length(ctx->pending);
101 int result;
103 while (true) {
104 int i;
106 result = ctx->next_job_id++;
107 if (result == 0) {
108 continue;
111 for (i=0; i<num_pending; i++) {
112 struct fncall_state *state = tevent_req_data(
113 ctx->pending[i], struct fncall_state);
115 if (result == state->job_id) {
116 break;
119 if (i == num_pending) {
120 return result;
125 static void fncall_unset_pending(struct tevent_req *req);
126 static int fncall_destructor(struct tevent_req *req);
128 static bool fncall_set_pending(struct tevent_req *req,
129 struct fncall_context *ctx,
130 struct tevent_context *ev)
132 struct tevent_req **pending;
133 int num_pending, orphaned_array_length;
135 num_pending = talloc_array_length(ctx->pending);
137 pending = talloc_realloc(ctx, ctx->pending, struct tevent_req *,
138 num_pending+1);
139 if (pending == NULL) {
140 return false;
142 pending[num_pending] = req;
143 num_pending += 1;
144 ctx->pending = pending;
145 talloc_set_destructor(req, fncall_destructor);
148 * Make sure that the orphaned array of fncall_state structs has
149 * enough space. A job can change from pending to orphaned in
150 * fncall_destructor, and to fail in a talloc destructor should be
151 * avoided if possible.
154 orphaned_array_length = talloc_array_length(ctx->orphaned);
155 if (num_pending > orphaned_array_length) {
156 struct fncall_state **orphaned;
158 orphaned = talloc_realloc(ctx, ctx->orphaned,
159 struct fncall_state *,
160 orphaned_array_length + 1);
161 if (orphaned == NULL) {
162 fncall_unset_pending(req);
163 return false;
165 ctx->orphaned = orphaned;
168 if (ctx->fde != NULL) {
169 return true;
172 ctx->fde = tevent_add_fd(ev, ctx->pending, ctx->sig_fd, TEVENT_FD_READ,
173 fncall_handler, ctx);
174 if (ctx->fde == NULL) {
175 fncall_unset_pending(req);
176 return false;
178 return true;
181 static void fncall_unset_pending(struct tevent_req *req)
183 struct fncall_state *state = tevent_req_data(req, struct fncall_state);
184 struct fncall_context *ctx = state->ctx;
185 int num_pending = talloc_array_length(ctx->pending);
186 int i;
188 if (num_pending == 1) {
189 TALLOC_FREE(ctx->fde);
190 TALLOC_FREE(ctx->pending);
191 return;
194 for (i=0; i<num_pending; i++) {
195 if (req == ctx->pending[i]) {
196 break;
199 if (i == num_pending) {
200 return;
202 if (num_pending > 1) {
203 ctx->pending[i] = ctx->pending[num_pending-1];
205 ctx->pending = talloc_realloc(NULL, ctx->pending, struct tevent_req *,
206 num_pending - 1);
209 static int fncall_destructor(struct tevent_req *req)
211 struct fncall_state *state = tevent_req_data(
212 req, struct fncall_state);
213 struct fncall_context *ctx = state->ctx;
215 fncall_unset_pending(req);
217 if (state->done) {
218 return 0;
222 * Keep around the state of the deleted request until the request has
223 * finished in the helper thread. fncall_handler will destroy it.
225 ctx->orphaned[ctx->num_orphaned] = talloc_move(ctx->orphaned, &state);
226 ctx->num_orphaned += 1;
228 return 0;
231 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
232 struct fncall_context *ctx,
233 void (*fn)(void *private_data),
234 void *private_data)
236 struct tevent_req *req;
237 struct fncall_state *state;
238 int ret;
240 req = tevent_req_create(mem_ctx, &state, struct fncall_state);
241 if (req == NULL) {
242 return NULL;
244 state->ctx = ctx;
245 state->job_id = fncall_next_job_id(state->ctx);
246 state->done = false;
249 * We need to keep the private data we handed out to the thread around
250 * as long as the job is not finished. This is a bit of an abstraction
251 * violation, because the "req->state1->subreq->state2" (we're
252 * "subreq", "req" is the request our caller creates) is broken to
253 * "ctx->state2->state1", but we are right now in the destructor for
254 * "subreq2", so what can we do. We need to keep state1 around,
255 * otherwise the helper thread will have no place to put its results.
258 state->private_parent = talloc_parent(private_data);
259 state->job_private = talloc_move(state, &private_data);
261 ret = pthreadpool_add_job(state->ctx->pool, state->job_id, fn,
262 state->job_private);
263 if (ret == -1) {
264 tevent_req_error(req, errno);
265 return tevent_req_post(req, ev);
267 if (!fncall_set_pending(req, state->ctx, ev)) {
268 tevent_req_nomem(NULL, req);
269 return tevent_req_post(req, ev);
271 return req;
274 static void fncall_handler(struct tevent_context *ev, struct tevent_fd *fde,
275 uint16_t flags, void *private_data)
277 struct fncall_context *ctx = talloc_get_type_abort(
278 private_data, struct fncall_context);
279 int i, num_pending;
280 int job_id;
282 job_id = pthreadpool_finished_job(ctx->pool);
283 if (job_id <= 0) {
284 return;
287 num_pending = talloc_array_length(ctx->pending);
289 for (i=0; i<num_pending; i++) {
290 struct fncall_state *state = tevent_req_data(
291 ctx->pending[i], struct fncall_state);
293 if (job_id == state->job_id) {
294 state->done = true;
295 talloc_move(state->private_parent,
296 &state->job_private);
297 tevent_req_done(ctx->pending[i]);
298 return;
302 for (i=0; i<ctx->num_orphaned; i++) {
303 if (job_id == ctx->orphaned[i]->job_id) {
304 break;
307 if (i == ctx->num_orphaned) {
308 return;
311 TALLOC_FREE(ctx->orphaned[i]);
313 if (i < ctx->num_orphaned-1) {
314 ctx->orphaned[i] = ctx->orphaned[ctx->num_orphaned-1];
316 ctx->num_orphaned -= 1;
319 int fncall_recv(struct tevent_req *req, int *perr)
321 if (tevent_req_is_unix_error(req, perr)) {
322 return -1;
324 return 0;
327 #else /* WITH_PTHREADPOOL */
329 struct fncall_context {
330 uint8_t dummy;
333 struct fncall_context *fncall_context_init(TALLOC_CTX *mem_ctx,
334 int max_threads)
336 return talloc(mem_ctx, struct fncall_context);
339 struct fncall_state {
340 uint8_t dummy;
343 struct tevent_req *fncall_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
344 struct fncall_context *ctx,
345 void (*fn)(void *private_data),
346 void *private_data)
348 struct tevent_req *req;
349 struct fncall_state *state;
351 req = tevent_req_create(mem_ctx, &state, struct fncall_state);
352 if (req == NULL) {
353 return NULL;
355 fn(private_data);
356 tevent_req_post(req, ev);
357 return req;
360 int fncall_recv(struct tevent_req *req, int *perr)
362 return 0;
365 #endif