?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
docs-xml: "cluster addresses" dns registration
[Samba.git] / source3 / lib / ctdbd_conn.c
blobd5fac572d3c94402798c63d2fcff999578502ab1
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 "replace.h"
22 #include <tevent.h>
23 #include "util_tdb.h"
24 #include "serverid.h"
25 #include "ctdbd_conn.h"
26 #include "system/select.h"
27 #include "lib/util/sys_rw_data.h"
28 #include "lib/util/iov_buf.h"
29 #include "lib/util/select.h"
30 #include "lib/util/debug.h"
31 #include "lib/util/talloc_stack.h"
32 #include "lib/util/genrand.h"
33 #include "lib/util/fault.h"
34 #include "lib/util/dlinklist.h"
35 #include "lib/util/tevent_unix.h"
36 #include "lib/util/sys_rw.h"
37 #include "lib/util/blocking.h"
38 #include "ctdb/include/ctdb_protocol.h"
39
40 /* paths to these include files come from --with-ctdb= in configure */
41
42 struct ctdbd_srvid_cb {
43 uint64_t srvid;
44 int (*cb)(struct tevent_context *ev,
45 uint32_t src_vnn, uint32_t dst_vnn,
46 uint64_t dst_srvid,
47 const uint8_t *msg, size_t msglen,
48 void *private_data);
49 void *private_data;
50 };
51
52 struct ctdb_pkt_send_state;
53 struct ctdb_pkt_recv_state;
54
55 struct ctdbd_connection {
56 uint32_t reqid;
57 uint32_t our_vnn;
58 uint64_t rand_srvid;
59 struct ctdbd_srvid_cb *callbacks;
60 int fd;
61 int timeout;
62
63 /* For async connections, enabled via ctdbd_setup_fde() */
64 struct tevent_fd *fde;
65
66 /* State to track in-progress read */
67 struct ctdb_read_state {
68 /* Receive buffer for the initial packet length */
69 uint32_t msglen;
70
71 /* iovec state for current read */
72 struct iovec iov;
73 struct iovec *iovs;
74 int iovcnt;
75
76 /* allocated receive buffer based on packet length */
77 struct ctdb_req_header *hdr;
78 } read_state;
79
80 /* Lists of pending async reads and writes */
81 struct ctdb_pkt_recv_state *recv_list;
82 struct ctdb_pkt_send_state *send_list;
83 };
84
85 static void ctdbd_async_socket_handler(struct tevent_context *ev,
86 struct tevent_fd *fde,
87 uint16_t flags,
88 void *private_data);
89
90 static bool ctdbd_conn_has_async_sends(struct ctdbd_connection *conn)
91 {
92 return (conn->send_list != NULL);
93 }
94
95 static bool ctdbd_conn_has_async_reqs(struct ctdbd_connection *conn)
96 {
97 return (conn->fde != NULL);
98 }
99
100 static uint32_t ctdbd_next_reqid(struct ctdbd_connection *conn)
101 {
102 conn->reqid += 1;
103 if (conn->reqid == 0) {
104 conn->reqid += 1;
105 }
106 return conn->reqid;
107 }
108
109 static int ctdbd_control(struct ctdbd_connection *conn,
110 uint32_t vnn, uint32_t opcode,
111 uint64_t srvid, uint32_t flags,
112 TDB_DATA data,
113 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
114 int32_t *cstatus);
115
116 /*
117 * exit on fatal communications errors with the ctdbd daemon
118 */
119 static void cluster_fatal(const char *why)
120 {
121 DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
122 /* we don't use smb_panic() as we don't want to delay to write
123 a core file. We need to release this process id immediately
124 so that someone else can take over without getting sharing
125 violations */
126 _exit(1);
127 }
128
129 /*
130 *
131 */
132 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
133 {
134 if (DEBUGLEVEL < 11) {
135 return;
136 }
137 DEBUGADD(11, ("len=%d, magic=%x, vers=%d, gen=%d, op=%d, reqid=%d\n",
138 (int)hdr->length, (int)hdr->ctdb_magic,
139 (int)hdr->ctdb_version, (int)hdr->generation,
140 (int)hdr->operation, (int)hdr->reqid));
141 }
142
143 /*
144 * Register a srvid with ctdbd
145 */
146 int register_with_ctdbd(struct ctdbd_connection *conn, uint64_t srvid,
147 int (*cb)(struct tevent_context *ev,
148 uint32_t src_vnn, uint32_t dst_vnn,
149 uint64_t dst_srvid,
150 const uint8_t *msg, size_t msglen,
151 void *private_data),
152 void *private_data)
153 {
154
155 int ret;
156 int32_t cstatus;
157 size_t num_callbacks;
158 struct ctdbd_srvid_cb *tmp;
159
160 ret = ctdbd_control_local(conn, CTDB_CONTROL_REGISTER_SRVID, srvid, 0,
161 tdb_null, NULL, NULL, &cstatus);
162 if (ret != 0) {
163 return ret;
164 }
165
166 num_callbacks = talloc_array_length(conn->callbacks);
167
168 tmp = talloc_realloc(conn, conn->callbacks, struct ctdbd_srvid_cb,
169 num_callbacks + 1);
170 if (tmp == NULL) {
171 return ENOMEM;
172 }
173 conn->callbacks = tmp;
174
175 conn->callbacks[num_callbacks] = (struct ctdbd_srvid_cb) {
176 .srvid = srvid, .cb = cb, .private_data = private_data
177 };
178
179 return 0;
180 }
181
182 static int ctdbd_msg_call_back(struct tevent_context *ev,
183 struct ctdbd_connection *conn,
184 struct ctdb_req_message_old *msg)
185 {
186 uint32_t msg_len;
187 size_t i, num_callbacks;
188
189 msg_len = msg->hdr.length;
190 if (msg_len < offsetof(struct ctdb_req_message_old, data)) {
191 DBG_DEBUG("len %"PRIu32" too small\n", msg_len);
192 return 0;
193 }
194 msg_len -= offsetof(struct ctdb_req_message_old, data);
195
196 if (msg_len < msg->datalen) {
197 DBG_DEBUG("msg_len=%"PRIu32" < msg->datalen=%"PRIu32"\n",
198 msg_len, msg->datalen);
199 return 0;
200 }
201
202 num_callbacks = talloc_array_length(conn->callbacks);
203
204 for (i=0; i<num_callbacks; i++) {
205 struct ctdbd_srvid_cb *cb = &conn->callbacks[i];
206
207 if ((cb->srvid == msg->srvid) && (cb->cb != NULL)) {
208 int ret;
209
210 ret = cb->cb(ev,
211 msg->hdr.srcnode, msg->hdr.destnode,
212 msg->srvid, msg->data, msg->datalen,
213 cb->private_data);
214 if (ret != 0) {
215 return ret;
216 }
217 }
218 }
219 return 0;
220 }
221
222 /*
223 * get our vnn from the cluster
224 */
225 static int get_cluster_vnn(struct ctdbd_connection *conn, uint32_t *vnn)
226 {
227 int32_t cstatus=-1;
228 int ret;
229 ret = ctdbd_control_local(conn, CTDB_CONTROL_GET_PNN, 0, 0,
230 tdb_null, NULL, NULL, &cstatus);
231 if (ret != 0) {
232 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
233 return ret;
234 }
235 *vnn = (uint32_t)cstatus;
236 return ret;
237 }
238
239 /*
240 * Are we active (i.e. not banned or stopped?)
241 */
242 static bool ctdbd_working(struct ctdbd_connection *conn, uint32_t vnn)
243 {
244 int32_t cstatus=-1;
245 TDB_DATA outdata;
246 struct ctdb_node_map_old *m;
247 bool ok = false;
248 uint32_t i;
249 int ret;
250
251 ret = ctdbd_control_local(conn, CTDB_CONTROL_GET_NODEMAP, 0, 0,
252 tdb_null, talloc_tos(), &outdata, &cstatus);
253 if (ret != 0) {
254 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
255 return false;
256 }
257 if ((cstatus != 0) || (outdata.dptr == NULL)) {
258 DEBUG(2, ("Received invalid ctdb data\n"));
259 return false;
260 }
261
262 m = (struct ctdb_node_map_old *)outdata.dptr;
263
264 for (i=0; i<m->num; i++) {
265 if (vnn == m->nodes[i].pnn) {
266 break;
267 }
268 }
269
270 if (i == m->num) {
271 DEBUG(2, ("Did not find ourselves (node %d) in nodemap\n",
272 (int)vnn));
273 goto fail;
274 }
275
276 if ((m->nodes[i].flags & NODE_FLAGS_INACTIVE) != 0) {
277 DEBUG(2, ("Node has status %x, not active\n",
278 (int)m->nodes[i].flags));
279 goto fail;
280 }
281
282 ok = true;
283 fail:
284 TALLOC_FREE(outdata.dptr);
285 return ok;
286 }
287
288 uint32_t ctdbd_vnn(const struct ctdbd_connection *conn)
289 {
290 return conn->our_vnn;
291 }
292
293 /*
294 * Get us a ctdb connection
295 */
296
297 static int ctdbd_connect(const char *sockname, int *pfd)
298 {
299 struct sockaddr_un addr = { 0, };
300 int fd;
301 socklen_t salen;
302 size_t namelen;
303
304 fd = socket(AF_UNIX, SOCK_STREAM, 0);
305 if (fd == -1) {
306 int err = errno;
307 DEBUG(3, ("Could not create socket: %s\n", strerror(err)));
308 return err;
309 }
310
311 addr.sun_family = AF_UNIX;
312
313 namelen = strlcpy(addr.sun_path, sockname, sizeof(addr.sun_path));
314 if (namelen >= sizeof(addr.sun_path)) {
315 DEBUG(3, ("%s: Socket name too long: %s\n", __func__,
316 sockname));
317 close(fd);
318 return ENAMETOOLONG;
319 }
320
321 salen = sizeof(struct sockaddr_un);
322
323 if (connect(fd, (struct sockaddr *)(void *)&addr, salen) == -1) {
324 int err = errno;
325 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
326 strerror(err)));
327 close(fd);
328 return err;
329 }
330
331 *pfd = fd;
332 return 0;
333 }
334
335 static int ctdb_read_packet(int fd, int timeout, TALLOC_CTX *mem_ctx,
336 struct ctdb_req_header **result)
337 {
338 struct ctdb_req_header *req;
339 uint32_t msglen;
340 ssize_t nread;
341
342 if (timeout != -1) {
343 struct pollfd pfd = { .fd = fd, .events = POLLIN };
344 int ret;
345
346 ret = sys_poll_intr(&pfd, 1, timeout);
347 if (ret == -1) {
348 return errno;
349 }
350 if (ret == 0) {
351 return ETIMEDOUT;
352 }
353 if (ret != 1) {
354 return EIO;
355 }
356 }
357
358 nread = read_data(fd, &msglen, sizeof(msglen));
359 if (nread == -1) {
360 return errno;
361 }
362 if (nread == 0) {
363 return EIO;
364 }
365
366 if (msglen < sizeof(struct ctdb_req_header)) {
367 return EIO;
368 }
369
370 req = talloc_size(mem_ctx, msglen);
371 if (req == NULL) {
372 return ENOMEM;
373 }
374 talloc_set_name_const(req, "struct ctdb_req_header");
375
376 req->length = msglen;
377
378 nread = read_data(fd, ((char *)req) + sizeof(msglen),
379 msglen - sizeof(msglen));
380 if (nread == -1) {
381 TALLOC_FREE(req);
382 return errno;
383 }
384 if (nread == 0) {
385 TALLOC_FREE(req);
386 return EIO;
387 }
388
389 *result = req;
390 return 0;
391 }
392
393 /*
394 * Read a full ctdbd request. If we have a messaging context, defer incoming
395 * messages that might come in between.
396 */
397
398 static int ctdb_read_req(struct ctdbd_connection *conn, uint32_t reqid,
399 TALLOC_CTX *mem_ctx, struct ctdb_req_header **result)
400 {
401 struct ctdb_req_header *hdr;
402 int ret;
403
404 next_pkt:
405
406 ret = ctdb_read_packet(conn->fd, conn->timeout, mem_ctx, &hdr);
407 if (ret != 0) {
408 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
409 cluster_fatal("failed to read data from ctdbd\n");
410 }
411
412 DEBUG(11, ("Received ctdb packet\n"));
413 ctdb_packet_dump(hdr);
414
415 if (hdr->operation == CTDB_REQ_MESSAGE) {
416 struct ctdb_req_message_old *msg = (struct ctdb_req_message_old *)hdr;
417
418 ret = ctdbd_msg_call_back(NULL, conn, msg);
419 if (ret != 0) {
420 TALLOC_FREE(hdr);
421 return ret;
422 }
423
424 TALLOC_FREE(hdr);
425 goto next_pkt;
426 }
427
428 if ((reqid != 0) && (hdr->reqid != reqid)) {
429 /* we got the wrong reply */
430 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
431 "been %u\n", hdr->reqid, reqid));
432 TALLOC_FREE(hdr);
433 goto next_pkt;
434 }
435
436 *result = talloc_move(mem_ctx, &hdr);
437
438 return 0;
439 }
440
441 /**
442 * This prepares conn for handling async requests
443 **/
444 int ctdbd_setup_fde(struct ctdbd_connection *conn, struct tevent_context *ev)
445 {
446 int ret;
447
448 ret = set_blocking(conn->fd, false);
449 if (ret == -1) {
450 return errno;
451 }
452
453 conn->fde = tevent_add_fd(ev,
454 conn,
455 conn->fd,
456 TEVENT_FD_READ,
457 ctdbd_async_socket_handler,
458 conn);
459 if (conn->fde == NULL) {
460 return ENOMEM;
461 }
462
463 return 0;
464 }
465
466 static int ctdbd_connection_destructor(struct ctdbd_connection *c);
467
468 /*
469 * Get us a ctdbd connection
470 */
471
472 static int ctdbd_init_connection_internal(TALLOC_CTX *mem_ctx,
473 const char *sockname, int timeout,
474 struct ctdbd_connection *conn)
475 {
476 int ret;
477
478 conn->timeout = timeout;
479 if (conn->timeout == 0) {
480 conn->timeout = -1;
481 }
482
483 ret = ctdbd_connect(sockname, &conn->fd);
484 if (ret != 0) {
485 DEBUG(1, ("ctdbd_connect failed: %s\n", strerror(ret)));
486 return ret;
487 }
488 talloc_set_destructor(conn, ctdbd_connection_destructor);
489
490 ret = get_cluster_vnn(conn, &conn->our_vnn);
491 if (ret != 0) {
492 DEBUG(10, ("get_cluster_vnn failed: %s\n", strerror(ret)));
493 return ret;
494 }
495
496 if (!ctdbd_working(conn, conn->our_vnn)) {
497 DEBUG(2, ("Node is not working, can not connect\n"));
498 return EIO;
499 }
500
501 generate_random_buffer((unsigned char *)&conn->rand_srvid,
502 sizeof(conn->rand_srvid));
503
504 ret = register_with_ctdbd(conn, conn->rand_srvid, NULL, NULL);
505 if (ret != 0) {
506 DEBUG(5, ("Could not register random srvid: %s\n",
507 strerror(ret)));
508 return ret;
509 }
510
511 return 0;
512 }
513
514 int ctdbd_init_connection(TALLOC_CTX *mem_ctx,
515 const char *sockname, int timeout,
516 struct ctdbd_connection **pconn)
517 {
518 struct ctdbd_connection *conn;
519 int ret;
520
521 if (!(conn = talloc_zero(mem_ctx, struct ctdbd_connection))) {
522 DEBUG(0, ("talloc failed\n"));
523 return ENOMEM;
524 }
525
526 ret = ctdbd_init_connection_internal(mem_ctx,
527 sockname,
528 timeout,
529 conn);
530 if (ret != 0) {
531 DBG_ERR("ctdbd_init_connection_internal failed (%s)\n",
532 strerror(ret));
533 goto fail;
534 }
535
536 *pconn = conn;
537 return 0;
538
539 fail:
540 TALLOC_FREE(conn);
541 return ret;
542 }
543
544 int ctdbd_reinit_connection(TALLOC_CTX *mem_ctx,
545 const char *sockname, int timeout,
546 struct ctdbd_connection *conn)
547 {
548 int ret;
549
550 ret = ctdbd_connection_destructor(conn);
551 if (ret != 0) {
552 DBG_ERR("ctdbd_connection_destructor failed\n");
553 return ret;
554 }
555
556 ret = ctdbd_init_connection_internal(mem_ctx,
557 sockname,
558 timeout,
559 conn);
560 if (ret != 0) {
561 DBG_ERR("ctdbd_init_connection_internal failed (%s)\n",
562 strerror(ret));
563 return ret;
564 }
565
566 return 0;
567 }
568
569 int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
570 {
571 return conn->fd;
572 }
573
574 /*
575 * Packet handler to receive and handle a ctdb message
576 */
577 static int ctdb_handle_message(struct tevent_context *ev,
578 struct ctdbd_connection *conn,
579 struct ctdb_req_header *hdr)
580 {
581 struct ctdb_req_message_old *msg;
582
583 if (hdr->operation != CTDB_REQ_MESSAGE) {
584 DEBUG(0, ("Received async msg of type %u, discarding\n",
585 hdr->operation));
586 return EINVAL;
587 }
588
589 msg = (struct ctdb_req_message_old *)hdr;
590
591 ctdbd_msg_call_back(ev, conn, msg);
592
593 return 0;
594 }
595
596 void ctdbd_socket_readable(struct tevent_context *ev,
597 struct ctdbd_connection *conn)
598 {
599 struct ctdb_req_header *hdr = NULL;
600 int ret;
601
602 ret = ctdb_read_packet(conn->fd, conn->timeout, talloc_tos(), &hdr);
603 if (ret != 0) {
604 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
605 cluster_fatal("failed to read data from ctdbd\n");
606 }
607
608 ret = ctdb_handle_message(ev, conn, hdr);
609
610 TALLOC_FREE(hdr);
611
612 if (ret != 0) {
613 DEBUG(10, ("could not handle incoming message: %s\n",
614 strerror(ret)));
615 }
616 }
617
618 static int ctdb_pkt_send_handler(struct ctdbd_connection *conn);
619 static int ctdb_pkt_recv_handler(struct ctdbd_connection *conn);
620
621 /* Used for async connection and async ctcb requests */
622 static void ctdbd_async_socket_handler(struct tevent_context *ev,
623 struct tevent_fd *fde,
624 uint16_t flags,
625 void *private_data)
626 {
627 struct ctdbd_connection *conn = talloc_get_type_abort(
628 private_data, struct ctdbd_connection);
629 int ret;
630
631 if ((flags & TEVENT_FD_READ) != 0) {
632 ret = ctdb_pkt_recv_handler(conn);
633 if (ret != 0) {
634 DBG_DEBUG("ctdb_read_iov_handler returned %s\n",
635 strerror(ret));
636 }
637 return;
638 }
639
640 if ((flags & TEVENT_FD_WRITE) != 0) {
641 ret = ctdb_pkt_send_handler(conn);
642 if (ret != 0) {
643 DBG_DEBUG("ctdb_write_iov_handler returned %s\n",
644 strerror(ret));
645 return;
646 }
647 return;
648 }
649
650 return;
651 }
652
653 int ctdbd_messaging_send_iov(struct ctdbd_connection *conn,
654 uint32_t dst_vnn, uint64_t dst_srvid,
655 const struct iovec *iov, int iovlen)
656 {
657 struct ctdb_req_message_old r;
658 struct iovec iov2[iovlen+1];
659 size_t buflen = iov_buflen(iov, iovlen);
660 ssize_t nwritten;
661
662 r.hdr.length = offsetof(struct ctdb_req_message_old, data) + buflen;
663 r.hdr.ctdb_magic = CTDB_MAGIC;
664 r.hdr.ctdb_version = CTDB_PROTOCOL;
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 = buflen;
672
673 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
674 ctdb_packet_dump(&r.hdr);
675
676 iov2[0].iov_base = &r;
677 iov2[0].iov_len = offsetof(struct ctdb_req_message_old, data);
678 memcpy(&iov2[1], iov, iovlen * sizeof(struct iovec));
679
680 nwritten = write_data_iov(conn->fd, iov2, iovlen+1);
681 if (nwritten == -1) {
682 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
683 cluster_fatal("cluster dispatch daemon msg write error\n");
684 }
685
686 return 0;
687 }
688
689 /*
690 * send/recv a generic ctdb control message
691 */
692 static int ctdbd_control(struct ctdbd_connection *conn,
693 uint32_t vnn, uint32_t opcode,
694 uint64_t srvid, uint32_t flags,
695 TDB_DATA data,
696 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
697 int32_t *cstatus)
698 {
699 struct ctdb_req_control_old req;
700 struct ctdb_req_header *hdr;
701 struct ctdb_reply_control_old *reply = NULL;
702 struct iovec iov[2];
703 ssize_t nwritten;
704 int ret;
705
706 if (ctdbd_conn_has_async_reqs(conn)) {
707 /*
708 * Can't use sync call while an async call is in flight. Adding
709 * this check as a safety net. We'll be using different
710 * connections for sync and async requests, so this shouldn't
711 * happen, but who knows...
712 */
713 DBG_ERR("Async ctdb req on sync connection\n");
714 return EINVAL;
715 }
716
717 ZERO_STRUCT(req);
718 req.hdr.length = offsetof(struct ctdb_req_control_old, data) + data.dsize;
719 req.hdr.ctdb_magic = CTDB_MAGIC;
720 req.hdr.ctdb_version = CTDB_PROTOCOL;
721 req.hdr.operation = CTDB_REQ_CONTROL;
722 req.hdr.reqid = ctdbd_next_reqid(conn);
723 req.hdr.destnode = vnn;
724 req.opcode = opcode;
725 req.srvid = srvid;
726 req.datalen = data.dsize;
727 req.flags = flags;
728
729 DBG_DEBUG("Sending ctdb packet reqid=%"PRIu32", vnn=%"PRIu32", "
730 "opcode=%"PRIu32", srvid=%"PRIu64"\n", req.hdr.reqid,
731 req.hdr.destnode, req.opcode, req.srvid);
732 ctdb_packet_dump(&req.hdr);
733
734 iov[0].iov_base = &req;
735 iov[0].iov_len = offsetof(struct ctdb_req_control_old, data);
736 iov[1].iov_base = data.dptr;
737 iov[1].iov_len = data.dsize;
738
739 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
740 if (nwritten == -1) {
741 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
742 cluster_fatal("cluster dispatch daemon msg write error\n");
743 }
744
745 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
746 if (cstatus) {
747 *cstatus = 0;
748 }
749 return 0;
750 }
751
752 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
753 if (ret != 0) {
754 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
755 return ret;
756 }
757
758 if (hdr->operation != CTDB_REPLY_CONTROL) {
759 DEBUG(0, ("received invalid reply\n"));
760 TALLOC_FREE(hdr);
761 return EIO;
762 }
763 reply = (struct ctdb_reply_control_old *)hdr;
764
765 if (outdata) {
766 if (!(outdata->dptr = (uint8_t *)talloc_memdup(
767 mem_ctx, reply->data, reply->datalen))) {
768 TALLOC_FREE(reply);
769 return ENOMEM;
770 }
771 outdata->dsize = reply->datalen;
772 }
773 if (cstatus) {
774 (*cstatus) = reply->status;
775 }
776
777 TALLOC_FREE(reply);
778 return ret;
779 }
780
781 /*
782 * see if a remote process exists
783 */
784 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32_t vnn,
785 pid_t pid, uint64_t unique_id)
786 {
787 uint8_t buf[sizeof(pid)+sizeof(unique_id)];
788 int32_t cstatus = 0;
789 int ret;
790
791 if (unique_id == SERVERID_UNIQUE_ID_NOT_TO_VERIFY) {
792 ret = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS,
793 0, 0,
794 (TDB_DATA) { .dptr = (uint8_t *)&pid,
795 .dsize = sizeof(pid) },
796 NULL, NULL, &cstatus);
797 if (ret != 0) {
798 return false;
799 }
800 return (cstatus == 0);
801 }
802
803 memcpy(buf, &pid, sizeof(pid));
804 memcpy(buf+sizeof(pid), &unique_id, sizeof(unique_id));
805
806 ret = ctdbd_control(conn, vnn, CTDB_CONTROL_CHECK_PID_SRVID, 0, 0,
807 (TDB_DATA) { .dptr = buf, .dsize = sizeof(buf) },
808 NULL, NULL, &cstatus);
809 if (ret != 0) {
810 return false;
811 }
812 return (cstatus == 0);
813 }
814
815 /*
816 * Get a db path
817 */
818 char *ctdbd_dbpath(struct ctdbd_connection *conn,
819 TALLOC_CTX *mem_ctx, uint32_t db_id)
820 {
821 int ret;
822 TDB_DATA data;
823 TDB_DATA rdata = {0};
824 int32_t cstatus = 0;
825
826 data.dptr = (uint8_t*)&db_id;
827 data.dsize = sizeof(db_id);
828
829 ret = ctdbd_control_local(conn, CTDB_CONTROL_GETDBPATH, 0, 0, data,
830 mem_ctx, &rdata, &cstatus);
831 if ((ret != 0) || cstatus != 0) {
832 DEBUG(0, (__location__ " ctdb_control for getdbpath failed: %s\n",
833 strerror(ret)));
834 return NULL;
835 }
836
837 return (char *)rdata.dptr;
838 }
839
840 /*
841 * attach to a ctdb database
842 */
843 int ctdbd_db_attach(struct ctdbd_connection *conn,
844 const char *name, uint32_t *db_id, bool persistent)
845 {
846 int ret;
847 TDB_DATA data;
848 int32_t cstatus;
849
850 data = string_term_tdb_data(name);
851
852 ret = ctdbd_control_local(conn,
853 persistent
854 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
855 : CTDB_CONTROL_DB_ATTACH,
856 0, 0, data, NULL, &data, &cstatus);
857 if (ret != 0) {
858 DEBUG(0, (__location__ " ctdb_control for db_attach "
859 "failed: %s\n", strerror(ret)));
860 return ret;
861 }
862
863 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
864 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
865 return EIO;
866 }
867
868 *db_id = *(uint32_t *)data.dptr;
869 talloc_free(data.dptr);
870
871 return 0;
872 }
873
874 /*
875 * force the migration of a record to this node
876 */
877 int ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id, TDB_DATA key)
878 {
879 struct ctdb_req_call_old req;
880 struct ctdb_req_header *hdr = NULL;
881 struct iovec iov[2];
882 ssize_t nwritten;
883 int ret;
884
885 if (ctdbd_conn_has_async_reqs(conn)) {
886 /*
887 * Can't use sync call while an async call is in flight. Adding
888 * this check as a safety net. We'll be using different
889 * connections for sync and async requests, so this shouldn't
890 * happen, but who knows...
891 */
892 DBG_ERR("Async ctdb req on sync connection\n");
893 return EINVAL;
894 }
895
896 ZERO_STRUCT(req);
897
898 req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
899 req.hdr.ctdb_magic = CTDB_MAGIC;
900 req.hdr.ctdb_version = CTDB_PROTOCOL;
901 req.hdr.operation = CTDB_REQ_CALL;
902 req.hdr.reqid = ctdbd_next_reqid(conn);
903 req.flags = CTDB_IMMEDIATE_MIGRATION;
904 req.callid = CTDB_NULL_FUNC;
905 req.db_id = db_id;
906 req.keylen = key.dsize;
907
908 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
909 ctdb_packet_dump(&req.hdr);
910
911 iov[0].iov_base = &req;
912 iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
913 iov[1].iov_base = key.dptr;
914 iov[1].iov_len = key.dsize;
915
916 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
917 if (nwritten == -1) {
918 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
919 cluster_fatal("cluster dispatch daemon msg write error\n");
920 }
921
922 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
923 if (ret != 0) {
924 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
925 goto fail;
926 }
927
928 if (hdr->operation != CTDB_REPLY_CALL) {
929 if (hdr->operation == CTDB_REPLY_ERROR) {
930 DBG_ERR("received error from ctdb\n");
931 } else {
932 DBG_ERR("received invalid reply\n");
933 }
934 ret = EIO;
935 goto fail;
936 }
937
938 fail:
939
940 TALLOC_FREE(hdr);
941 return ret;
942 }
943
944 /*
945 * Fetch a record and parse it
946 */
947 int ctdbd_parse(struct ctdbd_connection *conn, uint32_t db_id,
948 TDB_DATA key, bool local_copy,
949 void (*parser)(TDB_DATA key, TDB_DATA data,
950 void *private_data),
951 void *private_data)
952 {
953 struct ctdb_req_call_old req;
954 struct ctdb_req_header *hdr = NULL;
955 struct ctdb_reply_call_old *reply;
956 struct iovec iov[2];
957 ssize_t nwritten;
958 uint32_t flags;
959 int ret;
960
961 if (ctdbd_conn_has_async_reqs(conn)) {
962 /*
963 * Can't use sync call while an async call is in flight. Adding
964 * this check as a safety net. We'll be using different
965 * connections for sync and async requests, so this shouldn't
966 * happen, but who knows...
967 */
968 DBG_ERR("Async ctdb req on sync connection\n");
969 return EINVAL;
970 }
971
972 flags = local_copy ? CTDB_WANT_READONLY : 0;
973
974 ZERO_STRUCT(req);
975
976 req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
977 req.hdr.ctdb_magic = CTDB_MAGIC;
978 req.hdr.ctdb_version = CTDB_PROTOCOL;
979 req.hdr.operation = CTDB_REQ_CALL;
980 req.hdr.reqid = ctdbd_next_reqid(conn);
981 req.flags = flags;
982 req.callid = CTDB_FETCH_FUNC;
983 req.db_id = db_id;
984 req.keylen = key.dsize;
985
986 iov[0].iov_base = &req;
987 iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
988 iov[1].iov_base = key.dptr;
989 iov[1].iov_len = key.dsize;
990
991 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
992 if (nwritten == -1) {
993 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
994 cluster_fatal("cluster dispatch daemon msg write error\n");
995 }
996
997 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
998 if (ret != 0) {
999 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
1000 goto fail;
1001 }
1002
1003 if ((hdr == NULL) || (hdr->operation != CTDB_REPLY_CALL)) {
1004 DEBUG(0, ("received invalid reply\n"));
1005 ret = EIO;
1006 goto fail;
1007 }
1008 reply = (struct ctdb_reply_call_old *)hdr;
1009
1010 if (reply->datalen == 0) {
1011 /*
1012 * Treat an empty record as non-existing
1013 */
1014 ret = ENOENT;
1015 goto fail;
1016 }
1017
1018 parser(key, make_tdb_data(&reply->data[0], reply->datalen),
1019 private_data);
1020
1021 ret = 0;
1022 fail:
1023 TALLOC_FREE(hdr);
1024 return ret;
1025 }
1026
1027 /*
1028 Traverse a ctdb database. "conn" must be an otherwise unused
1029 ctdb_connection where no other messages but the traverse ones are
1030 expected.
1031 */
1032
1033 int ctdbd_traverse(struct ctdbd_connection *conn, uint32_t db_id,
1034 void (*fn)(TDB_DATA key, TDB_DATA data,
1035 void *private_data),
1036 void *private_data)
1037 {
1038 int ret;
1039 TDB_DATA key, data;
1040 struct ctdb_traverse_start t;
1041 int32_t cstatus;
1042
1043 if (ctdbd_conn_has_async_reqs(conn)) {
1044 /*
1045 * Can't use sync call while an async call is in flight. Adding
1046 * this check as a safety net. We'll be using different
1047 * connections for sync and async requests, so this shouldn't
1048 * happen, but who knows...
1049 */
1050 DBG_ERR("Async ctdb req on sync connection\n");
1051 return EINVAL;
1052 }
1053
1054 t.db_id = db_id;
1055 t.srvid = conn->rand_srvid;
1056 t.reqid = ctdbd_next_reqid(conn);
1057
1058 data.dptr = (uint8_t *)&t;
1059 data.dsize = sizeof(t);
1060
1061 ret = ctdbd_control_local(conn, CTDB_CONTROL_TRAVERSE_START,
1062 conn->rand_srvid,
1063 0, data, NULL, NULL, &cstatus);
1064
1065 if ((ret != 0) || (cstatus != 0)) {
1066 DEBUG(0,("ctdbd_control failed: %s, %d\n", strerror(ret),
1067 cstatus));
1068
1069 if (ret == 0) {
1070 /*
1071 * We need a mapping here
1072 */
1073 ret = EIO;
1074 }
1075 return ret;
1076 }
1077
1078 while (true) {
1079 struct ctdb_req_header *hdr = NULL;
1080 struct ctdb_req_message_old *m;
1081 struct ctdb_rec_data_old *d;
1082
1083 ret = ctdb_read_packet(conn->fd, conn->timeout, conn, &hdr);
1084 if (ret != 0) {
1085 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
1086 cluster_fatal("failed to read data from ctdbd\n");
1087 }
1088
1089 if (hdr->operation != CTDB_REQ_MESSAGE) {
1090 DEBUG(0, ("Got operation %u, expected a message\n",
1091 (unsigned)hdr->operation));
1092 return EIO;
1093 }
1094
1095 m = (struct ctdb_req_message_old *)hdr;
1096 d = (struct ctdb_rec_data_old *)&m->data[0];
1097 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1098 DEBUG(0, ("Got invalid traverse data of length %d\n",
1099 (int)m->datalen));
1100 return EIO;
1101 }
1102
1103 key.dsize = d->keylen;
1104 key.dptr = &d->data[0];
1105 data.dsize = d->datalen;
1106 data.dptr = &d->data[d->keylen];
1107
1108 if (key.dsize == 0 && data.dsize == 0) {
1109 /* end of traverse */
1110 return 0;
1111 }
1112
1113 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1114 DEBUG(0, ("Got invalid ltdb header length %d\n",
1115 (int)data.dsize));
1116 return EIO;
1117 }
1118 data.dsize -= sizeof(struct ctdb_ltdb_header);
1119 data.dptr += sizeof(struct ctdb_ltdb_header);
1120
1121 if (fn != NULL) {
1122 fn(key, data, private_data);
1123 }
1124 }
1125 return 0;
1126 }
1127
1128 /*
1129 This is used to canonicalize a ctdb_sock_addr structure.
1130 */
1131 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1132 struct sockaddr_storage *out)
1133 {
1134 memcpy(out, in, sizeof (*out));
1135
1136 #ifdef HAVE_IPV6
1137 if (in->ss_family == AF_INET6) {
1138 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1139 const struct sockaddr_in6 *in6 =
1140 (const struct sockaddr_in6 *)in;
1141 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1142 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1143 memset(out, 0, sizeof(*out));
1144 #ifdef HAVE_SOCK_SIN_LEN
1145 out4->sin_len = sizeof(*out);
1146 #endif
1147 out4->sin_family = AF_INET;
1148 out4->sin_port = in6->sin6_port;
1149 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
1150 }
1151 }
1152 #endif
1153 }
1154
1155 /*
1156 * Register us as a server for a particular tcp connection
1157 */
1158
1159 int ctdbd_register_ips(struct ctdbd_connection *conn,
1160 const struct sockaddr_storage *_server,
1161 const struct sockaddr_storage *_client,
1162 int (*cb)(struct tevent_context *ev,
1163 uint32_t src_vnn, uint32_t dst_vnn,
1164 uint64_t dst_srvid,
1165 const uint8_t *msg, size_t msglen,
1166 void *private_data),
1167 void *private_data)
1168 {
1169 struct ctdb_connection p;
1170 TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
1171 int ret;
1172 struct sockaddr_storage client;
1173 struct sockaddr_storage server;
1174
1175 /*
1176 * Only one connection so far
1177 */
1178
1179 smbd_ctdb_canonicalize_ip(_client, &client);
1180 smbd_ctdb_canonicalize_ip(_server, &server);
1181
1182 switch (client.ss_family) {
1183 case AF_INET:
1184 memcpy(&p.dst.ip, &server, sizeof(p.dst.ip));
1185 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1186 break;
1187 case AF_INET6:
1188 memcpy(&p.dst.ip6, &server, sizeof(p.dst.ip6));
1189 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1190 break;
1191 default:
1192 return EIO;
1193 }
1194
1195 /*
1196 * We want to be told about IP releases
1197 */
1198
1199 ret = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1200 cb, private_data);
1201 if (ret != 0) {
1202 return ret;
1203 }
1204
1205 /*
1206 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1207 * can send an extra ack to trigger a reset for our client, so it
1208 * immediately reconnects
1209 */
1210 ret = ctdbd_control(conn, CTDB_CURRENT_NODE,
1211 CTDB_CONTROL_TCP_CLIENT, 0,
1212 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1213 NULL);
1214 if (ret != 0) {
1215 return ret;
1216 }
1217 return 0;
1218 }
1219
1220 /*
1221 call a control on the local node
1222 */
1223 int ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1224 uint64_t srvid, uint32_t flags, TDB_DATA data,
1225 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1226 int32_t *cstatus)
1227 {
1228 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data,
1229 mem_ctx, outdata, cstatus);
1230 }
1231
1232 int ctdb_watch_us(struct ctdbd_connection *conn)
1233 {
1234 struct ctdb_notify_data_old reg_data;
1235 size_t struct_len;
1236 int ret;
1237 int32_t cstatus;
1238
1239 reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1240 reg_data.len = 1;
1241 reg_data.notify_data[0] = 0;
1242
1243 struct_len = offsetof(struct ctdb_notify_data_old,
1244 notify_data) + reg_data.len;
1245
1246 ret = ctdbd_control_local(
1247 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1248 make_tdb_data((uint8_t *)&reg_data, struct_len),
1249 NULL, NULL, &cstatus);
1250 if (ret != 0) {
1251 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1252 strerror(ret)));
1253 }
1254 return ret;
1255 }
1256
1257 int ctdb_unwatch(struct ctdbd_connection *conn)
1258 {
1259 uint64_t srvid = CTDB_SRVID_SAMBA_NOTIFY;
1260 int ret;
1261 int32_t cstatus;
1262
1263 ret = ctdbd_control_local(
1264 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1265 make_tdb_data((uint8_t *)&srvid, sizeof(srvid)),
1266 NULL, NULL, &cstatus);
1267 if (ret != 0) {
1268 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1269 strerror(ret)));
1270 }
1271 return ret;
1272 }
1273
1274 int ctdbd_probe(const char *sockname, int timeout)
1275 {
1276 /*
1277 * Do a very early check if ctdbd is around to avoid an abort and core
1278 * later
1279 */
1280 struct ctdbd_connection *conn = NULL;
1281 int ret;
1282
1283 ret = ctdbd_init_connection(talloc_tos(), sockname, timeout,
1284 &conn);
1285
1286 /*
1287 * We only care if we can connect.
1288 */
1289 TALLOC_FREE(conn);
1290
1291 return ret;
1292 }
1293
1294 struct ctdb_pkt_send_state {
1295 struct ctdb_pkt_send_state *prev, *next;
1296 struct tevent_context *ev;
1297 struct ctdbd_connection *conn;
1298
1299 /* ctdb request id */
1300 uint32_t reqid;
1301
1302 /* the associated tevent request */
1303 struct tevent_req *req;
1304
1305 /* iovec array with data to send */
1306 struct iovec _iov;
1307 struct iovec *iov;
1308 int iovcnt;
1309
1310 /* Initial packet length */
1311 size_t packet_len;
1312 };
1313
1314 static void ctdb_pkt_send_cleanup(struct tevent_req *req,
1315 enum tevent_req_state req_state);
1316
1317 /**
1318 * Asynchronously send a ctdb packet given as iovec array
1319 *
1320 * Note: the passed iov array is not const here. Similar
1321 * functions in samba take a const array and create a copy
1322 * before calling iov_advance() on the array.
1323 *
1324 * This function will modify the iov array! But
1325 * this is a static function and our only caller
1326 * ctdb_parse_send/recv is preparared for this to
1327 * happen!
1328 **/
1329 static struct tevent_req *ctdb_pkt_send_send(TALLOC_CTX *mem_ctx,
1330 struct tevent_context *ev,
1331 struct ctdbd_connection *conn,
1332 uint32_t reqid,
1333 struct iovec *iov,
1334 int iovcnt,
1335 enum dbwrap_req_state *req_state)
1336 {
1337 struct tevent_req *req = NULL;
1338 struct ctdb_pkt_send_state *state = NULL;
1339 ssize_t nwritten;
1340 bool ok;
1341
1342 DBG_DEBUG("sending async ctdb reqid [%" PRIu32 "]\n", reqid);
1343
1344 req = tevent_req_create(mem_ctx, &state, struct ctdb_pkt_send_state);
1345 if (req == NULL) {
1346 return NULL;
1347 }
1348
1349 *state = (struct ctdb_pkt_send_state) {
1350 .ev = ev,
1351 .conn = conn,
1352 .req = req,
1353 .reqid = reqid,
1354 .iov = iov,
1355 .iovcnt = iovcnt,
1356 .packet_len = iov_buflen(iov, iovcnt),
1357 };
1358
1359 tevent_req_set_cleanup_fn(req, ctdb_pkt_send_cleanup);
1360
1361 *req_state = DBWRAP_REQ_QUEUED;
1362
1363 if (ctdbd_conn_has_async_sends(conn)) {
1364 /*
1365 * Can't attempt direct write with messages already queued and
1366 * possibly in progress
1367 */
1368 DLIST_ADD_END(conn->send_list, state);
1369 return req;
1370 }
1371
1372 /*
1373 * Attempt a direct write. If this returns short, schedule the
1374 * remaining data as an async write, otherwise we're already done.
1375 */
1376
1377 nwritten = writev(conn->fd, state->iov, state->iovcnt);
1378 if (nwritten == state->packet_len) {
1379 DBG_DEBUG("Finished sending reqid [%" PRIu32 "]\n", reqid);
1380
1381 *req_state = DBWRAP_REQ_DISPATCHED;
1382 tevent_req_done(req);
1383 return tevent_req_post(req, ev);
1384 }
1385
1386 if (nwritten == -1) {
1387 if (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK) {
1388 cluster_fatal("cluster write error\n");
1389 }
1390 nwritten = 0;
1391 }
1392
1393 DBG_DEBUG("Posting async write of reqid [%" PRIu32"]"
1394 "after short write [%zd]\n", reqid, nwritten);
1395
1396 ok = iov_advance(&state->iov, &state->iovcnt, nwritten);
1397 if (!ok) {
1398 *req_state = DBWRAP_REQ_ERROR;
1399 tevent_req_error(req, EIO);
1400 return tevent_req_post(req, ev);
1401 }
1402
1403 /*
1404 * As this is the first async write req we post, we must enable
1405 * fd-writable events.
1406 */
1407 TEVENT_FD_WRITEABLE(conn->fde);
1408 DLIST_ADD_END(conn->send_list, state);
1409 return req;
1410 }
1411
1412 static int ctdb_pkt_send_state_destructor(struct ctdb_pkt_send_state *state)
1413 {
1414 struct ctdbd_connection *conn = state->conn;
1415
1416 if (conn == NULL) {
1417 return 0;
1418 }
1419
1420 if (state->req == NULL) {
1421 DBG_DEBUG("Removing cancelled reqid [%" PRIu32"]\n",
1422 state->reqid);
1423 state->conn = NULL;
1424 DLIST_REMOVE(conn->send_list, state);
1425 return 0;
1426 }
1427
1428 DBG_DEBUG("Reparenting cancelled reqid [%" PRIu32"]\n",
1429 state->reqid);
1430
1431 talloc_reparent(state->req, conn, state);
1432 state->req = NULL;
1433 return -1;
1434 }
1435
1436 static void ctdb_pkt_send_cleanup(struct tevent_req *req,
1437 enum tevent_req_state req_state)
1438 {
1439 struct ctdb_pkt_send_state *state = tevent_req_data(
1440 req, struct ctdb_pkt_send_state);
1441 struct ctdbd_connection *conn = state->conn;
1442 size_t missing_len = 0;
1443
1444 if (conn == NULL) {
1445 return;
1446 }
1447
1448 missing_len = iov_buflen(state->iov, state->iovcnt);
1449 if (state->packet_len == missing_len) {
1450 /*
1451 * We haven't yet started sending this one, so we can just
1452 * remove it from the pending list
1453 */
1454 missing_len = 0;
1455 }
1456 if (missing_len != 0) {
1457 uint8_t *buf = NULL;
1458
1459 if (req_state != TEVENT_REQ_RECEIVED) {
1460 /*
1461 * Wait til the req_state is TEVENT_REQ_RECEIVED, as
1462 * that will be the final state when the request state
1463 * is talloc_free'd from tallloc_req_received(). Which
1464 * ensures we only run the following code *ONCE*!
1465 */
1466 return;
1467 }
1468
1469 DBG_DEBUG("Cancelling in-flight reqid [%" PRIu32"]\n",
1470 state->reqid);
1471 /*
1472 * A request in progress of being sent. Reparent the iov buffer
1473 * so we can continue sending the request. See also the comment
1474 * in ctdbd_parse_send() when copying the key buffer.
1475 */
1476
1477 buf = iov_concat(state, state->iov, state->iovcnt);
1478 if (buf == NULL) {
1479 cluster_fatal("iov_concat error\n");
1480 return;
1481 }
1482
1483 state->iovcnt = 1;
1484 state->_iov.iov_base = buf;
1485 state->_iov.iov_len = missing_len;
1486 state->iov = &state->_iov;
1487
1488 talloc_set_destructor(state, ctdb_pkt_send_state_destructor);
1489 return;
1490 }
1491
1492 DBG_DEBUG("Removing pending reqid [%" PRIu32"]\n", state->reqid);
1493
1494 state->conn = NULL;
1495 DLIST_REMOVE(conn->send_list, state);
1496
1497 if (!ctdbd_conn_has_async_sends(conn)) {
1498 DBG_DEBUG("No more sends, disabling fd-writable events\n");
1499 TEVENT_FD_NOT_WRITEABLE(conn->fde);
1500 }
1501 }
1502
1503 static int ctdb_pkt_send_handler(struct ctdbd_connection *conn)
1504 {
1505 struct ctdb_pkt_send_state *state = NULL;
1506 ssize_t nwritten;
1507 ssize_t iovlen;
1508 bool ok;
1509
1510 DBG_DEBUG("send handler\n");
1511
1512 if (!ctdbd_conn_has_async_sends(conn)) {
1513 DBG_WARNING("Writable fd-event without pending send\n");
1514 TEVENT_FD_NOT_WRITEABLE(conn->fde);
1515 return 0;
1516 }
1517
1518 state = conn->send_list;
1519 iovlen = iov_buflen(state->iov, state->iovcnt);
1520
1521 nwritten = writev(conn->fd, state->iov, state->iovcnt);
1522 if (nwritten == -1) {
1523 if (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK) {
1524 DBG_ERR("writev failed: %s\n", strerror(errno));
1525 cluster_fatal("cluster write error\n");
1526 }
1527 DBG_DEBUG("recoverable writev error, retry\n");
1528 return 0;
1529 }
1530
1531 if (nwritten < iovlen) {
1532 DBG_DEBUG("short write\n");
1533
1534 ok = iov_advance(&state->iov, &state->iovcnt, nwritten);
1535 if (!ok) {
1536 DBG_ERR("iov_advance failed\n");
1537 if (state->req == NULL) {
1538 TALLOC_FREE(state);
1539 return 0;
1540 }
1541 tevent_req_error(state->req, EIO);
1542 return 0;
1543 }
1544 return 0;
1545 }
1546
1547 if (state->req == NULL) {
1548 DBG_DEBUG("Finished sending cancelled reqid [%" PRIu32 "]\n",
1549 state->reqid);
1550 TALLOC_FREE(state);
1551 return 0;
1552 }
1553
1554 DBG_DEBUG("Finished send request id [%" PRIu32 "]\n", state->reqid);
1555
1556 tevent_req_done(state->req);
1557 return 0;
1558 }
1559
1560 static int ctdb_pkt_send_recv(struct tevent_req *req)
1561 {
1562 int ret;
1563
1564 if (tevent_req_is_unix_error(req, &ret)) {
1565 tevent_req_received(req);
1566 return ret;
1567 }
1568
1569 tevent_req_received(req);
1570 return 0;
1571 }
1572
1573 struct ctdb_pkt_recv_state {
1574 struct ctdb_pkt_recv_state *prev, *next;
1575 struct tevent_context *ev;
1576 struct ctdbd_connection *conn;
1577
1578 /* ctdb request id */
1579 uint32_t reqid;
1580
1581 /* the associated tevent_req */
1582 struct tevent_req *req;
1583
1584 /* pointer to allocated ctdb packet buffer */
1585 struct ctdb_req_header *hdr;
1586 };
1587
1588 static void ctdb_pkt_recv_cleanup(struct tevent_req *req,
1589 enum tevent_req_state req_state);
1590
1591 static struct tevent_req *ctdb_pkt_recv_send(TALLOC_CTX *mem_ctx,
1592 struct tevent_context *ev,
1593 struct ctdbd_connection *conn,
1594 uint32_t reqid)
1595 {
1596 struct tevent_req *req = NULL;
1597 struct ctdb_pkt_recv_state *state = NULL;
1598
1599 req = tevent_req_create(mem_ctx, &state, struct ctdb_pkt_recv_state);
1600 if (req == NULL) {
1601 return NULL;
1602 }
1603
1604 *state = (struct ctdb_pkt_recv_state) {
1605 .ev = ev,
1606 .conn = conn,
1607 .reqid = reqid,
1608 .req = req,
1609 };
1610
1611 tevent_req_set_cleanup_fn(req, ctdb_pkt_recv_cleanup);
1612
1613 /*
1614 * fd-readable event is always set for the fde, no need to deal with
1615 * that here.
1616 */
1617
1618 DLIST_ADD_END(conn->recv_list, state);
1619 DBG_DEBUG("Posted receive reqid [%" PRIu32 "]\n", state->reqid);
1620
1621 return req;
1622 }
1623
1624 static void ctdb_pkt_recv_cleanup(struct tevent_req *req,
1625 enum tevent_req_state req_state)
1626 {
1627 struct ctdb_pkt_recv_state *state = tevent_req_data(
1628 req, struct ctdb_pkt_recv_state);
1629 struct ctdbd_connection *conn = state->conn;
1630
1631 if (conn == NULL) {
1632 return;
1633 }
1634 state->conn = NULL;
1635 DLIST_REMOVE(conn->recv_list, state);
1636 }
1637
1638 static int ctdb_pkt_recv_handler(struct ctdbd_connection *conn)
1639 {
1640 struct ctdb_pkt_recv_state *state = NULL;
1641 ssize_t nread;
1642 ssize_t iovlen;
1643 bool ok;
1644
1645 DBG_DEBUG("receive handler\n");
1646
1647 if (conn->read_state.iovs == NULL) {
1648 conn->read_state.iov.iov_base = &conn->read_state.msglen;
1649 conn->read_state.iov.iov_len = sizeof(conn->read_state.msglen);
1650 conn->read_state.iovs = &conn->read_state.iov;
1651 conn->read_state.iovcnt = 1;
1652 }
1653
1654 iovlen = iov_buflen(conn->read_state.iovs, conn->read_state.iovcnt);
1655
1656 DBG_DEBUG("iovlen [%zd]\n", iovlen);
1657
1658 nread = readv(conn->fd, conn->read_state.iovs, conn->read_state.iovcnt);
1659 if (nread == 0) {
1660 cluster_fatal("cluster read error, peer closed connection\n");
1661 }
1662 if (nread == -1) {
1663 if (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK) {
1664 cluster_fatal("cluster read error\n");
1665 }
1666 DBG_DEBUG("recoverable error from readv, retry\n");
1667 return 0;
1668 }
1669
1670 if (nread < iovlen) {
1671 DBG_DEBUG("iovlen [%zd] nread [%zd]\n", iovlen, nread);
1672 ok = iov_advance(&conn->read_state.iovs,
1673 &conn->read_state.iovcnt,
1674 nread);
1675 if (!ok) {
1676 return EIO;
1677 }
1678 return 0;
1679 }
1680
1681 conn->read_state.iovs = NULL;
1682 conn->read_state.iovcnt = 0;
1683
1684 if (conn->read_state.hdr == NULL) {
1685 /*
1686 * Going this way after reading the 4 initial byte message
1687 * length
1688 */
1689 uint32_t msglen = conn->read_state.msglen;
1690 uint8_t *readbuf = NULL;
1691 size_t readlen;
1692
1693 DBG_DEBUG("msglen: %" PRIu32 "\n", msglen);
1694
1695 if (msglen < sizeof(struct ctdb_req_header)) {
1696 DBG_ERR("short message %" PRIu32 "\n", msglen);
1697 return EIO;
1698 }
1699
1700 conn->read_state.hdr = talloc_size(conn, msglen);
1701 if (conn->read_state.hdr == NULL) {
1702 return ENOMEM;
1703 }
1704 conn->read_state.hdr->length = msglen;
1705 talloc_set_name_const(conn->read_state.hdr,
1706 "struct ctdb_req_header");
1707
1708 readbuf = (uint8_t *)conn->read_state.hdr + sizeof(msglen);
1709 readlen = msglen - sizeof(msglen);
1710
1711 conn->read_state.iov.iov_base = readbuf;
1712 conn->read_state.iov.iov_len = readlen;
1713 conn->read_state.iovs = &conn->read_state.iov;
1714 conn->read_state.iovcnt = 1;
1715
1716 DBG_DEBUG("Scheduled packet read size %zd\n", readlen);
1717 return 0;
1718 }
1719
1720 /*
1721 * Searching a list here is expected to be cheap, as messages are
1722 * exepcted to be coming in more or less ordered and we should find the
1723 * waiting request near the beginning of the list.
1724 */
1725 for (state = conn->recv_list; state != NULL; state = state->next) {
1726 if (state->reqid == conn->read_state.hdr->reqid) {
1727 break;
1728 }
1729 }
1730
1731 if (state == NULL) {
1732 DBG_ERR("Discarding async ctdb reqid %u\n",
1733 conn->read_state.hdr->reqid);
1734 TALLOC_FREE(conn->read_state.hdr);
1735 ZERO_STRUCT(conn->read_state);
1736 return EINVAL;
1737 }
1738
1739 DBG_DEBUG("Got reply for reqid [%" PRIu32 "]\n", state->reqid);
1740
1741 state->hdr = talloc_move(state, &conn->read_state.hdr);
1742 ZERO_STRUCT(conn->read_state);
1743 tevent_req_done(state->req);
1744 return 0;
1745 }
1746
1747 static int ctdb_pkt_recv_recv(struct tevent_req *req,
1748 TALLOC_CTX *mem_ctx,
1749 struct ctdb_req_header **_hdr)
1750 {
1751 struct ctdb_pkt_recv_state *state = tevent_req_data(
1752 req, struct ctdb_pkt_recv_state);
1753 int error;
1754
1755 if (tevent_req_is_unix_error(req, &error)) {
1756 DBG_ERR("ctdb_read_req failed %s\n", strerror(error));
1757 tevent_req_received(req);
1758 return error;
1759 }
1760
1761 *_hdr = talloc_move(mem_ctx, &state->hdr);
1762
1763 tevent_req_received(req);
1764 return 0;
1765 }
1766
1767 static int ctdbd_connection_destructor(struct ctdbd_connection *c)
1768 {
1769 TALLOC_FREE(c->fde);
1770 if (c->fd != -1) {
1771 close(c->fd);
1772 c->fd = -1;
1773 }
1774
1775 TALLOC_FREE(c->read_state.hdr);
1776 ZERO_STRUCT(c->read_state);
1777
1778 while (c->send_list != NULL) {
1779 struct ctdb_pkt_send_state *send_state = c->send_list;
1780 DLIST_REMOVE(c->send_list, send_state);
1781 send_state->conn = NULL;
1782 tevent_req_defer_callback(send_state->req, send_state->ev);
1783 tevent_req_error(send_state->req, EIO);
1784 }
1785
1786 while (c->recv_list != NULL) {
1787 struct ctdb_pkt_recv_state *recv_state = c->recv_list;
1788 DLIST_REMOVE(c->recv_list, recv_state);
1789 recv_state->conn = NULL;
1790 tevent_req_defer_callback(recv_state->req, recv_state->ev);
1791 tevent_req_error(recv_state->req, EIO);
1792 }
1793
1794 return 0;
1795 }
1796
1797 struct ctdbd_parse_state {
1798 struct tevent_context *ev;
1799 struct ctdbd_connection *conn;
1800 uint32_t reqid;
1801 TDB_DATA key;
1802 uint8_t _keybuf[64];
1803 struct ctdb_req_call_old ctdb_req;
1804 struct iovec iov[2];
1805 void (*parser)(TDB_DATA key,
1806 TDB_DATA data,
1807 void *private_data);
1808 void *private_data;
1809 enum dbwrap_req_state *req_state;
1810 };
1811
1812 static void ctdbd_parse_pkt_send_done(struct tevent_req *subreq);
1813 static void ctdbd_parse_done(struct tevent_req *subreq);
1814
1815 struct tevent_req *ctdbd_parse_send(TALLOC_CTX *mem_ctx,
1816 struct tevent_context *ev,
1817 struct ctdbd_connection *conn,
1818 uint32_t db_id,
1819 TDB_DATA key,
1820 bool local_copy,
1821 void (*parser)(TDB_DATA key,
1822 TDB_DATA data,
1823 void *private_data),
1824 void *private_data,
1825 enum dbwrap_req_state *req_state)
1826 {
1827 struct tevent_req *req = NULL;
1828 struct ctdbd_parse_state *state = NULL;
1829 uint32_t flags;
1830 uint32_t packet_length;
1831 struct tevent_req *subreq = NULL;
1832
1833 req = tevent_req_create(mem_ctx, &state, struct ctdbd_parse_state);
1834 if (req == NULL) {
1835 *req_state = DBWRAP_REQ_ERROR;
1836 return NULL;
1837 }
1838
1839 *state = (struct ctdbd_parse_state) {
1840 .ev = ev,
1841 .conn = conn,
1842 .reqid = ctdbd_next_reqid(conn),
1843 .parser = parser,
1844 .private_data = private_data,
1845 .req_state = req_state,
1846 };
1847
1848 flags = local_copy ? CTDB_WANT_READONLY : 0;
1849 packet_length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
1850
1851 /*
1852 * Copy the key into our state, as ctdb_pkt_send_cleanup() requires that
1853 * all passed iov elements have a lifetime longer that the tevent_req
1854 * returned by ctdb_pkt_send_send(). This is required continue sending a
1855 * the low level request into the ctdb socket, if a higher level
1856 * ('this') request is canceled (or talloc free'd) by the application
1857 * layer, without sending invalid packets to ctdb.
1858 */
1859 if (key.dsize > sizeof(state->_keybuf)) {
1860 state->key.dptr = talloc_memdup(state, key.dptr, key.dsize);
1861 if (tevent_req_nomem(state->key.dptr, req)) {
1862 return tevent_req_post(req, ev);
1863 }
1864 } else {
1865 memcpy(state->_keybuf, key.dptr, key.dsize);
1866 state->key.dptr = state->_keybuf;
1867 }
1868 state->key.dsize = key.dsize;
1869
1870 state->ctdb_req.hdr.length = packet_length;
1871 state->ctdb_req.hdr.ctdb_magic = CTDB_MAGIC;
1872 state->ctdb_req.hdr.ctdb_version = CTDB_PROTOCOL;
1873 state->ctdb_req.hdr.operation = CTDB_REQ_CALL;
1874 state->ctdb_req.hdr.reqid = state->reqid;
1875 state->ctdb_req.flags = flags;
1876 state->ctdb_req.callid = CTDB_FETCH_FUNC;
1877 state->ctdb_req.db_id = db_id;
1878 state->ctdb_req.keylen = state->key.dsize;
1879
1880 state->iov[0].iov_base = &state->ctdb_req;
1881 state->iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
1882 state->iov[1].iov_base = state->key.dptr;
1883 state->iov[1].iov_len = state->key.dsize;
1884
1885 /*
1886 * Note that ctdb_pkt_send_send()
1887 * will modify state->iov using
1888 * iov_advance() without making a copy.
1889 */
1890 subreq = ctdb_pkt_send_send(state,
1891 ev,
1892 conn,
1893 state->reqid,
1894 state->iov,
1895 ARRAY_SIZE(state->iov),
1896 req_state);
1897 if (tevent_req_nomem(subreq, req)) {
1898 *req_state = DBWRAP_REQ_ERROR;
1899 return tevent_req_post(req, ev);
1900 }
1901 tevent_req_set_callback(subreq, ctdbd_parse_pkt_send_done, req);
1902
1903 return req;
1904 }
1905
1906 static void ctdbd_parse_pkt_send_done(struct tevent_req *subreq)
1907 {
1908 struct tevent_req *req = tevent_req_callback_data(
1909 subreq, struct tevent_req);
1910 struct ctdbd_parse_state *state = tevent_req_data(
1911 req, struct ctdbd_parse_state);
1912 int ret;
1913
1914 ret = ctdb_pkt_send_recv(subreq);
1915 TALLOC_FREE(subreq);
1916 if (tevent_req_error(req, ret)) {
1917 DBG_DEBUG("ctdb_pkt_send_recv failed %s\n", strerror(ret));
1918 return;
1919 }
1920
1921 subreq = ctdb_pkt_recv_send(state,
1922 state->ev,
1923 state->conn,
1924 state->reqid);
1925 if (tevent_req_nomem(subreq, req)) {
1926 return;
1927 }
1928
1929 *state->req_state = DBWRAP_REQ_DISPATCHED;
1930 tevent_req_set_callback(subreq, ctdbd_parse_done, req);
1931 return;
1932 }
1933
1934 static void ctdbd_parse_done(struct tevent_req *subreq)
1935 {
1936 struct tevent_req *req = tevent_req_callback_data(
1937 subreq, struct tevent_req);
1938 struct ctdbd_parse_state *state = tevent_req_data(
1939 req, struct ctdbd_parse_state);
1940 struct ctdb_req_header *hdr = NULL;
1941 struct ctdb_reply_call_old *reply = NULL;
1942 int ret;
1943
1944 DBG_DEBUG("async parse request finished\n");
1945
1946 ret = ctdb_pkt_recv_recv(subreq, state, &hdr);
1947 TALLOC_FREE(subreq);
1948 if (tevent_req_error(req, ret)) {
1949 DBG_ERR("ctdb_pkt_recv_recv returned %s\n", strerror(ret));
1950 return;
1951 }
1952
1953 if (hdr->operation != CTDB_REPLY_CALL) {
1954 DBG_ERR("received invalid reply\n");
1955 ctdb_packet_dump(hdr);
1956 tevent_req_error(req, EIO);
1957 return;
1958 }
1959
1960 reply = (struct ctdb_reply_call_old *)hdr;
1961
1962 if (reply->datalen == 0) {
1963 /*
1964 * Treat an empty record as non-existing
1965 */
1966 tevent_req_error(req, ENOENT);
1967 return;
1968 }
1969
1970 state->parser(state->key,
1971 make_tdb_data(&reply->data[0], reply->datalen),
1972 state->private_data);
1973
1974 tevent_req_done(req);
1975 return;
1976 }
1977
1978 int ctdbd_parse_recv(struct tevent_req *req)
1979 {
1980 int error;
1981
1982 if (tevent_req_is_unix_error(req, &error)) {
1983 DBG_DEBUG("async parse returned %s\n", strerror(error));
1984 tevent_req_received(req);
1985 return error;
1986 }
1987
1988 tevent_req_received(req);
1989 return 0;
1990 }
Samba GIT mirror