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 this is called in the client, when data comes in from the daemon
84 static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
86 struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context);
87 struct ctdb_req_header *hdr;
89 if (cnt < sizeof(*hdr)) {
90 ctdb_set_error(ctdb, "Bad packet length %d in client\n", cnt);
93 hdr = (struct ctdb_req_header *)data;
94 if (cnt != hdr->length) {
95 ctdb_set_error(ctdb, "Bad header length %d expected %d in client\n",
100 if (hdr->ctdb_magic != CTDB_MAGIC) {
101 ctdb_set_error(ctdb, "Non CTDB packet rejected in client\n");
105 if (hdr->ctdb_version != CTDB_VERSION) {
106 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected in client\n", hdr->ctdb_version);
110 switch (hdr->operation) {
111 case CTDB_REPLY_CALL:
112 ctdb_reply_call(ctdb, hdr);
115 case CTDB_REQ_MESSAGE:
116 ctdb_request_message(ctdb, hdr);
119 case CTDB_REPLY_CONNECT_WAIT:
120 ctdb_reply_connect_wait(ctdb, hdr);
123 case CTDB_REPLY_FETCH_LOCK:
124 ctdb_reply_fetch_lock(ctdb, hdr);
128 printf("bogus operation code:%d\n",hdr->operation);
133 connect to a unix domain socket
135 static int ux_socket_connect(struct ctdb_context *ctdb)
137 struct sockaddr_un addr;
139 memset(&addr, 0, sizeof(addr));
140 addr.sun_family = AF_UNIX;
141 strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
143 ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
144 if (ctdb->daemon.sd == -1) {
148 if (connect(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
149 close(ctdb->daemon.sd);
150 ctdb->daemon.sd = -1;
154 ctdb->daemon.queue = ctdb_queue_setup(ctdb, ctdb, ctdb->daemon.sd,
156 ctdb_client_read_cb, ctdb);
163 make a recv call to the local ctdb daemon - called from client context
165 This is called when the program wants to wait for a ctdb_call to complete and get the
166 results. This call will block unless the call has already completed.
168 int ctdb_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
170 struct ctdb_record_handle *rec;
172 while (state->state < CTDB_CALL_DONE) {
173 event_loop_once(state->node->ctdb->ev);
175 if (state->state != CTDB_CALL_DONE) {
176 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
181 rec = state->fetch_private;
183 /* ugly hack to manage forced migration */
185 rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr);
186 rec->data->dsize = state->call.reply_data.dsize;
191 if (state->call.reply_data.dsize) {
192 call->reply_data.dptr = talloc_memdup(state->node->ctdb,
193 state->call.reply_data.dptr,
194 state->call.reply_data.dsize);
195 call->reply_data.dsize = state->call.reply_data.dsize;
197 call->reply_data.dptr = NULL;
198 call->reply_data.dsize = 0;
200 call->status = state->call.status;
210 destroy a ctdb_call in client
212 static int ctdb_client_call_destructor(struct ctdb_call_state *state)
214 idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
221 make a ctdb call to the local daemon - async send. Called from client context.
223 This constructs a ctdb_call request and queues it for processing.
224 This call never blocks.
226 struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db,
227 struct ctdb_call *call)
229 struct ctdb_call_state *state;
230 struct ctdb_context *ctdb = ctdb_db->ctdb;
231 struct ctdb_ltdb_header header;
236 /* if the domain socket is not yet open, open it */
237 if (ctdb->daemon.sd==-1) {
238 ux_socket_connect(ctdb);
241 ret = ctdb_ltdb_lock(ctdb_db, call->key);
243 printf("failed to lock ltdb record\n");
247 ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
249 ctdb_ltdb_unlock(ctdb_db, call->key);
254 if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
255 state = ctdb_call_local_send(ctdb_db, call, &header, &data);
256 ctdb_ltdb_unlock(ctdb_db, call->key);
261 state = talloc_zero(ctdb_db, struct ctdb_call_state);
263 printf("failed to allocate state\n");
264 ctdb_ltdb_unlock(ctdb_db, call->key);
268 talloc_steal(state, data.dptr);
270 len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
271 state->c = ctdbd_allocate_pkt(ctdb, len);
272 if (state->c == NULL) {
273 printf("failed to allocate packet\n");
274 ctdb_ltdb_unlock(ctdb_db, call->key);
277 talloc_set_name_const(state->c, "ctdbd req_call packet");
278 talloc_steal(state, state->c);
280 state->c->hdr.length = len;
281 state->c->hdr.ctdb_magic = CTDB_MAGIC;
282 state->c->hdr.ctdb_version = CTDB_VERSION;
283 state->c->hdr.operation = CTDB_REQ_CALL;
284 state->c->hdr.destnode = header.dmaster;
285 state->c->hdr.srcnode = ctdb->vnn;
286 /* this limits us to 16k outstanding messages - not unreasonable */
287 state->c->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
288 state->c->flags = call->flags;
289 state->c->db_id = ctdb_db->db_id;
290 state->c->callid = call->call_id;
291 state->c->keylen = call->key.dsize;
292 state->c->calldatalen = call->call_data.dsize;
293 memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
294 memcpy(&state->c->data[call->key.dsize],
295 call->call_data.dptr, call->call_data.dsize);
297 state->call.call_data.dptr = &state->c->data[call->key.dsize];
298 state->call.key.dptr = &state->c->data[0];
300 state->node = ctdb->nodes[header.dmaster];
301 state->state = CTDB_CALL_WAIT;
302 state->header = header;
303 state->ctdb_db = ctdb_db;
305 talloc_set_destructor(state, ctdb_client_call_destructor);
307 ctdb_client_queue_pkt(ctdb, &state->c->hdr);
309 /*XXX set up timeout to cleanup if server doesnt respond
310 event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
311 ctdb_call_timeout, state);
314 ctdb_ltdb_unlock(ctdb_db, call->key);
321 tell the daemon what messaging srvid we will use, and register the message
322 handler function in the client
324 int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid,
325 ctdb_message_fn_t handler,
329 struct ctdb_req_register c;
332 /* if the domain socket is not yet open, open it */
333 if (ctdb->daemon.sd==-1) {
334 ux_socket_connect(ctdb);
339 c.hdr.length = sizeof(c);
340 c.hdr.ctdb_magic = CTDB_MAGIC;
341 c.hdr.ctdb_version = CTDB_VERSION;
342 c.hdr.operation = CTDB_REQ_REGISTER;
345 res = ctdb_client_queue_pkt(ctdb, &c.hdr);
350 /* also need to register the handler with our ctdb structure */
351 return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private_data);
357 setup handler for receipt of ctdb messages from ctdb_send_message()
359 int ctdb_set_message_handler(struct ctdb_context *ctdb,
361 ctdb_message_fn_t handler,
364 if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
365 return ctdb_client_set_message_handler(ctdb, srvid, handler, private_data);
367 return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private_data);
372 send a message - from client context
374 int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn,
375 uint32_t srvid, TDB_DATA data)
377 struct ctdb_req_message *r;
380 len = offsetof(struct ctdb_req_message, data) + data.dsize;
381 r = ctdb->methods->allocate_pkt(ctdb, len);
382 CTDB_NO_MEMORY(ctdb, r);
383 talloc_set_name_const(r, "req_message packet");
386 r->hdr.ctdb_magic = CTDB_MAGIC;
387 r->hdr.ctdb_version = CTDB_VERSION;
388 r->hdr.operation = CTDB_REQ_MESSAGE;
389 r->hdr.destnode = vnn;
390 r->hdr.srcnode = ctdb->vnn;
393 r->datalen = data.dsize;
394 memcpy(&r->data[0], data.dptr, data.dsize);
396 res = ctdb_client_queue_pkt(ctdb, &r->hdr);
406 wait for all nodes to be connected - from client
408 static void ctdb_client_connect_wait(struct ctdb_context *ctdb)
410 struct ctdb_req_connect_wait r;
415 r.hdr.length = sizeof(r);
416 r.hdr.ctdb_magic = CTDB_MAGIC;
417 r.hdr.ctdb_version = CTDB_VERSION;
418 r.hdr.operation = CTDB_REQ_CONNECT_WAIT;
420 res = ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)&r.hdr, r.hdr.length);
422 printf("Failed to queue a connect wait request\n");
426 /* now we can go into the normal wait routine, as the reply packet
427 will update the ctdb->num_connected variable */
428 ctdb_daemon_connect_wait(ctdb);
432 wait for all nodes to be connected
434 void ctdb_connect_wait(struct ctdb_context *ctdb)
436 if (!(ctdb->flags & CTDB_FLAG_DAEMON_MODE)) {
437 ctdb_daemon_connect_wait(ctdb);
441 ctdb_client_connect_wait(ctdb);
445 struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db,
449 struct ctdb_call_state *state;
450 struct ctdb_context *ctdb = ctdb_db->ctdb;
451 struct ctdb_req_fetch_lock *req;
454 /* if the domain socket is not yet open, open it */
455 if (ctdb->daemon.sd==-1) {
456 ux_socket_connect(ctdb);
459 state = talloc_zero(ctdb_db, struct ctdb_call_state);
461 printf("failed to allocate state\n");
464 state->state = CTDB_CALL_WAIT;
465 state->ctdb_db = ctdb_db;
466 len = offsetof(struct ctdb_req_fetch_lock, key) + key.dsize;
467 state->c = ctdbd_allocate_pkt(ctdb, len);
468 if (state->c == NULL) {
469 printf("failed to allocate packet\n");
472 bzero(state->c, len);
473 talloc_set_name_const(state->c, "ctdbd req_fetch_lock packet");
474 talloc_steal(state, state->c);
476 req = (struct ctdb_req_fetch_lock *)state->c;
477 req->hdr.length = len;
478 req->hdr.ctdb_magic = CTDB_MAGIC;
479 req->hdr.ctdb_version = CTDB_VERSION;
480 req->hdr.operation = CTDB_REQ_FETCH_LOCK;
481 req->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
482 req->db_id = ctdb_db->db_id;
483 req->keylen = key.dsize;
484 memcpy(&req->key[0], key.dptr, key.dsize);
486 res = ctdb_client_queue_pkt(ctdb, &req->hdr);
498 make a recv call to the local ctdb daemon - called from client context
500 This is called when the program wants to wait for a ctdb_fetch_lock to complete and get the
501 results. This call will block unless the call has already completed.
503 int ctdb_client_fetch_lock_recv(struct ctdb_call_state *state, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data)
505 while (state->state < CTDB_CALL_DONE) {
506 event_loop_once(state->ctdb_db->ctdb->ev);
508 if (state->state != CTDB_CALL_DONE) {
509 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
514 data->dsize = state->call.reply_data.dsize;
515 data->dptr = talloc_memdup(mem_ctx, state->call.reply_data.dptr, data->dsize);
520 int ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db,
525 struct ctdb_ltdb_header header;
528 ret = ctdb_ltdb_lock(ctdb_db, key);
530 printf("failed to lock ltdb record\n");
531 return FETCH_LOCK_LOCKFAILED;
534 ret = ctdb_ltdb_fetch(ctdb_db, key, &header, ctdb_db, data);
536 ctdb_ltdb_unlock(ctdb_db, key);
537 return FETCH_LOCK_FETCHFAILED;
541 if (header.dmaster != ctdb_db->ctdb->vnn) {
542 struct ctdb_call_state *state;
544 state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key);
545 ret = ctdb_client_fetch_lock_recv(state, mem_ctx, key, data);
547 ctdb_ltdb_unlock(ctdb_db, key);
548 return FETCH_LOCK_DMASTERFAILED;
556 a helper function for the client that will store the new data for the
557 record and release the tdb chainlock
559 int ctdb_client_store_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key, TDB_DATA data)
562 struct ctdb_ltdb_header header;
564 /* should be avoided if possible hang header off rec ? */
565 ret = ctdb_ltdb_fetch(ctdb_db, key, &header, NULL, NULL);
567 ctdb_set_error(ctdb_db->ctdb, "Fetch of locally held record failed");
571 ret = ctdb_ltdb_store(ctdb_db, key, &header, data);
573 ctdb_ltdb_unlock(ctdb_db, key);