?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
libcli: Speed up sddl_decode_ace()
[samba.git] / source3 / lib / ctdbd_conn.c
blobabee0eb87b122fec348acbd47cfdb4e17d85278b
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/util_net.h"
28 #include "lib/util/sys_rw_data.h"
29 #include "lib/util/iov_buf.h"
30 #include "lib/util/select.h"
31 #include "lib/util/debug.h"
32 #include "lib/util/talloc_stack.h"
33 #include "lib/util/genrand.h"
34 #include "lib/util/fault.h"
35 #include "lib/util/dlinklist.h"
36 #include "lib/util/tevent_unix.h"
37 #include "lib/util/sys_rw.h"
38 #include "lib/util/blocking.h"
39 #include "ctdb/include/ctdb_protocol.h"
40 #include "lib/async_req/async_sock.h"
41 #include "lib/dbwrap/dbwrap.h"
42 #include "lib/dbwrap/dbwrap_rbt.h"
43
44 /* paths to these include files come from --with-ctdb= in configure */
45
46 struct ctdbd_srvid_cb {
47 uint64_t srvid;
48 int (*cb)(struct tevent_context *ev,
49 uint32_t src_vnn, uint32_t dst_vnn,
50 uint64_t dst_srvid,
51 const uint8_t *msg, size_t msglen,
52 void *private_data);
53 void *private_data;
54 };
55
56 struct ctdbd_connection {
57 uint32_t reqid;
58 uint32_t our_vnn;
59 uint64_t rand_srvid;
60 struct ctdbd_srvid_cb *callbacks;
61 int fd;
62 int timeout;
63
64 /*
65 * Outgoing queue for writev_send of asynchronous ctdb requests
66 */
67 struct tevent_queue *outgoing;
68 struct tevent_req **pending;
69 struct tevent_req *read_req;
70 };
71
72 static bool ctdbd_conn_has_async_reqs(struct ctdbd_connection *conn)
73 {
74 size_t len = talloc_array_length(conn->pending);
75 return (len != 0);
76 }
77
78 static uint32_t ctdbd_next_reqid(struct ctdbd_connection *conn)
79 {
80 conn->reqid += 1;
81 if (conn->reqid == 0) {
82 conn->reqid += 1;
83 }
84 return conn->reqid;
85 }
86
87 static int ctdbd_control(struct ctdbd_connection *conn,
88 uint32_t vnn, uint32_t opcode,
89 uint64_t srvid, uint32_t flags,
90 TDB_DATA data,
91 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
92 int32_t *cstatus);
93
94 /*
95 * exit on fatal communications errors with the ctdbd daemon
96 */
97 static void cluster_fatal(const char *why)
98 {
99 DEBUG(0,("cluster fatal event: %s - exiting immediately\n", why));
100 /* we don't use smb_panic() as we don't want to delay to write
101 a core file. We need to release this process id immediately
102 so that someone else can take over without getting sharing
103 violations */
104 _exit(1);
105 }
106
107 /*
108 *
109 */
110 static void ctdb_packet_dump(struct ctdb_req_header *hdr)
111 {
112 if (DEBUGLEVEL < 11) {
113 return;
114 }
115 DEBUGADD(11, ("len=%"PRIu32", magic=%"PRIu32", vers=%"PRIu32", "
116 "gen=%"PRIu32", op=%"PRIu32", reqid=%"PRIu32"\n",
117 hdr->length,
118 hdr->ctdb_magic,
119 hdr->ctdb_version,
120 hdr->generation,
121 hdr->operation,
122 hdr->reqid));
123 }
124
125 /*
126 * Register a srvid with ctdbd
127 */
128 int register_with_ctdbd(struct ctdbd_connection *conn, uint64_t srvid,
129 int (*cb)(struct tevent_context *ev,
130 uint32_t src_vnn, uint32_t dst_vnn,
131 uint64_t dst_srvid,
132 const uint8_t *msg, size_t msglen,
133 void *private_data),
134 void *private_data)
135 {
136 size_t num_callbacks = talloc_array_length(conn->callbacks);
137 struct ctdbd_srvid_cb *tmp;
138 bool need_register = true;
139 size_t i;
140
141 for (i = 0; i < num_callbacks; i++) {
142 struct ctdbd_srvid_cb *c = &conn->callbacks[i];
143
144 if (c->srvid == srvid) {
145 need_register = false;
146 break;
147 }
148 }
149
150 if (need_register) {
151 int ret;
152 int32_t cstatus;
153
154 ret = ctdbd_control_local(conn, CTDB_CONTROL_REGISTER_SRVID,
155 srvid, 0, tdb_null, NULL, NULL,
156 &cstatus);
157 if (ret != 0) {
158 return ret;
159 }
160 }
161
162
163 tmp = talloc_realloc(conn, conn->callbacks, struct ctdbd_srvid_cb,
164 num_callbacks + 1);
165 if (tmp == NULL) {
166 return ENOMEM;
167 }
168 conn->callbacks = tmp;
169
170 conn->callbacks[num_callbacks] = (struct ctdbd_srvid_cb) {
171 .srvid = srvid, .cb = cb, .private_data = private_data
172 };
173
174 return 0;
175 }
176
177 void deregister_from_ctdbd(struct ctdbd_connection *conn,
178 uint64_t srvid,
179 int (*cb)(struct tevent_context *ev,
180 uint32_t src_vnn,
181 uint32_t dst_vnn,
182 uint64_t dst_srvid,
183 const uint8_t *msg,
184 size_t msglen,
185 void *private_data),
186 void *private_data)
187 {
188 struct ctdbd_srvid_cb *cbs = conn->callbacks;
189 size_t i, num_callbacks = talloc_array_length(cbs);
190 bool need_deregister = false;
191 bool keep_registration = false;
192
193 if (num_callbacks == 0) {
194 return;
195 }
196
197 for (i = 0; i < num_callbacks;) {
198 struct ctdbd_srvid_cb *c = &cbs[i];
199
200 if (c->srvid != srvid) {
201 i++;
202 continue;
203 }
204
205 if ((c->cb == cb) && (c->private_data == private_data)) {
206 need_deregister = true;
207 ARRAY_DEL_ELEMENT(cbs, i, num_callbacks);
208 num_callbacks--;
209 continue;
210 }
211
212 keep_registration = true;
213 i++;
214 }
215
216 conn->callbacks = talloc_realloc(conn,
217 cbs,
218 struct ctdbd_srvid_cb,
219 num_callbacks);
220
221 if (keep_registration) {
222 need_deregister = false;
223 }
224
225 if (need_deregister) {
226 int ret;
227 int32_t cstatus;
228
229 ret = ctdbd_control_local(conn, CTDB_CONTROL_DEREGISTER_SRVID,
230 srvid, 0, tdb_null, NULL, NULL,
231 &cstatus);
232 if (ret != 0) {
233 /*
234 * If CTDB_CONTROL_DEREGISTER_SRVID fails we may still
235 * get messages later, but we don't have a callback
236 * anymore, we just ignore these.
237 */
238 }
239 }
240
241 return;
242 }
243
244 static int ctdbd_msg_call_back(struct tevent_context *ev,
245 struct ctdbd_connection *conn,
246 struct ctdb_req_message_old *msg)
247 {
248 uint32_t msg_len;
249 size_t i, num_callbacks;
250
251 msg_len = msg->hdr.length;
252 if (msg_len < offsetof(struct ctdb_req_message_old, data)) {
253 DBG_DEBUG("len %"PRIu32" too small\n", msg_len);
254 return 0;
255 }
256 msg_len -= offsetof(struct ctdb_req_message_old, data);
257
258 if (msg_len < msg->datalen) {
259 DBG_DEBUG("msg_len=%"PRIu32" < msg->datalen=%"PRIu32"\n",
260 msg_len, msg->datalen);
261 return 0;
262 }
263
264 num_callbacks = talloc_array_length(conn->callbacks);
265
266 for (i=0; i<num_callbacks; i++) {
267 struct ctdbd_srvid_cb *cb = &conn->callbacks[i];
268
269 if ((cb->srvid == msg->srvid) && (cb->cb != NULL)) {
270 int ret;
271
272 ret = cb->cb(ev,
273 msg->hdr.srcnode, msg->hdr.destnode,
274 msg->srvid, msg->data, msg->datalen,
275 cb->private_data);
276 if (ret != 0) {
277 return ret;
278 }
279 }
280 }
281 return 0;
282 }
283
284 /*
285 * get our vnn from the cluster
286 */
287 static int get_cluster_vnn(struct ctdbd_connection *conn, uint32_t *vnn)
288 {
289 int32_t cstatus=-1;
290 int ret;
291 ret = ctdbd_control_local(conn, CTDB_CONTROL_GET_PNN, 0, 0,
292 tdb_null, NULL, NULL, &cstatus);
293 if (ret != 0) {
294 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
295 return ret;
296 }
297 *vnn = (uint32_t)cstatus;
298 return ret;
299 }
300
301 static int ctdbd_control_get_nodemap(struct ctdbd_connection *conn,
302 TALLOC_CTX *mem_ctx,
303 struct ctdb_node_map_old **_nodemap)
304 {
305 int32_t cstatus=-1;
306 TDB_DATA outdata = {0};
307 int ret;
308
309 ret = ctdbd_control_local(conn, CTDB_CONTROL_GET_NODEMAP, 0, 0,
310 tdb_null, mem_ctx, &outdata, &cstatus);
311 if (ret != 0) {
312 DEBUG(1, ("ctdbd_control failed: %s\n", strerror(ret)));
313 return ret;
314 }
315 if ((cstatus != 0) || (outdata.dptr == NULL)) {
316 DEBUG(2, ("Received invalid ctdb data\n"));
317 return EINVAL;
318 }
319
320 *_nodemap = (struct ctdb_node_map_old *)outdata.dptr;
321 return 0;
322 }
323
324 /*
325 * Are we active (i.e. not banned or stopped?)
326 */
327 static bool ctdbd_working(struct ctdbd_connection *conn, uint32_t vnn)
328 {
329 struct ctdb_node_map_old *m = NULL;
330 bool ok = false;
331 uint32_t i;
332 int ret;
333
334 ret = ctdbd_control_get_nodemap(conn, talloc_tos(), &m);
335 if (ret != 0) {
336 DEBUG(1, ("ctdbd_control_get_nodemap() failed: %s\n", strerror(ret)));
337 return false;
338 }
339
340 for (i=0; i<m->num; i++) {
341 if (vnn == m->nodes[i].pnn) {
342 break;
343 }
344 }
345
346 if (i == m->num) {
347 DEBUG(2, ("Did not find ourselves (node %d) in nodemap\n",
348 (int)vnn));
349 goto fail;
350 }
351
352 if ((m->nodes[i].flags & NODE_FLAGS_INACTIVE) != 0) {
353 DEBUG(2, ("Node has status %x, not active\n",
354 (int)m->nodes[i].flags));
355 goto fail;
356 }
357
358 ok = true;
359 fail:
360 TALLOC_FREE(m);
361 return ok;
362 }
363
364 uint32_t ctdbd_vnn(const struct ctdbd_connection *conn)
365 {
366 return conn->our_vnn;
367 }
368
369 /*
370 * Get us a ctdb connection
371 */
372
373 static int ctdbd_connect(const char *sockname, int *pfd)
374 {
375 struct samba_sockaddr addr = {
376 .sa_socklen = sizeof(struct sockaddr_un),
377 .u = {
378 .un = {
379 .sun_family = AF_UNIX,
380 },
381 },
382 };
383 int fd;
384 size_t namelen;
385 int ret;
386
387 fd = socket(AF_UNIX, SOCK_STREAM, 0);
388 if (fd == -1) {
389 int err = errno;
390 DEBUG(3, ("Could not create socket: %s\n", strerror(err)));
391 return err;
392 }
393
394 namelen = strlcpy(addr.u.un.sun_path,
395 sockname,
396 sizeof(addr.u.un.sun_path));
397 if (namelen >= sizeof(addr.u.un.sun_path)) {
398 DEBUG(3, ("%s: Socket name too long: %s\n", __func__,
399 sockname));
400 close(fd);
401 return ENAMETOOLONG;
402 }
403
404 ret = connect(fd, &addr.u.sa, addr.sa_socklen);
405 if (ret == -1) {
406 int err = errno;
407 DEBUG(1, ("connect(%s) failed: %s\n", sockname,
408 strerror(err)));
409 close(fd);
410 return err;
411 }
412
413 *pfd = fd;
414 return 0;
415 }
416
417 static int ctdb_read_packet(int fd, int timeout, TALLOC_CTX *mem_ctx,
418 struct ctdb_req_header **result)
419 {
420 struct ctdb_req_header *req;
421 uint32_t msglen;
422 ssize_t nread;
423
424 if (timeout != -1) {
425 struct pollfd pfd = { .fd = fd, .events = POLLIN };
426 int ret;
427
428 ret = sys_poll_intr(&pfd, 1, timeout);
429 if (ret == -1) {
430 return errno;
431 }
432 if (ret == 0) {
433 return ETIMEDOUT;
434 }
435 if (ret != 1) {
436 return EIO;
437 }
438 }
439
440 nread = read_data(fd, &msglen, sizeof(msglen));
441 if (nread == -1) {
442 return errno;
443 }
444 if (nread == 0) {
445 return EIO;
446 }
447
448 if (msglen < sizeof(struct ctdb_req_header)) {
449 return EIO;
450 }
451
452 req = talloc_size(mem_ctx, msglen);
453 if (req == NULL) {
454 return ENOMEM;
455 }
456 talloc_set_name_const(req, "struct ctdb_req_header");
457
458 req->length = msglen;
459
460 nread = read_data(fd, ((char *)req) + sizeof(msglen),
461 msglen - sizeof(msglen));
462 if (nread == -1) {
463 TALLOC_FREE(req);
464 return errno;
465 }
466 if (nread == 0) {
467 TALLOC_FREE(req);
468 return EIO;
469 }
470
471 *result = req;
472 return 0;
473 }
474
475 /*
476 * Read a full ctdbd request. If we have a messaging context, defer incoming
477 * messages that might come in between.
478 */
479
480 static int ctdb_read_req(struct ctdbd_connection *conn, uint32_t reqid,
481 TALLOC_CTX *mem_ctx, struct ctdb_req_header **result)
482 {
483 struct ctdb_req_header *hdr = NULL;
484 int ret;
485
486 next_pkt:
487
488 ret = ctdb_read_packet(conn->fd, conn->timeout, mem_ctx, &hdr);
489 if (ret != 0) {
490 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
491 cluster_fatal("failed to read data from ctdbd\n");
492 return -1;
493 }
494 SMB_ASSERT(hdr != NULL);
495
496 DEBUG(11, ("Received ctdb packet\n"));
497 ctdb_packet_dump(hdr);
498
499 if (hdr->operation == CTDB_REQ_MESSAGE) {
500 struct ctdb_req_message_old *msg = (struct ctdb_req_message_old *)hdr;
501
502 ret = ctdbd_msg_call_back(NULL, conn, msg);
503 if (ret != 0) {
504 TALLOC_FREE(hdr);
505 return ret;
506 }
507
508 TALLOC_FREE(hdr);
509 goto next_pkt;
510 }
511
512 if ((reqid != 0) && (hdr->reqid != reqid)) {
513 /* we got the wrong reply */
514 DEBUG(0,("Discarding mismatched ctdb reqid %u should have "
515 "been %u\n", hdr->reqid, reqid));
516 TALLOC_FREE(hdr);
517 goto next_pkt;
518 }
519
520 *result = talloc_move(mem_ctx, &hdr);
521
522 return 0;
523 }
524
525 static int ctdbd_connection_destructor(struct ctdbd_connection *c);
526
527 /*
528 * Get us a ctdbd connection
529 */
530
531 static int ctdbd_init_connection_internal(TALLOC_CTX *mem_ctx,
532 const char *sockname, int timeout,
533 struct ctdbd_connection *conn)
534 {
535 int ret;
536
537 conn->timeout = timeout;
538 if (conn->timeout == 0) {
539 conn->timeout = -1;
540 }
541
542 ret = ctdbd_connect(sockname, &conn->fd);
543 if (ret != 0) {
544 DEBUG(1, ("ctdbd_connect failed: %s\n", strerror(ret)));
545 return ret;
546 }
547 talloc_set_destructor(conn, ctdbd_connection_destructor);
548
549 ret = get_cluster_vnn(conn, &conn->our_vnn);
550 if (ret != 0) {
551 DEBUG(10, ("get_cluster_vnn failed: %s\n", strerror(ret)));
552 return ret;
553 }
554
555 if (!ctdbd_working(conn, conn->our_vnn)) {
556 DEBUG(2, ("Node is not working, can not connect\n"));
557 return EIO;
558 }
559
560 generate_random_buffer((unsigned char *)&conn->rand_srvid,
561 sizeof(conn->rand_srvid));
562
563 ret = register_with_ctdbd(conn, conn->rand_srvid, NULL, NULL);
564 if (ret != 0) {
565 DEBUG(5, ("Could not register random srvid: %s\n",
566 strerror(ret)));
567 return ret;
568 }
569
570 return 0;
571 }
572
573 int ctdbd_init_connection(TALLOC_CTX *mem_ctx,
574 const char *sockname, int timeout,
575 struct ctdbd_connection **pconn)
576 {
577 struct ctdbd_connection *conn;
578 int ret;
579
580 if (!(conn = talloc_zero(mem_ctx, struct ctdbd_connection))) {
581 DEBUG(0, ("talloc failed\n"));
582 return ENOMEM;
583 }
584
585 ret = ctdbd_init_connection_internal(mem_ctx,
586 sockname,
587 timeout,
588 conn);
589 if (ret != 0) {
590 DBG_ERR("ctdbd_init_connection_internal failed (%s)\n",
591 strerror(ret));
592 goto fail;
593 }
594
595 *pconn = conn;
596 return 0;
597
598 fail:
599 TALLOC_FREE(conn);
600 return ret;
601 }
602
603 int ctdbd_reinit_connection(TALLOC_CTX *mem_ctx,
604 const char *sockname, int timeout,
605 struct ctdbd_connection *conn)
606 {
607 int ret;
608
609 ret = ctdbd_connection_destructor(conn);
610 if (ret != 0) {
611 DBG_ERR("ctdbd_connection_destructor failed\n");
612 return ret;
613 }
614
615 ret = ctdbd_init_connection_internal(mem_ctx,
616 sockname,
617 timeout,
618 conn);
619 if (ret != 0) {
620 DBG_ERR("ctdbd_init_connection_internal failed (%s)\n",
621 strerror(ret));
622 return ret;
623 }
624
625 return 0;
626 }
627
628 int ctdbd_init_async_connection(
629 TALLOC_CTX *mem_ctx,
630 const char *sockname,
631 int timeout,
632 struct ctdbd_connection **pconn)
633 {
634 struct ctdbd_connection *conn = NULL;
635 int ret;
636
637 *pconn = NULL;
638
639 ret = ctdbd_init_connection(mem_ctx, sockname, timeout, &conn);
640 if (ret != 0) {
641 return ret;
642 }
643
644 ret = set_blocking(conn->fd, false);
645 if (ret == -1) {
646 int err = errno;
647 SMB_ASSERT(err != 0);
648 TALLOC_FREE(conn);
649 return err;
650 }
651
652 conn->outgoing = tevent_queue_create(conn, "ctdb async outgoing");
653 if (conn->outgoing == NULL) {
654 TALLOC_FREE(conn);
655 return ENOMEM;
656 }
657
658 *pconn = conn;
659 return 0;
660 }
661
662 int ctdbd_conn_get_fd(struct ctdbd_connection *conn)
663 {
664 return conn->fd;
665 }
666
667 /*
668 * Packet handler to receive and handle a ctdb message
669 */
670 static int ctdb_handle_message(struct tevent_context *ev,
671 struct ctdbd_connection *conn,
672 struct ctdb_req_header *hdr)
673 {
674 struct ctdb_req_message_old *msg;
675
676 if (hdr->operation != CTDB_REQ_MESSAGE) {
677 DEBUG(0, ("Received async msg of type %u, discarding\n",
678 hdr->operation));
679 return EINVAL;
680 }
681
682 msg = (struct ctdb_req_message_old *)hdr;
683
684 ctdbd_msg_call_back(ev, conn, msg);
685
686 return 0;
687 }
688
689 void ctdbd_socket_readable(struct tevent_context *ev,
690 struct ctdbd_connection *conn)
691 {
692 struct ctdb_req_header *hdr = NULL;
693 int ret;
694
695 ret = ctdb_read_packet(conn->fd, conn->timeout, talloc_tos(), &hdr);
696 if (ret != 0) {
697 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
698 cluster_fatal("failed to read data from ctdbd\n");
699 }
700 SMB_ASSERT(hdr != NULL);
701
702 ret = ctdb_handle_message(ev, conn, hdr);
703
704 TALLOC_FREE(hdr);
705
706 if (ret != 0) {
707 DEBUG(10, ("could not handle incoming message: %s\n",
708 strerror(ret)));
709 }
710 }
711
712 int ctdbd_messaging_send_iov(struct ctdbd_connection *conn,
713 uint32_t dst_vnn, uint64_t dst_srvid,
714 const struct iovec *iov, int iovlen)
715 {
716 struct ctdb_req_message_old r;
717 struct iovec iov2[iovlen+1];
718 size_t buflen = iov_buflen(iov, iovlen);
719 ssize_t nwritten;
720
721 r.hdr.length = offsetof(struct ctdb_req_message_old, data) + buflen;
722 r.hdr.ctdb_magic = CTDB_MAGIC;
723 r.hdr.ctdb_version = CTDB_PROTOCOL;
724 r.hdr.generation = 1;
725 r.hdr.operation = CTDB_REQ_MESSAGE;
726 r.hdr.destnode = dst_vnn;
727 r.hdr.srcnode = conn->our_vnn;
728 r.hdr.reqid = 0;
729 r.srvid = dst_srvid;
730 r.datalen = buflen;
731
732 DEBUG(10, ("ctdbd_messaging_send: Sending ctdb packet\n"));
733 ctdb_packet_dump(&r.hdr);
734
735 iov2[0].iov_base = &r;
736 iov2[0].iov_len = offsetof(struct ctdb_req_message_old, data);
737 memcpy(&iov2[1], iov, iovlen * sizeof(struct iovec));
738
739 nwritten = write_data_iov(conn->fd, iov2, iovlen+1);
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 return 0;
746 }
747
748 /*
749 * send/recv a generic ctdb control message
750 */
751 static int ctdbd_control(struct ctdbd_connection *conn,
752 uint32_t vnn, uint32_t opcode,
753 uint64_t srvid, uint32_t flags,
754 TDB_DATA data,
755 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
756 int32_t *cstatus)
757 {
758 struct ctdb_req_control_old req;
759 struct ctdb_req_header *hdr;
760 struct ctdb_reply_control_old *reply = NULL;
761 struct iovec iov[2];
762 ssize_t nwritten;
763 int ret;
764
765 if (ctdbd_conn_has_async_reqs(conn)) {
766 /*
767 * Can't use sync call while an async call is in flight. Adding
768 * this check as a safety net. We'll be using different
769 * connections for sync and async requests, so this shouldn't
770 * happen, but who knows...
771 */
772 DBG_ERR("Async ctdb req on sync connection\n");
773 return EINVAL;
774 }
775
776 ZERO_STRUCT(req);
777 req.hdr.length = offsetof(struct ctdb_req_control_old, data) + data.dsize;
778 req.hdr.ctdb_magic = CTDB_MAGIC;
779 req.hdr.ctdb_version = CTDB_PROTOCOL;
780 req.hdr.operation = CTDB_REQ_CONTROL;
781 req.hdr.reqid = ctdbd_next_reqid(conn);
782 req.hdr.destnode = vnn;
783 req.opcode = opcode;
784 req.srvid = srvid;
785 req.datalen = data.dsize;
786 req.flags = flags;
787
788 DBG_DEBUG("Sending ctdb packet reqid=%"PRIu32", vnn=%"PRIu32", "
789 "opcode=%"PRIu32", srvid=%"PRIu64"\n", req.hdr.reqid,
790 req.hdr.destnode, req.opcode, req.srvid);
791 ctdb_packet_dump(&req.hdr);
792
793 iov[0].iov_base = &req;
794 iov[0].iov_len = offsetof(struct ctdb_req_control_old, data);
795 iov[1].iov_base = data.dptr;
796 iov[1].iov_len = data.dsize;
797
798 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
799 if (nwritten == -1) {
800 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
801 cluster_fatal("cluster dispatch daemon msg write error\n");
802 }
803
804 if (flags & CTDB_CTRL_FLAG_NOREPLY) {
805 if (cstatus) {
806 *cstatus = 0;
807 }
808 return 0;
809 }
810
811 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
812 if (ret != 0) {
813 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
814 return ret;
815 }
816
817 if (hdr->operation != CTDB_REPLY_CONTROL) {
818 DEBUG(0, ("received invalid reply\n"));
819 TALLOC_FREE(hdr);
820 return EIO;
821 }
822 reply = (struct ctdb_reply_control_old *)hdr;
823
824 if (outdata) {
825 if (!(outdata->dptr = (uint8_t *)talloc_memdup(
826 mem_ctx, reply->data, reply->datalen))) {
827 TALLOC_FREE(reply);
828 return ENOMEM;
829 }
830 outdata->dsize = reply->datalen;
831 }
832 if (cstatus) {
833 (*cstatus) = reply->status;
834 }
835
836 TALLOC_FREE(reply);
837 return ret;
838 }
839
840 /*
841 * see if a remote process exists
842 */
843 bool ctdbd_process_exists(struct ctdbd_connection *conn, uint32_t vnn,
844 pid_t pid, uint64_t unique_id)
845 {
846 uint8_t buf[sizeof(pid)+sizeof(unique_id)];
847 int32_t cstatus = 0;
848 int ret;
849
850 if (unique_id == SERVERID_UNIQUE_ID_NOT_TO_VERIFY) {
851 ret = ctdbd_control(conn, vnn, CTDB_CONTROL_PROCESS_EXISTS,
852 0, 0,
853 (TDB_DATA) { .dptr = (uint8_t *)&pid,
854 .dsize = sizeof(pid) },
855 NULL, NULL, &cstatus);
856 if (ret != 0) {
857 return false;
858 }
859 return (cstatus == 0);
860 }
861
862 memcpy(buf, &pid, sizeof(pid));
863 memcpy(buf+sizeof(pid), &unique_id, sizeof(unique_id));
864
865 ret = ctdbd_control(conn, vnn, CTDB_CONTROL_CHECK_PID_SRVID, 0, 0,
866 (TDB_DATA) { .dptr = buf, .dsize = sizeof(buf) },
867 NULL, NULL, &cstatus);
868 if (ret != 0) {
869 return false;
870 }
871 return (cstatus == 0);
872 }
873
874 /*
875 * Get a db path
876 */
877 char *ctdbd_dbpath(struct ctdbd_connection *conn,
878 TALLOC_CTX *mem_ctx, uint32_t db_id)
879 {
880 int ret;
881 TDB_DATA data;
882 TDB_DATA rdata = {0};
883 int32_t cstatus = 0;
884
885 data.dptr = (uint8_t*)&db_id;
886 data.dsize = sizeof(db_id);
887
888 ret = ctdbd_control_local(conn, CTDB_CONTROL_GETDBPATH, 0, 0, data,
889 mem_ctx, &rdata, &cstatus);
890 if ((ret != 0) || cstatus != 0) {
891 DEBUG(0, (__location__ " ctdb_control for getdbpath failed: %s\n",
892 strerror(ret)));
893 TALLOC_FREE(rdata.dptr);
894 }
895
896 return (char *)rdata.dptr;
897 }
898
899 /*
900 * attach to a ctdb database
901 */
902 int ctdbd_db_attach(struct ctdbd_connection *conn,
903 const char *name, uint32_t *db_id, bool persistent)
904 {
905 int ret;
906 TDB_DATA data = {0};
907 int32_t cstatus;
908
909 data = string_term_tdb_data(name);
910
911 ret = ctdbd_control_local(conn,
912 persistent
913 ? CTDB_CONTROL_DB_ATTACH_PERSISTENT
914 : CTDB_CONTROL_DB_ATTACH,
915 0, 0, data, NULL, &data, &cstatus);
916 if (ret != 0) {
917 DEBUG(0, (__location__ " ctdb_control for db_attach "
918 "failed: %s\n", strerror(ret)));
919 return ret;
920 }
921
922 if (cstatus != 0 || data.dsize != sizeof(uint32_t)) {
923 DEBUG(0,(__location__ " ctdb_control for db_attach failed\n"));
924 TALLOC_FREE(data.dptr);
925 return EIO;
926 }
927
928 *db_id = *(uint32_t *)data.dptr;
929 talloc_free(data.dptr);
930
931 return 0;
932 }
933
934 /*
935 * force the migration of a record to this node
936 */
937 int ctdbd_migrate(struct ctdbd_connection *conn, uint32_t db_id, TDB_DATA key)
938 {
939 struct ctdb_req_call_old req;
940 struct ctdb_req_header *hdr = NULL;
941 struct iovec iov[2];
942 ssize_t nwritten;
943 int ret;
944
945 if (ctdbd_conn_has_async_reqs(conn)) {
946 /*
947 * Can't use sync call while an async call is in flight. Adding
948 * this check as a safety net. We'll be using different
949 * connections for sync and async requests, so this shouldn't
950 * happen, but who knows...
951 */
952 DBG_ERR("Async ctdb req on sync connection\n");
953 return EINVAL;
954 }
955
956 ZERO_STRUCT(req);
957
958 req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
959 req.hdr.ctdb_magic = CTDB_MAGIC;
960 req.hdr.ctdb_version = CTDB_PROTOCOL;
961 req.hdr.operation = CTDB_REQ_CALL;
962 req.hdr.reqid = ctdbd_next_reqid(conn);
963 req.flags = CTDB_IMMEDIATE_MIGRATION;
964 req.callid = CTDB_NULL_FUNC;
965 req.db_id = db_id;
966 req.keylen = key.dsize;
967
968 DEBUG(10, ("ctdbd_migrate: Sending ctdb packet\n"));
969 ctdb_packet_dump(&req.hdr);
970
971 iov[0].iov_base = &req;
972 iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
973 iov[1].iov_base = key.dptr;
974 iov[1].iov_len = key.dsize;
975
976 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
977 if (nwritten == -1) {
978 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
979 cluster_fatal("cluster dispatch daemon msg write error\n");
980 }
981
982 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
983 if (ret != 0) {
984 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
985 goto fail;
986 }
987
988 if (hdr->operation != CTDB_REPLY_CALL) {
989 if (hdr->operation == CTDB_REPLY_ERROR) {
990 DBG_ERR("received error from ctdb\n");
991 } else {
992 DBG_ERR("received invalid reply\n");
993 }
994 ret = EIO;
995 goto fail;
996 }
997
998 fail:
999
1000 TALLOC_FREE(hdr);
1001 return ret;
1002 }
1003
1004 /*
1005 * Fetch a record and parse it
1006 */
1007 int ctdbd_parse(struct ctdbd_connection *conn, uint32_t db_id,
1008 TDB_DATA key, bool local_copy,
1009 void (*parser)(TDB_DATA key, TDB_DATA data,
1010 void *private_data),
1011 void *private_data)
1012 {
1013 struct ctdb_req_call_old req;
1014 struct ctdb_req_header *hdr = NULL;
1015 struct ctdb_reply_call_old *reply;
1016 struct iovec iov[2];
1017 ssize_t nwritten;
1018 uint32_t flags;
1019 int ret;
1020
1021 if (ctdbd_conn_has_async_reqs(conn)) {
1022 /*
1023 * Can't use sync call while an async call is in flight. Adding
1024 * this check as a safety net. We'll be using different
1025 * connections for sync and async requests, so this shouldn't
1026 * happen, but who knows...
1027 */
1028 DBG_ERR("Async ctdb req on sync connection\n");
1029 return EINVAL;
1030 }
1031
1032 flags = local_copy ? CTDB_WANT_READONLY : 0;
1033
1034 ZERO_STRUCT(req);
1035
1036 req.hdr.length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
1037 req.hdr.ctdb_magic = CTDB_MAGIC;
1038 req.hdr.ctdb_version = CTDB_PROTOCOL;
1039 req.hdr.operation = CTDB_REQ_CALL;
1040 req.hdr.reqid = ctdbd_next_reqid(conn);
1041 req.flags = flags;
1042 req.callid = CTDB_FETCH_FUNC;
1043 req.db_id = db_id;
1044 req.keylen = key.dsize;
1045
1046 iov[0].iov_base = &req;
1047 iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
1048 iov[1].iov_base = key.dptr;
1049 iov[1].iov_len = key.dsize;
1050
1051 nwritten = write_data_iov(conn->fd, iov, ARRAY_SIZE(iov));
1052 if (nwritten == -1) {
1053 DEBUG(3, ("write_data_iov failed: %s\n", strerror(errno)));
1054 cluster_fatal("cluster dispatch daemon msg write error\n");
1055 }
1056
1057 ret = ctdb_read_req(conn, req.hdr.reqid, NULL, &hdr);
1058 if (ret != 0) {
1059 DEBUG(10, ("ctdb_read_req failed: %s\n", strerror(ret)));
1060 goto fail;
1061 }
1062
1063 if ((hdr == NULL) || (hdr->operation != CTDB_REPLY_CALL)) {
1064 DEBUG(0, ("received invalid reply\n"));
1065 ret = EIO;
1066 goto fail;
1067 }
1068 reply = (struct ctdb_reply_call_old *)hdr;
1069
1070 if (reply->datalen == 0) {
1071 /*
1072 * Treat an empty record as non-existing
1073 */
1074 ret = ENOENT;
1075 goto fail;
1076 }
1077
1078 parser(key, make_tdb_data(&reply->data[0], reply->datalen),
1079 private_data);
1080
1081 ret = 0;
1082 fail:
1083 TALLOC_FREE(hdr);
1084 return ret;
1085 }
1086
1087 /*
1088 Traverse a ctdb database. "conn" must be an otherwise unused
1089 ctdb_connection where no other messages but the traverse ones are
1090 expected.
1091 */
1092
1093 int ctdbd_traverse(struct ctdbd_connection *conn, uint32_t db_id,
1094 void (*fn)(TDB_DATA key, TDB_DATA data,
1095 void *private_data),
1096 void *private_data)
1097 {
1098 int ret;
1099 TDB_DATA key, data;
1100 struct ctdb_traverse_start t;
1101 int32_t cstatus = 0;
1102
1103 if (ctdbd_conn_has_async_reqs(conn)) {
1104 /*
1105 * Can't use sync call while an async call is in flight. Adding
1106 * this check as a safety net. We'll be using different
1107 * connections for sync and async requests, so this shouldn't
1108 * happen, but who knows...
1109 */
1110 DBG_ERR("Async ctdb req on sync connection\n");
1111 return EINVAL;
1112 }
1113
1114 t.db_id = db_id;
1115 t.srvid = conn->rand_srvid;
1116 t.reqid = ctdbd_next_reqid(conn);
1117
1118 data.dptr = (uint8_t *)&t;
1119 data.dsize = sizeof(t);
1120
1121 ret = ctdbd_control_local(conn, CTDB_CONTROL_TRAVERSE_START,
1122 conn->rand_srvid,
1123 0, data, NULL, NULL, &cstatus);
1124
1125 if ((ret != 0) || (cstatus != 0)) {
1126 DEBUG(0,("ctdbd_control failed: %s, %d\n", strerror(ret),
1127 cstatus));
1128
1129 if (ret == 0) {
1130 /*
1131 * We need a mapping here
1132 */
1133 ret = EIO;
1134 }
1135 return ret;
1136 }
1137
1138 while (true) {
1139 struct ctdb_req_header *hdr = NULL;
1140 struct ctdb_req_message_old *m;
1141 struct ctdb_rec_data_old *d;
1142
1143 ret = ctdb_read_packet(conn->fd, conn->timeout, conn, &hdr);
1144 if (ret != 0) {
1145 DBG_ERR("ctdb_read_packet failed: %s\n", strerror(ret));
1146 cluster_fatal("failed to read data from ctdbd\n");
1147 }
1148 SMB_ASSERT(hdr != NULL);
1149
1150 if (hdr->operation != CTDB_REQ_MESSAGE) {
1151 DEBUG(0, ("Got operation %u, expected a message\n",
1152 (unsigned)hdr->operation));
1153 return EIO;
1154 }
1155
1156 m = (struct ctdb_req_message_old *)hdr;
1157 d = (struct ctdb_rec_data_old *)&m->data[0];
1158 if (m->datalen < sizeof(uint32_t) || m->datalen != d->length) {
1159 DEBUG(0, ("Got invalid traverse data of length %d\n",
1160 (int)m->datalen));
1161 return EIO;
1162 }
1163
1164 key.dsize = d->keylen;
1165 key.dptr = &d->data[0];
1166 data.dsize = d->datalen;
1167 data.dptr = &d->data[d->keylen];
1168
1169 if (key.dsize == 0 && data.dsize == 0) {
1170 /* end of traverse */
1171 return 0;
1172 }
1173
1174 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
1175 DEBUG(0, ("Got invalid ltdb header length %d\n",
1176 (int)data.dsize));
1177 return EIO;
1178 }
1179 data.dsize -= sizeof(struct ctdb_ltdb_header);
1180 data.dptr += sizeof(struct ctdb_ltdb_header);
1181
1182 if (fn != NULL) {
1183 fn(key, data, private_data);
1184 }
1185 }
1186 return 0;
1187 }
1188
1189 /*
1190 This is used to canonicalize a ctdb_sock_addr structure.
1191 */
1192 static void smbd_ctdb_canonicalize_ip(const struct sockaddr_storage *in,
1193 struct sockaddr_storage *out)
1194 {
1195 memcpy(out, in, sizeof (*out));
1196
1197 #ifdef HAVE_IPV6
1198 if (in->ss_family == AF_INET6) {
1199 const char prefix[12] = { 0,0,0,0,0,0,0,0,0,0,0xff,0xff };
1200 const struct sockaddr_in6 *in6 =
1201 (const struct sockaddr_in6 *)in;
1202 struct sockaddr_in *out4 = (struct sockaddr_in *)out;
1203 if (memcmp(&in6->sin6_addr, prefix, 12) == 0) {
1204 memset(out, 0, sizeof(*out));
1205 #ifdef HAVE_SOCK_SIN_LEN
1206 out4->sin_len = sizeof(*out);
1207 #endif
1208 out4->sin_family = AF_INET;
1209 out4->sin_port = in6->sin6_port;
1210 memcpy(&out4->sin_addr, &in6->sin6_addr.s6_addr[12], 4);
1211 }
1212 }
1213 #endif
1214 }
1215
1216 /*
1217 * Register us as a server for a particular tcp connection
1218 */
1219
1220 int ctdbd_register_ips(struct ctdbd_connection *conn,
1221 const struct sockaddr_storage *_server,
1222 const struct sockaddr_storage *_client,
1223 int (*cb)(struct tevent_context *ev,
1224 uint32_t src_vnn, uint32_t dst_vnn,
1225 uint64_t dst_srvid,
1226 const uint8_t *msg, size_t msglen,
1227 void *private_data),
1228 void *private_data)
1229 {
1230 struct ctdb_connection p;
1231 TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
1232 int ret;
1233 struct sockaddr_storage client;
1234 struct sockaddr_storage server;
1235
1236 /*
1237 * Only one connection so far
1238 */
1239
1240 smbd_ctdb_canonicalize_ip(_client, &client);
1241 smbd_ctdb_canonicalize_ip(_server, &server);
1242
1243 ZERO_STRUCT(p);
1244 switch (client.ss_family) {
1245 case AF_INET:
1246 memcpy(&p.dst.ip, &server, sizeof(p.dst.ip));
1247 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1248 break;
1249 case AF_INET6:
1250 memcpy(&p.dst.ip6, &server, sizeof(p.dst.ip6));
1251 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1252 break;
1253 default:
1254 return EIO;
1255 }
1256
1257 /*
1258 * We want to be told about IP releases
1259 */
1260
1261 ret = register_with_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1262 cb, private_data);
1263 if (ret != 0) {
1264 return ret;
1265 }
1266
1267 /*
1268 * inform ctdb of our tcp connection, so if IP takeover happens ctdb
1269 * can send an extra ack to trigger a reset for our client, so it
1270 * immediately reconnects
1271 */
1272 ret = ctdbd_control_local(conn,
1273 CTDB_CONTROL_TCP_CLIENT, 0,
1274 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1275 NULL);
1276 if (ret != 0) {
1277 return ret;
1278 }
1279 return 0;
1280 }
1281
1282 void ctdbd_unregister_ips(struct ctdbd_connection *conn,
1283 const struct sockaddr_storage *_server,
1284 const struct sockaddr_storage *_client,
1285 int (*cb)(struct tevent_context *ev,
1286 uint32_t src_vnn,
1287 uint32_t dst_vnn,
1288 uint64_t dst_srvid,
1289 const uint8_t *msg,
1290 size_t msglen,
1291 void *private_data),
1292 void *private_data)
1293 {
1294 struct ctdb_connection p = {};
1295 TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
1296 int ret;
1297 struct sockaddr_storage client;
1298 struct sockaddr_storage server;
1299
1300 /*
1301 * Only one connection so far
1302 */
1303
1304 smbd_ctdb_canonicalize_ip(_client, &client);
1305 smbd_ctdb_canonicalize_ip(_server, &server);
1306
1307 switch (client.ss_family) {
1308 case AF_INET:
1309 memcpy(&p.dst.ip, &server, sizeof(p.dst.ip));
1310 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1311 break;
1312 case AF_INET6:
1313 memcpy(&p.dst.ip6, &server, sizeof(p.dst.ip6));
1314 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1315 break;
1316 default:
1317 return;
1318 }
1319
1320 /*
1321 * We no longer want to be told about IP releases
1322 * for the given callback/private_data combination
1323 */
1324 deregister_from_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1325 cb, private_data);
1326
1327 /*
1328 * inform ctdb of our tcp connection is no longer active
1329 */
1330 ret = ctdbd_control_local(conn,
1331 CTDB_CONTROL_TCP_CLIENT_DISCONNECTED, 0,
1332 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1333 NULL);
1334 if (ret != 0) {
1335 /*
1336 * We ignore errors here, as we'll just
1337 * no longer have a callback handler
1338 * registered and messages may just be ignored
1339 */
1340 }
1341
1342 return;
1343 }
1344
1345 void ctdbd_passed_ips(struct ctdbd_connection *conn,
1346 const struct sockaddr_storage *_server,
1347 const struct sockaddr_storage *_client,
1348 int (*cb)(struct tevent_context *ev,
1349 uint32_t src_vnn,
1350 uint32_t dst_vnn,
1351 uint64_t dst_srvid,
1352 const uint8_t *msg,
1353 size_t msglen,
1354 void *private_data),
1355 void *private_data)
1356 {
1357 struct ctdb_connection p;
1358 TDB_DATA data = { .dptr = (uint8_t *)&p, .dsize = sizeof(p) };
1359 int ret;
1360 struct sockaddr_storage client;
1361 struct sockaddr_storage server;
1362
1363 /*
1364 * Only one connection so far
1365 */
1366
1367 smbd_ctdb_canonicalize_ip(_client, &client);
1368 smbd_ctdb_canonicalize_ip(_server, &server);
1369
1370 ZERO_STRUCT(p);
1371 switch (client.ss_family) {
1372 case AF_INET:
1373 memcpy(&p.dst.ip, &server, sizeof(p.dst.ip));
1374 memcpy(&p.src.ip, &client, sizeof(p.src.ip));
1375 break;
1376 case AF_INET6:
1377 memcpy(&p.dst.ip6, &server, sizeof(p.dst.ip6));
1378 memcpy(&p.src.ip6, &client, sizeof(p.src.ip6));
1379 break;
1380 default:
1381 return;
1382 }
1383
1384 /*
1385 * We no longer want to be told about IP releases
1386 * for the given callback/private_data combination
1387 */
1388 deregister_from_ctdbd(conn, CTDB_SRVID_RELEASE_IP,
1389 cb, private_data);
1390
1391 /*
1392 * inform ctdb of our tcp connection is now passed to
1393 * another process.
1394 */
1395 ret = ctdbd_control_local(conn,
1396 CTDB_CONTROL_TCP_CLIENT_PASSED, 0,
1397 CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL,
1398 NULL);
1399 if (ret != 0) {
1400 /*
1401 * We ignore errors here, as we'll just
1402 * no longer have a callback handler
1403 * registered and messages may just be ignored
1404 */
1405 }
1406
1407 return;
1408 }
1409
1410 static int ctdbd_control_get_public_ips(struct ctdbd_connection *conn,
1411 uint32_t vnn,
1412 uint32_t flags,
1413 TALLOC_CTX *mem_ctx,
1414 struct ctdb_public_ip_list_old **_ips)
1415 {
1416 struct ctdb_public_ip_list_old *ips = NULL;
1417 TDB_DATA outdata;
1418 int32_t cstatus = -1;
1419 size_t min_dsize;
1420 size_t max_ips;
1421 int ret;
1422
1423 *_ips = NULL;
1424
1425 ret = ctdbd_control(conn,
1426 vnn,
1427 CTDB_CONTROL_GET_PUBLIC_IPS,
1428 0, /* srvid */
1429 flags,
1430 tdb_null, /* indata */
1431 mem_ctx,
1432 &outdata,
1433 &cstatus);
1434 if (ret != 0 || cstatus != 0) {
1435 DBG_ERR("ctdb_control for getpublicips failed ret:%d cstatus:%d\n",
1436 ret, (int)cstatus);
1437 return -1;
1438 }
1439
1440 min_dsize = offsetof(struct ctdb_public_ip_list_old, ips);
1441 if (outdata.dsize < min_dsize) {
1442 DBG_ERR("outdata.dsize=%zu < min_dsize=%zu\n",
1443 outdata.dsize, min_dsize);
1444 return -1;
1445 }
1446 max_ips = (outdata.dsize - min_dsize)/sizeof(struct ctdb_public_ip);
1447 ips = (struct ctdb_public_ip_list_old *)outdata.dptr;
1448 if ((size_t)ips->num > max_ips) {
1449 DBG_ERR("ips->num=%zu > max_ips=%zu\n",
1450 (size_t)ips->num, max_ips);
1451 return -1;
1452 }
1453
1454 *_ips = ips;
1455 return 0;
1456 }
1457
1458 static struct samba_sockaddr ctdbd_sock_addr_to_samba(const ctdb_sock_addr *c)
1459 {
1460 struct samba_sockaddr s = {};
1461
1462 switch (c->sa.sa_family) {
1463 case AF_INET:
1464 s.u.in = c->ip;
1465 break;
1466 case AF_INET6:
1467 /*
1468 * ctdb always requires HAVE_IPV6,
1469 * so we don't need an ifdef here.
1470 */
1471 s.u.in6 = c->ip6;
1472 break;
1473 default:
1474 /*
1475 * ctdb_sock_addr only supports ipv4 and ipv6
1476 */
1477 smb_panic(__location__);
1478 break;
1479 }
1480
1481 return s;
1482 }
1483
1484 int ctdbd_public_ip_foreach(struct ctdbd_connection *conn,
1485 int (*cb)(uint32_t total_ip_count,
1486 const struct sockaddr_storage *ip,
1487 bool is_movable_ip,
1488 void *private_data),
1489 void *private_data)
1490 {
1491 uint32_t i;
1492 struct ctdb_public_ip_list_old *ips = NULL;
1493 int ret = ENOMEM;
1494 TALLOC_CTX *frame = talloc_stackframe();
1495
1496 ret = ctdbd_control_get_public_ips(conn, CTDB_CURRENT_NODE, 0, frame, &ips);
1497 if (ret < 0) {
1498 ret = EIO;
1499 goto out_free;
1500 }
1501
1502 for (i=0; i < ips->num; i++) {
1503 const ctdb_sock_addr *addr = &ips->ips[i].addr;
1504 struct samba_sockaddr tmp = ctdbd_sock_addr_to_samba(addr);
1505
1506 ret = cb(ips->num,
1507 &tmp.u.ss,
1508 true, /* all ctdb public ips are movable */
1509 private_data);
1510 if (ret != 0) {
1511 goto out_free;
1512 }
1513 }
1514
1515 ret = 0;
1516 out_free:
1517 TALLOC_FREE(frame);
1518 return ret;
1519 }
1520
1521 static int count_ips(struct db_record *rec, void *private_data)
1522 {
1523 return 0;
1524 }
1525
1526 static int collect_ips(struct db_record *rec, void *private_data)
1527 {
1528 struct ctdb_public_ip_list_old *ips = talloc_get_type_abort(
1529 private_data, struct ctdb_public_ip_list_old);
1530 struct ctdb_public_ip *ip;
1531 TDB_DATA val = dbwrap_record_get_value(rec);
1532
1533 SMB_ASSERT(val.dsize == sizeof(*ip));
1534
1535 ip = (struct ctdb_public_ip *)val.dptr;
1536 ips->ips[ips->num] = *ip;
1537 ips->num += 1;
1538
1539 return 0;
1540 }
1541
1542 static int ctdbd_control_get_all_public_ips(struct ctdbd_connection *conn,
1543 const struct ctdb_node_map_old *nodemap,
1544 TALLOC_CTX *mem_ctx,
1545 struct ctdb_public_ip_list_old **_ips)
1546 {
1547 TALLOC_CTX *frame = talloc_stackframe();
1548 uint32_t ni;
1549 struct ctdb_public_ip_list_old *ips = NULL;
1550 struct db_context *rbt = NULL;
1551 NTSTATUS status;
1552 size_t len;
1553 int ret;
1554 int count;
1555
1556 rbt = db_open_rbt(frame);
1557 if (rbt == NULL) {
1558 DBG_WARNING("db_open_rbt() failed\n");
1559 TALLOC_FREE(frame);
1560 return -1;
1561 }
1562
1563 for (ni=0; ni < nodemap->num; ni++) {
1564 const struct ctdb_node_and_flags *n = &nodemap->nodes[ni];
1565 uint32_t j;
1566
1567 if (n->flags & NODE_FLAGS_INACTIVE) {
1568 continue;
1569 }
1570
1571 ret = ctdbd_control_get_public_ips(conn,
1572 n->pnn,
1573 0,
1574 frame,
1575 &ips);
1576 if (ret != 0) {
1577 TALLOC_FREE(frame);
1578 return -1;
1579 }
1580
1581 for (j=0; j<ips->num; j++) {
1582 struct ctdb_public_ip ip;
1583 TDB_DATA key;
1584 TDB_DATA val;
1585
1586 ip.pnn = ips->ips[j].pnn;
1587 ip.addr = ips->ips[j].addr;
1588
1589 key = make_tdb_data((uint8_t *)&ip.addr, sizeof(ip.addr));
1590 val = make_tdb_data((uint8_t *)&ip, sizeof(ip));
1591
1592 if (n->pnn == ip.pnn) {
1593 /*
1594 * Node claims IP is hosted on it, so
1595 * save that information
1596 */
1597 status = dbwrap_store(rbt, key, val,
1598 TDB_REPLACE);
1599 if (!NT_STATUS_IS_OK(status)) {
1600 TALLOC_FREE(frame);
1601 return -1;
1602 }
1603 } else {
1604 /*
1605 * Node thinks IP is hosted elsewhere,
1606 * so overwrite with CTDB_UNKNOWN_PNN
1607 * if there's no existing entry
1608 */
1609 bool exists = dbwrap_exists(rbt, key);
1610 if (!exists) {
1611 ip.pnn = CTDB_UNKNOWN_PNN;
1612 status = dbwrap_store(rbt, key, val,
1613 TDB_INSERT);
1614 if (!NT_STATUS_IS_OK(status)) {
1615 TALLOC_FREE(frame);
1616 return -1;
1617 }
1618 }
1619 }
1620 }
1621
1622 TALLOC_FREE(ips);
1623 }
1624
1625 status = dbwrap_traverse_read(rbt, count_ips, NULL, &count);
1626 if (!NT_STATUS_IS_OK(status)) {
1627 TALLOC_FREE(frame);
1628 return -1;
1629 }
1630
1631 len = offsetof(struct ctdb_public_ip_list_old, ips) +
1632 count*sizeof(struct ctdb_public_ip);
1633 ips = talloc_zero_size(mem_ctx, len);
1634 if (ips == NULL) {
1635 TALLOC_FREE(frame);
1636 return -1;
1637 }
1638 talloc_set_type(ips, struct ctdb_public_ip_list_old);
1639 talloc_reparent(mem_ctx, frame, ips);
1640
1641 status = dbwrap_traverse_read(rbt, collect_ips, ips, &count);
1642 if (!NT_STATUS_IS_OK(status)) {
1643 TALLOC_FREE(frame);
1644 return -1;
1645 }
1646
1647 if ((unsigned int)count != ips->num) {
1648 TALLOC_FREE(frame);
1649 return -1;
1650 }
1651
1652 *_ips = talloc_move(mem_ctx, &ips);
1653 TALLOC_FREE(frame);
1654 return 0;
1655 }
1656
1657 /*
1658 * This includes all node and/or public ips
1659 * of the whole cluster.
1660 *
1661 * node ips have:
1662 * - a valid pinned_pnn value.
1663 * - current_pnn is valid if the node is healthy
1664 *
1665 * public ips have:
1666 * - pinned_pnn as CTDB_UNKNOWN_PNN
1667 * - current_pnn is valid if a node healthy and hosting this ip.
1668 */
1669 int ctdbd_all_ip_foreach(struct ctdbd_connection *conn,
1670 bool include_node_ips,
1671 bool include_public_ips,
1672 int (*cb)(uint32_t total_ip_count,
1673 const struct sockaddr_storage *ip,
1674 uint32_t pinned_pnn,
1675 uint32_t current_pnn,
1676 void *private_data),
1677 void *private_data)
1678 {
1679 TALLOC_CTX *frame = talloc_stackframe();
1680 struct ctdb_node_map_old *nodemap = NULL;
1681 struct ctdb_public_ip_list_old *ips = NULL;
1682 int ret = ENOMEM;
1683 uint32_t total_ip_count = 0;
1684 uint32_t i;
1685
1686 ret = ctdbd_control_get_nodemap(conn, frame, &nodemap);
1687 if (ret != 0) {
1688 DBG_WARNING("ctdbd_control_get_nodemap() failed: %s\n", strerror(ret));
1689 TALLOC_FREE(frame);
1690 return -1;
1691 }
1692
1693 for (i=0; include_node_ips && i < nodemap->num; i++) {
1694 const struct ctdb_node_and_flags *n = &nodemap->nodes[i];
1695
1696 if (n->flags & NODE_FLAGS_DELETED) {
1697 continue;
1698 }
1699
1700 total_ip_count += 1;
1701 }
1702
1703 if (include_public_ips) {
1704 ret = ctdbd_control_get_all_public_ips(conn, nodemap,
1705 frame, &ips);
1706 if (ret < 0) {
1707 ret = EIO;
1708 goto out_free;
1709 }
1710
1711 total_ip_count += ips->num;
1712 }
1713
1714 for (i=0; include_node_ips && i < nodemap->num; i++) {
1715 const struct ctdb_node_and_flags *n = &nodemap->nodes[i];
1716 struct samba_sockaddr tmp = ctdbd_sock_addr_to_samba(&n->addr);
1717 uint32_t pinned_pnn = n->pnn;
1718 uint32_t current_pnn = n->pnn;
1719
1720 if (n->flags & NODE_FLAGS_DELETED) {
1721 continue;
1722 }
1723
1724 if (n->flags & (NODE_FLAGS_INACTIVE|NODE_FLAGS_DISABLED)) {
1725 /*
1726 * The ip address is not available
1727 * unless the node is up and
1728 * healthy.
1729 */
1730 current_pnn = CTDB_UNKNOWN_PNN;
1731 }
1732
1733 ret = cb(total_ip_count,
1734 &tmp.u.ss,
1735 pinned_pnn,
1736 current_pnn,
1737 private_data);
1738 if (ret != 0) {
1739 goto out_free;
1740 }
1741 }
1742
1743 for (i=0; include_public_ips && i < ips->num; i++) {
1744 const ctdb_sock_addr *addr = &ips->ips[i].addr;
1745 struct samba_sockaddr tmp = ctdbd_sock_addr_to_samba(addr);
1746 /* all ctdb public ips are movable and not pinned */
1747 uint32_t pinned_pnn = CTDB_UNKNOWN_PNN;
1748 uint32_t current_pnn = ips->ips[i].pnn;
1749 uint32_t ni;
1750
1751 for (ni=0; ni < nodemap->num; ni++) {
1752 const struct ctdb_node_and_flags *n = &nodemap->nodes[ni];
1753
1754 if (n->pnn != current_pnn) {
1755 continue;
1756 }
1757
1758 if (n->flags & (NODE_FLAGS_INACTIVE|NODE_FLAGS_DISABLED)) {
1759 current_pnn = CTDB_UNKNOWN_PNN;
1760 }
1761 break;
1762 }
1763
1764 ret = cb(total_ip_count,
1765 &tmp.u.ss,
1766 pinned_pnn,
1767 current_pnn,
1768 private_data);
1769 if (ret != 0) {
1770 goto out_free;
1771 }
1772 }
1773
1774 ret = 0;
1775 out_free:
1776 TALLOC_FREE(frame);
1777 return ret;
1778 }
1779
1780 /*
1781 call a control on the local node
1782 */
1783 int ctdbd_control_local(struct ctdbd_connection *conn, uint32_t opcode,
1784 uint64_t srvid, uint32_t flags, TDB_DATA data,
1785 TALLOC_CTX *mem_ctx, TDB_DATA *outdata,
1786 int32_t *cstatus)
1787 {
1788 return ctdbd_control(conn, CTDB_CURRENT_NODE, opcode, srvid, flags, data,
1789 mem_ctx, outdata, cstatus);
1790 }
1791
1792 int ctdb_watch_us(struct ctdbd_connection *conn)
1793 {
1794 struct ctdb_notify_data_old reg_data;
1795 size_t struct_len;
1796 int ret;
1797 int32_t cstatus;
1798
1799 reg_data.srvid = CTDB_SRVID_SAMBA_NOTIFY;
1800 reg_data.len = 1;
1801 reg_data.notify_data[0] = 0;
1802
1803 struct_len = offsetof(struct ctdb_notify_data_old,
1804 notify_data) + reg_data.len;
1805
1806 ret = ctdbd_control_local(
1807 conn, CTDB_CONTROL_REGISTER_NOTIFY, conn->rand_srvid, 0,
1808 make_tdb_data((uint8_t *)&reg_data, struct_len),
1809 NULL, NULL, &cstatus);
1810 if (ret != 0) {
1811 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1812 strerror(ret)));
1813 }
1814 return ret;
1815 }
1816
1817 int ctdb_unwatch(struct ctdbd_connection *conn)
1818 {
1819 uint64_t srvid = CTDB_SRVID_SAMBA_NOTIFY;
1820 int ret;
1821 int32_t cstatus;
1822
1823 ret = ctdbd_control_local(
1824 conn, CTDB_CONTROL_DEREGISTER_NOTIFY, conn->rand_srvid, 0,
1825 make_tdb_data((uint8_t *)&srvid, sizeof(srvid)),
1826 NULL, NULL, &cstatus);
1827 if (ret != 0) {
1828 DEBUG(1, ("ctdbd_control_local failed: %s\n",
1829 strerror(ret)));
1830 }
1831 return ret;
1832 }
1833
1834 int ctdbd_probe(const char *sockname, int timeout)
1835 {
1836 /*
1837 * Do a very early check if ctdbd is around to avoid an abort and core
1838 * later
1839 */
1840 struct ctdbd_connection *conn = NULL;
1841 int ret;
1842
1843 ret = ctdbd_init_connection(talloc_tos(), sockname, timeout,
1844 &conn);
1845
1846 /*
1847 * We only care if we can connect.
1848 */
1849 TALLOC_FREE(conn);
1850
1851 return ret;
1852 }
1853
1854 static int ctdbd_connection_destructor(struct ctdbd_connection *c)
1855 {
1856 if (c->fd != -1) {
1857 close(c->fd);
1858 c->fd = -1;
1859 }
1860 return 0;
1861 }
1862
1863 void ctdbd_prep_hdr_next_reqid(
1864 struct ctdbd_connection *conn, struct ctdb_req_header *hdr)
1865 {
1866 *hdr = (struct ctdb_req_header) {
1867 .ctdb_magic = CTDB_MAGIC,
1868 .ctdb_version = CTDB_PROTOCOL,
1869 .reqid = ctdbd_next_reqid(conn),
1870 .destnode = CTDB_CURRENT_NODE,
1871 };
1872 }
1873
1874 struct ctdbd_pkt_read_state {
1875 uint8_t *pkt;
1876 };
1877
1878 static ssize_t ctdbd_pkt_read_more(
1879 uint8_t *buf, size_t buflen, void *private_data);
1880 static void ctdbd_pkt_read_done(struct tevent_req *subreq);
1881
1882 static struct tevent_req *ctdbd_pkt_read_send(
1883 TALLOC_CTX *mem_ctx, struct tevent_context *ev, int fd)
1884 {
1885 struct tevent_req *req = NULL, *subreq = NULL;
1886 struct ctdbd_pkt_read_state *state = NULL;
1887
1888 req = tevent_req_create(mem_ctx, &state, struct ctdbd_pkt_read_state);
1889 if (req == NULL) {
1890 return NULL;
1891 }
1892 subreq = read_packet_send(state, ev, fd, 4, ctdbd_pkt_read_more, NULL);
1893 if (tevent_req_nomem(subreq, req)) {
1894 return tevent_req_post(req, ev);
1895 }
1896 tevent_req_set_callback(subreq, ctdbd_pkt_read_done, req);
1897 return req;
1898 }
1899
1900 static ssize_t ctdbd_pkt_read_more(
1901 uint8_t *buf, size_t buflen, void *private_data)
1902 {
1903 uint32_t msglen;
1904 if (buflen < 4) {
1905 return -1;
1906 }
1907 if (buflen > 4) {
1908 return 0; /* Been here, done */
1909 }
1910 memcpy(&msglen, buf, 4);
1911
1912 if (msglen < sizeof(struct ctdb_req_header)) {
1913 return -1;
1914 }
1915 return msglen - sizeof(msglen);
1916 }
1917
1918 static void ctdbd_pkt_read_done(struct tevent_req *subreq)
1919 {
1920 struct tevent_req *req = tevent_req_callback_data(
1921 subreq, struct tevent_req);
1922 struct ctdbd_pkt_read_state *state = tevent_req_data(
1923 req, struct ctdbd_pkt_read_state);
1924 ssize_t nread;
1925 int err;
1926
1927 nread = read_packet_recv(subreq, state, &state->pkt, &err);
1928 TALLOC_FREE(subreq);
1929 if (nread == -1) {
1930 tevent_req_error(req, err);
1931 return;
1932 }
1933 tevent_req_done(req);
1934 }
1935
1936 static int ctdbd_pkt_read_recv(
1937 struct tevent_req *req, TALLOC_CTX *mem_ctx, uint8_t **pkt)
1938 {
1939 struct ctdbd_pkt_read_state *state = tevent_req_data(
1940 req, struct ctdbd_pkt_read_state);
1941 int err;
1942
1943 if (tevent_req_is_unix_error(req, &err)) {
1944 return err;
1945 }
1946 *pkt = talloc_move(mem_ctx, &state->pkt);
1947 tevent_req_received(req);
1948 return 0;
1949 }
1950
1951 static bool ctdbd_conn_receive_next(struct ctdbd_connection *conn);
1952 static void ctdbd_conn_received(struct tevent_req *subreq);
1953
1954 struct ctdbd_req_state {
1955 struct ctdbd_connection *conn;
1956 struct tevent_context *ev;
1957 uint32_t reqid;
1958 struct ctdb_req_header *reply;
1959 };
1960
1961 static void ctdbd_req_unset_pending(struct tevent_req *req)
1962 {
1963 struct ctdbd_req_state *state = tevent_req_data(
1964 req, struct ctdbd_req_state);
1965 struct ctdbd_connection *conn = state->conn;
1966 size_t num_pending = talloc_array_length(conn->pending);
1967 size_t i, num_after;
1968
1969 tevent_req_set_cleanup_fn(req, NULL);
1970
1971 if (num_pending == 1) {
1972 /*
1973 * conn->read_req is a child of conn->pending
1974 */
1975 TALLOC_FREE(conn->pending);
1976 conn->read_req = NULL;
1977 return;
1978 }
1979
1980 for (i=0; i<num_pending; i++) {
1981 if (req == conn->pending[i]) {
1982 break;
1983 }
1984 }
1985 if (i == num_pending) {
1986 /*
1987 * Something's seriously broken. Just returning here is the
1988 * right thing nevertheless, the point of this routine is to
1989 * remove ourselves from conn->pending.
1990 */
1991 return;
1992 }
1993
1994 num_after = num_pending - i - 1;
1995 if (num_after > 0) {
1996 memmove(&conn->pending[i],
1997 &conn->pending[i] + 1,
1998 sizeof(*conn->pending) * num_after);
1999 }
2000 conn->pending = talloc_realloc(
2001 NULL, conn->pending, struct tevent_req *, num_pending - 1);
2002 }
2003
2004 static void ctdbd_req_cleanup(
2005 struct tevent_req *req, enum tevent_req_state req_state)
2006 {
2007 ctdbd_req_unset_pending(req);
2008 }
2009
2010 static bool ctdbd_req_set_pending(struct tevent_req *req)
2011 {
2012 struct ctdbd_req_state *state = tevent_req_data(
2013 req, struct ctdbd_req_state);
2014 struct ctdbd_connection *conn = state->conn;
2015 struct tevent_req **pending = NULL;
2016 size_t num_pending = talloc_array_length(conn->pending);
2017 bool ok;
2018
2019 pending = talloc_realloc(
2020 conn, conn->pending, struct tevent_req *, num_pending + 1);
2021 if (pending == NULL) {
2022 return false;
2023 }
2024 pending[num_pending] = req;
2025 conn->pending = pending;
2026
2027 tevent_req_set_cleanup_fn(req, ctdbd_req_cleanup);
2028
2029 ok = ctdbd_conn_receive_next(conn);
2030 if (!ok) {
2031 ctdbd_req_unset_pending(req);
2032 return false;
2033 }
2034
2035 return true;
2036 }
2037
2038 static bool ctdbd_conn_receive_next(struct ctdbd_connection *conn)
2039 {
2040 size_t num_pending = talloc_array_length(conn->pending);
2041 struct tevent_req *req = NULL;
2042 struct ctdbd_req_state *state = NULL;
2043
2044 if (conn->read_req != NULL) {
2045 return true;
2046 }
2047 if (num_pending == 0) {
2048 /*
2049 * done for now
2050 */
2051 return true;
2052 }
2053
2054 req = conn->pending[0];
2055 state = tevent_req_data(req, struct ctdbd_req_state);
2056
2057 conn->read_req = ctdbd_pkt_read_send(
2058 conn->pending, state->ev, conn->fd);
2059 if (conn->read_req == NULL) {
2060 return false;
2061 }
2062 tevent_req_set_callback(conn->read_req, ctdbd_conn_received, conn);
2063 return true;
2064 }
2065
2066 static void ctdbd_conn_received(struct tevent_req *subreq)
2067 {
2068 struct ctdbd_connection *conn = tevent_req_callback_data(
2069 subreq, struct ctdbd_connection);
2070 TALLOC_CTX *frame = talloc_stackframe();
2071 uint8_t *pkt = NULL;
2072 int ret;
2073 struct ctdb_req_header *hdr = NULL;
2074 uint32_t reqid;
2075 struct tevent_req *req = NULL;
2076 struct ctdbd_req_state *state = NULL;
2077 size_t i, num_pending;
2078 bool ok;
2079
2080 SMB_ASSERT(subreq == conn->read_req);
2081 conn->read_req = NULL;
2082
2083 ret = ctdbd_pkt_read_recv(subreq, frame, &pkt);
2084 TALLOC_FREE(subreq);
2085 if (ret != 0) {
2086 cluster_fatal("ctdbd_pkt_read failed\n");
2087 }
2088
2089 hdr = (struct ctdb_req_header *)pkt;
2090 reqid = hdr->reqid;
2091 num_pending = talloc_array_length(conn->pending);
2092
2093 for (i=0; i<num_pending; i++) {
2094 req = conn->pending[i];
2095 state = tevent_req_data(req, struct ctdbd_req_state);
2096 if (state->reqid == reqid) {
2097 break;
2098 }
2099 }
2100
2101 if (i == num_pending) {
2102 /* not found */
2103 TALLOC_FREE(frame);
2104 return;
2105 }
2106
2107 state->reply = talloc_move(state, &hdr);
2108 tevent_req_defer_callback(req, state->ev);
2109 tevent_req_done(req);
2110
2111 TALLOC_FREE(frame);
2112
2113 ok = ctdbd_conn_receive_next(conn);
2114 if (!ok) {
2115 cluster_fatal("ctdbd_conn_receive_next failed\n");
2116 }
2117 }
2118
2119 static void ctdbd_req_written(struct tevent_req *subreq);
2120
2121 struct tevent_req *ctdbd_req_send(
2122 TALLOC_CTX *mem_ctx,
2123 struct tevent_context *ev,
2124 struct ctdbd_connection *conn,
2125 struct iovec *iov,
2126 size_t num_iov)
2127 {
2128 struct tevent_req *req = NULL, *subreq = NULL;
2129 struct ctdbd_req_state *state = NULL;
2130 struct ctdb_req_header *hdr = NULL;
2131 bool ok;
2132
2133 req = tevent_req_create(mem_ctx, &state, struct ctdbd_req_state);
2134 if (req == NULL) {
2135 return NULL;
2136 }
2137 state->conn = conn;
2138 state->ev = ev;
2139
2140 if ((num_iov == 0) ||
2141 (iov[0].iov_len < sizeof(struct ctdb_req_header))) {
2142 tevent_req_error(req, EINVAL);
2143 return tevent_req_post(req, ev);
2144 }
2145 hdr = iov[0].iov_base;
2146 state->reqid = hdr->reqid;
2147
2148 ok = ctdbd_req_set_pending(req);
2149 if (!ok) {
2150 tevent_req_oom(req);
2151 return tevent_req_post(req, ev);
2152 }
2153
2154 subreq = writev_send(
2155 state, ev, conn->outgoing, conn->fd, false, iov, num_iov);
2156 if (tevent_req_nomem(subreq, req)) {
2157 return tevent_req_post(req, ev);
2158 }
2159 tevent_req_set_callback(subreq, ctdbd_req_written, req);
2160
2161 return req;
2162 }
2163
2164 static void ctdbd_req_written(struct tevent_req *subreq)
2165 {
2166 struct tevent_req *req = tevent_req_callback_data(
2167 subreq, struct tevent_req);
2168 ssize_t nwritten;
2169 int err;
2170
2171 nwritten = writev_recv(subreq, &err);
2172 TALLOC_FREE(subreq);
2173 if (nwritten == -1) {
2174 tevent_req_error(req, err);
2175 return;
2176 }
2177 }
2178
2179 int ctdbd_req_recv(
2180 struct tevent_req *req,
2181 TALLOC_CTX *mem_ctx,
2182 struct ctdb_req_header **reply)
2183 {
2184 struct ctdbd_req_state *state = tevent_req_data(
2185 req, struct ctdbd_req_state);
2186 int err;
2187
2188 if (tevent_req_is_unix_error(req, &err)) {
2189 return err;
2190 }
2191 *reply = talloc_move(mem_ctx, &state->reply);
2192 tevent_req_received(req);
2193 return 0;
2194 }
2195
2196 struct ctdbd_parse_state {
2197 struct tevent_context *ev;
2198 struct ctdbd_connection *conn;
2199 uint32_t reqid;
2200 TDB_DATA key;
2201 uint8_t _keybuf[64];
2202 struct ctdb_req_call_old ctdb_req;
2203 struct iovec iov[2];
2204 void (*parser)(TDB_DATA key,
2205 TDB_DATA data,
2206 void *private_data);
2207 void *private_data;
2208 };
2209
2210 static void ctdbd_parse_done(struct tevent_req *subreq);
2211
2212 struct tevent_req *ctdbd_parse_send(TALLOC_CTX *mem_ctx,
2213 struct tevent_context *ev,
2214 struct ctdbd_connection *conn,
2215 uint32_t db_id,
2216 TDB_DATA key,
2217 bool local_copy,
2218 void (*parser)(TDB_DATA key,
2219 TDB_DATA data,
2220 void *private_data),
2221 void *private_data,
2222 enum dbwrap_req_state *req_state)
2223 {
2224 struct tevent_req *req = NULL;
2225 struct ctdbd_parse_state *state = NULL;
2226 uint32_t flags;
2227 uint32_t packet_length;
2228 struct tevent_req *subreq = NULL;
2229
2230 req = tevent_req_create(mem_ctx, &state, struct ctdbd_parse_state);
2231 if (req == NULL) {
2232 *req_state = DBWRAP_REQ_ERROR;
2233 return NULL;
2234 }
2235
2236 *req_state = DBWRAP_REQ_DISPATCHED;
2237
2238 *state = (struct ctdbd_parse_state) {
2239 .ev = ev,
2240 .conn = conn,
2241 .reqid = ctdbd_next_reqid(conn),
2242 .parser = parser,
2243 .private_data = private_data,
2244 };
2245
2246 flags = local_copy ? CTDB_WANT_READONLY : 0;
2247 packet_length = offsetof(struct ctdb_req_call_old, data) + key.dsize;
2248
2249 /*
2250 * Copy the key into our state, as ctdb_pkt_send_cleanup() requires that
2251 * all passed iov elements have a lifetime longer that the tevent_req
2252 * returned by ctdb_pkt_send_send(). This is required continue sending a
2253 * the low level request into the ctdb socket, if a higher level
2254 * ('this') request is canceled (or talloc free'd) by the application
2255 * layer, without sending invalid packets to ctdb.
2256 */
2257 if (key.dsize > sizeof(state->_keybuf)) {
2258 state->key.dptr = talloc_memdup(state, key.dptr, key.dsize);
2259 if (tevent_req_nomem(state->key.dptr, req)) {
2260 return tevent_req_post(req, ev);
2261 }
2262 } else {
2263 memcpy(state->_keybuf, key.dptr, key.dsize);
2264 state->key.dptr = state->_keybuf;
2265 }
2266 state->key.dsize = key.dsize;
2267
2268 state->ctdb_req.hdr.length = packet_length;
2269 state->ctdb_req.hdr.ctdb_magic = CTDB_MAGIC;
2270 state->ctdb_req.hdr.ctdb_version = CTDB_PROTOCOL;
2271 state->ctdb_req.hdr.operation = CTDB_REQ_CALL;
2272 state->ctdb_req.hdr.reqid = state->reqid;
2273 state->ctdb_req.flags = flags;
2274 state->ctdb_req.callid = CTDB_FETCH_FUNC;
2275 state->ctdb_req.db_id = db_id;
2276 state->ctdb_req.keylen = state->key.dsize;
2277
2278 state->iov[0].iov_base = &state->ctdb_req;
2279 state->iov[0].iov_len = offsetof(struct ctdb_req_call_old, data);
2280 state->iov[1].iov_base = state->key.dptr;
2281 state->iov[1].iov_len = state->key.dsize;
2282
2283 subreq = ctdbd_req_send(
2284 state, ev, conn, state->iov, ARRAY_SIZE(state->iov));
2285 if (tevent_req_nomem(subreq, req)) {
2286 *req_state = DBWRAP_REQ_ERROR;
2287 return tevent_req_post(req, ev);
2288 }
2289 tevent_req_set_callback(subreq, ctdbd_parse_done, req);
2290
2291 return req;
2292 }
2293
2294 static void ctdbd_parse_done(struct tevent_req *subreq)
2295 {
2296 struct tevent_req *req = tevent_req_callback_data(
2297 subreq, struct tevent_req);
2298 struct ctdbd_parse_state *state = tevent_req_data(
2299 req, struct ctdbd_parse_state);
2300 struct ctdb_req_header *hdr = NULL;
2301 struct ctdb_reply_call_old *reply = NULL;
2302 int ret;
2303
2304 ret = ctdbd_req_recv(subreq, state, &hdr);
2305 TALLOC_FREE(subreq);
2306 if (tevent_req_error(req, ret)) {
2307 DBG_DEBUG("ctdb_req_recv failed %s\n", strerror(ret));
2308 return;
2309 }
2310 SMB_ASSERT(hdr != NULL);
2311
2312 if (hdr->operation != CTDB_REPLY_CALL) {
2313 DBG_ERR("received invalid reply\n");
2314 ctdb_packet_dump(hdr);
2315 tevent_req_error(req, EIO);
2316 return;
2317 }
2318
2319 reply = (struct ctdb_reply_call_old *)hdr;
2320
2321 if (reply->datalen == 0) {
2322 /*
2323 * Treat an empty record as non-existing
2324 */
2325 tevent_req_error(req, ENOENT);
2326 return;
2327 }
2328
2329 state->parser(state->key,
2330 make_tdb_data(&reply->data[0], reply->datalen),
2331 state->private_data);
2332
2333 tevent_req_done(req);
2334 return;
2335 }
2336
2337 int ctdbd_parse_recv(struct tevent_req *req)
2338 {
2339 return tevent_req_simple_recv_unix(req);
2340 }
Samba git mirror