source: vendor/current/ctdb/server/ctdb_server.c

Last change on this file was 988, checked in by Silvan Scherrer, 9 years ago

Samba Server: update vendor to version 4.4.3
File size: 13.4 KB
Line 
1/*
2 ctdb main protocol 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 "replace.h"
21#include "system/network.h"
22#include "system/filesys.h"
23
24#include <talloc.h>
25#include <tevent.h>
26
27#include "lib/util/dlinklist.h"
28#include "lib/util/debug.h"
29#include "lib/util/samba_util.h"
30
31#include "ctdb_private.h"
32#include "ctdb_client.h"
33
34#include "common/common.h"
35#include "common/logging.h"
36
37/*
38 choose the transport we will use
39*/
40int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
41{
42 ctdb->transport = talloc_strdup(ctdb, transport);
43 CTDB_NO_MEMORY(ctdb, ctdb->transport);
44
45 return 0;
46}
47
48/*
49 Check whether an ip is a valid node ip
50 Returns the node id for this ip address or -1
51*/
52int ctdb_ip_to_nodeid(struct ctdb_context *ctdb, const ctdb_sock_addr *nodeip)
53{
54 int nodeid;
55
56 for (nodeid=0;nodeid<ctdb->num_nodes;nodeid++) {
57 if (ctdb->nodes[nodeid]->flags & NODE_FLAGS_DELETED) {
58 continue;
59 }
60 if (ctdb_same_ip(&ctdb->nodes[nodeid]->address, nodeip)) {
61 return nodeid;
62 }
63 }
64
65 return -1;
66}
67
68/*
69 choose the recovery lock file
70*/
71int ctdb_set_recovery_lock_file(struct ctdb_context *ctdb, const char *file)
72{
73 if (ctdb->recovery_lock_file != NULL) {
74 talloc_free(ctdb->recovery_lock_file);
75 ctdb->recovery_lock_file = NULL;
76 }
77
78 if (file == NULL) {
79 DEBUG(DEBUG_ALERT,("Recovery lock file set to \"\". Disabling recovery lock checking\n"));
80 return 0;
81 }
82
83 ctdb->recovery_lock_file = talloc_strdup(ctdb, file);
84 CTDB_NO_MEMORY(ctdb, ctdb->recovery_lock_file);
85
86 return 0;
87}
88
89/* Load a nodes list file into a nodes array */
90static int convert_node_map_to_list(struct ctdb_context *ctdb,
91 TALLOC_CTX *mem_ctx,
92 struct ctdb_node_map_old *node_map,
93 struct ctdb_node ***nodes,
94 uint32_t *num_nodes)
95{
96 int i;
97
98 *nodes = talloc_zero_array(mem_ctx,
99 struct ctdb_node *, node_map->num);
100 CTDB_NO_MEMORY(ctdb, *nodes);
101 *num_nodes = node_map->num;
102
103 for (i = 0; i < node_map->num; i++) {
104 struct ctdb_node *node;
105
106 node = talloc_zero(*nodes, struct ctdb_node);
107 CTDB_NO_MEMORY(ctdb, node);
108 (*nodes)[i] = node;
109
110 node->address = node_map->nodes[i].addr;
111 node->name = talloc_asprintf(node, "%s:%u",
112 ctdb_addr_to_str(&node->address),
113 ctdb_addr_to_port(&node->address));
114
115 node->flags = node_map->nodes[i].flags;
116 if (!(node->flags & NODE_FLAGS_DELETED)) {
117 node->flags = NODE_FLAGS_UNHEALTHY;
118 }
119 node->flags |= NODE_FLAGS_DISCONNECTED;
120
121 node->pnn = i;
122 node->ctdb = ctdb;
123 node->dead_count = 0;
124 }
125
126 return 0;
127}
128
129/* Load the nodes list from a file */
130void ctdb_load_nodes_file(struct ctdb_context *ctdb)
131{
132 struct ctdb_node_map_old *node_map;
133 int ret;
134
135 node_map = ctdb_read_nodes_file(ctdb, ctdb->nodes_file);
136 if (node_map == NULL) {
137 goto fail;
138 }
139
140 TALLOC_FREE(ctdb->nodes);
141 ret = convert_node_map_to_list(ctdb, ctdb, node_map,
142 &ctdb->nodes, &ctdb->num_nodes);
143 if (ret == -1) {
144 goto fail;
145 }
146
147 talloc_free(node_map);
148 return;
149
150fail:
151 DEBUG(DEBUG_ERR, ("Failed to load nodes file \"%s\"\n",
152 ctdb->nodes_file));
153 talloc_free(node_map);
154 exit(1);
155}
156
157/*
158 setup the local node address
159*/
160int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
161{
162 ctdb->address = talloc(ctdb, ctdb_sock_addr);
163 CTDB_NO_MEMORY(ctdb, ctdb->address);
164
165 if (ctdb_parse_address(ctdb, address, ctdb->address) != 0) {
166 return -1;
167 }
168
169 ctdb->name = talloc_asprintf(ctdb, "%s:%u",
170 ctdb_addr_to_str(ctdb->address),
171 ctdb_addr_to_port(ctdb->address));
172 return 0;
173}
174
175
176/*
177 return the number of active nodes
178*/
179uint32_t ctdb_get_num_active_nodes(struct ctdb_context *ctdb)
180{
181 int i;
182 uint32_t count=0;
183 for (i=0; i < ctdb->num_nodes; i++) {
184 if (!(ctdb->nodes[i]->flags & NODE_FLAGS_INACTIVE)) {
185 count++;
186 }
187 }
188 return count;
189}
190
191
192/*
193 called when we need to process a packet. This can be a requeued packet
194 after a lockwait, or a real packet from another node
195*/
196void ctdb_input_pkt(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
197{
198 TALLOC_CTX *tmp_ctx;
199
200 /* place the packet as a child of the tmp_ctx. We then use
201 talloc_free() below to free it. If any of the calls want
202 to keep it, then they will steal it somewhere else, and the
203 talloc_free() will only free the tmp_ctx */
204 tmp_ctx = talloc_new(ctdb);
205 talloc_steal(tmp_ctx, hdr);
206
207 DEBUG(DEBUG_DEBUG,(__location__ " ctdb request %u of type %u length %u from "
208 "node %u to %u\n", hdr->reqid, hdr->operation, hdr->length,
209 hdr->srcnode, hdr->destnode));
210
211 switch (hdr->operation) {
212 case CTDB_REQ_CALL:
213 case CTDB_REPLY_CALL:
214 case CTDB_REQ_DMASTER:
215 case CTDB_REPLY_DMASTER:
216 /* we don't allow these calls when banned */
217 if (ctdb->nodes[ctdb->pnn]->flags & NODE_FLAGS_BANNED) {
218 DEBUG(DEBUG_DEBUG,(__location__ " ctdb operation %u"
219 " request %u"
220 " length %u from node %u to %u while node"
221 " is banned\n",
222 hdr->operation, hdr->reqid,
223 hdr->length,
224 hdr->srcnode, hdr->destnode));
225 goto done;
226 }
227
228 /* for ctdb_call inter-node operations verify that the
229 remote node that sent us the call is running in the
230 same generation instance as this node
231 */
232 if (ctdb->vnn_map->generation != hdr->generation) {
233 DEBUG(DEBUG_DEBUG,(__location__ " ctdb operation %u"
234 " request %u"
235 " length %u from node %u to %u had an"
236 " invalid generation id:%u while our"
237 " generation id is:%u\n",
238 hdr->operation, hdr->reqid,
239 hdr->length,
240 hdr->srcnode, hdr->destnode,
241 hdr->generation, ctdb->vnn_map->generation));
242 goto done;
243 }
244 }
245
246 switch (hdr->operation) {
247 case CTDB_REQ_CALL:
248 CTDB_INCREMENT_STAT(ctdb, node.req_call);
249 ctdb_request_call(ctdb, hdr);
250 break;
251
252 case CTDB_REPLY_CALL:
253 CTDB_INCREMENT_STAT(ctdb, node.reply_call);
254 ctdb_reply_call(ctdb, hdr);
255 break;
256
257 case CTDB_REPLY_ERROR:
258 CTDB_INCREMENT_STAT(ctdb, node.reply_error);
259 ctdb_reply_error(ctdb, hdr);
260 break;
261
262 case CTDB_REQ_DMASTER:
263 CTDB_INCREMENT_STAT(ctdb, node.req_dmaster);
264 ctdb_request_dmaster(ctdb, hdr);
265 break;
266
267 case CTDB_REPLY_DMASTER:
268 CTDB_INCREMENT_STAT(ctdb, node.reply_dmaster);
269 ctdb_reply_dmaster(ctdb, hdr);
270 break;
271
272 case CTDB_REQ_MESSAGE:
273 CTDB_INCREMENT_STAT(ctdb, node.req_message);
274 ctdb_request_message(ctdb, hdr);
275 break;
276
277 case CTDB_REQ_CONTROL:
278 CTDB_INCREMENT_STAT(ctdb, node.req_control);
279 ctdb_request_control(ctdb, hdr);
280 break;
281
282 case CTDB_REPLY_CONTROL:
283 CTDB_INCREMENT_STAT(ctdb, node.reply_control);
284 ctdb_reply_control(ctdb, hdr);
285 break;
286
287 case CTDB_REQ_KEEPALIVE:
288 CTDB_INCREMENT_STAT(ctdb, keepalive_packets_recv);
289 break;
290
291 default:
292 DEBUG(DEBUG_CRIT,("%s: Packet with unknown operation %u\n",
293 __location__, hdr->operation));
294 break;
295 }
296
297done:
298 talloc_free(tmp_ctx);
299}
300
301
302/*
303 called by the transport layer when a node is dead
304*/
305void ctdb_node_dead(struct ctdb_node *node)
306{
307 if (node->flags & NODE_FLAGS_DISCONNECTED) {
308 DEBUG(DEBUG_INFO,("%s: node %s is already marked disconnected: %u connected\n",
309 node->ctdb->name, node->name,
310 node->ctdb->num_connected));
311 return;
312 }
313 node->ctdb->num_connected--;
314 node->flags |= NODE_FLAGS_DISCONNECTED | NODE_FLAGS_UNHEALTHY;
315 node->rx_cnt = 0;
316 node->dead_count = 0;
317
318 DEBUG(DEBUG_NOTICE,("%s: node %s is dead: %u connected\n",
319 node->ctdb->name, node->name, node->ctdb->num_connected));
320 ctdb_daemon_cancel_controls(node->ctdb, node);
321
322 if (node->ctdb->methods == NULL) {
323 DEBUG(DEBUG_ERR,(__location__ " Can not restart transport while shutting down daemon.\n"));
324 return;
325 }
326
327 node->ctdb->methods->restart(node);
328}
329
330/*
331 called by the transport layer when a node is connected
332*/
333void ctdb_node_connected(struct ctdb_node *node)
334{
335 if (!(node->flags & NODE_FLAGS_DISCONNECTED)) {
336 DEBUG(DEBUG_INFO,("%s: node %s is already marked connected: %u connected\n",
337 node->ctdb->name, node->name,
338 node->ctdb->num_connected));
339 return;
340 }
341 node->ctdb->num_connected++;
342 node->dead_count = 0;
343 node->flags &= ~NODE_FLAGS_DISCONNECTED;
344 node->flags |= NODE_FLAGS_UNHEALTHY;
345 DEBUG(DEBUG_NOTICE,
346 ("%s: connected to %s - %u connected\n",
347 node->ctdb->name, node->name, node->ctdb->num_connected));
348}
349
350struct queue_next {
351 struct ctdb_context *ctdb;
352 struct ctdb_req_header *hdr;
353};
354
355
356/*
357 triggered when a deferred packet is due
358 */
359static void queue_next_trigger(struct tevent_context *ev,
360 struct tevent_timer *te,
361 struct timeval t, void *private_data)
362{
363 struct queue_next *q = talloc_get_type(private_data, struct queue_next);
364 ctdb_input_pkt(q->ctdb, q->hdr);
365 talloc_free(q);
366}
367
368/*
369 defer a packet, so it is processed on the next event loop
370 this is used for sending packets to ourselves
371 */
372static void ctdb_defer_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
373{
374 struct queue_next *q;
375 q = talloc(ctdb, struct queue_next);
376 if (q == NULL) {
377 DEBUG(DEBUG_ERR,(__location__ " Failed to allocate deferred packet\n"));
378 return;
379 }
380 q->ctdb = ctdb;
381 q->hdr = talloc_memdup(ctdb, hdr, hdr->length);
382 if (q->hdr == NULL) {
383 DEBUG(DEBUG_ERR,("Error copying deferred packet to self\n"));
384 return;
385 }
386#if 0
387 /* use this to put packets directly into our recv function */
388 ctdb_input_pkt(q->ctdb, q->hdr);
389#else
390 tevent_add_timer(ctdb->ev, q, timeval_zero(), queue_next_trigger, q);
391#endif
392}
393
394
395/*
396 broadcast a packet to all nodes
397*/
398static void ctdb_broadcast_packet_all(struct ctdb_context *ctdb,
399 struct ctdb_req_header *hdr)
400{
401 int i;
402 for (i=0; i < ctdb->num_nodes; i++) {
403 if (ctdb->nodes[i]->flags & NODE_FLAGS_DELETED) {
404 continue;
405 }
406 hdr->destnode = ctdb->nodes[i]->pnn;
407 ctdb_queue_packet(ctdb, hdr);
408 }
409}
410
411/*
412 broadcast a packet to all nodes in the current vnnmap
413*/
414static void ctdb_broadcast_packet_vnnmap(struct ctdb_context *ctdb,
415 struct ctdb_req_header *hdr)
416{
417 int i;
418 for (i=0;i<ctdb->vnn_map->size;i++) {
419 hdr->destnode = ctdb->vnn_map->map[i];
420 ctdb_queue_packet(ctdb, hdr);
421 }
422}
423
424/*
425 broadcast a packet to all connected nodes
426*/
427static void ctdb_broadcast_packet_connected(struct ctdb_context *ctdb,
428 struct ctdb_req_header *hdr)
429{
430 int i;
431 for (i=0; i < ctdb->num_nodes; i++) {
432 if (ctdb->nodes[i]->flags & NODE_FLAGS_DELETED) {
433 continue;
434 }
435 if (!(ctdb->nodes[i]->flags & NODE_FLAGS_DISCONNECTED)) {
436 hdr->destnode = ctdb->nodes[i]->pnn;
437 ctdb_queue_packet(ctdb, hdr);
438 }
439 }
440}
441
442/*
443 queue a packet or die
444*/
445void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
446{
447 struct ctdb_node *node;
448
449 switch (hdr->destnode) {
450 case CTDB_BROADCAST_ALL:
451 ctdb_broadcast_packet_all(ctdb, hdr);
452 return;
453 case CTDB_BROADCAST_VNNMAP:
454 ctdb_broadcast_packet_vnnmap(ctdb, hdr);
455 return;
456 case CTDB_BROADCAST_CONNECTED:
457 ctdb_broadcast_packet_connected(ctdb, hdr);
458 return;
459 }
460
461 CTDB_INCREMENT_STAT(ctdb, node_packets_sent);
462
463 if (!ctdb_validate_pnn(ctdb, hdr->destnode)) {
464 DEBUG(DEBUG_CRIT,(__location__ " cant send to node %u that does not exist\n",
465 hdr->destnode));
466 return;
467 }
468
469 node = ctdb->nodes[hdr->destnode];
470
471 if (node->flags & NODE_FLAGS_DELETED) {
472 DEBUG(DEBUG_ERR, (__location__ " Can not queue packet to DELETED node %d\n", hdr->destnode));
473 return;
474 }
475
476 if (node->pnn == ctdb->pnn) {
477 ctdb_defer_packet(ctdb, hdr);
478 return;
479 }
480
481 if (ctdb->methods == NULL) {
482 DEBUG(DEBUG_ALERT, (__location__ " Can not queue packet. "
483 "Transport is DOWN\n"));
484 return;
485 }
486
487 node->tx_cnt++;
488 if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
489 ctdb_fatal(ctdb, "Unable to queue packet\n");
490 }
491}
492
493
494
495
496/*
497 a valgrind hack to allow us to get opcode specific backtraces
498 very ugly, and relies on no compiler optimisation!
499*/
500void ctdb_queue_packet_opcode(struct ctdb_context *ctdb, struct ctdb_req_header *hdr, unsigned opcode)
501{
502 switch (opcode) {
503#define DO_OP(x) case x: ctdb_queue_packet(ctdb, hdr); break
504 DO_OP(1);
505 DO_OP(2);
506 DO_OP(3);
507 DO_OP(4);
508 DO_OP(5);
509 DO_OP(6);
510 DO_OP(7);
511 DO_OP(8);
512 DO_OP(9);
513 DO_OP(10);
514 DO_OP(11);
515 DO_OP(12);
516 DO_OP(13);
517 DO_OP(14);
518 DO_OP(15);
519 DO_OP(16);
520 DO_OP(17);
521 DO_OP(18);
522 DO_OP(19);
523 DO_OP(20);
524 DO_OP(21);
525 DO_OP(22);
526 DO_OP(23);
527 DO_OP(24);
528 DO_OP(25);
529 DO_OP(26);
530 DO_OP(27);
531 DO_OP(28);
532 DO_OP(29);
533 DO_OP(30);
534 DO_OP(31);
535 DO_OP(32);
536 DO_OP(33);
537 DO_OP(34);
538 DO_OP(35);
539 DO_OP(36);
540 DO_OP(37);
541 DO_OP(38);
542 DO_OP(39);
543 DO_OP(40);
544 DO_OP(41);
545 DO_OP(42);
546 DO_OP(43);
547 DO_OP(44);
548 DO_OP(45);
549 DO_OP(46);
550 DO_OP(47);
551 DO_OP(48);
552 DO_OP(49);
553 DO_OP(50);
554 DO_OP(51);
555 DO_OP(52);
556 DO_OP(53);
557 DO_OP(54);
558 DO_OP(55);
559 DO_OP(56);
560 DO_OP(57);
561 DO_OP(58);
562 DO_OP(59);
563 DO_OP(60);
564 DO_OP(61);
565 DO_OP(62);
566 DO_OP(63);
567 DO_OP(64);
568 DO_OP(65);
569 DO_OP(66);
570 DO_OP(67);
571 DO_OP(68);
572 DO_OP(69);
573 DO_OP(70);
574 DO_OP(71);
575 DO_OP(72);
576 DO_OP(73);
577 DO_OP(74);
578 DO_OP(75);
579 DO_OP(76);
580 DO_OP(77);
581 DO_OP(78);
582 DO_OP(79);
583 DO_OP(80);
584 DO_OP(81);
585 DO_OP(82);
586 DO_OP(83);
587 DO_OP(84);
588 DO_OP(85);
589 DO_OP(86);
590 DO_OP(87);
591 DO_OP(88);
592 DO_OP(89);
593 DO_OP(90);
594 DO_OP(91);
595 DO_OP(92);
596 DO_OP(93);
597 DO_OP(94);
598 DO_OP(95);
599 DO_OP(96);
600 DO_OP(97);
601 DO_OP(98);
602 DO_OP(99);
603 DO_OP(100);
604 default:
605 ctdb_queue_packet(ctdb, hdr);
606 break;
607 }
608}
Note: See TracBrowser for help on using the repository browser.