merge from peter
[vlendec/samba-autobuild/.git] / ctdb / common / ctdb.c
1 /* 
2    ctdb main protocol code
3
4    Copyright (C) Andrew Tridgell  2006
5
6    This library is free software; you can redistribute it and/or
7    modify it under the terms of the GNU Lesser General Public
8    License as published by the Free Software Foundation; either
9    version 2 of the License, or (at your option) any later version.
10
11    This library 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 GNU
14    Lesser General Public License for more details.
15
16    You should have received a copy of the GNU Lesser General Public
17    License along with this library; if not, write to the Free Software
18    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19 */
20
21 #include "includes.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 "../include/ctdb_private.h"
28
29 /*
30   choose the transport we will use
31 */
32 int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
33 {
34         ctdb->transport = talloc_strdup(ctdb, transport);
35         return 0;
36 }
37
38
39 /*
40   set some ctdb flags
41 */
42 void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags)
43 {
44         ctdb->flags |= flags;
45 }
46
47 /*
48   clear some ctdb flags
49 */
50 void ctdb_clear_flags(struct ctdb_context *ctdb, unsigned flags)
51 {
52         ctdb->flags &= ~flags;
53 }
54
55 /*
56   set max acess count before a dmaster migration
57 */
58 void ctdb_set_max_lacount(struct ctdb_context *ctdb, unsigned count)
59 {
60         ctdb->max_lacount = count;
61 }
62
63 /*
64   set the directory for the local databases
65 */
66 int ctdb_set_tdb_dir(struct ctdb_context *ctdb, const char *dir)
67 {
68         if (dir == NULL) {
69                 ctdb->db_directory = talloc_asprintf(ctdb, "ctdb-%u", ctdb_get_vnn(ctdb));
70         } else {
71                 ctdb->db_directory = talloc_strdup(ctdb, dir);
72         }
73         if (ctdb->db_directory == NULL) {
74                 return -1;
75         }
76         return 0;
77 }
78
79 /*
80   add a node to the list of active nodes
81 */
82 static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr)
83 {
84         struct ctdb_node *node, **nodep;
85
86         nodep = talloc_realloc(ctdb, ctdb->nodes, struct ctdb_node *, ctdb->num_nodes+1);
87         CTDB_NO_MEMORY(ctdb, nodep);
88
89         ctdb->nodes = nodep;
90         nodep = &ctdb->nodes[ctdb->num_nodes];
91         (*nodep) = talloc_zero(ctdb->nodes, struct ctdb_node);
92         CTDB_NO_MEMORY(ctdb, *nodep);
93         node = *nodep;
94
95         if (ctdb_parse_address(ctdb, node, nstr, &node->address) != 0) {
96                 return -1;
97         }
98         node->ctdb = ctdb;
99         node->name = talloc_asprintf(node, "%s:%u", 
100                                      node->address.address, 
101                                      node->address.port);
102         /* for now we just set the vnn to the line in the file - this
103            will change! */
104         node->vnn = ctdb->num_nodes;
105
106         if (ctdb_same_address(&ctdb->address, &node->address)) {
107                 ctdb->vnn = node->vnn;
108         }
109
110         ctdb->num_nodes++;
111
112         return 0;
113 }
114
115 /*
116   setup the node list from a file
117 */
118 int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist)
119 {
120         char **lines;
121         int nlines;
122         int i;
123
124         lines = file_lines_load(nlist, &nlines, ctdb);
125         if (lines == NULL) {
126                 ctdb_set_error(ctdb, "Failed to load nlist '%s'\n", nlist);
127                 return -1;
128         }
129
130         for (i=0;i<nlines;i++) {
131                 if (ctdb_add_node(ctdb, lines[i]) != 0) {
132                         talloc_free(lines);
133                         return -1;
134                 }
135         }
136         
137         talloc_free(lines);
138         return 0;
139 }
140
141 /*
142   setup the local node address
143 */
144 int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
145 {
146         if (ctdb_parse_address(ctdb, ctdb, address, &ctdb->address) != 0) {
147                 return -1;
148         }
149         
150         ctdb->name = talloc_asprintf(ctdb, "%s:%u", 
151                                      ctdb->address.address, 
152                                      ctdb->address.port);
153         return 0;
154 }
155
156
157 /*
158   setup the local socket name
159 */
160 int ctdb_set_socketname(struct ctdb_context *ctdb, const char *socketname)
161 {
162         ctdb->daemon.name = talloc_strdup(ctdb, socketname);
163         return 0;
164 }
165
166 /*
167   add a node to the list of active nodes
168 */
169 int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id)
170 {
171         struct ctdb_registered_call *call;
172
173         call = talloc(ctdb_db, struct ctdb_registered_call);
174         call->fn = fn;
175         call->id = id;
176
177         DLIST_ADD(ctdb_db->calls, call);        
178         return 0;
179 }
180
181 /*
182   return the vnn of this node
183 */
184 uint32_t ctdb_get_vnn(struct ctdb_context *ctdb)
185 {
186         return ctdb->vnn;
187 }
188
189 /*
190   return the number of nodes
191 */
192 uint32_t ctdb_get_num_nodes(struct ctdb_context *ctdb)
193 {
194         return ctdb->num_nodes;
195 }
196
197
198 /*
199   called by the transport layer when a packet comes in
200 */
201 void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
202 {
203         struct ctdb_req_header *hdr = (struct ctdb_req_header *)data;
204         TALLOC_CTX *tmp_ctx;
205
206         ctdb->status.node_packets_recv++;
207
208         /* place the packet as a child of the tmp_ctx. We then use
209            talloc_free() below to free it. If any of the calls want
210            to keep it, then they will steal it somewhere else, and the
211            talloc_free() will only free the tmp_ctx */
212         tmp_ctx = talloc_new(ctdb);
213         talloc_steal(tmp_ctx, hdr);
214
215         if (length < sizeof(*hdr)) {
216                 ctdb_set_error(ctdb, "Bad packet length %d\n", length);
217                 goto done;
218         }
219         if (length != hdr->length) {
220                 ctdb_set_error(ctdb, "Bad header length %d expected %d\n", 
221                                hdr->length, length);
222                 goto done;
223         }
224
225         if (hdr->ctdb_magic != CTDB_MAGIC) {
226                 ctdb_set_error(ctdb, "Non CTDB packet rejected\n");
227                 goto done;
228         }
229
230         if (hdr->ctdb_version != CTDB_VERSION) {
231                 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
232                 goto done;
233         }
234
235         DEBUG(3,(__location__ " ctdb request %d of type %d length %d from "
236                  "node %d to %d\n", hdr->reqid, hdr->operation, hdr->length,
237                  hdr->srcnode, hdr->destnode));
238
239         switch (hdr->operation) {
240         case CTDB_REQ_CALL:
241                 ctdb->status.count.req_call++;
242                 ctdb_request_call(ctdb, hdr);
243                 break;
244
245         case CTDB_REPLY_CALL:
246                 ctdb->status.count.reply_call++;
247                 ctdb_reply_call(ctdb, hdr);
248                 break;
249
250         case CTDB_REPLY_ERROR:
251                 ctdb->status.count.reply_error++;
252                 ctdb_reply_error(ctdb, hdr);
253                 break;
254
255         case CTDB_REPLY_REDIRECT:
256                 ctdb->status.count.reply_redirect++;
257                 ctdb_reply_redirect(ctdb, hdr);
258                 break;
259
260         case CTDB_REQ_DMASTER:
261                 ctdb->status.count.req_dmaster++;
262                 ctdb_request_dmaster(ctdb, hdr);
263                 break;
264
265         case CTDB_REPLY_DMASTER:
266                 ctdb->status.count.reply_dmaster++;
267                 ctdb_reply_dmaster(ctdb, hdr);
268                 break;
269
270         case CTDB_REQ_MESSAGE:
271                 ctdb->status.count.req_message++;
272                 ctdb_request_message(ctdb, hdr);
273                 break;
274
275         case CTDB_REQ_FINISHED:
276                 ctdb->status.count.req_finished++;
277                 ctdb_request_finished(ctdb, hdr);
278                 break;
279
280         case CTDB_REQ_CONTROL:
281                 ctdb->status.count.req_control++;
282                 ctdb_request_control(ctdb, hdr);
283                 break;
284
285         case CTDB_REPLY_CONTROL:
286                 ctdb->status.count.reply_control++;
287                 ctdb_reply_control(ctdb, hdr);
288                 break;
289
290         default:
291                 DEBUG(0,("%s: Packet with unknown operation %d\n", 
292                          __location__, hdr->operation));
293                 break;
294         }
295
296 done:
297         talloc_free(tmp_ctx);
298 }
299
300 /*
301   called by the transport layer when a packet comes in
302 */
303 void ctdb_recv_raw_pkt(void *p, uint8_t *data, uint32_t length)
304 {
305         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
306         ctdb_recv_pkt(ctdb, data, length);
307 }
308
309 /*
310   called by the transport layer when a node is dead
311 */
312 static void ctdb_node_dead(struct ctdb_node *node)
313 {
314         node->ctdb->num_connected--;
315         DEBUG(1,("%s: node %s is dead: %d connected\n", 
316                  node->ctdb->name, node->name, node->ctdb->num_connected));
317 }
318
319 /*
320   called by the transport layer when a node is connected
321 */
322 static void ctdb_node_connected(struct ctdb_node *node)
323 {
324         node->ctdb->num_connected++;
325         DEBUG(1,("%s: connected to %s - %d connected\n", 
326                  node->ctdb->name, node->name, node->ctdb->num_connected));
327 }
328
329 /*
330   wait for all nodes to be connected
331 */
332 void ctdb_daemon_connect_wait(struct ctdb_context *ctdb)
333 {
334         int expected = ctdb->num_nodes - 1;
335         if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) {
336                 expected++;
337         }
338         while (ctdb->num_connected != expected) {
339                 DEBUG(3,("ctdb_connect_wait: waiting for %d nodes (have %d)\n", 
340                          expected, ctdb->num_connected));
341                 event_loop_once(ctdb->ev);
342         }
343         DEBUG(3,("ctdb_connect_wait: got all %d nodes\n", expected));
344 }
345
346 struct queue_next {
347         struct ctdb_context *ctdb;
348         struct ctdb_req_header *hdr;
349 };
350
351
352 /*
353   trigered when a deferred packet is due
354  */
355 static void queue_next_trigger(struct event_context *ev, struct timed_event *te, 
356                                struct timeval t, void *private_data)
357 {
358         struct queue_next *q = talloc_get_type(private_data, struct queue_next);
359         ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length);
360         talloc_free(q);
361 }       
362
363 /*
364   defer a packet, so it is processed on the next event loop
365   this is used for sending packets to ourselves
366  */
367 static void ctdb_defer_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
368 {
369         struct queue_next *q;
370         q = talloc(ctdb, struct queue_next);
371         if (q == NULL) {
372                 DEBUG(0,(__location__ " Failed to allocate deferred packet\n"));
373                 return;
374         }
375         q->ctdb = ctdb;
376         q->hdr = talloc_memdup(ctdb, hdr, hdr->length);
377         if (q->hdr == NULL) {
378                 DEBUG(0,("Error copying deferred packet to self\n"));
379                 return;
380         }
381 #if 0
382         /* use this to put packets directly into our recv function */
383         ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length);
384         talloc_free(q);
385 #else
386         event_add_timed(ctdb->ev, q, timeval_zero(), queue_next_trigger, q);
387 #endif
388 }
389
390 /*
391   queue a packet or die
392 */
393 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
394 {
395         struct ctdb_node *node;
396         ctdb->status.node_packets_sent++;
397         node = ctdb->nodes[hdr->destnode];
398         if (hdr->destnode == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
399                 ctdb_defer_packet(ctdb, hdr);
400         } else if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
401                 ctdb_fatal(ctdb, "Unable to queue packet\n");
402         }
403 }
404
405
406 static const struct ctdb_upcalls ctdb_upcalls = {
407         .recv_pkt       = ctdb_recv_pkt,
408         .node_dead      = ctdb_node_dead,
409         .node_connected = ctdb_node_connected
410 };
411
412 /*
413   initialise the ctdb daemon. 
414
415   NOTE: In current code the daemon does not fork. This is for testing purposes only
416   and to simplify the code.
417 */
418 struct ctdb_context *ctdb_init(struct event_context *ev)
419 {
420         struct ctdb_context *ctdb;
421
422         ctdb = talloc_zero(ev, struct ctdb_context);
423         ctdb->ev = ev;
424         ctdb->upcalls = &ctdb_upcalls;
425         ctdb->idr = idr_init(ctdb);
426         ctdb->max_lacount = CTDB_DEFAULT_MAX_LACOUNT;
427
428         return ctdb;
429 }
430