2 Unix SMB/CIFS implementation.
3 Samba internal messaging functions
4 Copyright (C) 2007 by Volker Lendecke
5 Copyright (C) 2007 by Andrew Tridgell
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.
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.
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/>.
23 #ifdef CLUSTER_SUPPORT
25 #include "librpc/gen_ndr/messaging.h"
26 #include "librpc/gen_ndr/ndr_messaging.h"
28 /* paths to these include files come from --with-ctdb= in configure */
30 #include "ctdb_private.h"
32 struct ctdbd_connection
{
33 struct messaging_context
*msg_ctx
;
37 struct packet_context
*pkt
;
40 void (*release_ip_handler
)(const char *ip_addr
, void *private_data
);
41 void *release_ip_priv
;
44 static NTSTATUS
ctdbd_control(struct ctdbd_connection
*conn
,
45 uint32_t vnn
, uint32 opcode
,
46 uint64_t srvid
, TDB_DATA data
,
47 TALLOC_CTX
*mem_ctx
, TDB_DATA
*outdata
,
51 * exit on fatal communications errors with the ctdbd daemon
53 static void cluster_fatal(const char *why
)
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
66 static void ctdb_packet_dump(struct ctdb_req_header
*hdr
)
68 if (DEBUGLEVEL
< 10) {
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
));
78 * Register a srvid with ctdbd
80 static NTSTATUS
register_with_ctdbd(struct ctdbd_connection
*conn
,
85 return ctdbd_control(conn
, CTDB_CURRENT_NODE
,
86 CTDB_CONTROL_REGISTER_SRVID
, srvid
,
87 tdb_null
, NULL
, NULL
, &cstatus
);
91 * get our vnn from the cluster
93 static NTSTATUS
get_cluster_vnn(struct ctdbd_connection
*conn
, uint32
*vnn
)
97 status
= ctdbd_control(conn
,
98 CTDB_CURRENT_NODE
, CTDB_CONTROL_GET_PNN
, 0,
99 tdb_null
, NULL
, NULL
, &cstatus
);
100 if (!NT_STATUS_IS_OK(status
)) {
101 cluster_fatal("ctdbd_control failed\n");
103 *vnn
= (uint32_t)cstatus
;
107 uint32
ctdbd_vnn(const struct ctdbd_connection
*conn
)
109 return conn
->our_vnn
;
113 * Get us a ctdb connection
116 static NTSTATUS
ctdbd_connect(TALLOC_CTX
*mem_ctx
,
117 struct packet_context
**presult
)
119 struct packet_context
*result
;
120 const char *sockname
= lp_ctdbd_socket();
121 struct sockaddr_un addr
;
124 if (!sockname
|| !*sockname
) {
125 sockname
= CTDB_PATH
;
128 fd
= socket(AF_UNIX
, SOCK_STREAM
, 0);
130 DEBUG(3, ("Could not create socket: %s\n", strerror(errno
)));
131 return map_nt_error_from_unix(errno
);
135 addr
.sun_family
= AF_UNIX
;
136 strncpy(addr
.sun_path
, sockname
, sizeof(addr
.sun_path
));
138 if (connect(fd
, (struct sockaddr
*)&addr
, sizeof(addr
)) == -1) {
139 DEBUG(0, ("connect(%s) failed: %s\n", sockname
,
142 return map_nt_error_from_unix(errno
);
145 if (!(result
= packet_init(mem_ctx
, fd
))) {
147 return NT_STATUS_NO_MEMORY
;
155 * Do we have a complete ctdb packet in the queue?
158 static bool ctdb_req_complete(const struct data_blob
*data
,
164 if (data
->length
< sizeof(msglen
)) {
168 msglen
= *((uint32
*)data
->data
);
170 DEBUG(10, ("msglen = %d\n", msglen
));
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");
179 if (data
->length
>= msglen
) {
188 * State necessary to defer an incoming message while we are waiting for a
192 struct deferred_msg_state
{
193 struct messaging_context
*msg_ctx
;
194 struct messaging_rec
*rec
;
198 * Timed event handler for the deferred message
201 static void deferred_message_dispatch(struct event_context
*event_ctx
,
202 struct timed_event
*te
,
203 const struct timeval
*now
,
206 struct deferred_msg_state
*state
= talloc_get_type_abort(
207 private_data
, struct deferred_msg_state
);
209 messaging_dispatch_rec(state
->msg_ctx
, state
->rec
);
214 struct req_pull_state
{
220 * Pull a ctdb request out of the incoming packet queue
223 static NTSTATUS
ctdb_req_pull(const struct data_blob
*data
,
226 struct req_pull_state
*state
= (struct req_pull_state
*)private_data
;
228 state
->req
= data_blob_talloc(state
->mem_ctx
, data
->data
,
230 if (state
->req
.data
== NULL
) {
231 return NT_STATUS_NO_MEMORY
;
237 * Fetch a messaging_rec from an incoming ctdb style message
240 static struct messaging_rec
*ctdb_pull_messaging_rec(TALLOC_CTX
*mem_ctx
,
241 size_t overall_length
,
242 struct ctdb_req_message
*msg
)
244 struct messaging_rec
*result
;
246 enum ndr_err_code ndr_err
;
248 if ((overall_length
< offsetof(struct ctdb_req_message
, data
))
250 < offsetof(struct ctdb_req_message
, data
) + msg
->datalen
)) {
252 cluster_fatal("got invalid msg length");
255 if (!(result
= TALLOC_P(mem_ctx
, struct messaging_rec
))) {
256 DEBUG(0, ("talloc failed\n"));
260 blob
= data_blob_const(msg
->data
, msg
->datalen
);
262 ndr_err
= ndr_pull_struct_blob(
263 &blob
, result
, result
,
264 (ndr_pull_flags_fn_t
)ndr_pull_messaging_rec
);
266 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
267 DEBUG(0, ("ndr_pull_struct_blob failed: %s\n",
268 ndr_errstr(ndr_err
)));
273 if (DEBUGLEVEL
>= 10) {
274 DEBUG(10, ("ctdb_pull_messaging_rec:\n"));
275 NDR_PRINT_DEBUG(messaging_rec
, result
);
282 * Read a full ctdbd request. If we have a messaging context, defer incoming
283 * messages that might come in between.
286 static NTSTATUS
ctdb_read_req(struct ctdbd_connection
*conn
, uint32 reqid
,
287 TALLOC_CTX
*mem_ctx
, void *result
)
289 struct ctdb_req_header
*hdr
;
290 struct req_pull_state state
;
295 status
= packet_fd_read_sync(conn
->pkt
);
297 if (NT_STATUS_EQUAL(status
, NT_STATUS_NETWORK_BUSY
)) {
300 } else if (NT_STATUS_EQUAL(status
, NT_STATUS_RETRY
)) {
305 if (!NT_STATUS_IS_OK(status
)) {
306 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status
)));
307 cluster_fatal("ctdbd died\n");
313 state
.mem_ctx
= mem_ctx
;
315 if (!packet_handler(conn
->pkt
, ctdb_req_complete
, ctdb_req_pull
,
320 DEBUG(10, ("not enough data from ctdb socket, retrying\n"));
324 if (!NT_STATUS_IS_OK(status
)) {
325 DEBUG(0, ("Could not read packet: %s\n", nt_errstr(status
)));
326 cluster_fatal("ctdbd died\n");
329 hdr
= (struct ctdb_req_header
*)state
.req
.data
;
331 DEBUG(10, ("Received ctdb packet\n"));
332 ctdb_packet_dump(hdr
);
334 if (hdr
->operation
== CTDB_REQ_MESSAGE
) {
335 struct timed_event
*evt
;
336 struct deferred_msg_state
*msg_state
;
337 struct ctdb_req_message
*msg
= (struct ctdb_req_message
*)hdr
;
339 if (conn
->msg_ctx
== NULL
) {
340 DEBUG(1, ("Got a message without having a msg ctx, "
341 "dropping msg %llu\n",
342 (long long unsigned)msg
->srvid
));
346 if ((conn
->release_ip_handler
!= NULL
)
347 && (msg
->srvid
== CTDB_SRVID_RELEASE_IP
)) {
348 /* must be dispatched immediately */
349 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
350 conn
->release_ip_handler((const char *)msg
->data
,
351 conn
->release_ip_priv
);
356 if (!(msg_state
= TALLOC_P(NULL
, struct deferred_msg_state
))) {
357 DEBUG(0, ("talloc failed\n"));
362 if (!(msg_state
->rec
= ctdb_pull_messaging_rec(
363 msg_state
, state
.req
.length
, msg
))) {
364 DEBUG(0, ("ctdbd_pull_messaging_rec failed\n"));
365 TALLOC_FREE(msg_state
);
372 msg_state
->msg_ctx
= conn
->msg_ctx
;
375 * We're waiting for a call reply, but an async message has
376 * crossed. Defer dispatching to the toplevel event loop.
378 evt
= event_add_timed(conn
->msg_ctx
->event_ctx
,
379 conn
->msg_ctx
->event_ctx
,
381 "deferred_message_dispatch",
382 deferred_message_dispatch
,
385 DEBUG(0, ("event_add_timed failed\n"));
386 TALLOC_FREE(msg_state
);
394 if (hdr
->reqid
!= reqid
) {
395 /* we got the wrong reply */
396 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
397 "been %u\n", hdr
->reqid
, reqid
));
402 *((void **)result
) = talloc_move(mem_ctx
, &hdr
);
408 * Get us a ctdbd connection
411 NTSTATUS
ctdbd_init_connection(TALLOC_CTX
*mem_ctx
,
412 struct ctdbd_connection
**pconn
)
414 struct ctdbd_connection
*conn
;
417 if (!(conn
= TALLOC_ZERO_P(mem_ctx
, struct ctdbd_connection
))) {
418 DEBUG(0, ("talloc failed\n"));
419 return NT_STATUS_NO_MEMORY
;
422 status
= ctdbd_connect(conn
, &conn
->pkt
);
424 if (!NT_STATUS_IS_OK(status
)) {
425 DEBUG(10, ("ctdbd_connect failed: %s\n", nt_errstr(status
)));
429 status
= get_cluster_vnn(conn
, &conn
->our_vnn
);
431 if (!NT_STATUS_IS_OK(status
)) {
432 DEBUG(10, ("get_cluster_vnn failed: %s\n", nt_errstr(status
)));
436 generate_random_buffer((unsigned char *)&conn
->rand_srvid
,
437 sizeof(conn
->rand_srvid
));
439 status
= register_with_ctdbd(conn
, conn
->rand_srvid
);
441 if (!NT_STATUS_IS_OK(status
)) {
442 DEBUG(5, ("Could not register random srvid: %s\n",
456 * Get us a ctdbd connection and register us as a process
459 NTSTATUS
ctdbd_messaging_connection(TALLOC_CTX
*mem_ctx
,
460 struct ctdbd_connection
**pconn
)
462 struct ctdbd_connection
*conn
;
465 status
= ctdbd_init_connection(mem_ctx
, &conn
);
467 if (!NT_STATUS_IS_OK(status
)) {
471 status
= register_with_ctdbd(conn
, (uint64_t)sys_getpid());
472 if (!NT_STATUS_IS_OK(status
)) {
476 status
= register_with_ctdbd(conn
, MSG_SRVID_SAMBA
);
477 if (!NT_STATUS_IS_OK(status
)) {
490 * Packet handler to receive and handle a ctdb message
492 static NTSTATUS
ctdb_handle_message(const struct data_blob
*data
,
495 struct ctdbd_connection
*conn
= talloc_get_type_abort(
496 private_data
, struct ctdbd_connection
);
497 struct ctdb_req_message
*msg
;
498 struct messaging_rec
*msg_rec
;
500 msg
= (struct ctdb_req_message
*)data
->data
;
502 if (msg
->hdr
.operation
!= CTDB_REQ_MESSAGE
) {
503 DEBUG(0, ("Received async msg of type %u, discarding\n",
504 msg
->hdr
.operation
));
505 return NT_STATUS_INVALID_PARAMETER
;
508 if ((conn
->release_ip_handler
!= NULL
)
509 && (msg
->srvid
== CTDB_SRVID_RELEASE_IP
)) {
510 /* must be dispatched immediately */
511 DEBUG(10, ("received CTDB_SRVID_RELEASE_IP\n"));
512 conn
->release_ip_handler((const char *)msg
->data
,
513 conn
->release_ip_priv
);
517 SMB_ASSERT(conn
->msg_ctx
!= NULL
);
519 if (msg
->srvid
== CTDB_SRVID_RECONFIGURE
) {
520 DEBUG(0,("Got cluster reconfigure message\n"));
522 * when the cluster is reconfigured, we need to clean the brl
525 messaging_send(conn
->msg_ctx
, procid_self(),
526 MSG_SMB_BRL_VALIDATE
, &data_blob_null
);
529 * it's possible that we have just rejoined the cluster after
530 * an outage. In that case our pending locks could have been
531 * removed from the lockdb, so retry them once more
533 message_send_all(conn
->msg_ctx
, MSG_SMB_UNLOCK
, NULL
, 0, NULL
);
539 /* only messages to our pid or the broadcast are valid here */
540 if (msg
->srvid
!= sys_getpid() && msg
->srvid
!= MSG_SRVID_SAMBA
) {
541 DEBUG(0,("Got unexpected message with srvid=%llu\n",
542 (unsigned long long)msg
->srvid
));
546 if (!(msg_rec
= ctdb_pull_messaging_rec(NULL
, data
->length
, msg
))) {
547 DEBUG(10, ("ctdb_pull_messaging_rec failed\n"));
548 return NT_STATUS_NO_MEMORY
;
551 messaging_dispatch_rec(conn
->msg_ctx
, msg_rec
);
553 TALLOC_FREE(msg_rec
);
558 * The ctdbd socket is readable asynchronuously
561 static void ctdbd_socket_handler(struct event_context
*event_ctx
,
562 struct fd_event
*event
,
566 struct ctdbd_connection
*conn
= talloc_get_type_abort(
567 private_data
, struct ctdbd_connection
);
571 status
= packet_fd_read(conn
->pkt
);
573 if (!NT_STATUS_IS_OK(status
)) {
574 DEBUG(0, ("packet_fd_read failed: %s\n", nt_errstr(status
)));
575 cluster_fatal("ctdbd died\n");
578 while (packet_handler(conn
->pkt
, ctdb_req_complete
,
579 ctdb_handle_message
, conn
, &status
)) {
580 if (!NT_STATUS_IS_OK(status
)) {
581 DEBUG(10, ("could not handle incoming message: %s\n",
588 * Prepare a ctdbd connection to receive messages
591 NTSTATUS
ctdbd_register_msg_ctx(struct ctdbd_connection
*conn
,
592 struct messaging_context
*msg_ctx
)
594 SMB_ASSERT(conn
->msg_ctx
== NULL
);
595 SMB_ASSERT(conn
->fde
== NULL
);
597 if (!(conn
->fde
= event_add_fd(msg_ctx
->event_ctx
, conn
,
598 packet_get_fd(conn
->pkt
),
600 ctdbd_socket_handler
,
602 DEBUG(0, ("event_add_fd failed\n"));
603 return NT_STATUS_NO_MEMORY
;
606 conn
->msg_ctx
= msg_ctx
;
612 * Send a messaging message across a ctdbd
615 NTSTATUS
ctdbd_messaging_send(struct ctdbd_connection
*conn
,
616 uint32 dst_vnn
, uint64 dst_srvid
,
617 struct messaging_rec
*msg
)
619 struct ctdb_req_message r
;
623 enum ndr_err_code ndr_err
;
625 if (!(mem_ctx
= talloc_init("ctdbd_messaging_send"))) {
626 DEBUG(0, ("talloc failed\n"));
627 return NT_STATUS_NO_MEMORY
;
630 ndr_err
= ndr_push_struct_blob(
632 (ndr_push_flags_fn_t
)ndr_push_messaging_rec
);
634 if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err
)) {
635 DEBUG(0, ("ndr_push_struct_blob failed: %s\n",
636 ndr_errstr(ndr_err
)));
637 status
= ndr_map_error2ntstatus(ndr_err
);
641 r
.hdr
.length
= offsetof(struct ctdb_req_message
, data
) + blob
.length
;
642 r
.hdr
.ctdb_magic
= CTDB_MAGIC
;
643 r
.hdr
.ctdb_version
= CTDB_VERSION
;
644 r
.hdr
.generation
= 1;
645 r
.hdr
.operation
= CTDB_REQ_MESSAGE
;
646 r
.hdr
.destnode
= dst_vnn
;
647 r
.hdr
.srcnode
= conn
->our_vnn
;
650 r
.datalen
= blob
.length
;
652 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
653 ctdb_packet_dump(&r
.hdr
);
655 status
= packet_send(
657 data_blob_const(&r
, offsetof(struct ctdb_req_message
, data
)),
660 if (!NT_STATUS_IS_OK(status
)) {
661 DEBUG(0, ("packet_send failed: %s\n", nt_errstr(status
)));
665 status
= packet_flush(conn
->pkt
);
667 if (!NT_STATUS_IS_OK(status
)) {
668 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status
)));
669 cluster_fatal("cluster dispatch daemon msg write error\n");
672 status
= NT_STATUS_OK
;
674 TALLOC_FREE(mem_ctx
);
679 * send/recv a generic ctdb control message
681 static NTSTATUS
ctdbd_control(struct ctdbd_connection
*conn
,
682 uint32_t vnn
, uint32 opcode
,
683 uint64_t srvid
, TDB_DATA data
,
684 TALLOC_CTX
*mem_ctx
, TDB_DATA
*outdata
,
687 struct ctdb_req_control req
;
688 struct ctdb_reply_control
*reply
= NULL
;
689 struct ctdbd_connection
*new_conn
= NULL
;
693 status
= ctdbd_init_connection(NULL
, &new_conn
);
695 if (!NT_STATUS_IS_OK(status
)) {
696 DEBUG(10, ("Could not init temp connection: %s\n",
705 req
.hdr
.length
= offsetof(struct ctdb_req_control
, data
) + data
.dsize
;
706 req
.hdr
.ctdb_magic
= CTDB_MAGIC
;
707 req
.hdr
.ctdb_version
= CTDB_VERSION
;
708 req
.hdr
.operation
= CTDB_REQ_CONTROL
;
709 req
.hdr
.reqid
= ++conn
->reqid
;
710 req
.hdr
.destnode
= vnn
;
713 req
.datalen
= data
.dsize
;
715 DEBUG(10, ("ctdbd_control: Sending ctdb packet\n"));
716 ctdb_packet_dump(&req
.hdr
);
718 status
= packet_send(
720 data_blob_const(&req
, offsetof(struct ctdb_req_control
, data
)),
721 data_blob_const(data
.dptr
, data
.dsize
));
723 if (!NT_STATUS_IS_OK(status
)) {
724 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status
)));
728 status
= packet_flush(conn
->pkt
);
730 if (!NT_STATUS_IS_OK(status
)) {
731 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status
)));
732 cluster_fatal("cluster dispatch daemon control write error\n");
735 status
= ctdb_read_req(conn
, req
.hdr
.reqid
, NULL
, (void *)&reply
);
737 if (!NT_STATUS_IS_OK(status
)) {
738 DEBUG(10, ("ctdb_read_req failed: %s\n", nt_errstr(status
)));
742 if (reply
->hdr
.operation
!= CTDB_REPLY_CONTROL
) {
743 DEBUG(0, ("received invalid reply\n"));
748 if (!(outdata
->dptr
= (uint8
*)talloc_memdup(
749 mem_ctx
, reply
->data
, reply
->datalen
))) {
751 return NT_STATUS_NO_MEMORY
;
753 outdata
->dsize
= reply
->datalen
;
756 (*cstatus
) = reply
->status
;
759 status
= NT_STATUS_OK
;
762 TALLOC_FREE(new_conn
);
768 * see if a remote process exists
770 bool ctdbd_process_exists(struct ctdbd_connection
*conn
, uint32 vnn
, pid_t pid
)
776 data
.dptr
= (uint8_t*)&pid
;
777 data
.dsize
= sizeof(pid
);
779 status
= ctdbd_control(conn
, vnn
, CTDB_CONTROL_PROCESS_EXISTS
, 0,
780 data
, NULL
, NULL
, &cstatus
);
781 if (!NT_STATUS_IS_OK(status
)) {
782 DEBUG(0, (__location__
" ctdb_control for process_exists "
793 char *ctdbd_dbpath(struct ctdbd_connection
*conn
,
794 TALLOC_CTX
*mem_ctx
, uint32_t db_id
)
800 data
.dptr
= (uint8_t*)&db_id
;
801 data
.dsize
= sizeof(db_id
);
803 status
= ctdbd_control(conn
, CTDB_CURRENT_NODE
,
804 CTDB_CONTROL_GETDBPATH
, 0, data
,
805 mem_ctx
, &data
, &cstatus
);
806 if (!NT_STATUS_IS_OK(status
) || cstatus
!= 0) {
807 DEBUG(0,(__location__
" ctdb_control for getdbpath failed\n"));
811 return (char *)data
.dptr
;
815 * attach to a ctdb database
817 NTSTATUS
ctdbd_db_attach(struct ctdbd_connection
*conn
,
818 const char *name
, uint32_t *db_id
, int tdb_flags
)
824 data
.dptr
= (uint8_t*)name
;
825 data
.dsize
= strlen(name
)+1;
827 status
= ctdbd_control(conn
, CTDB_CURRENT_NODE
,
828 CTDB_CONTROL_DB_ATTACH
, 0, data
,
829 NULL
, &data
, &cstatus
);
830 if (!NT_STATUS_IS_OK(status
)) {
831 DEBUG(0, (__location__
" ctdb_control for db_attach "
832 "failed: %s\n", nt_errstr(status
)));
836 if (cstatus
!= 0 || data
.dsize
!= sizeof(uint32_t)) {
837 DEBUG(0,(__location__
" ctdb_control for db_attach failed\n"));
838 return NT_STATUS_INTERNAL_ERROR
;
841 *db_id
= *(uint32_t *)data
.dptr
;
842 talloc_free(data
.dptr
);
844 if (!(tdb_flags
& TDB_SEQNUM
)) {
848 data
.dptr
= (uint8_t *)db_id
;
849 data
.dsize
= sizeof(*db_id
);
851 status
= ctdbd_control(conn
, CTDB_CURRENT_NODE
,
852 CTDB_CONTROL_ENABLE_SEQNUM
, 0, data
,
853 NULL
, NULL
, &cstatus
);
854 if (!NT_STATUS_IS_OK(status
) || cstatus
!= 0) {
855 DEBUG(0,(__location__
" ctdb_control for enable seqnum "
857 return NT_STATUS_IS_OK(status
) ? NT_STATUS_INTERNAL_ERROR
:
865 * force the migration of a record to this node
867 NTSTATUS
ctdbd_migrate(struct ctdbd_connection
*conn
, uint32 db_id
,
870 struct ctdb_req_call req
;
871 struct ctdb_reply_call
*reply
;
876 req
.hdr
.length
= offsetof(struct ctdb_req_call
, data
) + key
.dsize
;
877 req
.hdr
.ctdb_magic
= CTDB_MAGIC
;
878 req
.hdr
.ctdb_version
= CTDB_VERSION
;
879 req
.hdr
.operation
= CTDB_REQ_CALL
;
880 req
.hdr
.reqid
= ++conn
->reqid
;
881 req
.flags
= CTDB_IMMEDIATE_MIGRATION
;
882 req
.callid
= CTDB_NULL_FUNC
;
884 req
.keylen
= key
.dsize
;
886 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
887 ctdb_packet_dump(&req
.hdr
);
889 status
= packet_send(
891 data_blob_const(&req
, offsetof(struct ctdb_req_call
, data
)),
892 data_blob_const(key
.dptr
, key
.dsize
));
894 if (!NT_STATUS_IS_OK(status
)) {
895 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status
)));
899 status
= packet_flush(conn
->pkt
);
901 if (!NT_STATUS_IS_OK(status
)) {
902 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status
)));
903 cluster_fatal("cluster dispatch daemon control write error\n");
906 status
= ctdb_read_req(conn
, req
.hdr
.reqid
, NULL
, (void *)&reply
);
908 if (!NT_STATUS_IS_OK(status
)) {
909 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status
)));
913 if (reply
->hdr
.operation
!= CTDB_REPLY_CALL
) {
914 DEBUG(0, ("received invalid reply\n"));
915 status
= NT_STATUS_INTERNAL_ERROR
;
919 status
= NT_STATUS_OK
;
927 * remotely fetch a record without locking it or forcing a migration
929 NTSTATUS
ctdbd_fetch(struct ctdbd_connection
*conn
, uint32 db_id
,
930 TDB_DATA key
, TALLOC_CTX
*mem_ctx
, TDB_DATA
*data
)
932 struct ctdb_req_call req
;
933 struct ctdb_reply_call
*reply
;
938 req
.hdr
.length
= offsetof(struct ctdb_req_call
, data
) + key
.dsize
;
939 req
.hdr
.ctdb_magic
= CTDB_MAGIC
;
940 req
.hdr
.ctdb_version
= CTDB_VERSION
;
941 req
.hdr
.operation
= CTDB_REQ_CALL
;
942 req
.hdr
.reqid
= ++conn
->reqid
;
944 req
.callid
= CTDB_FETCH_FUNC
;
946 req
.keylen
= key
.dsize
;
948 status
= packet_send(
950 data_blob_const(&req
, offsetof(struct ctdb_req_call
, data
)),
951 data_blob_const(key
.dptr
, key
.dsize
));
953 if (!NT_STATUS_IS_OK(status
)) {
954 DEBUG(3, ("packet_send failed: %s\n", nt_errstr(status
)));
958 status
= packet_flush(conn
->pkt
);
960 if (!NT_STATUS_IS_OK(status
)) {
961 DEBUG(3, ("write to ctdbd failed: %s\n", nt_errstr(status
)));
962 cluster_fatal("cluster dispatch daemon control write error\n");
965 status
= ctdb_read_req(conn
, req
.hdr
.reqid
, NULL
, (void *)&reply
);
967 if (!NT_STATUS_IS_OK(status
)) {
968 DEBUG(0, ("ctdb_read_req failed: %s\n", nt_errstr(status
)));
972 if (reply
->hdr
.operation
!= CTDB_REPLY_CALL
) {
973 DEBUG(0, ("received invalid reply\n"));
974 status
= NT_STATUS_INTERNAL_ERROR
;
978 data
->dsize
= reply
->datalen
;
979 if (data
->dsize
== 0) {
984 data
->dptr
= (uint8
*)talloc_memdup(mem_ctx
, &reply
->data
[0],
986 if (data
->dptr
== NULL
) {
987 DEBUG(0, ("talloc failed\n"));
988 status
= NT_STATUS_NO_MEMORY
;
993 status
= NT_STATUS_OK
;
999 struct ctdbd_traverse_state
{
1000 void (*fn
)(TDB_DATA key
, TDB_DATA data
, void *private_data
);
1005 * Handle a traverse record coming in on the ctdbd connection
1008 static NTSTATUS
ctdb_traverse_handler(const struct data_blob
*blob
,
1011 struct ctdbd_traverse_state
*state
=
1012 (struct ctdbd_traverse_state
*)private_data
;
1014 struct ctdb_req_message
*m
;
1015 struct ctdb_rec_data
*d
;
1018 m
= (struct ctdb_req_message
*)blob
->data
;
1020 if (blob
->length
< sizeof(*m
) || m
->hdr
.length
!= blob
->length
) {
1021 DEBUG(0, ("Got invalid message of length %d\n",
1022 (int)blob
->length
));
1023 return NT_STATUS_UNEXPECTED_IO_ERROR
;
1026 d
= (struct ctdb_rec_data
*)&m
->data
[0];
1027 if (m
->datalen
< sizeof(uint32_t) || m
->datalen
!= d
->length
) {
1028 DEBUG(0, ("Got invalid traverse data of length %d\n",
1030 return NT_STATUS_UNEXPECTED_IO_ERROR
;
1033 key
.dsize
= d
->keylen
;
1034 key
.dptr
= &d
->data
[0];
1035 data
.dsize
= d
->datalen
;
1036 data
.dptr
= &d
->data
[d
->keylen
];
1038 if (key
.dsize
== 0 && data
.dsize
== 0) {
1039 /* end of traverse */
1040 return NT_STATUS_END_OF_FILE
;
1043 if (data
.dsize
< sizeof(struct ctdb_ltdb_header
)) {
1044 DEBUG(0, ("Got invalid ltdb header length %d\n",
1046 return NT_STATUS_UNEXPECTED_IO_ERROR
;
1048 data
.dsize
-= sizeof(struct ctdb_ltdb_header
);
1049 data
.dptr
+= sizeof(struct ctdb_ltdb_header
);
1052 state
->fn(key
, data
, state
->private_data
);
1055 return NT_STATUS_OK
;
1059 Traverse a ctdb database. This uses a kind-of hackish way to open a second
1060 connection to ctdbd to avoid the hairy recursive and async problems with
1064 NTSTATUS
ctdbd_traverse(uint32 db_id
,
1065 void (*fn
)(TDB_DATA key
, TDB_DATA data
,
1066 void *private_data
),
1069 struct ctdbd_connection
*conn
;
1073 struct ctdb_traverse_start t
;
1075 struct ctdbd_traverse_state state
;
1077 status
= ctdbd_init_connection(NULL
, &conn
);
1080 t
.srvid
= conn
->rand_srvid
;
1081 t
.reqid
= ++conn
->reqid
;
1083 data
.dptr
= (uint8_t *)&t
;
1084 data
.dsize
= sizeof(t
);
1086 status
= ctdbd_control(conn
, CTDB_CURRENT_NODE
,
1087 CTDB_CONTROL_TRAVERSE_START
, conn
->rand_srvid
,
1088 data
, NULL
, NULL
, &cstatus
);
1090 if (!NT_STATUS_IS_OK(status
) || (cstatus
!= 0)) {
1092 DEBUG(0,("ctdbd_control failed: %s, %d\n", nt_errstr(status
),
1095 if (NT_STATUS_IS_OK(status
)) {
1097 * We need a mapping here
1099 status
= NT_STATUS_UNSUCCESSFUL
;
1105 state
.private_data
= private_data
;
1109 status
= NT_STATUS_OK
;
1111 if (packet_handler(conn
->pkt
, ctdb_req_complete
,
1112 ctdb_traverse_handler
, &state
, &status
)) {
1114 if (NT_STATUS_EQUAL(status
, NT_STATUS_END_OF_FILE
)) {
1115 status
= NT_STATUS_OK
;
1120 * There might be more in the queue
1125 if (!NT_STATUS_IS_OK(status
)) {
1129 status
= packet_fd_read_sync(conn
->pkt
);
1131 if (NT_STATUS_EQUAL(status
, NT_STATUS_RETRY
)) {
1133 * There might be more in the queue
1138 if (NT_STATUS_EQUAL(status
, NT_STATUS_END_OF_FILE
)) {
1139 status
= NT_STATUS_OK
;
1142 if (!NT_STATUS_IS_OK(status
)) {
1143 DEBUG(0, ("packet_fd_read_sync failed: %s\n", nt_errstr(status
)));
1144 cluster_fatal("ctdbd died\n");
1154 * Register us as a server for a particular tcp connection
1157 NTSTATUS
ctdbd_register_ips(struct ctdbd_connection
*conn
,
1158 const struct sockaddr_in
*server
,
1159 const struct sockaddr_in
*client
,
1160 void (*release_ip_handler
)(const char *ip_addr
,
1161 void *private_data
),
1164 struct ctdb_control_tcp p
;
1169 * Only one connection so far
1171 SMB_ASSERT(conn
->release_ip_handler
== NULL
);
1174 * We want to be told about IP releases
1177 status
= register_with_ctdbd(conn
, CTDB_SRVID_RELEASE_IP
);
1178 if (!NT_STATUS_IS_OK(status
)) {
1186 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1187 * can send an extra ack to trigger a reset for our client, so it
1188 * immediately reconnects
1190 data
.dptr
= (uint8_t *)&p
;
1191 data
.dsize
= sizeof(p
);
1193 return ctdbd_control(conn
, CTDB_CURRENT_NODE
,
1194 CTDB_CONTROL_TCP_CLIENT
,
1195 CTDB_CTRL_FLAG_NOREPLY
, data
, NULL
, NULL
, NULL
);
1199 * We want to handle reconfigure events
1201 NTSTATUS
ctdbd_register_reconfigure(struct ctdbd_connection
*conn
)
1203 return register_with_ctdbd(conn
, CTDB_SRVID_RECONFIGURE
);
1208 NTSTATUS
ctdbd_init_connection(TALLOC_CTX
*mem_ctx
,
1209 struct ctdbd_connection
**pconn
)
1211 return NT_STATUS_NOT_IMPLEMENTED
;