Line data Source code
1 : /*
2 : * Unix SMB/CIFS implementation.
3 : * threadpool implementation based on pthreads
4 : * Copyright (C) Volker Lendecke 2009,2011
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 "replace.h"
21 : #include "system/filesys.h"
22 : #include "pthreadpool_tevent.h"
23 : #include "pthreadpool.h"
24 : #include "lib/util/tevent_unix.h"
25 : #include "lib/util/dlinklist.h"
26 :
27 : struct pthreadpool_tevent_job_state;
28 :
29 : /*
30 : * We need one pthreadpool_tevent_glue object per unique combintaion of tevent
31 : * contexts and pthreadpool_tevent objects. Maintain a list of used tevent
32 : * contexts in a pthreadpool_tevent.
33 : */
34 : struct pthreadpool_tevent_glue {
35 : struct pthreadpool_tevent_glue *prev, *next;
36 : struct pthreadpool_tevent *pool; /* back-pointer to owning object. */
37 : /* Tuple we are keeping track of in this list. */
38 : struct tevent_context *ev;
39 : struct tevent_threaded_context *tctx;
40 : /* Pointer to link object owned by *ev. */
41 : struct pthreadpool_tevent_glue_ev_link *ev_link;
42 : };
43 :
44 : /*
45 : * The pthreadpool_tevent_glue_ev_link and its destructor ensure we remove the
46 : * tevent context from our list of active event contexts if the event context
47 : * is destroyed.
48 : * This structure is talloc()'ed from the struct tevent_context *, and is a
49 : * back-pointer allowing the related struct pthreadpool_tevent_glue object
50 : * to be removed from the struct pthreadpool_tevent glue list if the owning
51 : * tevent_context is talloc_free()'ed.
52 : */
53 : struct pthreadpool_tevent_glue_ev_link {
54 : struct pthreadpool_tevent_glue *glue;
55 : };
56 :
57 : struct pthreadpool_tevent {
58 : struct pthreadpool *pool;
59 : struct pthreadpool_tevent_glue *glue_list;
60 :
61 : struct pthreadpool_tevent_job_state *jobs;
62 : };
63 :
64 : struct pthreadpool_tevent_job_state {
65 : struct pthreadpool_tevent_job_state *prev, *next;
66 : struct pthreadpool_tevent *pool;
67 : struct tevent_context *ev;
68 : struct tevent_immediate *im;
69 : struct tevent_req *req;
70 :
71 : void (*fn)(void *private_data);
72 : void *private_data;
73 : };
74 :
75 : static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
76 :
77 : static int pthreadpool_tevent_job_signal(int jobid,
78 : void (*job_fn)(void *private_data),
79 : void *job_private_data,
80 : void *private_data);
81 :
82 76686 : int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
83 : struct pthreadpool_tevent **presult)
84 : {
85 : struct pthreadpool_tevent *pool;
86 : int ret;
87 :
88 76686 : pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
89 76686 : if (pool == NULL) {
90 0 : return ENOMEM;
91 : }
92 :
93 76686 : ret = pthreadpool_init(max_threads, &pool->pool,
94 : pthreadpool_tevent_job_signal, pool);
95 76686 : if (ret != 0) {
96 0 : TALLOC_FREE(pool);
97 0 : return ret;
98 : }
99 :
100 76686 : talloc_set_destructor(pool, pthreadpool_tevent_destructor);
101 :
102 76686 : *presult = pool;
103 76686 : return 0;
104 : }
105 :
106 20173 : size_t pthreadpool_tevent_max_threads(struct pthreadpool_tevent *pool)
107 : {
108 20173 : if (pool->pool == NULL) {
109 0 : return 0;
110 : }
111 :
112 20173 : return pthreadpool_max_threads(pool->pool);
113 : }
114 :
115 40242 : size_t pthreadpool_tevent_queued_jobs(struct pthreadpool_tevent *pool)
116 : {
117 40242 : if (pool->pool == NULL) {
118 0 : return 0;
119 : }
120 :
121 40242 : return pthreadpool_queued_jobs(pool->pool);
122 : }
123 :
124 109993 : static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
125 : {
126 : struct pthreadpool_tevent_job_state *state, *next;
127 109993 : struct pthreadpool_tevent_glue *glue = NULL;
128 : int ret;
129 :
130 109993 : ret = pthreadpool_stop(pool->pool);
131 109993 : if (ret != 0) {
132 0 : return ret;
133 : }
134 :
135 109994 : for (state = pool->jobs; state != NULL; state = next) {
136 1 : next = state->next;
137 1 : DLIST_REMOVE(pool->jobs, state);
138 1 : state->pool = NULL;
139 : }
140 :
141 : /*
142 : * Delete all the registered
143 : * tevent_context/tevent_threaded_context
144 : * pairs.
145 : */
146 113615 : for (glue = pool->glue_list; glue != NULL; glue = pool->glue_list) {
147 : /* The glue destructor removes it from the list */
148 3622 : TALLOC_FREE(glue);
149 : }
150 109993 : pool->glue_list = NULL;
151 :
152 109993 : ret = pthreadpool_destroy(pool->pool);
153 109993 : if (ret != 0) {
154 0 : return ret;
155 : }
156 109993 : pool->pool = NULL;
157 :
158 109993 : return 0;
159 : }
160 :
161 3899 : static int pthreadpool_tevent_glue_destructor(
162 : struct pthreadpool_tevent_glue *glue)
163 : {
164 3899 : if (glue->pool->glue_list != NULL) {
165 3899 : DLIST_REMOVE(glue->pool->glue_list, glue);
166 : }
167 :
168 : /* Ensure the ev_link destructor knows we're gone */
169 3899 : glue->ev_link->glue = NULL;
170 :
171 3899 : TALLOC_FREE(glue->ev_link);
172 3899 : TALLOC_FREE(glue->tctx);
173 :
174 3899 : return 0;
175 : }
176 :
177 : /*
178 : * Destructor called either explicitly from
179 : * pthreadpool_tevent_glue_destructor(), or indirectly
180 : * when owning tevent_context is destroyed.
181 : *
182 : * When called from pthreadpool_tevent_glue_destructor()
183 : * ev_link->glue is already NULL, so this does nothing.
184 : *
185 : * When called from talloc_free() of the owning
186 : * tevent_context we must ensure we also remove the
187 : * linked glue object from the list inside
188 : * struct pthreadpool_tevent.
189 : */
190 3899 : static int pthreadpool_tevent_glue_link_destructor(
191 : struct pthreadpool_tevent_glue_ev_link *ev_link)
192 : {
193 3899 : TALLOC_FREE(ev_link->glue);
194 3899 : return 0;
195 : }
196 :
197 204816 : static int pthreadpool_tevent_register_ev(struct pthreadpool_tevent *pool,
198 : struct tevent_context *ev)
199 : {
200 204816 : struct pthreadpool_tevent_glue *glue = NULL;
201 204816 : struct pthreadpool_tevent_glue_ev_link *ev_link = NULL;
202 :
203 : /*
204 : * See if this tevent_context was already registered by
205 : * searching the glue object list. If so we have nothing
206 : * to do here - we already have a tevent_context/tevent_threaded_context
207 : * pair.
208 : */
209 204823 : for (glue = pool->glue_list; glue != NULL; glue = glue->next) {
210 201915 : if (glue->ev == ev) {
211 172870 : return 0;
212 : }
213 : }
214 :
215 : /*
216 : * Event context not yet registered - create a new glue
217 : * object containing a tevent_context/tevent_threaded_context
218 : * pair and put it on the list to remember this registration.
219 : * We also need a link object to ensure the event context
220 : * can't go away without us knowing about it.
221 : */
222 2908 : glue = talloc_zero(pool, struct pthreadpool_tevent_glue);
223 2908 : if (glue == NULL) {
224 0 : return ENOMEM;
225 : }
226 2908 : *glue = (struct pthreadpool_tevent_glue) {
227 : .pool = pool,
228 : .ev = ev,
229 : };
230 2908 : talloc_set_destructor(glue, pthreadpool_tevent_glue_destructor);
231 :
232 : /*
233 : * Now allocate the link object to the event context. Note this
234 : * is allocated OFF THE EVENT CONTEXT ITSELF, so if the event
235 : * context is freed we are able to cleanup the glue object
236 : * in the link object destructor.
237 : */
238 :
239 2908 : ev_link = talloc_zero(ev, struct pthreadpool_tevent_glue_ev_link);
240 2908 : if (ev_link == NULL) {
241 0 : TALLOC_FREE(glue);
242 0 : return ENOMEM;
243 : }
244 2908 : ev_link->glue = glue;
245 2908 : talloc_set_destructor(ev_link, pthreadpool_tevent_glue_link_destructor);
246 :
247 2908 : glue->ev_link = ev_link;
248 :
249 : #ifdef HAVE_PTHREAD
250 2908 : glue->tctx = tevent_threaded_context_create(glue, ev);
251 2908 : if (glue->tctx == NULL) {
252 0 : TALLOC_FREE(ev_link);
253 0 : TALLOC_FREE(glue);
254 0 : return ENOMEM;
255 : }
256 : #endif
257 :
258 2908 : DLIST_ADD(pool->glue_list, glue);
259 2873 : return 0;
260 : }
261 :
262 : static void pthreadpool_tevent_job_fn(void *private_data);
263 : static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
264 : struct tevent_immediate *im,
265 : void *private_data);
266 :
267 204566 : static int pthreadpool_tevent_job_state_destructor(
268 : struct pthreadpool_tevent_job_state *state)
269 : {
270 204566 : if (state->pool == NULL) {
271 175495 : return 0;
272 : }
273 :
274 : /*
275 : * We should never be called with state->req == NULL,
276 : * state->pool must be cleared before the 2nd talloc_free().
277 : */
278 1 : if (state->req == NULL) {
279 0 : abort();
280 : }
281 :
282 : /*
283 : * We need to reparent to a long term context.
284 : */
285 1 : (void)talloc_reparent(state->req, NULL, state);
286 1 : state->req = NULL;
287 1 : return -1;
288 : }
289 :
290 204816 : struct tevent_req *pthreadpool_tevent_job_send(
291 : TALLOC_CTX *mem_ctx, struct tevent_context *ev,
292 : struct pthreadpool_tevent *pool,
293 : void (*fn)(void *private_data), void *private_data)
294 : {
295 : struct tevent_req *req;
296 : struct pthreadpool_tevent_job_state *state;
297 : int ret;
298 :
299 204816 : req = tevent_req_create(mem_ctx, &state,
300 : struct pthreadpool_tevent_job_state);
301 204816 : if (req == NULL) {
302 0 : return NULL;
303 : }
304 204816 : state->pool = pool;
305 204816 : state->ev = ev;
306 204816 : state->req = req;
307 204816 : state->fn = fn;
308 204816 : state->private_data = private_data;
309 :
310 204816 : if (pool == NULL) {
311 0 : tevent_req_error(req, EINVAL);
312 0 : return tevent_req_post(req, ev);
313 : }
314 204816 : if (pool->pool == NULL) {
315 0 : tevent_req_error(req, EINVAL);
316 0 : return tevent_req_post(req, ev);
317 : }
318 :
319 204816 : state->im = tevent_create_immediate(state);
320 204816 : if (tevent_req_nomem(state->im, req)) {
321 0 : return tevent_req_post(req, ev);
322 : }
323 :
324 204816 : ret = pthreadpool_tevent_register_ev(pool, ev);
325 204816 : if (tevent_req_error(req, ret)) {
326 0 : return tevent_req_post(req, ev);
327 : }
328 :
329 204816 : ret = pthreadpool_add_job(pool->pool, 0,
330 : pthreadpool_tevent_job_fn,
331 : state);
332 204816 : if (tevent_req_error(req, ret)) {
333 2 : return tevent_req_post(req, ev);
334 : }
335 :
336 : /*
337 : * Once the job is scheduled, we need to protect
338 : * our memory.
339 : */
340 204814 : talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
341 :
342 204814 : DLIST_ADD_END(pool->jobs, state);
343 :
344 175743 : return req;
345 : }
346 :
347 204813 : static void pthreadpool_tevent_job_fn(void *private_data)
348 : {
349 204813 : struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
350 : private_data, struct pthreadpool_tevent_job_state);
351 204813 : state->fn(state->private_data);
352 204565 : }
353 :
354 204565 : static int pthreadpool_tevent_job_signal(int jobid,
355 : void (*job_fn)(void *private_data),
356 : void *job_private_data,
357 : void *private_data)
358 : {
359 204565 : struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
360 : job_private_data, struct pthreadpool_tevent_job_state);
361 204565 : struct tevent_threaded_context *tctx = NULL;
362 204565 : struct pthreadpool_tevent_glue *g = NULL;
363 :
364 204565 : if (state->pool == NULL) {
365 : /* The pthreadpool_tevent is already gone */
366 0 : return 0;
367 : }
368 :
369 : #ifdef HAVE_PTHREAD
370 204565 : for (g = state->pool->glue_list; g != NULL; g = g->next) {
371 204566 : if (g->ev == state->ev) {
372 204566 : tctx = g->tctx;
373 204566 : break;
374 : }
375 : }
376 :
377 204565 : if (tctx == NULL) {
378 0 : abort();
379 : }
380 : #endif
381 :
382 175495 : if (tctx != NULL) {
383 : /* with HAVE_PTHREAD */
384 204565 : tevent_threaded_schedule_immediate(tctx, state->im,
385 : pthreadpool_tevent_job_done,
386 : state);
387 : } else {
388 : /* without HAVE_PTHREAD */
389 0 : tevent_schedule_immediate(state->im, state->ev,
390 : pthreadpool_tevent_job_done,
391 : state);
392 : }
393 :
394 204561 : return 0;
395 : }
396 :
397 204565 : static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
398 : struct tevent_immediate *im,
399 : void *private_data)
400 : {
401 204565 : struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
402 : private_data, struct pthreadpool_tevent_job_state);
403 :
404 204565 : if (state->pool != NULL) {
405 204565 : DLIST_REMOVE(state->pool->jobs, state);
406 204565 : state->pool = NULL;
407 : }
408 :
409 204565 : if (state->req == NULL) {
410 : /*
411 : * There was a talloc_free() state->req
412 : * while the job was pending,
413 : * which mean we're reparented on a longterm
414 : * talloc context.
415 : *
416 : * We just cleanup here...
417 : */
418 0 : talloc_free(state);
419 0 : return;
420 : }
421 :
422 204565 : tevent_req_done(state->req);
423 : }
424 :
425 204566 : int pthreadpool_tevent_job_recv(struct tevent_req *req)
426 : {
427 204566 : return tevent_req_simple_recv_unix(req);
428 : }
|
