lib: Add pthreadpool_tevent

This is a replacement for fncall.[ch] without having to go through
a pipe for job completion signalling

Signed-off-by: Volker Lendecke <vl@samba.org>
Reviewed-by: Stefan Metzmacher <metze@samba.org>
Reviewed-by: Jeremy Allison <jra@samba.org>

diff --git a/source3/lib/pthreadpool/pthreadpool_tevent.c b/source3/lib/pthreadpool/pthreadpool_tevent.c
new file mode 100644 (file)
index 0000000..a66c36a
--- /dev/null
+++ b/source3/lib/pthreadpool/pthreadpool_tevent.c
@@ -0,0 +1,240 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * threadpool implementation based on pthreads
+ * Copyright (C) Volker Lendecke 2009,2011
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "replace.h"
+#include "pthreadpool_tevent.h"
+#include "pthreadpool.h"
+#include "lib/util/tevent_unix.h"
+#include "lib/util/dlinklist.h"
+
+struct pthreadpool_tevent_job_state;
+
+struct pthreadpool_tevent {
+       struct pthreadpool *pool;
+
+       struct pthreadpool_tevent_job_state *jobs;
+};
+
+static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool);
+
+static int pthreadpool_tevent_job_signal(int jobid,
+                                        void (*job_fn)(void *private_data),
+                                        void *job_private_data,
+                                        void *private_data);
+
+int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
+                           struct pthreadpool_tevent **presult)
+{
+       struct pthreadpool_tevent *pool;
+       int ret;
+
+       pool = talloc_zero(mem_ctx, struct pthreadpool_tevent);
+       if (pool == NULL) {
+               return ENOMEM;
+       }
+
+       ret = pthreadpool_init(max_threads, &pool->pool,
+                              pthreadpool_tevent_job_signal, pool);
+       if (ret != 0) {
+               TALLOC_FREE(pool);
+               return ret;
+       }
+
+       talloc_set_destructor(pool, pthreadpool_tevent_destructor);
+
+       *presult = pool;
+       return 0;
+}
+
+static int pthreadpool_tevent_destructor(struct pthreadpool_tevent *pool)
+{
+       int ret;
+
+       ret = pthreadpool_destroy(pool->pool);
+       if (ret != 0) {
+               return ret;
+       }
+       pool->pool = NULL;
+
+       if (pool->jobs != NULL) {
+               abort();
+       }
+
+       return 0;
+}
+
+struct pthreadpool_tevent_job_state {
+       struct pthreadpool_tevent_job_state *prev, *next;
+       struct pthreadpool_tevent *pool;
+       struct tevent_context *ev;
+       struct tevent_threaded_context *tctx;
+       struct tevent_immediate *im;
+       struct tevent_req *req;
+
+       void (*fn)(void *private_data);
+       void *private_data;
+};
+
+static void pthreadpool_tevent_job_fn(void *private_data);
+static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
+                                       struct tevent_immediate *im,
+                                       void *private_data);
+
+static int pthreadpool_tevent_job_state_destructor(struct pthreadpool_tevent_job_state *state)
+{
+       if (state->pool == NULL) {
+               return 0;
+       }
+
+       /*
+        * We should never be called with state->req == NULL,
+        * state->pool must be cleared before the 2nd talloc_free().
+        */
+       if (state->req == NULL) {
+               abort();
+       }
+
+       /*
+        * We need to reparent to a long term context.
+        */
+       (void)talloc_reparent(state->req, state->pool, state);
+       state->req = NULL;
+       return -1;
+}
+
+struct tevent_req *pthreadpool_tevent_job_send(
+       TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+       struct pthreadpool_tevent *pool,
+       void (*fn)(void *private_data), void *private_data)
+{
+       struct tevent_req *req;
+       struct pthreadpool_tevent_job_state *state;
+       int ret;
+
+       req = tevent_req_create(mem_ctx, &state,
+                               struct pthreadpool_tevent_job_state);
+       if (req == NULL) {
+               return NULL;
+       }
+       state->pool = pool;
+       state->ev = ev;
+       state->req = req;
+       state->fn = fn;
+       state->private_data = private_data;
+
+       state->im = tevent_create_immediate(state);
+       if (tevent_req_nomem(state->im, req)) {
+               return tevent_req_post(req, ev);
+       }
+
+#ifdef HAVE_PTHREAD
+       state->tctx = tevent_threaded_context_create(state, ev);
+       if (state->tctx == NULL && errno == ENOSYS) {
+               /*
+                * Samba build with pthread support but
+                * tevent without???
+                */
+               tevent_req_error(req, ENOSYS);
+               return tevent_req_post(req, ev);
+       }
+       if (tevent_req_nomem(state->tctx, req)) {
+               return tevent_req_post(req, ev);
+       }
+#endif
+
+       ret = pthreadpool_add_job(pool->pool, 0,
+                                 pthreadpool_tevent_job_fn,
+                                 state);
+       if (tevent_req_error(req, ret)) {
+               return tevent_req_post(req, ev);
+       }
+
+       /*
+        * Once the job is scheduled, we need to protect
+        * our memory.
+        */
+       talloc_set_destructor(state, pthreadpool_tevent_job_state_destructor);
+
+       DLIST_ADD_END(pool->jobs, state);
+
+       return req;
+}
+
+static void pthreadpool_tevent_job_fn(void *private_data)
+{
+       struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
+               private_data, struct pthreadpool_tevent_job_state);
+       state->fn(state->private_data);
+}
+
+static int pthreadpool_tevent_job_signal(int jobid,
+                                        void (*job_fn)(void *private_data),
+                                        void *job_private_data,
+                                        void *private_data)
+{
+       struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
+               job_private_data, struct pthreadpool_tevent_job_state);
+
+       if (state->tctx != NULL) {
+               /* with HAVE_PTHREAD */
+               tevent_threaded_schedule_immediate(state->tctx, state->im,
+                                                  pthreadpool_tevent_job_done,
+                                                  state);
+       } else {
+               /* without HAVE_PTHREAD */
+               tevent_schedule_immediate(state->im, state->ev,
+                                         pthreadpool_tevent_job_done,
+                                         state);
+       }
+
+       return 0;
+}
+
+static void pthreadpool_tevent_job_done(struct tevent_context *ctx,
+                                       struct tevent_immediate *im,
+                                       void *private_data)
+{
+       struct pthreadpool_tevent_job_state *state = talloc_get_type_abort(
+               private_data, struct pthreadpool_tevent_job_state);
+
+       DLIST_REMOVE(state->pool->jobs, state);
+       state->pool = NULL;
+
+       TALLOC_FREE(state->tctx);
+
+       if (state->req == NULL) {
+               /*
+                * There was a talloc_free() state->req
+                * while the job was pending,
+                * which mean we're reparented on a longterm
+                * talloc context.
+                *
+                * We just cleanup here...
+                */
+               talloc_free(state);
+               return;
+       }
+
+       tevent_req_done(state->req);
+}
+
+int pthreadpool_tevent_job_recv(struct tevent_req *req)
+{
+       return tevent_req_simple_recv_unix(req);
+}

diff --git a/source3/lib/pthreadpool/pthreadpool_tevent.h b/source3/lib/pthreadpool/pthreadpool_tevent.h
new file mode 100644 (file)
index 0000000..de74a34
--- /dev/null
+++ b/source3/lib/pthreadpool/pthreadpool_tevent.h
@@ -0,0 +1,37 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * threadpool implementation based on pthreads
+ * Copyright (C) Volker Lendecke 2016
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef __PTHREADPOOL_TEVENT_H__
+#define __PTHREADPOOL_TEVENT_H__
+
+#include <tevent.h>
+
+struct pthreadpool_tevent;
+
+int pthreadpool_tevent_init(TALLOC_CTX *mem_ctx, unsigned max_threads,
+                           struct pthreadpool_tevent **presult);
+
+struct tevent_req *pthreadpool_tevent_job_send(
+       TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+       struct pthreadpool_tevent *pool,
+       void (*fn)(void *private_data), void *private_data);
+
+int pthreadpool_tevent_job_recv(struct tevent_req *req);
+
+#endif

diff --git a/source3/lib/pthreadpool/wscript_build b/source3/lib/pthreadpool/wscript_build
index aa02850ea303d6bbdb5a2bf352c7a5ce74e440dd..8195af7c3227817d2064daca3f868e07ef0b07b7 100644 (file)
--- a/source3/lib/pthreadpool/wscript_build
+++ b/source3/lib/pthreadpool/wscript_build
@@ -2,12 +2,18 @@
 
 if bld.env.WITH_PTHREADPOOL:
     bld.SAMBA3_SUBSYSTEM('PTHREADPOOL',
-                         source='pthreadpool.c pthreadpool_pipe.c',
-                         deps='pthread rt replace')
+                         source='''pthreadpool.c
+                                   pthreadpool_pipe.c
+                                   pthreadpool_tevent.c
+                                ''',
+                         deps='pthread rt replace tevent-util')
 else:
     bld.SAMBA3_SUBSYSTEM('PTHREADPOOL',
-                         source='pthreadpool_sync.c pthreadpool_pipe.c',
-                         deps='replace')
+                         source='''pthreadpool_sync.c
+                                   pthreadpool_pipe.c
+                                   pthreadpool_tevent.c
+                                ''',
+                         deps='replace tevent-util')
 
 
 bld.SAMBA3_BINARY('pthreadpooltest',