Merge branch 'master' of git://git.samba.org/tridge/ctdb
[metze/ctdb/wip.git] / server / ctdb_daemon.c
1 /* 
2    ctdb daemon code
3
4    Copyright (C) Andrew Tridgell  2006
5
6    This program is free software; you can redistribute it and/or modify
7    it under the terms of the GNU General Public License as published by
8    the Free Software Foundation; either version 3 of the License, or
9    (at your option) any later version.
10    
11    This program is distributed in the hope that it will be useful,
12    but WITHOUT ANY WARRANTY; without even the implied warranty of
13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14    GNU General Public License for more details.
15    
16    You should have received a copy of the GNU General Public License
17    along with this program; if not, see <http://www.gnu.org/licenses/>.
18 */
19
20 #include "includes.h"
21 #include "db_wrap.h"
22 #include "lib/tdb/include/tdb.h"
23 #include "lib/events/events.h"
24 #include "lib/util/dlinklist.h"
25 #include "system/network.h"
26 #include "system/filesys.h"
27 #include "system/wait.h"
28 #include "../include/ctdb.h"
29 #include "../include/ctdb_private.h"
30
31 static void daemon_incoming_packet(void *, struct ctdb_req_header *);
32
33 /*
34   handler for when a node changes its flags
35 */
36 static void flag_change_handler(struct ctdb_context *ctdb, uint64_t srvid, 
37                                 TDB_DATA data, void *private_data)
38 {
39         struct ctdb_node_flag_change *c = (struct ctdb_node_flag_change *)data.dptr;
40
41         if (data.dsize != sizeof(*c) || !ctdb_validate_pnn(ctdb, c->pnn)) {
42                 DEBUG(DEBUG_CRIT,(__location__ "Invalid data in ctdb_node_flag_change\n"));
43                 return;
44         }
45
46         if (!ctdb_validate_pnn(ctdb, c->pnn)) {
47                 DEBUG(DEBUG_CRIT,("Bad pnn %u in flag_change_handler\n", c->pnn));
48                 return;
49         }
50
51         /* don't get the disconnected flag from the other node */
52         ctdb->nodes[c->pnn]->flags = 
53                 (ctdb->nodes[c->pnn]->flags&NODE_FLAGS_DISCONNECTED) 
54                 | (c->new_flags & ~NODE_FLAGS_DISCONNECTED);    
55         DEBUG(DEBUG_INFO,("Node flags for node %u are now 0x%x\n", c->pnn, ctdb->nodes[c->pnn]->flags));
56
57         /* make sure we don't hold any IPs when we shouldn't */
58         if (c->pnn == ctdb->pnn &&
59             (ctdb->nodes[c->pnn]->flags & (NODE_FLAGS_INACTIVE|NODE_FLAGS_BANNED))) {
60                 ctdb_release_all_ips(ctdb);
61         }
62 }
63
64 static void print_exit_message(void)
65 {
66         DEBUG(DEBUG_NOTICE,("CTDB daemon shutting down\n"));
67 }
68
69
70 /* called when the "startup" event script has finished */
71 static void ctdb_start_transport(struct ctdb_context *ctdb)
72 {
73         /* start the transport running */
74         if (ctdb->methods->start(ctdb) != 0) {
75                 DEBUG(DEBUG_ALERT,("transport failed to start!\n"));
76                 ctdb_fatal(ctdb, "transport failed to start");
77         }
78
79         /* start the recovery daemon process */
80         if (ctdb_start_recoverd(ctdb) != 0) {
81                 DEBUG(DEBUG_ALERT,("Failed to start recovery daemon\n"));
82                 exit(11);
83         }
84
85         /* Make sure we log something when the daemon terminates */
86         atexit(print_exit_message);
87
88         /* a handler for when nodes are disabled/enabled */
89         ctdb_register_message_handler(ctdb, ctdb, CTDB_SRVID_NODE_FLAGS_CHANGED, 
90                                       flag_change_handler, NULL);
91
92         /* start monitoring for connected/disconnected nodes */
93         ctdb_start_keepalive(ctdb);
94
95         /* start monitoring for node health */
96         ctdb_start_monitoring(ctdb);
97
98         /* start periodic update of tcp tickle lists */
99         ctdb_start_tcp_tickle_update(ctdb);
100 }
101
102 static void block_signal(int signum)
103 {
104         struct sigaction act;
105
106         memset(&act, 0, sizeof(act));
107
108         act.sa_handler = SIG_IGN;
109         sigemptyset(&act.sa_mask);
110         sigaddset(&act.sa_mask, signum);
111         sigaction(signum, &act, NULL);
112 }
113
114
115 /*
116   send a packet to a client
117  */
118 static int daemon_queue_send(struct ctdb_client *client, struct ctdb_req_header *hdr)
119 {
120         client->ctdb->statistics.client_packets_sent++;
121         return ctdb_queue_send(client->queue, (uint8_t *)hdr, hdr->length);
122 }
123
124 /*
125   message handler for when we are in daemon mode. This redirects the message
126   to the right client
127  */
128 static void daemon_message_handler(struct ctdb_context *ctdb, uint64_t srvid, 
129                                     TDB_DATA data, void *private_data)
130 {
131         struct ctdb_client *client = talloc_get_type(private_data, struct ctdb_client);
132         struct ctdb_req_message *r;
133         int len;
134
135         /* construct a message to send to the client containing the data */
136         len = offsetof(struct ctdb_req_message, data) + data.dsize;
137         r = ctdbd_allocate_pkt(ctdb, ctdb, CTDB_REQ_MESSAGE, 
138                                len, struct ctdb_req_message);
139         CTDB_NO_MEMORY_VOID(ctdb, r);
140
141         talloc_set_name_const(r, "req_message packet");
142
143         r->srvid         = srvid;
144         r->datalen       = data.dsize;
145         memcpy(&r->data[0], data.dptr, data.dsize);
146
147         daemon_queue_send(client, &r->hdr);
148
149         talloc_free(r);
150 }
151                                            
152
153 /*
154   this is called when the ctdb daemon received a ctdb request to 
155   set the srvid from the client
156  */
157 int daemon_register_message_handler(struct ctdb_context *ctdb, uint32_t client_id, uint64_t srvid)
158 {
159         struct ctdb_client *client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client);
160         int res;
161         if (client == NULL) {
162                 DEBUG(DEBUG_ERR,("Bad client_id in daemon_request_register_message_handler\n"));
163                 return -1;
164         }
165         res = ctdb_register_message_handler(ctdb, client, srvid, daemon_message_handler, client);
166         if (res != 0) {
167                 DEBUG(DEBUG_ERR,(__location__ " Failed to register handler %llu in daemon\n", 
168                          (unsigned long long)srvid));
169         } else {
170                 DEBUG(DEBUG_INFO,(__location__ " Registered message handler for srvid=%llu\n", 
171                          (unsigned long long)srvid));
172         }
173
174         /* this is a hack for Samba - we now know the pid of the Samba client */
175         if ((srvid & 0xFFFFFFFF) == srvid &&
176             kill(srvid, 0) == 0) {
177                 client->pid = srvid;
178                 DEBUG(DEBUG_INFO,(__location__ " Registered PID %u for client %u\n",
179                          (unsigned)client->pid, client_id));
180         }
181         return res;
182 }
183
184 /*
185   this is called when the ctdb daemon received a ctdb request to 
186   remove a srvid from the client
187  */
188 int daemon_deregister_message_handler(struct ctdb_context *ctdb, uint32_t client_id, uint64_t srvid)
189 {
190         struct ctdb_client *client = ctdb_reqid_find(ctdb, client_id, struct ctdb_client);
191         if (client == NULL) {
192                 DEBUG(DEBUG_ERR,("Bad client_id in daemon_request_deregister_message_handler\n"));
193                 return -1;
194         }
195         return ctdb_deregister_message_handler(ctdb, srvid, client);
196 }
197
198
199 /*
200   destroy a ctdb_client
201 */
202 static int ctdb_client_destructor(struct ctdb_client *client)
203 {
204         ctdb_takeover_client_destructor_hook(client);
205         ctdb_reqid_remove(client->ctdb, client->client_id);
206         client->ctdb->statistics.num_clients--;
207         return 0;
208 }
209
210
211 /*
212   this is called when the ctdb daemon received a ctdb request message
213   from a local client over the unix domain socket
214  */
215 static void daemon_request_message_from_client(struct ctdb_client *client, 
216                                                struct ctdb_req_message *c)
217 {
218         TDB_DATA data;
219         int res;
220
221         /* maybe the message is for another client on this node */
222         if (ctdb_get_pnn(client->ctdb)==c->hdr.destnode) {
223                 ctdb_request_message(client->ctdb, (struct ctdb_req_header *)c);
224                 return;
225         }
226
227         /* its for a remote node */
228         data.dptr = &c->data[0];
229         data.dsize = c->datalen;
230         res = ctdb_daemon_send_message(client->ctdb, c->hdr.destnode,
231                                        c->srvid, data);
232         if (res != 0) {
233                 DEBUG(DEBUG_ERR,(__location__ " Failed to send message to remote node %u\n",
234                          c->hdr.destnode));
235         }
236 }
237
238
239 struct daemon_call_state {
240         struct ctdb_client *client;
241         uint32_t reqid;
242         struct ctdb_call *call;
243         struct timeval start_time;
244 };
245
246 /* 
247    complete a call from a client 
248 */
249 static void daemon_call_from_client_callback(struct ctdb_call_state *state)
250 {
251         struct daemon_call_state *dstate = talloc_get_type(state->async.private_data, 
252                                                            struct daemon_call_state);
253         struct ctdb_reply_call *r;
254         int res;
255         uint32_t length;
256         struct ctdb_client *client = dstate->client;
257
258         talloc_steal(client, dstate);
259         talloc_steal(dstate, dstate->call);
260
261         res = ctdb_daemon_call_recv(state, dstate->call);
262         if (res != 0) {
263                 DEBUG(DEBUG_ERR, (__location__ " ctdbd_call_recv() returned error\n"));
264                 client->ctdb->statistics.pending_calls--;
265                 ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
266                 return;
267         }
268
269         length = offsetof(struct ctdb_reply_call, data) + dstate->call->reply_data.dsize;
270         r = ctdbd_allocate_pkt(client->ctdb, dstate, CTDB_REPLY_CALL, 
271                                length, struct ctdb_reply_call);
272         if (r == NULL) {
273                 DEBUG(DEBUG_ERR, (__location__ " Failed to allocate reply_call in ctdb daemon\n"));
274                 client->ctdb->statistics.pending_calls--;
275                 ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
276                 return;
277         }
278         r->hdr.reqid        = dstate->reqid;
279         r->datalen          = dstate->call->reply_data.dsize;
280         memcpy(&r->data[0], dstate->call->reply_data.dptr, r->datalen);
281
282         res = daemon_queue_send(client, &r->hdr);
283         if (res != 0) {
284                 DEBUG(DEBUG_ERR, (__location__ " Failed to queue packet from daemon to client\n"));
285         }
286         ctdb_latency(&client->ctdb->statistics.max_call_latency, dstate->start_time);
287         talloc_free(dstate);
288         client->ctdb->statistics.pending_calls--;
289 }
290
291 struct ctdb_daemon_packet_wrap {
292         struct ctdb_context *ctdb;
293         uint32_t client_id;
294 };
295
296 /*
297   a wrapper to catch disconnected clients
298  */
299 static void daemon_incoming_packet_wrap(void *p, struct ctdb_req_header *hdr)
300 {
301         struct ctdb_client *client;
302         struct ctdb_daemon_packet_wrap *w = talloc_get_type(p, 
303                                                             struct ctdb_daemon_packet_wrap);
304         if (w == NULL) {
305                 DEBUG(DEBUG_CRIT,(__location__ " Bad packet type '%s'\n", talloc_get_name(p)));
306                 return;
307         }
308
309         client = ctdb_reqid_find(w->ctdb, w->client_id, struct ctdb_client);
310         if (client == NULL) {
311                 DEBUG(DEBUG_ERR,(__location__ " Packet for disconnected client %u\n",
312                          w->client_id));
313                 talloc_free(w);
314                 return;
315         }
316         talloc_free(w);
317
318         /* process it */
319         daemon_incoming_packet(client, hdr);    
320 }
321
322
323 /*
324   this is called when the ctdb daemon received a ctdb request call
325   from a local client over the unix domain socket
326  */
327 static void daemon_request_call_from_client(struct ctdb_client *client, 
328                                             struct ctdb_req_call *c)
329 {
330         struct ctdb_call_state *state;
331         struct ctdb_db_context *ctdb_db;
332         struct daemon_call_state *dstate;
333         struct ctdb_call *call;
334         struct ctdb_ltdb_header header;
335         TDB_DATA key, data;
336         int ret;
337         struct ctdb_context *ctdb = client->ctdb;
338         struct ctdb_daemon_packet_wrap *w;
339
340         ctdb->statistics.total_calls++;
341         ctdb->statistics.pending_calls++;
342
343         ctdb_db = find_ctdb_db(client->ctdb, c->db_id);
344         if (!ctdb_db) {
345                 DEBUG(DEBUG_ERR, (__location__ " Unknown database in request. db_id==0x%08x",
346                           c->db_id));
347                 ctdb->statistics.pending_calls--;
348                 return;
349         }
350
351         key.dptr = c->data;
352         key.dsize = c->keylen;
353
354         w = talloc(ctdb, struct ctdb_daemon_packet_wrap);
355         CTDB_NO_MEMORY_VOID(ctdb, w);   
356
357         w->ctdb = ctdb;
358         w->client_id = client->client_id;
359
360         ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, key, &header, 
361                                            (struct ctdb_req_header *)c, &data,
362                                            daemon_incoming_packet_wrap, w, True);
363         if (ret == -2) {
364                 /* will retry later */
365                 ctdb->statistics.pending_calls--;
366                 return;
367         }
368
369         talloc_free(w);
370
371         if (ret != 0) {
372                 DEBUG(DEBUG_ERR,(__location__ " Unable to fetch record\n"));
373                 ctdb->statistics.pending_calls--;
374                 return;
375         }
376
377         dstate = talloc(client, struct daemon_call_state);
378         if (dstate == NULL) {
379                 ctdb_ltdb_unlock(ctdb_db, key);
380                 DEBUG(DEBUG_ERR,(__location__ " Unable to allocate dstate\n"));
381                 ctdb->statistics.pending_calls--;
382                 return;
383         }
384         dstate->start_time = timeval_current();
385         dstate->client = client;
386         dstate->reqid  = c->hdr.reqid;
387         talloc_steal(dstate, data.dptr);
388
389         call = dstate->call = talloc_zero(dstate, struct ctdb_call);
390         if (call == NULL) {
391                 ctdb_ltdb_unlock(ctdb_db, key);
392                 DEBUG(DEBUG_ERR,(__location__ " Unable to allocate call\n"));
393                 ctdb->statistics.pending_calls--;
394                 ctdb_latency(&ctdb->statistics.max_call_latency, dstate->start_time);
395                 return;
396         }
397
398         call->call_id = c->callid;
399         call->key = key;
400         call->call_data.dptr = c->data + c->keylen;
401         call->call_data.dsize = c->calldatalen;
402         call->flags = c->flags;
403
404         if (header.dmaster == ctdb->pnn) {
405                 state = ctdb_call_local_send(ctdb_db, call, &header, &data);
406         } else {
407                 state = ctdb_daemon_call_send_remote(ctdb_db, call, &header);
408         }
409
410         ctdb_ltdb_unlock(ctdb_db, key);
411
412         if (state == NULL) {
413                 DEBUG(DEBUG_ERR,(__location__ " Unable to setup call send\n"));
414                 ctdb->statistics.pending_calls--;
415                 ctdb_latency(&ctdb->statistics.max_call_latency, dstate->start_time);
416                 return;
417         }
418         talloc_steal(state, dstate);
419         talloc_steal(client, state);
420
421         state->async.fn = daemon_call_from_client_callback;
422         state->async.private_data = dstate;
423 }
424
425
426 static void daemon_request_control_from_client(struct ctdb_client *client, 
427                                                struct ctdb_req_control *c);
428
429 /* data contains a packet from the client */
430 static void daemon_incoming_packet(void *p, struct ctdb_req_header *hdr)
431 {
432         struct ctdb_client *client = talloc_get_type(p, struct ctdb_client);
433         TALLOC_CTX *tmp_ctx;
434         struct ctdb_context *ctdb = client->ctdb;
435
436         /* place the packet as a child of a tmp_ctx. We then use
437            talloc_free() below to free it. If any of the calls want
438            to keep it, then they will steal it somewhere else, and the
439            talloc_free() will be a no-op */
440         tmp_ctx = talloc_new(client);
441         talloc_steal(tmp_ctx, hdr);
442
443         if (hdr->ctdb_magic != CTDB_MAGIC) {
444                 ctdb_set_error(client->ctdb, "Non CTDB packet rejected in daemon\n");
445                 goto done;
446         }
447
448         if (hdr->ctdb_version != CTDB_VERSION) {
449                 ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected in daemon\n", hdr->ctdb_version);
450                 goto done;
451         }
452
453         switch (hdr->operation) {
454         case CTDB_REQ_CALL:
455                 ctdb->statistics.client.req_call++;
456                 daemon_request_call_from_client(client, (struct ctdb_req_call *)hdr);
457                 break;
458
459         case CTDB_REQ_MESSAGE:
460                 ctdb->statistics.client.req_message++;
461                 daemon_request_message_from_client(client, (struct ctdb_req_message *)hdr);
462                 break;
463
464         case CTDB_REQ_CONTROL:
465                 ctdb->statistics.client.req_control++;
466                 daemon_request_control_from_client(client, (struct ctdb_req_control *)hdr);
467                 break;
468
469         default:
470                 DEBUG(DEBUG_CRIT,(__location__ " daemon: unrecognized operation %u\n",
471                          hdr->operation));
472         }
473
474 done:
475         talloc_free(tmp_ctx);
476 }
477
478 /*
479   called when the daemon gets a incoming packet
480  */
481 static void ctdb_daemon_read_cb(uint8_t *data, size_t cnt, void *args)
482 {
483         struct ctdb_client *client = talloc_get_type(args, struct ctdb_client);
484         struct ctdb_req_header *hdr;
485
486         if (cnt == 0) {
487                 talloc_free(client);
488                 return;
489         }
490
491         client->ctdb->statistics.client_packets_recv++;
492
493         if (cnt < sizeof(*hdr)) {
494                 ctdb_set_error(client->ctdb, "Bad packet length %u in daemon\n", 
495                                (unsigned)cnt);
496                 return;
497         }
498         hdr = (struct ctdb_req_header *)data;
499         if (cnt != hdr->length) {
500                 ctdb_set_error(client->ctdb, "Bad header length %u expected %u\n in daemon", 
501                                (unsigned)hdr->length, (unsigned)cnt);
502                 return;
503         }
504
505         if (hdr->ctdb_magic != CTDB_MAGIC) {
506                 ctdb_set_error(client->ctdb, "Non CTDB packet rejected\n");
507                 return;
508         }
509
510         if (hdr->ctdb_version != CTDB_VERSION) {
511                 ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected in daemon\n", hdr->ctdb_version);
512                 return;
513         }
514
515         DEBUG(DEBUG_DEBUG,(__location__ " client request %u of type %u length %u from "
516                  "node %u to %u\n", hdr->reqid, hdr->operation, hdr->length,
517                  hdr->srcnode, hdr->destnode));
518
519         /* it is the responsibility of the incoming packet function to free 'data' */
520         daemon_incoming_packet(client, hdr);
521 }
522
523 static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde, 
524                          uint16_t flags, void *private_data)
525 {
526         struct sockaddr_in addr;
527         socklen_t len;
528         int fd;
529         struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
530         struct ctdb_client *client;
531
532         memset(&addr, 0, sizeof(addr));
533         len = sizeof(addr);
534         fd = accept(ctdb->daemon.sd, (struct sockaddr *)&addr, &len);
535         if (fd == -1) {
536                 return;
537         }
538
539         set_nonblocking(fd);
540         set_close_on_exec(fd);
541
542         client = talloc_zero(ctdb, struct ctdb_client);
543         client->ctdb = ctdb;
544         client->fd = fd;
545         client->client_id = ctdb_reqid_new(ctdb, client);
546         ctdb->statistics.num_clients++;
547
548         client->queue = ctdb_queue_setup(ctdb, client, fd, CTDB_DS_ALIGNMENT, 
549                                          ctdb_daemon_read_cb, client);
550
551         talloc_set_destructor(client, ctdb_client_destructor);
552 }
553
554
555
556 /*
557   create a unix domain socket and bind it
558   return a file descriptor open on the socket 
559 */
560 static int ux_socket_bind(struct ctdb_context *ctdb)
561 {
562         struct sockaddr_un addr;
563
564         ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
565         if (ctdb->daemon.sd == -1) {
566                 return -1;
567         }
568
569         set_nonblocking(ctdb->daemon.sd);
570         set_close_on_exec(ctdb->daemon.sd);
571
572 #if 0
573         /* AIX doesn't like this :( */
574         if (fchown(ctdb->daemon.sd, geteuid(), getegid()) != 0 ||
575             fchmod(ctdb->daemon.sd, 0700) != 0) {
576                 DEBUG(DEBUG_CRIT,("Unable to secure ctdb socket '%s', ctdb->daemon.name\n"));
577                 goto failed;
578         }
579 #endif
580
581         set_nonblocking(ctdb->daemon.sd);
582
583         memset(&addr, 0, sizeof(addr));
584         addr.sun_family = AF_UNIX;
585         strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
586
587         if (bind(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
588                 DEBUG(DEBUG_CRIT,("Unable to bind on ctdb socket '%s'\n", ctdb->daemon.name));
589                 goto failed;
590         }       
591         if (listen(ctdb->daemon.sd, 10) != 0) {
592                 DEBUG(DEBUG_CRIT,("Unable to listen on ctdb socket '%s'\n", ctdb->daemon.name));
593                 goto failed;
594         }
595
596         return 0;
597
598 failed:
599         close(ctdb->daemon.sd);
600         ctdb->daemon.sd = -1;
601         return -1;      
602 }
603
604 /*
605   delete the socket on exit - called on destruction of autofree context
606  */
607 static int unlink_destructor(const char *name)
608 {
609         unlink(name);
610         return 0;
611 }
612
613 /*
614   start the protocol going as a daemon
615 */
616 int ctdb_start_daemon(struct ctdb_context *ctdb, bool do_fork)
617 {
618         int res, ret = -1;
619         struct fd_event *fde;
620         const char *domain_socket_name;
621
622         /* get rid of any old sockets */
623         unlink(ctdb->daemon.name);
624
625         /* create a unix domain stream socket to listen to */
626         res = ux_socket_bind(ctdb);
627         if (res!=0) {
628                 DEBUG(DEBUG_ALERT,(__location__ " Failed to open CTDB unix domain socket\n"));
629                 exit(10);
630         }
631
632         if (do_fork && fork()) {
633                 return 0;
634         }
635
636         tdb_reopen_all(False);
637
638         if (do_fork) {
639                 setsid();
640                 close(0);
641                 if (open("/dev/null", O_RDONLY) != 0) {
642                         DEBUG(DEBUG_ALERT,(__location__ " Failed to setup stdin on /dev/null\n"));
643                         exit(11);
644                 }
645         }
646         block_signal(SIGPIPE);
647
648         if (ctdb->do_setsched) {
649                 /* try to set us up as realtime */
650                 ctdb_set_scheduler(ctdb);
651         }
652
653         /* ensure the socket is deleted on exit of the daemon */
654         domain_socket_name = talloc_strdup(talloc_autofree_context(), ctdb->daemon.name);
655         talloc_set_destructor(domain_socket_name, unlink_destructor);   
656
657         ctdb->ev = event_context_init(NULL);
658
659         ctdb_set_child_logging(ctdb);
660
661         /* force initial recovery for election */
662         ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
663
664         if (strcmp(ctdb->transport, "tcp") == 0) {
665                 int ctdb_tcp_init(struct ctdb_context *);
666                 ret = ctdb_tcp_init(ctdb);
667         }
668 #ifdef USE_INFINIBAND
669         if (strcmp(ctdb->transport, "ib") == 0) {
670                 int ctdb_ibw_init(struct ctdb_context *);
671                 ret = ctdb_ibw_init(ctdb);
672         }
673 #endif
674         if (ret != 0) {
675                 DEBUG(DEBUG_ERR,("Failed to initialise transport '%s'\n", ctdb->transport));
676                 return -1;
677         }
678
679         /* initialise the transport  */
680         if (ctdb->methods->initialise(ctdb) != 0) {
681                 ctdb_fatal(ctdb, "transport failed to initialise");
682         }
683
684         /* attach to any existing persistent databases */
685         if (ctdb_attach_persistent(ctdb) != 0) {
686                 ctdb_fatal(ctdb, "Failed to attach to persistent databases\n");         
687         }
688
689         /* start frozen, then let the first election sort things out */
690         if (!ctdb_blocking_freeze(ctdb)) {
691                 ctdb_fatal(ctdb, "Failed to get initial freeze\n");
692         }
693
694         /* now start accepting clients, only can do this once frozen */
695         fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, 
696                            EVENT_FD_READ|EVENT_FD_AUTOCLOSE, 
697                            ctdb_accept_client, ctdb);
698
699         /* tell all other nodes we've just started up */
700         ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_ALL,
701                                  0, CTDB_CONTROL_STARTUP, 0,
702                                  CTDB_CTRL_FLAG_NOREPLY,
703                                  tdb_null, NULL, NULL);
704
705         /* release any IPs we hold from previous runs of the daemon */
706         ctdb_release_all_ips(ctdb);
707
708         /* start the transport going */
709         ctdb_start_transport(ctdb);
710
711         /* go into a wait loop to allow other nodes to complete */
712         event_loop_wait(ctdb->ev);
713
714         DEBUG(DEBUG_CRIT,("event_loop_wait() returned. this should not happen\n"));
715         exit(1);
716 }
717
718 /*
719   allocate a packet for use in daemon<->daemon communication
720  */
721 struct ctdb_req_header *_ctdb_transport_allocate(struct ctdb_context *ctdb,
722                                                  TALLOC_CTX *mem_ctx, 
723                                                  enum ctdb_operation operation, 
724                                                  size_t length, size_t slength,
725                                                  const char *type)
726 {
727         int size;
728         struct ctdb_req_header *hdr;
729
730         length = MAX(length, slength);
731         size = (length+(CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1);
732
733         hdr = (struct ctdb_req_header *)ctdb->methods->allocate_pkt(mem_ctx, size);
734         if (hdr == NULL) {
735                 DEBUG(DEBUG_ERR,("Unable to allocate transport packet for operation %u of length %u\n",
736                          operation, (unsigned)length));
737                 return NULL;
738         }
739         talloc_set_name_const(hdr, type);
740         memset(hdr, 0, slength);
741         hdr->length       = length;
742         hdr->operation    = operation;
743         hdr->ctdb_magic   = CTDB_MAGIC;
744         hdr->ctdb_version = CTDB_VERSION;
745         hdr->generation   = ctdb->vnn_map->generation;
746         hdr->srcnode      = ctdb->pnn;
747
748         return hdr;     
749 }
750
751 struct daemon_control_state {
752         struct daemon_control_state *next, *prev;
753         struct ctdb_client *client;
754         struct ctdb_req_control *c;
755         uint32_t reqid;
756         struct ctdb_node *node;
757 };
758
759 /*
760   callback when a control reply comes in
761  */
762 static void daemon_control_callback(struct ctdb_context *ctdb,
763                                     int32_t status, TDB_DATA data, 
764                                     const char *errormsg,
765                                     void *private_data)
766 {
767         struct daemon_control_state *state = talloc_get_type(private_data, 
768                                                              struct daemon_control_state);
769         struct ctdb_client *client = state->client;
770         struct ctdb_reply_control *r;
771         size_t len;
772
773         /* construct a message to send to the client containing the data */
774         len = offsetof(struct ctdb_reply_control, data) + data.dsize;
775         if (errormsg) {
776                 len += strlen(errormsg);
777         }
778         r = ctdbd_allocate_pkt(ctdb, state, CTDB_REPLY_CONTROL, len, 
779                                struct ctdb_reply_control);
780         CTDB_NO_MEMORY_VOID(ctdb, r);
781
782         r->hdr.reqid     = state->reqid;
783         r->status        = status;
784         r->datalen       = data.dsize;
785         r->errorlen = 0;
786         memcpy(&r->data[0], data.dptr, data.dsize);
787         if (errormsg) {
788                 r->errorlen = strlen(errormsg);
789                 memcpy(&r->data[r->datalen], errormsg, r->errorlen);
790         }
791
792         daemon_queue_send(client, &r->hdr);
793
794         talloc_free(state);
795 }
796
797 /*
798   fail all pending controls to a disconnected node
799  */
800 void ctdb_daemon_cancel_controls(struct ctdb_context *ctdb, struct ctdb_node *node)
801 {
802         struct daemon_control_state *state;
803         while ((state = node->pending_controls)) {
804                 DLIST_REMOVE(node->pending_controls, state);
805                 daemon_control_callback(ctdb, (uint32_t)-1, tdb_null, 
806                                         "node is disconnected", state);
807         }
808 }
809
810 /*
811   destroy a daemon_control_state
812  */
813 static int daemon_control_destructor(struct daemon_control_state *state)
814 {
815         if (state->node) {
816                 DLIST_REMOVE(state->node->pending_controls, state);
817         }
818         return 0;
819 }
820
821 /*
822   this is called when the ctdb daemon received a ctdb request control
823   from a local client over the unix domain socket
824  */
825 static void daemon_request_control_from_client(struct ctdb_client *client, 
826                                                struct ctdb_req_control *c)
827 {
828         TDB_DATA data;
829         int res;
830         struct daemon_control_state *state;
831         TALLOC_CTX *tmp_ctx = talloc_new(client);
832
833         if (c->hdr.destnode == CTDB_CURRENT_NODE) {
834                 c->hdr.destnode = client->ctdb->pnn;
835         }
836
837         state = talloc(client, struct daemon_control_state);
838         CTDB_NO_MEMORY_VOID(client->ctdb, state);
839
840         state->client = client;
841         state->c = talloc_steal(state, c);
842         state->reqid = c->hdr.reqid;
843         if (ctdb_validate_pnn(client->ctdb, c->hdr.destnode)) {
844                 state->node = client->ctdb->nodes[c->hdr.destnode];
845                 DLIST_ADD(state->node->pending_controls, state);
846         } else {
847                 state->node = NULL;
848         }
849
850         talloc_set_destructor(state, daemon_control_destructor);
851
852         if (c->flags & CTDB_CTRL_FLAG_NOREPLY) {
853                 talloc_steal(tmp_ctx, state);
854         }
855         
856         data.dptr = &c->data[0];
857         data.dsize = c->datalen;
858         res = ctdb_daemon_send_control(client->ctdb, c->hdr.destnode,
859                                        c->srvid, c->opcode, client->client_id,
860                                        c->flags,
861                                        data, daemon_control_callback,
862                                        state);
863         if (res != 0) {
864                 DEBUG(DEBUG_ERR,(__location__ " Failed to send control to remote node %u\n",
865                          c->hdr.destnode));
866         }
867
868         talloc_free(tmp_ctx);
869 }
870
871 /*
872   register a call function
873 */
874 int ctdb_daemon_set_call(struct ctdb_context *ctdb, uint32_t db_id,
875                          ctdb_fn_t fn, int id)
876 {
877         struct ctdb_registered_call *call;
878         struct ctdb_db_context *ctdb_db;
879
880         ctdb_db = find_ctdb_db(ctdb, db_id);
881         if (ctdb_db == NULL) {
882                 return -1;
883         }
884
885         call = talloc(ctdb_db, struct ctdb_registered_call);
886         call->fn = fn;
887         call->id = id;
888
889         DLIST_ADD(ctdb_db->calls, call);        
890         return 0;
891 }
892
893
894
895 /*
896   this local messaging handler is ugly, but is needed to prevent
897   recursion in ctdb_send_message() when the destination node is the
898   same as the source node
899  */
900 struct ctdb_local_message {
901         struct ctdb_context *ctdb;
902         uint64_t srvid;
903         TDB_DATA data;
904 };
905
906 static void ctdb_local_message_trigger(struct event_context *ev, struct timed_event *te, 
907                                        struct timeval t, void *private_data)
908 {
909         struct ctdb_local_message *m = talloc_get_type(private_data, 
910                                                        struct ctdb_local_message);
911         int res;
912
913         res = ctdb_dispatch_message(m->ctdb, m->srvid, m->data);
914         if (res != 0) {
915                 DEBUG(DEBUG_ERR, (__location__ " Failed to dispatch message for srvid=%llu\n", 
916                           (unsigned long long)m->srvid));
917         }
918         talloc_free(m);
919 }
920
921 static int ctdb_local_message(struct ctdb_context *ctdb, uint64_t srvid, TDB_DATA data)
922 {
923         struct ctdb_local_message *m;
924         m = talloc(ctdb, struct ctdb_local_message);
925         CTDB_NO_MEMORY(ctdb, m);
926
927         m->ctdb = ctdb;
928         m->srvid = srvid;
929         m->data  = data;
930         m->data.dptr = talloc_memdup(m, m->data.dptr, m->data.dsize);
931         if (m->data.dptr == NULL) {
932                 talloc_free(m);
933                 return -1;
934         }
935
936         /* this needs to be done as an event to prevent recursion */
937         event_add_timed(ctdb->ev, m, timeval_zero(), ctdb_local_message_trigger, m);
938         return 0;
939 }
940
941 /*
942   send a ctdb message
943 */
944 int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t pnn,
945                              uint64_t srvid, TDB_DATA data)
946 {
947         struct ctdb_req_message *r;
948         int len;
949
950         /* see if this is a message to ourselves */
951         if (pnn == ctdb->pnn) {
952                 return ctdb_local_message(ctdb, srvid, data);
953         }
954
955         len = offsetof(struct ctdb_req_message, data) + data.dsize;
956         r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REQ_MESSAGE, len,
957                                     struct ctdb_req_message);
958         CTDB_NO_MEMORY(ctdb, r);
959
960         r->hdr.destnode  = pnn;
961         r->srvid         = srvid;
962         r->datalen       = data.dsize;
963         memcpy(&r->data[0], data.dptr, data.dsize);
964
965         ctdb_queue_packet(ctdb, &r->hdr);
966
967         talloc_free(r);
968         return 0;
969 }
970