lib: Add messages_dgm_ref.[ch]

We only have one messaging_dgm context per process. But we will use this from
two completely independent messaging subsystems which are independently
initialized. We need to coordinate creation and destruction, do this via
talloc.

I know this looks like a step back, but when in the future we have really just
one messaging subsystem, this can go again. My immediate goal is to make
source3 and source4 transport-compatible, and this looks like a quick way
towards that goal.

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

source3/lib/messages_dgm_ref.c

@@ -0,0 +1,121 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * Samba internal messaging functions
+ * Copyright (C) 2014 by Volker Lendecke
+ *
+ * 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 <talloc.h>
+#include "replace.h"
+#include "messages_dgm.h"
+#include "messages_dgm_ref.h"
+#include "lib/util/debug.h"
+#include "lib/util/dlinklist.h"
+
+struct msg_dgm_ref {
+	struct msg_dgm_ref *prev, *next;
+	void (*recv_cb)(const uint8_t *msg, size_t msg_len,
+			int *fds, size_t num_fds, void *private_data);
+	void *recv_cb_private_data;
+};
+
+static pid_t dgm_pid = 0;
+static struct msg_dgm_ref *refs = NULL;
+
+static int msg_dgm_ref_destructor(struct msg_dgm_ref *r);
+static void msg_dgm_ref_recv(const uint8_t *msg, size_t msg_len,
+			     int *fds, size_t num_fds, void *private_data);
+
+void *messaging_dgm_ref(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+			uint64_t unique,
+			const char *socket_dir,
+			const char *lockfile_dir,
+			void (*recv_cb)(const uint8_t *msg, size_t msg_len,
+					int *fds, size_t num_fds,
+					void *private_data),
+			void *recv_cb_private_data,
+			int *err)
+{
+	struct msg_dgm_ref *result, *tmp_refs;
+
+	result = talloc(mem_ctx, struct msg_dgm_ref);
+	if (result == NULL) {
+		*err = ENOMEM;
+		return NULL;
+	}
+
+	tmp_refs = refs;
+
+	if ((refs != NULL) && (dgm_pid != getpid())) {
+		/*
+		 * Have to reinit after fork
+		 */
+		messaging_dgm_destroy();
+		refs = NULL;
+	}
+
+	if (refs == NULL) {
+		int ret;
+
+		ret = messaging_dgm_init(ev, unique, socket_dir, lockfile_dir,
+					 msg_dgm_ref_recv, NULL);
+		if (ret != 0) {
+			DEBUG(10, ("messaging_dgm_init failed: %s\n",
+				   strerror(ret)));
+			TALLOC_FREE(result);
+			*err = ret;
+			return NULL;
+		}
+		dgm_pid = getpid();
+	}
+
+	refs = tmp_refs;
+
+	result->recv_cb = recv_cb;
+	result->recv_cb_private_data = recv_cb_private_data;
+	DLIST_ADD(refs, result);
+	talloc_set_destructor(result, msg_dgm_ref_destructor);
+
+	return result;
+}
+
+static void msg_dgm_ref_recv(const uint8_t *msg, size_t msg_len,
+			     int *fds, size_t num_fds, void *private_data)
+{
+	struct msg_dgm_ref *r, *next;
+
+	/*
+	 * We have to broadcast incoming messages to all refs. The first ref
+	 * that grabs the fd's will get them.
+	 */
+	for (r = refs; r != NULL; r = next) {
+		next = r->next;
+		r->recv_cb(msg, msg_len, fds, num_fds,
+			   r->recv_cb_private_data);
+	}
+}
+
+static int msg_dgm_ref_destructor(struct msg_dgm_ref *r)
+{
+	if (refs == NULL) {
+		abort();
+	}
+	DLIST_REMOVE(refs, r);
+
+	if (refs == NULL) {
+		messaging_dgm_destroy();
+	}
+	return 0;
+}

source3/lib/messages_dgm_ref.h

@@ -0,0 +1,37 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * Samba internal messaging functions
+ * Copyright (C) 2014 by Volker Lendecke
+ *
+ * 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 _MESSAGES_DGM_REF_H_
+#define _MESSAGES_DGM_REF_H_
+
+#include <talloc.h>
+#include <tevent.h>
+#include "replace.h"
+
+void *messaging_dgm_ref(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+			uint64_t unique,
+			const char *socket_dir,
+			const char *lockfile_dir,
+			void (*recv_cb)(const uint8_t *msg, size_t msg_len,
+					int *fds, size_t num_fds,
+					void *private_data),
+			void *recv_cb_private_data,
+			int *err);
+
+#endif

source3/wscript_build

@@ -309,7 +309,7 @@ bld.SAMBA3_SUBSYSTEM('TDB_LIB',
                      deps='dbwrap samba-cluster-support')
 
 bld.SAMBA3_LIBRARY('messages_dgm',
-                   source='''lib/messages_dgm.c''',
+                   source='''lib/messages_dgm.c lib/messages_dgm_ref.c''',
                    deps='talloc UNIX_MSG POLL_FUNCS_TEVENT samba-debug',
                    private_library=True)