Context Navigation

← Previous Revision
Latest Revision
Next Revision →
Blame
Revision Log

ctdbd_conn.c@ 421

Visit:

Last change on this file since 421 was 414, checked in by Herwig Bauernfeind, 16 years ago
Samba 3.5.0: Initial import
File size: 31.2 KB

Line
1	/*
2	Unix SMB/CIFS implementation.
3	Samba internal messaging functions
4	Copyright (C) 2007 by Volker Lendecke
5	Copyright (C) 2007 by Andrew Tridgell
6
7	This program is free software; you can redistribute it and/or modify
8	it under the terms of the GNU General Public License as published by
9	the Free Software Foundation; either version 3 of the License, or
10	(at your option) any later version.
11
12	This program is distributed in the hope that it will be useful,
13	but WITHOUT ANY WARRANTY; without even the implied warranty of
14	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	GNU General Public License for more details.
16
17	You should have received a copy of the GNU General Public License
18	along with this program. If not, see <http://www.gnu.org/licenses/>.
19	*/
20
21	#include "includes.h"
22
23	#ifdef CLUSTER_SUPPORT
24
25	#include "librpc/gen_ndr/messaging.h"
26	#include "librpc/gen_ndr/ndr_messaging.h"
27
28	/* paths to these include files come from --with-ctdb= in configure */
29	#include "ctdb.h"
30	#include "ctdb_private.h"
31
32	struct ctdbd_connection {
33	struct messaging_context *msg_ctx;
34	uint32 reqid;
35	uint32 our_vnn;
36	uint64 rand_srvid;
37	struct packet_context *pkt;
38	struct fd_event *fde;
39
40	void (release_ip_handler)(const char ip_addr, void *private_data);
41	void *release_ip_priv;
42	};
43
44	static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
45	uint32_t vnn, uint32 opcode,
46	uint64_t srvid, uint32_t flags, TDB_DATA data,
47	TALLOC_CTX mem_ctx, TDB_DATA outdata,
48	int *cstatus);
49
50	/*
51	* exit on fatal communications errors with the ctdbd daemon
52	*/
53	static void cluster_fatal(const char *why)
54	{
55	DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
56	/* we don't use smb_panic() as we don't want to delay to write
57	a core file. We need to release this process id immediately
58	so that someone else can take over without getting sharing
59	violations */
60	_exit(0);
61	}
62
63	/*
64	*
65	*/
66	static void ctdb_packet_dump(struct ctdb_req_header *hdr)
67	{
68	if (DEBUGLEVEL < 10) {
69	return;
70	}
71	DEBUGADD(10, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
72	(int)hdr->length, (int)hdr->ctdb_magic,
73	(int)hdr->ctdb_version, (int)hdr->generation,
74	(int)hdr->operation, (int)hdr->reqid));
75	}
76
77	/*
78	* Register a srvid with ctdbd
79	*/
80	static NTSTATUS register_with_ctdbd(struct ctdbd_connection *conn,
81	uint64_t srvid)
82	{
83
84	int cstatus;
85	return ctdbd_control(conn, CTDB_CURRENT_NODE,
86	CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
87	tdb_null, NULL, NULL, &cstatus);
88	}
89
90	/*
91	* get our vnn from the cluster
92	*/
93	static NTSTATUS get_cluster_vnn(struct ctdbd_connection conn, uint32 vnn)
94	{
95	int32_t cstatus=-1;
96	NTSTATUS status;
97	status = ctdbd_control(conn,
98	CTDB_CURRENT_NODE, CTDB_CONTROL_GET_PNN, 0, 0,
99	tdb_null, NULL, NULL, &cstatus);
100	if (!NT_STATUS_IS_OK(status)) {
101	cluster_fatal("ctdbd_control failed\n");
102	}
103	*vnn = (uint32_t)cstatus;
104	return status;
105	}
106
107	uint32 ctdbd_vnn(const struct ctdbd_connection *conn)
108	{
109	return conn->our_vnn;
110	}
111
112	/*
113	* Get us a ctdb connection
114	*/
115
116	static NTSTATUS ctdbd_connect(TALLOC_CTX *mem_ctx,
117	struct packet_context **presult)
118	{
119	struct packet_context *result;
120	const char *sockname = lp_ctdbd_socket();
121	struct sockaddr_un addr;
122	int fd;
123
124	if (!sockname \|\| !*sockname) {
125	sockname = CTDB_PATH;
126	}
127
128	fd = socket(AF_UNIX, SOCK_STREAM, 0);
129	if (fd == -1) {
130	DEBUG(3, ("Could not create socket: %s\n", strerror(errno)));
131	return map_nt_error_from_unix(errno);
132	}
133
134	ZERO_STRUCT(addr);
135	addr.sun_family = AF_UNIX;
136	strncpy(addr.sun_path, sockname, sizeof(addr.sun_path));
137
138	if (sys_connect(fd, (struct sockaddr *)&addr) == -1) {
139	DEBUG(1, ("connect(%s) failed: %s\n", sockname,
140	strerror(errno)));
141	close(fd);
142	return map_nt_error_from_unix(errno);
143	}
144
145	if (!(result = packet_init(mem_ctx, fd))) {
146	close(fd);
147	return NT_STATUS_NO_MEMORY;
148	}
149
150	*presult = result;
151	return NT_STATUS_OK;
152	}
153
154	/*
155	* Do we have a complete ctdb packet in the queue?
156	*/
157
158	static bool ctdb_req_complete(const uint8_t *buf, size_t available,
159	size_t *length,
160	void *private_data)
161	{
162	uint32 msglen;
163
164	if (available < sizeof(msglen)) {
165	return False;
166	}
167
168	msglen = ((uint32 )buf);
169
170	DEBUG(10, ("msglen = %d\n", msglen));
171
172	if (msglen < sizeof(struct ctdb_req_header)) {
173	DEBUG(0, ("Got invalid msglen: %d, expected at least %d for "
174	"the req_header\n", (int)msglen,
175	(int)sizeof(struct ctdb_req_header)));
176	cluster_fatal("ctdbd protocol error\n");
177	}
178
179	if (available < msglen) {
180	return false;
181	}
182
183	*length = msglen;
184	return true;
185	}
186
187	/*
188	* State necessary to defer an incoming message while we are waiting for a
189	* ctdb reply.
190	*/
191
192	struct deferred_msg_state {
193	struct messaging_context *msg_ctx;
194	struct messaging_rec *rec;
195	};
196
197	/*
198	* Timed event handler for the deferred message
199	*/
200
201	static void deferred_message_dispatch(struct event_context *event_ctx,
202	struct timed_event *te,
203	struct timeval now,
204	void *private_data)
205	{
206	struct deferred_msg_state *state = talloc_get_type_abort(
207	private_data, struct deferred_msg_state);
208
209	messaging_dispatch_rec(state->msg_ctx, state->rec);
210	TALLOC_FREE(state);
211	TALLOC_FREE(te);
212	}
213
214	struct req_pull_state {
215	TALLOC_CTX *mem_ctx;
216	DATA_BLOB req;
217	};
218
219	/*
220	* Pull a ctdb request out of the incoming packet queue
221	*/
222
223	static NTSTATUS ctdb_req_pull(uint8_t *buf, size_t length,
224	void *private_data)
225	{
226	struct req_pull_state state = (struct req_pull_state )private_data;
227
228	state->req.data = talloc_move(state->mem_ctx, &buf);
229	state->req.length = length;
230	return NT_STATUS_OK;
231	}
232
233	/*
234	* Fetch a messaging_rec from an incoming ctdb style message
235	*/
236
237	static struct messaging_rec ctdb_pull_messaging_rec(TALLOC_CTX mem_ctx,
238	size_t overall_length,
239	struct ctdb_req_message *msg)
240	{
241	struct messaging_rec *result;
242	DATA_BLOB blob;
243	enum ndr_err_code ndr_err;
244
245	if ((overall_length < offsetof(struct ctdb_req_message, data))
246	\|\| (overall_length
247	< offsetof(struct ctdb_req_message, data) + msg->datalen)) {
248
249	cluster_fatal("got invalid msg length");
250	}
251
252	if (!(result = TALLOC_P(mem_ctx, struct messaging_rec))) {
253	DEBUG(0, ("talloc failed\n"));
254	return NULL;
255	}
256
257	blob = data_blob_const(msg->data, msg->datalen);
258
259	ndr_err = ndr_pull_struct_blob(
260	&blob, result, NULL, result,
261	(ndr_pull_flags_fn_t)ndr_pull_messaging_rec);
262
263	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
264	DEBUG(0, ("ndr_pull_struct_blob failed: %s\n",
265	ndr_errstr(ndr_err)));
266	TALLOC_FREE(result);
267	return NULL;
268	}
269
270	if (DEBUGLEVEL >= 10) {
271	DEBUG(10, ("ctdb_pull_messaging_rec:\n"));
272	NDR_PRINT_DEBUG(messaging_rec, result);
273	}
274
275	return result;
276	}
277
278	static NTSTATUS ctdb_packet_fd_read_sync(struct packet_context *ctx)
279	{
280	struct timeval timeout;
281	struct timeval *ptimeout;
282
283	timeout = timeval_set(lp_ctdb_timeout(), 0);
284	ptimeout = (timeout.tv_sec != 0) ? &timeout : NULL;
285
286	return packet_fd_read_sync(ctx, ptimeout);
287	}
288
289	/*
290	* Read a full ctdbd request. If we have a messaging context, defer incoming
291	* messages that might come in between.
292	*/
293
294	static NTSTATUS ctdb_read_req(struct ctdbd_connection *conn, uint32 reqid,
295	TALLOC_CTX mem_ctx, void result)
296	{
297	struct ctdb_req_header *hdr;
298	struct req_pull_state state;
299	NTSTATUS status;
300
301	again:
302
303	status = ctdb_packet_fd_read_sync(conn->pkt);
304
305	if (NT_STATUS_EQUAL(status, NT_STATUS_NETWORK_BUSY)) {
306	/* EAGAIN */
307	goto again;
308	} else if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
309	/* EAGAIN */
310	goto again;
311	}
312
313	if (!NT_STATUS_IS_OK(status)) {
314	DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
315	cluster_fatal("ctdbd died\n");
316	}
317
318	next_pkt:
319
320	ZERO_STRUCT(state);
321	state.mem_ctx = mem_ctx;
322
323	if (!packet_handler(conn->pkt, ctdb_req_complete, ctdb_req_pull,
324	&state, &status)) {
325	/*
326	* Not enough data
327	*/
328	DEBUG(10, ("not enough data from ctdb socket, retrying\n"));
329	goto again;
330	}
331
332	if (!NT_STATUS_IS_OK(status)) {
333	DEBUG(0, ("Could not read packet: %s\n", nt_errstr(status)));
334	cluster_fatal("ctdbd died\n");
335	}
336
337	hdr = (struct ctdb_req_header *)state.req.data;
338
339	DEBUG(10, ("Received ctdb packet\n"));
340	ctdb_packet_dump(hdr);
341
342	if (hdr->operation == CTDB_REQ_MESSAGE) {
343	struct timed_event *evt;
344	struct deferred_msg_state *msg_state;
345	struct ctdb_req_message msg = (struct ctdb_req_message )hdr;
346
347	if (conn->msg_ctx == NULL) {
348	DEBUG(1, ("Got a message without having a msg ctx, "
349	"dropping msg %llu\n",
350	(long long unsigned)msg->srvid));
351	goto next_pkt;
352	}
353
354	if ((conn->release_ip_handler != NULL)
355	&& (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
356	/* must be dispatched immediately */
357	DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
358	conn->release_ip_handler((const char *)msg->data,
359	conn->release_ip_priv);
360	TALLOC_FREE(hdr);
361	goto next_pkt;
362	}
363
364	if (msg->srvid == CTDB_SRVID_RECONFIGURE) {
365	DEBUG(0,("Got cluster reconfigure message in ctdb_read_req\n"));
366	messaging_send(conn->msg_ctx, procid_self(),
367	MSG_SMB_BRL_VALIDATE, &data_blob_null);
368	TALLOC_FREE(hdr);
369	goto next_pkt;
370	}
371
372	if (!(msg_state = TALLOC_P(talloc_autofree_context(), struct deferred_msg_state))) {
373	DEBUG(0, ("talloc failed\n"));
374	TALLOC_FREE(hdr);
375	goto next_pkt;
376	}
377
378	if (!(msg_state->rec = ctdb_pull_messaging_rec(
379	msg_state, state.req.length, msg))) {
380	DEBUG(0, ("ctdbd_pull_messaging_rec failed\n"));
381	TALLOC_FREE(msg_state);
382	TALLOC_FREE(hdr);
383	goto next_pkt;
384	}
385
386	TALLOC_FREE(hdr);
387
388	msg_state->msg_ctx = conn->msg_ctx;
389
390	/*
391	* We're waiting for a call reply, but an async message has
392	* crossed. Defer dispatching to the toplevel event loop.
393	*/
394	evt = event_add_timed(conn->msg_ctx->event_ctx,
395	conn->msg_ctx->event_ctx,
396	timeval_zero(),
397	deferred_message_dispatch,
398	msg_state);
399	if (evt == NULL) {
400	DEBUG(0, ("event_add_timed failed\n"));
401	TALLOC_FREE(msg_state);
402	TALLOC_FREE(hdr);
403	goto next_pkt;
404	}
405
406	goto next_pkt;
407	}
408
409	if (hdr->reqid != reqid) {
410	/* we got the wrong reply */
411	DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
412	"been %u\n", hdr->reqid, reqid));
413	TALLOC_FREE(hdr);
414	goto again;
415	}
416
417	((void *)result) = talloc_move(mem_ctx, &hdr);
418
419	return NT_STATUS_OK;
420	}
421
422	/*
423	* Get us a ctdbd connection
424	*/
425
426	NTSTATUS ctdbd_init_connection(TALLOC_CTX *mem_ctx,
427	struct ctdbd_connection **pconn)
428	{
429	struct ctdbd_connection *conn;
430	NTSTATUS status;
431
432	if (!(conn = TALLOC_ZERO_P(mem_ctx, struct ctdbd_connection))) {
433	DEBUG(0, ("talloc failed\n"));
434	return NT_STATUS_NO_MEMORY;
435	}
436
437	status = ctdbd_connect(conn, &conn->pkt);
438
439	if (!NT_STATUS_IS_OK(status)) {
440	DEBUG(10, ("ctdbd_connect failed: %s\n", nt_errstr(status)));
441	goto fail;
442	}
443
444	status = get_cluster_vnn(conn, &conn->our_vnn);
445
446	if (!NT_STATUS_IS_OK(status)) {
447	DEBUG(10, ("get_cluster_vnn failed: %s\n", nt_errstr(status)));
448	goto fail;
449	}
450
451	generate_random_buffer((unsigned char *)&conn->rand_srvid,
452	sizeof(conn->rand_srvid));
453
454	status = register_with_ctdbd(conn, conn->rand_srvid);
455
456	if (!NT_STATUS_IS_OK(status)) {
457	DEBUG(5, ("Could not register random srvid: %s\n",
458	nt_errstr(status)));
459	goto fail;
460	}
461
462	*pconn = conn;
463	return NT_STATUS_OK;
464
465	fail:
466	TALLOC_FREE(conn);
467	return status;
468	}
469
470	/*
471	* Get us a ctdbd connection and register us as a process
472	*/
473
474	NTSTATUS ctdbd_messaging_connection(TALLOC_CTX *mem_ctx,
475	struct ctdbd_connection **pconn)
476	{
477	struct ctdbd_connection *conn;
478	NTSTATUS status;
479
480	status = ctdbd_init_connection(mem_ctx, &conn);
481
482	if (!NT_STATUS_IS_OK(status)) {
483	return status;
484	}
485
486	status = register_with_ctdbd(conn, (uint64_t)sys_getpid());
487	if (!NT_STATUS_IS_OK(status)) {
488	goto fail;
489	}
490
491	status = register_with_ctdbd(conn, MSG_SRVID_SAMBA);
492	if (!NT_STATUS_IS_OK(status)) {
493	goto fail;
494	}
495
496	*pconn = conn;
497	return NT_STATUS_OK;
498
499	fail:
500	TALLOC_FREE(conn);
501	return status;
502	}
503
504	/*
505	* Packet handler to receive and handle a ctdb message
506	*/
507	static NTSTATUS ctdb_handle_message(uint8_t *buf, size_t length,
508	void *private_data)
509	{
510	struct ctdbd_connection *conn = talloc_get_type_abort(
511	private_data, struct ctdbd_connection);
512	struct ctdb_req_message *msg;
513	struct messaging_rec *msg_rec;
514
515	msg = (struct ctdb_req_message *)buf;
516
517	if (msg->hdr.operation != CTDB_REQ_MESSAGE) {
518	DEBUG(0, ("Received async msg of type %u, discarding\n",
519	msg->hdr.operation));
520	TALLOC_FREE(buf);
521	return NT_STATUS_INVALID_PARAMETER;
522	}
523
524	if ((conn->release_ip_handler != NULL)
525	&& (msg->srvid == CTDB_SRVID_RELEASE_IP)) {
526	/* must be dispatched immediately */
527	DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
528	conn->release_ip_handler((const char *)msg->data,
529	conn->release_ip_priv);
530	TALLOC_FREE(buf);
531	return NT_STATUS_OK;
532	}
533
534	SMB_ASSERT(conn->msg_ctx != NULL);
535
536	if (msg->srvid == CTDB_SRVID_RECONFIGURE) {
537	DEBUG(0,("Got cluster reconfigure message\n"));
538	/*
539	* when the cluster is reconfigured, we need to clean the brl
540	* database
541	*/
542	messaging_send(conn->msg_ctx, procid_self(),
543	MSG_SMB_BRL_VALIDATE, &data_blob_null);
544
545	/*
546	* it's possible that we have just rejoined the cluster after
547	* an outage. In that case our pending locks could have been
548	* removed from the lockdb, so retry them once more
549	*/
550	message_send_all(conn->msg_ctx, MSG_SMB_UNLOCK, NULL, 0, NULL);
551
552	TALLOC_FREE(buf);
553
554	return NT_STATUS_OK;
555	}
556
557	/* only messages to our pid or the broadcast are valid here */
558	if (msg->srvid != sys_getpid() && msg->srvid != MSG_SRVID_SAMBA) {
559	DEBUG(0,("Got unexpected message with srvid=%llu\n",
560	(unsigned long long)msg->srvid));
561	TALLOC_FREE(buf);
562	return NT_STATUS_OK;
563	}
564
565	if (!(msg_rec = ctdb_pull_messaging_rec(NULL, length, msg))) {
566	DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
567	TALLOC_FREE(buf);
568	return NT_STATUS_NO_MEMORY;
569	}
570
571	messaging_dispatch_rec(conn->msg_ctx, msg_rec);
572
573	TALLOC_FREE(msg_rec);
574	TALLOC_FREE(buf);
575	return NT_STATUS_OK;
576	}
577
578	/*
579	* The ctdbd socket is readable asynchronuously
580	*/
581
582	static void ctdbd_socket_handler(struct event_context *event_ctx,
583	struct fd_event *event,
584	uint16 flags,
585	void *private_data)
586	{
587	struct ctdbd_connection *conn = talloc_get_type_abort(
588	private_data, struct ctdbd_connection);
589
590	NTSTATUS status;
591
592	status = packet_fd_read(conn->pkt);
593
594	if (!NT_STATUS_IS_OK(status)) {
595	DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status)));
596	cluster_fatal("ctdbd died\n");
597	}
598
599	while (packet_handler(conn->pkt, ctdb_req_complete,
600	ctdb_handle_message, conn, &status)) {
601	if (!NT_STATUS_IS_OK(status)) {
602	DEBUG(10, ("could not handle incoming message: %s\n",
603	nt_errstr(status)));
604	}
605	}
606	}
607
608	/*
609	* Prepare a ctdbd connection to receive messages
610	*/
611
612	NTSTATUS ctdbd_register_msg_ctx(struct ctdbd_connection *conn,
613	struct messaging_context *msg_ctx)
614	{
615	SMB_ASSERT(conn->msg_ctx == NULL);
616	SMB_ASSERT(conn->fde == NULL);
617
618	if (!(conn->fde = event_add_fd(msg_ctx->event_ctx, conn,
619	packet_get_fd(conn->pkt),
620	EVENT_FD_READ,
621	ctdbd_socket_handler,
622	conn))) {
623	DEBUG(0, ("event_add_fd failed\n"));
624	return NT_STATUS_NO_MEMORY;
625	}
626
627	conn->msg_ctx = msg_ctx;
628
629	return NT_STATUS_OK;
630	}
631
632	/*
633	* Send a messaging message across a ctdbd
634	*/
635
636	NTSTATUS ctdbd_messaging_send(struct ctdbd_connection *conn,
637	uint32 dst_vnn, uint64 dst_srvid,
638	struct messaging_rec *msg)
639	{
640	struct ctdb_req_message r;
641	TALLOC_CTX *mem_ctx;
642	DATA_BLOB blob;
643	NTSTATUS status;
644	enum ndr_err_code ndr_err;
645
646	if (!(mem_ctx = talloc_init("ctdbd_messaging_send"))) {
647	DEBUG(0, ("talloc failed\n"));
648	return NT_STATUS_NO_MEMORY;
649	}
650
651	ndr_err = ndr_push_struct_blob(
652	&blob, mem_ctx, NULL, msg,
653	(ndr_push_flags_fn_t)ndr_push_messaging_rec);
654
655	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
656	DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
657	ndr_errstr(ndr_err)));
658	status = ndr_map_error2ntstatus(ndr_err);
659	goto fail;
660	}
661
662	r.hdr.length = offsetof(struct ctdb_req_message, data) + blob.length;
663	r.hdr.ctdb_magic = CTDB_MAGIC;
664	r.hdr.ctdb_version = CTDB_VERSION;
665	r.hdr.generation = 1;
666	r.hdr.operation = CTDB_REQ_MESSAGE;
667	r.hdr.destnode = dst_vnn;
668	r.hdr.srcnode = conn->our_vnn;
669	r.hdr.reqid = 0;
670	r.srvid = dst_srvid;
671	r.datalen = blob.length;
672
673	DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
674	ctdb_packet_dump(&r.hdr);
675
676	status = packet_send(
677	conn->pkt, 2,
678	data_blob_const(&r, offsetof(struct ctdb_req_message, data)),
679	blob);
680
681	if (!NT_STATUS_IS_OK(status)) {
682	DEBUG(0, ("packet_send failed: %s\n", nt_errstr(status)));
683	goto fail;
684	}
685
686	status = packet_flush(conn->pkt);
687
688	if (!NT_STATUS_IS_OK(status)) {
689	DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
690	cluster_fatal("cluster dispatch daemon msg write error\n");
691	}
692
693	status = NT_STATUS_OK;
694	fail:
695	TALLOC_FREE(mem_ctx);
696	return status;
697	}
698
699	/*
700	* send/recv a generic ctdb control message
701	*/
702	static NTSTATUS ctdbd_control(struct ctdbd_connection *conn,
703	uint32_t vnn, uint32 opcode,
704	uint64_t srvid, uint32_t flags,
705	TDB_DATA data,
706	TALLOC_CTX mem_ctx, TDB_DATA outdata,
707	int *cstatus)
708	{
709	struct ctdb_req_control req;
710	struct ctdb_reply_control *reply = NULL;
711	struct ctdbd_connection *new_conn = NULL;
712	NTSTATUS status;
713
714	/* the samba3 ctdb code can't handle NOREPLY yet */
715	flags &= ~CTDB_CTRL_FLAG_NOREPLY;
716
717	if (conn == NULL) {
718	status = ctdbd_init_connection(NULL, &new_conn);
719
720	if (!NT_STATUS_IS_OK(status)) {
721	DEBUG(10, ("Could not init temp connection: %s\n",
722	nt_errstr(status)));
723	goto fail;
724	}
725
726	conn = new_conn;
727	}
728
729	ZERO_STRUCT(req);
730	req.hdr.length = offsetof(struct ctdb_req_control, data) + data.dsize;
731	req.hdr.ctdb_magic = CTDB_MAGIC;
732	req.hdr.ctdb_version = CTDB_VERSION;
733	req.hdr.operation = CTDB_REQ_CONTROL;
734	req.hdr.reqid = ++conn->reqid;
735	req.hdr.destnode = vnn;
736	req.opcode = opcode;
737	req.srvid = srvid;
738	req.datalen = data.dsize;
739
740	DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
741	ctdb_packet_dump(&req.hdr);
742
743	status = packet_send(
744	conn->pkt, 2,
745	data_blob_const(&req, offsetof(struct ctdb_req_control, data)),
746	data_blob_const(data.dptr, data.dsize));
747
748	if (!NT_STATUS_IS_OK(status)) {
749	DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
750	goto fail;
751	}
752
753	status = packet_flush(conn->pkt);
754
755	if (!NT_STATUS_IS_OK(status)) {
756	DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
757	cluster_fatal("cluster dispatch daemon control write error\n");
758	}
759
760	if (flags & CTDB_CTRL_FLAG_NOREPLY) {
761	TALLOC_FREE(new_conn);
762	return NT_STATUS_OK;
763	}
764
765	status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
766
767	if (!NT_STATUS_IS_OK(status)) {
768	DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
769	goto fail;
770	}
771
772	if (reply->hdr.operation != CTDB_REPLY_CONTROL) {
773	DEBUG(0, ("received invalid reply\n"));
774	goto fail;
775	}
776
777	if (outdata) {
778	if (!(outdata->dptr = (uint8 *)talloc_memdup(
779	mem_ctx, reply->data, reply->datalen))) {
780	TALLOC_FREE(reply);
781	return NT_STATUS_NO_MEMORY;
782	}
783	outdata->dsize = reply->datalen;
784	}
785	if (cstatus) {
786	(*cstatus) = reply->status;
787	}
788
789	status = NT_STATUS_OK;
790
791	fail:
792	TALLOC_FREE(new_conn);
793	TALLOC_FREE(reply);
794	return status;
795	}
796
797	/*
798	* see if a remote process exists
799	*/
800	bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32 vnn, pid_t pid)
801	{
802	NTSTATUS status;
803	TDB_DATA data;
804	int32_t cstatus;
805
806	data.dptr = (uint8_t*)&pid;
807	data.dsize = sizeof(pid);
808
809	status = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS, 0, 0,
810	data, NULL, NULL, &cstatus);
811	if (!NT_STATUS_IS_OK(status)) {
812	DEBUG(0, (__location__ " ctdb_control for process_exists "
813	"failed\n"));
814	return False;
815	}
816
817	return cstatus == 0;
818	}
819
820	/*
821	* Get a db path
822	*/
823	char ctdbd_dbpath(struct ctdbd_connection conn,
824	TALLOC_CTX *mem_ctx, uint32_t db_id)
825	{
826	NTSTATUS status;
827	TDB_DATA data;
828	int32_t cstatus;
829
830	data.dptr = (uint8_t*)&db_id;
831	data.dsize = sizeof(db_id);
832
833	status = ctdbd_control(conn, CTDB_CURRENT_NODE,
834	CTDB_CONTROL_GETDBPATH, 0, 0, data,
835	mem_ctx, &data, &cstatus);
836	if (!NT_STATUS_IS_OK(status) \|\| cstatus != 0) {
837	DEBUG(0,(__location__ " ctdb_control for getdbpath failed\n"));
838	return NULL;
839	}
840
841	return (char *)data.dptr;
842	}
843
844	/*
845	* attach to a ctdb database
846	*/
847	NTSTATUS ctdbd_db_attach(struct ctdbd_connection *conn,
848	const char name, uint32_t db_id, int tdb_flags)
849	{
850	NTSTATUS status;
851	TDB_DATA data;
852	int32_t cstatus;
853	bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
854
855	data.dptr = (uint8_t*)name;
856	data.dsize = strlen(name)+1;
857
858	status = ctdbd_control(conn, CTDB_CURRENT_NODE,
859	persistent
860	? CTDB_CONTROL_DB_ATTACH_PERSISTENT
861	: CTDB_CONTROL_DB_ATTACH,
862	0, 0, data, NULL, &data, &cstatus);
863	if (!NT_STATUS_IS_OK(status)) {
864	DEBUG(0, (__location__ " ctdb_control for db_attach "
865	"failed: %s\n", nt_errstr(status)));
866	return status;
867	}
868
869	if (cstatus != 0 \|\| data.dsize != sizeof(uint32_t)) {
870	DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
871	return NT_STATUS_INTERNAL_ERROR;
872	}
873
874	db_id = (uint32_t *)data.dptr;
875	talloc_free(data.dptr);
876
877	if (!(tdb_flags & TDB_SEQNUM)) {
878	return NT_STATUS_OK;
879	}
880
881	data.dptr = (uint8_t *)db_id;
882	data.dsize = sizeof(*db_id);
883
884	status = ctdbd_control(conn, CTDB_CURRENT_NODE,
885	CTDB_CONTROL_ENABLE_SEQNUM, 0, 0, data,
886	NULL, NULL, &cstatus);
887	if (!NT_STATUS_IS_OK(status) \|\| cstatus != 0) {
888	DEBUG(0,(__location__ " ctdb_control for enable seqnum "
889	"failed\n"));
890	return NT_STATUS_IS_OK(status) ? NT_STATUS_INTERNAL_ERROR :
891	status;
892	}
893
894	return NT_STATUS_OK;
895	}
896
897	/*
898	* force the migration of a record to this node
899	*/
900	NTSTATUS ctdbd_migrate(struct ctdbd_connection *conn, uint32 db_id,
901	TDB_DATA key)
902	{
903	struct ctdb_req_call req;
904	struct ctdb_reply_call *reply;
905	NTSTATUS status;
906
907	ZERO_STRUCT(req);
908
909	req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
910	req.hdr.ctdb_magic = CTDB_MAGIC;
911	req.hdr.ctdb_version = CTDB_VERSION;
912	req.hdr.operation = CTDB_REQ_CALL;
913	req.hdr.reqid = ++conn->reqid;
914	req.flags = CTDB_IMMEDIATE_MIGRATION;
915	req.callid = CTDB_NULL_FUNC;
916	req.db_id = db_id;
917	req.keylen = key.dsize;
918
919	DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
920	ctdb_packet_dump(&req.hdr);
921
922	status = packet_send(
923	conn->pkt, 2,
924	data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
925	data_blob_const(key.dptr, key.dsize));
926
927	if (!NT_STATUS_IS_OK(status)) {
928	DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
929	return status;
930	}
931
932	status = packet_flush(conn->pkt);
933
934	if (!NT_STATUS_IS_OK(status)) {
935	DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
936	cluster_fatal("cluster dispatch daemon control write error\n");
937	}
938
939	status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
940
941	if (!NT_STATUS_IS_OK(status)) {
942	DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
943	goto fail;
944	}
945
946	if (reply->hdr.operation != CTDB_REPLY_CALL) {
947	DEBUG(0, ("received invalid reply\n"));
948	status = NT_STATUS_INTERNAL_ERROR;
949	goto fail;
950	}
951
952	status = NT_STATUS_OK;
953	fail:
954
955	TALLOC_FREE(reply);
956	return status;
957	}
958
959	/*
960	* remotely fetch a record without locking it or forcing a migration
961	*/
962	NTSTATUS ctdbd_fetch(struct ctdbd_connection *conn, uint32 db_id,
963	TDB_DATA key, TALLOC_CTX mem_ctx, TDB_DATA data)
964	{
965	struct ctdb_req_call req;
966	struct ctdb_reply_call *reply;
967	NTSTATUS status;
968
969	ZERO_STRUCT(req);
970
971	req.hdr.length = offsetof(struct ctdb_req_call, data) + key.dsize;
972	req.hdr.ctdb_magic = CTDB_MAGIC;
973	req.hdr.ctdb_version = CTDB_VERSION;
974	req.hdr.operation = CTDB_REQ_CALL;
975	req.hdr.reqid = ++conn->reqid;
976	req.flags = 0;
977	req.callid = CTDB_FETCH_FUNC;
978	req.db_id = db_id;
979	req.keylen = key.dsize;
980
981	status = packet_send(
982	conn->pkt, 2,
983	data_blob_const(&req, offsetof(struct ctdb_req_call, data)),
984	data_blob_const(key.dptr, key.dsize));
985
986	if (!NT_STATUS_IS_OK(status)) {
987	DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status)));
988	return status;
989	}
990
991	status = packet_flush(conn->pkt);
992
993	if (!NT_STATUS_IS_OK(status)) {
994	DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status)));
995	cluster_fatal("cluster dispatch daemon control write error\n");
996	}
997
998	status = ctdb_read_req(conn, req.hdr.reqid, NULL, (void *)&reply);
999
1000	if (!NT_STATUS_IS_OK(status)) {
1001	DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status)));
1002	goto fail;
1003	}
1004
1005	if (reply->hdr.operation != CTDB_REPLY_CALL) {
1006	DEBUG(0, ("received invalid reply\n"));
1007	status = NT_STATUS_INTERNAL_ERROR;
1008	goto fail;
1009	}
1010
1011	data->dsize = reply->datalen;
1012	if (data->dsize == 0) {
1013	data->dptr = NULL;
1014	goto done;
1015	}
1016
1017	data->dptr = (uint8 *)talloc_memdup(mem_ctx, &reply->data[0],
1018	reply->datalen);
1019	if (data->dptr == NULL) {
1020	DEBUG(0, ("talloc failed\n"));
1021	status = NT_STATUS_NO_MEMORY;
1022	goto fail;
1023	}
1024
1025	done:
1026	status = NT_STATUS_OK;
1027	fail:
1028	TALLOC_FREE(reply);
1029	return status;
1030	}
1031
1032	struct ctdbd_traverse_state {
1033	void (fn)(TDB_DATA key, TDB_DATA data, void private_data);
1034	void *private_data;
1035	};
1036
1037	/*
1038	* Handle a traverse record coming in on the ctdbd connection
1039	*/
1040
1041	static NTSTATUS ctdb_traverse_handler(uint8_t *buf, size_t length,
1042	void *private_data)
1043	{
1044	struct ctdbd_traverse_state *state =
1045	(struct ctdbd_traverse_state *)private_data;
1046
1047	struct ctdb_req_message *m;
1048	struct ctdb_rec_data *d;
1049	TDB_DATA key, data;
1050
1051	m = (struct ctdb_req_message *)buf;
1052
1053	if (length < sizeof(*m) \|\| m->hdr.length != length) {
1054	DEBUG(0, ("Got invalid message of length %d\n", (int)length));
1055	TALLOC_FREE(buf);
1056	return NT_STATUS_UNEXPECTED_IO_ERROR;
1057	}
1058
1059	d = (struct ctdb_rec_data *)&m->data[0];
1060	if (m->datalen < sizeof(uint32_t) \|\| m->datalen != d->length) {
1061	DEBUG(0, ("Got invalid traverse data of length %d\n",
1062	(int)m->datalen));
1063	TALLOC_FREE(buf);
1064	return NT_STATUS_UNEXPECTED_IO_ERROR;
1065	}
1066
1067	key.dsize = d->keylen;
1068	key.dptr = &d->data[0];
1069	data.dsize = d->datalen;
1070	data.dptr = &d->data[d->keylen];
1071
1072	if (key.dsize == 0 && data.dsize == 0) {
1073	/* end of traverse */
1074	return NT_STATUS_END_OF_FILE;
1075	}
1076
1077	if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1078	DEBUG(0, ("Got invalid ltdb header length %d\n",
1079	(int)data.dsize));
1080	TALLOC_FREE(buf);
1081	return NT_STATUS_UNEXPECTED_IO_ERROR;
1082	}
1083	data.dsize -= sizeof(struct ctdb_ltdb_header);
1084	data.dptr += sizeof(struct ctdb_ltdb_header);
1085
1086	if (state->fn) {
1087	state->fn(key, data, state->private_data);
1088	}
1089
1090	TALLOC_FREE(buf);
1091	return NT_STATUS_OK;
1092	}
1093
1094	/*
1095	Traverse a ctdb database. This uses a kind-of hackish way to open a second
1096	connection to ctdbd to avoid the hairy recursive and async problems with
1097	everything in-line.
1098	*/
1099
1100	NTSTATUS ctdbd_traverse(uint32 db_id,
1101	void (*fn)(TDB_DATA key, TDB_DATA data,
1102	void *private_data),
1103	void *private_data)
1104	{
1105	struct ctdbd_connection *conn;
1106	NTSTATUS status;
1107
1108	TDB_DATA data;
1109	struct ctdb_traverse_start t;
1110	int cstatus;
1111	struct ctdbd_traverse_state state;
1112
1113	status = ctdbd_init_connection(NULL, &conn);
1114	if (!NT_STATUS_IS_OK(status)) {
1115	DEBUG(0, ("ctdbd_init_connection failed: %s\n",
1116	nt_errstr(status)));
1117	return status;
1118	}
1119
1120	t.db_id = db_id;
1121	t.srvid = conn->rand_srvid;
1122	t.reqid = ++conn->reqid;
1123
1124	data.dptr = (uint8_t *)&t;
1125	data.dsize = sizeof(t);
1126
1127	status = ctdbd_control(conn, CTDB_CURRENT_NODE,
1128	CTDB_CONTROL_TRAVERSE_START, conn->rand_srvid, 0,
1129	data, NULL, NULL, &cstatus);
1130
1131	if (!NT_STATUS_IS_OK(status) \|\| (cstatus != 0)) {
1132
1133	DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status),
1134	cstatus));
1135
1136	if (NT_STATUS_IS_OK(status)) {
1137	/*
1138	* We need a mapping here
1139	*/
1140	status = NT_STATUS_UNSUCCESSFUL;
1141	}
1142	goto done;
1143	}
1144
1145	state.fn = fn;
1146	state.private_data = private_data;
1147
1148	while (True) {
1149
1150	status = NT_STATUS_OK;
1151
1152	if (packet_handler(conn->pkt, ctdb_req_complete,
1153	ctdb_traverse_handler, &state, &status)) {
1154
1155	if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1156	status = NT_STATUS_OK;
1157	break;
1158	}
1159
1160	/*
1161	* There might be more in the queue
1162	*/
1163	continue;
1164	}
1165
1166	if (!NT_STATUS_IS_OK(status)) {
1167	break;
1168	}
1169
1170	status = ctdb_packet_fd_read_sync(conn->pkt);
1171
1172	if (NT_STATUS_EQUAL(status, NT_STATUS_RETRY)) {
1173	/*
1174	* There might be more in the queue
1175	*/
1176	continue;
1177	}
1178
1179	if (NT_STATUS_EQUAL(status, NT_STATUS_END_OF_FILE)) {
1180	status = NT_STATUS_OK;
1181	break;
1182	}
1183
1184	if (!NT_STATUS_IS_OK(status)) {
1185	DEBUG(0, ("packet_fd_read_sync failed: %s\n", nt_errstr(status)));
1186	cluster_fatal("ctdbd died\n");
1187	}
1188	}
1189
1190	done:
1191	TALLOC_FREE(conn);
1192	return status;
1193	}
1194
1195	/*
1196	This is used to canonicalize a ctdb_sock_addr structure.
1197	*/
1198	static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1199	struct sockaddr_storage *out)
1200	{
1201	memcpy(out, in, sizeof (*out));
1202
1203	#ifdef HAVE_IPV6
1204	if (in->ss_family == AF_INET6) {
1205	const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1206	const struct sockaddr_in6 in6 = (struct sockaddr_in6 )in;
1207	struct sockaddr_in out4 = (struct sockaddr_in )out;
1208	if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1209	memset(out, 0, sizeof(*out));
1210	#ifdef HAVE_SOCK_SIN_LEN
1211	out4->sin_len = sizeof(*out);
1212	#endif
1213	out4->sin_family = AF_INET;
1214	out4->sin_port = in6->sin6_port;
1215	memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr32[3], 4);
1216	}
1217	}
1218	#endif
1219	}
1220
1221	/*
1222	* Register us as a server for a particular tcp connection
1223	*/
1224
1225	NTSTATUS ctdbd_register_ips(struct ctdbd_connection *conn,
1226	const struct sockaddr_storage *_server,
1227	const struct sockaddr_storage *_client,
1228	void (release_ip_handler)(const char ip_addr,
1229	void *private_data),
1230	void *private_data)
1231	{
1232	/*
1233	* we still use ctdb_control_tcp for ipv4
1234	* because we want to work against older ctdb
1235	* versions at runtime
1236	*/
1237	struct ctdb_control_tcp p4;
1238	#ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1239	struct ctdb_control_tcp_addr p;
1240	#endif
1241	TDB_DATA data;
1242	NTSTATUS status;
1243	struct sockaddr_storage client;
1244	struct sockaddr_storage server;
1245
1246	/*
1247	* Only one connection so far
1248	*/
1249	SMB_ASSERT(conn->release_ip_handler == NULL);
1250
1251	smbd_ctdb_canonicalize_ip(_client, &client);
1252	smbd_ctdb_canonicalize_ip(_server, &server);
1253
1254	switch (client.ss_family) {
1255	case AF_INET:
1256	p4.dest = (struct sockaddr_in )&server;
1257	p4.src = (struct sockaddr_in )&client;
1258	data.dptr = (uint8_t *)&p4;
1259	data.dsize = sizeof(p4);
1260	break;
1261	#ifdef HAVE_STRUCT_CTDB_CONTROL_TCP_ADDR
1262	case AF_INET6:
1263	p.dest.ip6 = (struct sockaddr_in6 )&server;
1264	p.src.ip6 = (struct sockaddr_in6 )&client;
1265	data.dptr = (uint8_t *)&p;
1266	data.dsize = sizeof(p);
1267	break;
1268	#endif
1269	default:
1270	return NT_STATUS_INTERNAL_ERROR;
1271	}
1272
1273	conn->release_ip_handler = release_ip_handler;
1274
1275	/*
1276	* We want to be told about IP releases
1277	*/
1278
1279	status = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP);
1280	if (!NT_STATUS_IS_OK(status)) {
1281	return status;
1282	}
1283
1284	/*
1285	* inform ctdb of our tcp connection, so if IP takeover happens ctdb
1286	* can send an extra ack to trigger a reset for our client, so it
1287	* immediately reconnects