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 this is called in the client, when data comes in from the daemon
54 static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
56 struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context);
57 struct ctdb_req_header *hdr;
59 if (cnt < sizeof(*hdr)) {
60 ctdb_set_error(ctdb, "Bad packet length %d\n", cnt);
63 hdr = (struct ctdb_req_header *)data;
64 if (cnt != hdr->length) {
65 ctdb_set_error(ctdb, "Bad header length %d expected %d\n",
70 if (hdr->ctdb_magic != CTDB_MAGIC) {
71 ctdb_set_error(ctdb, "Non CTDB packet rejected\n");
75 if (hdr->ctdb_version != CTDB_VERSION) {
76 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
80 switch (hdr->operation) {
82 ctdb_reply_call(ctdb, hdr);
85 case CTDB_REQ_MESSAGE:
86 ctdb_request_message(ctdb, hdr);
89 case CTDB_REPLY_CONNECT_WAIT:
90 ctdb_reply_connect_wait(ctdb, hdr);
96 connect to a unix domain socket
98 static int ux_socket_connect(struct ctdb_context *ctdb)
100 struct sockaddr_un addr;
102 memset(&addr, 0, sizeof(addr));
103 addr.sun_family = AF_UNIX;
104 strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
106 ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
107 if (ctdb->daemon.sd == -1) {
111 if (connect(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
112 close(ctdb->daemon.sd);
113 ctdb->daemon.sd = -1;
117 ctdb->daemon.queue = ctdb_queue_setup(ctdb, ctdb, ctdb->daemon.sd,
119 ctdb_client_read_cb, ctdb);
126 make a recv call to the local ctdb daemon - called from client context
128 This is called when the program wants to wait for a ctdb_call to complete and get the
129 results. This call will block unless the call has already completed.
131 int ctdb_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
133 struct ctdb_record_handle *rec;
135 while (state->state < CTDB_CALL_DONE) {
136 event_loop_once(state->node->ctdb->ev);
138 if (state->state != CTDB_CALL_DONE) {
139 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
144 rec = state->fetch_private;
146 /* ugly hack to manage forced migration */
148 rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr);
149 rec->data->dsize = state->call.reply_data.dsize;
154 if (state->call.reply_data.dsize) {
155 call->reply_data.dptr = talloc_memdup(state->node->ctdb,
156 state->call.reply_data.dptr,
157 state->call.reply_data.dsize);
158 call->reply_data.dsize = state->call.reply_data.dsize;
160 call->reply_data.dptr = NULL;
161 call->reply_data.dsize = 0;
163 call->status = state->call.status;
173 destroy a ctdb_call in client
175 static int ctdb_client_call_destructor(struct ctdb_call_state *state)
177 idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
184 make a ctdb call to the local daemon - async send. Called from client context.
186 This constructs a ctdb_call request and queues it for processing.
187 This call never blocks.
189 struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db,
190 struct ctdb_call *call)
192 struct ctdb_call_state *state;
193 struct ctdb_context *ctdb = ctdb_db->ctdb;
194 struct ctdb_ltdb_header header;
199 /* if the domain socket is not yet open, open it */
200 if (ctdb->daemon.sd==-1) {
201 ux_socket_connect(ctdb);
204 ret = ctdb_ltdb_lock(ctdb_db, call->key);
206 printf("failed to lock ltdb record\n");
210 ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
212 ctdb_ltdb_unlock(ctdb_db, call->key);
217 if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
218 state = ctdb_call_local_send(ctdb_db, call, &header, &data);
219 ctdb_ltdb_unlock(ctdb_db, call->key);
224 state = talloc_zero(ctdb_db, struct ctdb_call_state);
226 printf("failed to allocate state\n");
227 ctdb_ltdb_unlock(ctdb_db, call->key);
231 talloc_steal(state, data.dptr);
233 len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
234 state->c = ctdbd_allocate_pkt(ctdb, len);
235 if (state->c == NULL) {
236 printf("failed to allocate packet\n");
237 ctdb_ltdb_unlock(ctdb_db, call->key);
240 talloc_set_name_const(state->c, "ctdbd req_call packet");
241 talloc_steal(state, state->c);
243 state->c->hdr.length = len;
244 state->c->hdr.ctdb_magic = CTDB_MAGIC;
245 state->c->hdr.ctdb_version = CTDB_VERSION;
246 state->c->hdr.operation = CTDB_REQ_CALL;
247 state->c->hdr.destnode = header.dmaster;
248 state->c->hdr.srcnode = ctdb->vnn;
249 /* this limits us to 16k outstanding messages - not unreasonable */
250 state->c->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF);
251 state->c->flags = call->flags;
252 state->c->db_id = ctdb_db->db_id;
253 state->c->callid = call->call_id;
254 state->c->keylen = call->key.dsize;
255 state->c->calldatalen = call->call_data.dsize;
256 memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
257 memcpy(&state->c->data[call->key.dsize],
258 call->call_data.dptr, call->call_data.dsize);
260 state->call.call_data.dptr = &state->c->data[call->key.dsize];
261 state->call.key.dptr = &state->c->data[0];
263 state->node = ctdb->nodes[header.dmaster];
264 state->state = CTDB_CALL_WAIT;
265 state->header = header;
266 state->ctdb_db = ctdb_db;
268 talloc_set_destructor(state, ctdb_client_call_destructor);
270 ctdb_client_queue_pkt(ctdb, &state->c->hdr);
272 /*XXX set up timeout to cleanup if server doesnt respond
273 event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
274 ctdb_call_timeout, state);
277 ctdb_ltdb_unlock(ctdb_db, call->key);
284 tell the daemon what messaging srvid we will use, and register the message
285 handler function in the client
287 int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid,
288 ctdb_message_fn_t handler,
292 struct ctdb_req_register c;
295 /* if the domain socket is not yet open, open it */
296 if (ctdb->daemon.sd==-1) {
297 ux_socket_connect(ctdb);
302 c.hdr.length = sizeof(c);
303 c.hdr.ctdb_magic = CTDB_MAGIC;
304 c.hdr.ctdb_version = CTDB_VERSION;
305 c.hdr.operation = CTDB_REQ_REGISTER;
308 res = ctdb_client_queue_pkt(ctdb, &c.hdr);
313 /* also need to register the handler with our ctdb structure */
314 return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private);
320 setup handler for receipt of ctdb messages from ctdb_send_message()
322 int ctdb_set_message_handler(struct ctdb_context *ctdb,
324 ctdb_message_fn_t handler,
327 if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) {
328 return ctdb_client_set_message_handler(ctdb, srvid, handler, private);
330 return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private);
335 send a message - from client context
337 int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn,
338 uint32_t srvid, TDB_DATA data)
340 struct ctdb_req_message *r;
343 len = offsetof(struct ctdb_req_message, data) + data.dsize;
344 r = ctdb->methods->allocate_pkt(ctdb, len);
345 CTDB_NO_MEMORY(ctdb, r);
346 talloc_set_name_const(r, "req_message packet");
349 r->hdr.ctdb_magic = CTDB_MAGIC;
350 r->hdr.ctdb_version = CTDB_VERSION;
351 r->hdr.operation = CTDB_REQ_MESSAGE;
352 r->hdr.destnode = vnn;
353 r->hdr.srcnode = ctdb->vnn;
356 r->datalen = data.dsize;
357 memcpy(&r->data[0], data.dptr, data.dsize);
359 res = ctdb_client_queue_pkt(ctdb, &r->hdr);
369 wait for all nodes to be connected - from client
371 static void ctdb_client_connect_wait(struct ctdb_context *ctdb)
373 struct ctdb_req_connect_wait r;
378 r.hdr.length = sizeof(r);
379 r.hdr.ctdb_magic = CTDB_MAGIC;
380 r.hdr.ctdb_version = CTDB_VERSION;
381 r.hdr.operation = CTDB_REQ_CONNECT_WAIT;
383 res = ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)&r.hdr, r.hdr.length);
385 printf("Failed to queue a connect wait request\n");
389 /* now we can go into the normal wait routine, as the reply packet
390 will update the ctdb->num_connected variable */
391 ctdb_daemon_connect_wait(ctdb);
395 wait for all nodes to be connected
397 void ctdb_connect_wait(struct ctdb_context *ctdb)
399 if (!(ctdb->flags & CTDB_FLAG_DAEMON_MODE)) {
400 ctdb_daemon_connect_wait(ctdb);
404 ctdb_client_connect_wait(ctdb);