4 Copyright (C) Andrew Tridgell 2007
5 Copyright (C) Ronnie Sahlberg 2007
7 This library is free software; you can redistribute it and/or
8 modify it under the terms of the GNU Lesser General Public
9 License as published by the Free Software Foundation; either
10 version 2 of the License, or (at your option) any later version.
12 This library 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 GNU
15 Lesser General Public License for more details.
17 You should have received a copy of the GNU Lesser General Public
18 License along with this library; if not, write to the Free Software
19 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
24 #include "lib/tdb/include/tdb.h"
25 #include "lib/events/events.h"
26 #include "lib/util/dlinklist.h"
27 #include "system/network.h"
28 #include "system/filesys.h"
29 #include "../include/ctdb.h"
30 #include "../include/ctdb_private.h"
33 queue a packet for sending from client to daemon
35 static int ctdb_client_queue_pkt(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
37 return ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)hdr, hdr->length);
42 handle a connect wait reply packet
44 static void ctdb_reply_connect_wait(struct ctdb_context *ctdb,
45 struct ctdb_req_header *hdr)
47 struct ctdb_reply_connect_wait *r = (struct ctdb_reply_connect_wait *)hdr;
48 ctdb->num_connected = r->num_connected;
52 called in the client when we receive a CTDB_REPLY_FETCH_LOCK from the daemon
54 This packet comes in response to a CTDB_REQ_FETCH_LOCK request packet. It
55 contains any reply data from the call
57 void ctdb_reply_fetch_lock(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
59 struct ctdb_reply_fetch_lock *c = (struct ctdb_reply_fetch_lock *)hdr;
60 struct ctdb_call_state *state;
62 state = idr_find(ctdb->idr, hdr->reqid);
63 if (state == NULL) return;
65 state->call.reply_data.dptr = c->data;
66 state->call.reply_data.dsize = c->datalen;
67 state->call.status = c->state;
69 talloc_steal(state, c);
71 /* get an extra reference here - this prevents the free in ctdb_recv_pkt()
72 from freeing the data */
73 (void)talloc_reference(state, c);
75 state->state = CTDB_CALL_DONE;
76 if (state->async.fn) {
77 state->async.fn(state);
82 called in the client when we receive a CTDB_REPLY_STORE_UNLOCK from the daemon
84 This packet comes in response to a CTDB_REQ_STORE_UNLOCK request packet. It
85 contains any reply data from the call
87 void ctdb_reply_store_unlock(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
89 struct ctdb_reply_store_unlock *c = (struct ctdb_reply_store_unlock *)hdr;
90 struct ctdb_call_state *state;
92 state = idr_find(ctdb->idr, hdr->reqid);
93 if (state == NULL) return;
95 state->call.status = c->state;
97 talloc_steal(state, c);
99 /* get an extra reference here - this prevents the free in ctdb_recv_pkt()
100 from freeing the data */
101 (void)talloc_reference(state, c);
103 state->state = CTDB_CALL_DONE;
104 if (state->async.fn) {
105 state->async.fn(state);
109 this is called in the client, when data comes in from the daemon
111 static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
113 struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context);
114 struct ctdb_req_header *hdr;
116 if (cnt < sizeof(*hdr)) {
117 ctdb_set_error(ctdb, "Bad packet length %d in client\n", cnt);
120 hdr = (struct ctdb_req_header *)data;
121 if (cnt != hdr->length) {
122 ctdb_set_error(ctdb, "Bad header length %d expected %d in client\n",
127 if (hdr->ctdb_magic != CTDB_MAGIC) {
128 ctdb_set_error(ctdb, "Non CTDB packet rejected in client\n");
132 if (hdr->ctdb_version != CTDB_VERSION) {
133 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected in client\n", hdr->ctdb_version);
137 switch (hdr->operation) {
138 case CTDB_REPLY_CALL:
139 ctdb_reply_call(ctdb, hdr);
142 case CTDB_REQ_MESSAGE:
143 ctdb_request_message(ctdb, hdr);
146 case CTDB_REPLY_CONNECT_WAIT:
147 ctdb_reply_connect_wait(ctdb, hdr);
150 case CTDB_REPLY_FETCH_LOCK:
151 ctdb_reply_fetch_lock(ctdb, hdr);
154 case CTDB_REPLY_STORE_UNLOCK:
155 ctdb_reply_store_unlock(ctdb, hdr);
159 printf("bogus operation code:%d\n",hdr->operation);
164 connect to a unix domain socket
166 static int ux_socket_connect(struct ctdb_context *ctdb)
168 struct sockaddr_un addr;
170 memset(&addr, 0, sizeof(addr));
171 addr.sun_family = AF_UNIX;
172 strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
174 ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
175 if (ctdb->daemon.sd == -1) {
179 if (connect(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
180 close(ctdb->daemon.sd);
181 ctdb->daemon.sd = -1;
185 ctdb->daemon.queue = ctdb_queue_setup(ctdb, ctdb, ctdb->daemon.sd,
187 ctdb_client_read_cb, ctdb);
194 make a recv call to the local ctdb daemon - called from client context
196 This is called when the program wants to wait for a ctdb_call to complete and get the
197 results. This call will block unless the call has already completed.
199 int ctdb_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
201 struct ctdb_record_handle *rec;
203 while (state->state < CTDB_CALL_DONE) {
204 event_loop_once(state->node->ctdb->ev);
206 if (state->state != CTDB_CALL_DONE) {
207 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
212 rec = state->fetch_private;
214 /* ugly hack to manage forced migration */
216 rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr);
217 rec->data->dsize = state->call.reply_data.dsize;
222 if (state->call.reply_data.dsize) {
223 call->reply_data.dptr = talloc_memdup(state->node->ctdb,
224 state->call.reply_data.dptr,
225 state->call.reply_data.dsize);
226 call->reply_data.dsize = state->call.reply_data.dsize;
228 call->reply_data.dptr = NULL;
229 call->reply_data.dsize = 0;
231 call->status = state->call.status;
241 destroy a ctdb_call in client
243 static int ctdb_client_call_destructor(struct ctdb_call_state *state)
245 idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
252 make a ctdb call to the local daemon - async send. Called from client context.
254 This constructs a ctdb_call request and queues it for processing.
255 This call never blocks.
257 struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db,
258 struct ctdb_call *call)
260 struct ctdb_call_state *state;
261 struct ctdb_context *ctdb = ctdb_db->ctdb;
262 struct ctdb_ltdb_header header;
267 /* if the domain socket is not yet open, open it */
268 if (ctdb->daemon.sd==-1) {
269 ux_socket_connect(ctdb);
272 ret = ctdb_ltdb_lock(ctdb_db, call->key);
274 printf("failed to lock ltdb record\n");
278 ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
280 ctdb_ltdb_unlock(ctdb_db, call->key);
285 if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
286 state = ctdb_call_local_send(ctdb_db, call, &header, &data);
287 ctdb_ltdb_unlock(ctdb_db, call->key);
292 state = talloc_zero(ctdb_db, struct ctdb_call_state);
294 printf("failed to allocate state\n");
295 ctdb_ltdb_unlock(ctdb_db, call->key);
299 talloc_steal(state, data.dptr);
301 len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
302 state->c = ctdbd_allocate_pkt(ctdb, len);
303 if (state->c == NULL) {
304 printf("failed to allocate packet\n");
305 ctdb_ltdb_unlock(ctdb_db, call->key);
308 talloc_set_name_const(state->c, "ctdbd req_call packet");
309 talloc_steal(state, state->c);
311 state->c->hdr.length = len;
312 state->c->hdr.ctdb_magic = CTDB_MAGIC;
313 state->c->hdr.ctdb_version = CTDB_VERSION;
314 state->c->hdr.operation = CTDB_REQ_CALL;
315 state->c->hdr.destnode = header.dmaster;
316 state->c->hdr.srcnode = ctdb->vnn;
317 /* this limits us to 16k outstanding messages - not unreasonable */
318 state->c->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
319 state->c->flags = call->flags;
320 state->c->db_id = ctdb_db->db_id;
321 state->c->callid = call->call_id;
322 state->c->keylen = call->key.dsize;
323 state->c->calldatalen = call->call_data.dsize;
324 memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
325 memcpy(&state->c->data[call->key.dsize],
326 call->call_data.dptr, call->call_data.dsize);
328 state->call.call_data.dptr = &state->c->data[call->key.dsize];
329 state->call.key.dptr = &state->c->data[0];
331 state->node = ctdb->nodes[header.dmaster];
332 state->state = CTDB_CALL_WAIT;
333 state->header = header;
334 state->ctdb_db = ctdb_db;
336 talloc_set_destructor(state, ctdb_client_call_destructor);
338 ctdb_client_queue_pkt(ctdb, &state->c->hdr);
340 /*XXX set up timeout to cleanup if server doesnt respond
341 event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
342 ctdb_call_timeout, state);
345 ctdb_ltdb_unlock(ctdb_db, call->key);
352 tell the daemon what messaging srvid we will use, and register the message
353 handler function in the client
355 int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid,
356 ctdb_message_fn_t handler,
360 struct ctdb_req_register c;
363 /* if the domain socket is not yet open, open it */
364 if (ctdb->daemon.sd==-1) {
365 ux_socket_connect(ctdb);
370 c.hdr.length = sizeof(c);
371 c.hdr.ctdb_magic = CTDB_MAGIC;
372 c.hdr.ctdb_version = CTDB_VERSION;
373 c.hdr.operation = CTDB_REQ_REGISTER;
376 res = ctdb_client_queue_pkt(ctdb, &c.hdr);
381 /* also need to register the handler with our ctdb structure */
382 return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private_data);
388 setup handler for receipt of ctdb messages from ctdb_send_message()
390 int ctdb_set_message_handler(struct ctdb_context *ctdb,
392 ctdb_message_fn_t handler,
395 if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
396 return ctdb_client_set_message_handler(ctdb, srvid, handler, private_data);
398 return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private_data);
403 send a message - from client context
405 int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn,
406 uint32_t srvid, TDB_DATA data)
408 struct ctdb_req_message *r;
411 len = offsetof(struct ctdb_req_message, data) + data.dsize;
412 r = ctdb->methods->allocate_pkt(ctdb, len);
413 CTDB_NO_MEMORY(ctdb, r);
414 talloc_set_name_const(r, "req_message packet");
417 r->hdr.ctdb_magic = CTDB_MAGIC;
418 r->hdr.ctdb_version = CTDB_VERSION;
419 r->hdr.operation = CTDB_REQ_MESSAGE;
420 r->hdr.destnode = vnn;
421 r->hdr.srcnode = ctdb->vnn;
424 r->datalen = data.dsize;
425 memcpy(&r->data[0], data.dptr, data.dsize);
427 res = ctdb_client_queue_pkt(ctdb, &r->hdr);
437 wait for all nodes to be connected - from client
439 static void ctdb_client_connect_wait(struct ctdb_context *ctdb)
441 struct ctdb_req_connect_wait r;
446 r.hdr.length = sizeof(r);
447 r.hdr.ctdb_magic = CTDB_MAGIC;
448 r.hdr.ctdb_version = CTDB_VERSION;
449 r.hdr.operation = CTDB_REQ_CONNECT_WAIT;
451 res = ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)&r.hdr, r.hdr.length);
453 printf("Failed to queue a connect wait request\n");
457 /* now we can go into the normal wait routine, as the reply packet
458 will update the ctdb->num_connected variable */
459 ctdb_daemon_connect_wait(ctdb);
463 wait for all nodes to be connected
465 void ctdb_connect_wait(struct ctdb_context *ctdb)
467 if (!(ctdb->flags & CTDB_FLAG_DAEMON_MODE)) {
468 ctdb_daemon_connect_wait(ctdb);
472 ctdb_client_connect_wait(ctdb);
476 struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db,
480 struct ctdb_call_state *state;
481 struct ctdb_context *ctdb = ctdb_db->ctdb;
482 struct ctdb_req_fetch_lock *req;
485 /* if the domain socket is not yet open, open it */
486 if (ctdb->daemon.sd==-1) {
487 ux_socket_connect(ctdb);
490 state = talloc_zero(ctdb_db, struct ctdb_call_state);
492 printf("failed to allocate state\n");
495 state->state = CTDB_CALL_WAIT;
496 state->ctdb_db = ctdb_db;
497 len = offsetof(struct ctdb_req_fetch_lock, key) + key.dsize;
498 state->c = ctdbd_allocate_pkt(ctdb, len);
499 if (state->c == NULL) {
500 printf("failed to allocate packet\n");
503 ZERO_STRUCT(*state->c);
504 talloc_set_name_const(state->c, "ctdbd req_fetch_lock packet");
505 talloc_steal(state, state->c);
507 req = (struct ctdb_req_fetch_lock *)state->c;
508 req->hdr.length = len;
509 req->hdr.ctdb_magic = CTDB_MAGIC;
510 req->hdr.ctdb_version = CTDB_VERSION;
511 req->hdr.operation = CTDB_REQ_FETCH_LOCK;
512 req->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
513 req->db_id = ctdb_db->db_id;
514 req->keylen = key.dsize;
515 memcpy(&req->key[0], key.dptr, key.dsize);
517 res = ctdb_client_queue_pkt(ctdb, &req->hdr);
528 struct ctdb_call_state *ctdb_client_store_unlock_send(
529 struct ctdb_record_handle *rh,
533 struct ctdb_call_state *state;
534 struct ctdb_db_context *ctdb_db = talloc_get_type(rh->ctdb_db, struct ctdb_db_context);
535 struct ctdb_context *ctdb = ctdb_db->ctdb;
536 struct ctdb_req_store_unlock *req;
539 /* if the domain socket is not yet open, open it */
540 if (ctdb->daemon.sd==-1) {
541 ux_socket_connect(ctdb);
544 state = talloc_zero(ctdb_db, struct ctdb_call_state);
546 printf("failed to allocate state\n");
549 state->state = CTDB_CALL_WAIT;
550 state->ctdb_db = ctdb_db;
551 len = offsetof(struct ctdb_req_store_unlock, data) + rh->key.dsize + data.dsize;
552 state->c = ctdbd_allocate_pkt(ctdb, len);
553 if (state->c == NULL) {
554 printf("failed to allocate packet\n");
557 ZERO_STRUCT(*state->c);
558 talloc_set_name_const(state->c, "ctdbd req_store_unlock packet");
559 talloc_steal(state, state->c);
561 req = (struct ctdb_req_store_unlock *)state->c;
562 req->hdr.length = len;
563 req->hdr.ctdb_magic = CTDB_MAGIC;
564 req->hdr.ctdb_version = CTDB_VERSION;
565 req->hdr.operation = CTDB_REQ_STORE_UNLOCK;
566 req->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
567 req->db_id = ctdb_db->db_id;
568 req->keylen = rh->key.dsize;
569 req->datalen = data.dsize;
570 memcpy(&req->data[0], rh->key.dptr, rh->key.dsize);
571 memcpy(&req->data[req->keylen], data.dptr, data.dsize);
573 res = ctdb_client_queue_pkt(ctdb, &req->hdr);
584 make a recv call to the local ctdb daemon - called from client context
586 This is called when the program wants to wait for a ctdb_fetch_lock to complete and get the
587 results. This call will block unless the call has already completed.
589 struct ctdb_record_handle *ctdb_client_fetch_lock_recv(struct ctdb_call_state *state, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data)
591 struct ctdb_record_handle *rec;
593 while (state->state < CTDB_CALL_DONE) {
594 event_loop_once(state->ctdb_db->ctdb->ev);
596 if (state->state != CTDB_CALL_DONE) {
597 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
602 rec = talloc(mem_ctx, struct ctdb_record_handle);
603 CTDB_NO_MEMORY_NULL(state->ctdb_db->ctdb, rec);
605 rec->ctdb_db = state->ctdb_db;
607 rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize);
608 rec->data = talloc(rec, TDB_DATA);
609 rec->data->dsize = state->call.reply_data.dsize;
610 rec->data->dptr = talloc_memdup(rec, state->call.reply_data.dptr, rec->data->dsize);
619 make a recv call to the local ctdb daemon - called from client context
621 This is called when the program wants to wait for a ctdb_store_unlock to complete and get the
622 results. This call will block unless the call has already completed.
624 int ctdb_client_store_unlock_recv(struct ctdb_call_state *state, struct ctdb_record_handle *rec)
626 while (state->state < CTDB_CALL_DONE) {
627 event_loop_once(state->ctdb_db->ctdb->ev);
629 if (state->state != CTDB_CALL_DONE) {
630 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
637 struct ctdb_record_handle *ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db,
642 struct ctdb_call_state *state;
643 struct ctdb_record_handle *rec;
645 state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key);
646 rec = ctdb_client_fetch_lock_recv(state, mem_ctx, key, data);
651 int ctdb_client_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data)
653 struct ctdb_call_state *state;
656 state = ctdb_client_store_unlock_send(rec, rec, data);
657 res = ctdb_client_store_unlock_recv(state, rec);