merge from tridge
[martins/samba.git] / ctdb / common / ctdb.c
index 559b0ed9a637683b0a10aba717d3ed921e2de1be..6ae32f5a0fb7df749d039848c331eccb6db47ef5 100644 (file)
 */
 int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
 {
-       int ctdb_tcp_init(struct ctdb_context *ctdb);
-#ifdef USE_INFINIBAND
-       int ctdb_ibw_init(struct ctdb_context *ctdb);
-#endif /* USE_INFINIBAND */
-
-       if (strcmp(transport, "tcp") == 0) {
-               return ctdb_tcp_init(ctdb);
-       }
-#ifdef USE_INFINIBAND
-       if (strcmp(transport, "ib") == 0) {
-               return ctdb_ibw_init(ctdb);
-       }
-#endif /* USE_INFINIBAND */
-
-       ctdb_set_error(ctdb, "Unknown transport '%s'\n", transport);
-       return -1;
+       ctdb->transport = talloc_strdup(ctdb, transport);
+       return 0;
 }
 
+
 /*
   set some ctdb flags
 */
@@ -73,6 +60,22 @@ void ctdb_set_max_lacount(struct ctdb_context *ctdb, unsigned count)
        ctdb->max_lacount = count;
 }
 
+/*
+  set the directory for the local databases
+*/
+int ctdb_set_tdb_dir(struct ctdb_context *ctdb, const char *dir)
+{
+       if (dir == NULL) {
+               ctdb->db_directory = talloc_asprintf(ctdb, "ctdb-%u", ctdb_get_vnn(ctdb));
+       } else {
+               ctdb->db_directory = talloc_strdup(ctdb, dir);
+       }
+       if (ctdb->db_directory == NULL) {
+               return -1;
+       }
+       return 0;
+}
+
 /*
   add a node to the list of active nodes
 */
@@ -100,13 +103,9 @@ static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr)
           will change! */
        node->vnn = ctdb->num_nodes;
 
-       if (ctdb->methods->add_node(node) != 0) {
-               talloc_free(node);
-               return -1;
-       }
-
        if (ctdb_same_address(&ctdb->address, &node->address)) {
                ctdb->vnn = node->vnn;
+               node->flags |= NODE_FLAGS_CONNECTED;
        }
 
        ctdb->num_nodes++;
@@ -155,6 +154,16 @@ int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
        return 0;
 }
 
+
+/*
+  setup the local socket name
+*/
+int ctdb_set_socketname(struct ctdb_context *ctdb, const char *socketname)
+{
+       ctdb->daemon.name = talloc_strdup(ctdb, socketname);
+       return 0;
+}
+
 /*
   add a node to the list of active nodes
 */
@@ -192,63 +201,131 @@ uint32_t ctdb_get_num_nodes(struct ctdb_context *ctdb)
 */
 void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
 {
-       struct ctdb_req_header *hdr;
+       struct ctdb_req_header *hdr = (struct ctdb_req_header *)data;
+       TALLOC_CTX *tmp_ctx;
+
+       ctdb->status.node_packets_recv++;
+
+       /* place the packet as a child of the tmp_ctx. We then use
+          talloc_free() below to free it. If any of the calls want
+          to keep it, then they will steal it somewhere else, and the
+          talloc_free() will only free the tmp_ctx */
+       tmp_ctx = talloc_new(ctdb);
+       talloc_steal(tmp_ctx, hdr);
 
        if (length < sizeof(*hdr)) {
                ctdb_set_error(ctdb, "Bad packet length %d\n", length);
-               return;
+               goto done;
        }
-       hdr = (struct ctdb_req_header *)data;
        if (length != hdr->length) {
                ctdb_set_error(ctdb, "Bad header length %d expected %d\n", 
                               hdr->length, length);
-               return;
+               goto done;
        }
 
        if (hdr->ctdb_magic != CTDB_MAGIC) {
                ctdb_set_error(ctdb, "Non CTDB packet rejected\n");
-               return;
+               goto done;
        }
 
        if (hdr->ctdb_version != CTDB_VERSION) {
                ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
-               return;
+               goto done;
        }
 
+       DEBUG(3,(__location__ " ctdb request %d of type %d length %d from "
+                "node %d to %d\n", hdr->reqid, hdr->operation, hdr->length,
+                hdr->srcnode, hdr->destnode));
+
        switch (hdr->operation) {
        case CTDB_REQ_CALL:
+               /* verify that the remote node that sent us the call
+                  is running in the same generation instance as this node
+               */
+               if (ctdb->vnn_map->generation != hdr->generation) {
+                       DEBUG(0,(__location__ " ctdb request %d of type"
+                               " %d length %d from node %d to %d had an"
+                               " invalid generation id:%d while our"
+                               " generation id is:%d\n", 
+                               hdr->reqid, hdr->operation, hdr->length, 
+                               hdr->srcnode, hdr->destnode, 
+                               ctdb->vnn_map->generation, 
+                               hdr->generation));
+                       break;
+               }
+               /* if we are in recovery mode we discard all traffic
+                  until the cluster has recovered.
+               */
+               if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
+                       DEBUG(0,(__location__ " ctdb request %d of type"
+                               " %d length %d from node %d to %d"
+                               " while we are in recovery mode\n", 
+                               hdr->reqid, hdr->operation, hdr->length, 
+                                hdr->srcnode, hdr->destnode));
+                       break;
+               }
+
+               ctdb->status.count.req_call++;
                ctdb_request_call(ctdb, hdr);
                break;
 
        case CTDB_REPLY_CALL:
+               ctdb->status.count.reply_call++;
                ctdb_reply_call(ctdb, hdr);
                break;
 
        case CTDB_REPLY_ERROR:
+               ctdb->status.count.reply_error++;
                ctdb_reply_error(ctdb, hdr);
                break;
 
-       case CTDB_REPLY_REDIRECT:
-               ctdb_reply_redirect(ctdb, hdr);
-               break;
-
        case CTDB_REQ_DMASTER:
+               ctdb->status.count.req_dmaster++;
                ctdb_request_dmaster(ctdb, hdr);
                break;
 
        case CTDB_REPLY_DMASTER:
+               ctdb->status.count.reply_dmaster++;
                ctdb_reply_dmaster(ctdb, hdr);
                break;
 
        case CTDB_REQ_MESSAGE:
+               ctdb->status.count.req_message++;
                ctdb_request_message(ctdb, hdr);
                break;
 
+       case CTDB_REQ_FINISHED:
+               ctdb->status.count.req_finished++;
+               ctdb_request_finished(ctdb, hdr);
+               break;
+
+       case CTDB_REQ_CONTROL:
+               ctdb->status.count.req_control++;
+               ctdb_request_control(ctdb, hdr);
+               break;
+
+       case CTDB_REPLY_CONTROL:
+               ctdb->status.count.reply_control++;
+               ctdb_reply_control(ctdb, hdr);
+               break;
+
        default:
-               printf("Packet with unknown operation %d\n", hdr->operation);
+               DEBUG(0,("%s: Packet with unknown operation %d\n", 
+                        __location__, hdr->operation));
                break;
        }
-       talloc_free(hdr);
+
+done:
+       talloc_free(tmp_ctx);
+}
+
+/*
+  called by the transport layer when a packet comes in
+*/
+void ctdb_recv_raw_pkt(void *p, uint8_t *data, uint32_t length)
+{
+       struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
+       ctdb_recv_pkt(ctdb, data, length);
 }
 
 /*
@@ -257,8 +334,9 @@ void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
 static void ctdb_node_dead(struct ctdb_node *node)
 {
        node->ctdb->num_connected--;
-       printf("%s: node %s is dead: %d connected\n", 
-              node->ctdb->name, node->name, node->ctdb->num_connected);
+       node->flags &= ~NODE_FLAGS_CONNECTED;
+       DEBUG(1,("%s: node %s is dead: %d connected\n", 
+                node->ctdb->name, node->name, node->ctdb->num_connected));
 }
 
 /*
@@ -267,8 +345,9 @@ static void ctdb_node_dead(struct ctdb_node *node)
 static void ctdb_node_connected(struct ctdb_node *node)
 {
        node->ctdb->num_connected++;
-       printf("%s: connected to %s - %d connected\n", 
-              node->ctdb->name, node->name, node->ctdb->num_connected);
+       node->flags |= NODE_FLAGS_CONNECTED;
+       DEBUG(1,("%s: connected to %s - %d connected\n", 
+                node->ctdb->name, node->name, node->ctdb->num_connected));
 }
 
 /*
@@ -281,33 +360,76 @@ void ctdb_daemon_connect_wait(struct ctdb_context *ctdb)
                expected++;
        }
        while (ctdb->num_connected != expected) {
+               DEBUG(3,("ctdb_connect_wait: waiting for %d nodes (have %d)\n", 
+                        expected, ctdb->num_connected));
                event_loop_once(ctdb->ev);
        }
+       DEBUG(3,("ctdb_connect_wait: got all %d nodes\n", expected));
 }
 
+struct queue_next {
+       struct ctdb_context *ctdb;
+       struct ctdb_req_header *hdr;
+};
+
+
 /*
-  wait until we're the only node left
-*/
-void ctdb_wait_loop(struct ctdb_context *ctdb)
+  trigered when a deferred packet is due
+ */
+static void queue_next_trigger(struct event_context *ev, struct timed_event *te, 
+                              struct timeval t, void *private_data)
 {
-       int expected = 0;
-       if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) {
-               expected++;
+       struct queue_next *q = talloc_get_type(private_data, struct queue_next);
+       ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length);
+       talloc_free(q);
+}      
+
+/*
+  defer a packet, so it is processed on the next event loop
+  this is used for sending packets to ourselves
+ */
+static void ctdb_defer_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+       struct queue_next *q;
+       q = talloc(ctdb, struct queue_next);
+       if (q == NULL) {
+               DEBUG(0,(__location__ " Failed to allocate deferred packet\n"));
+               return;
        }
-       while (ctdb->num_connected > expected) {
-               event_loop_once(ctdb->ev);
+       q->ctdb = ctdb;
+       q->hdr = talloc_memdup(ctdb, hdr, hdr->length);
+       if (q->hdr == NULL) {
+               DEBUG(0,("Error copying deferred packet to self\n"));
+               return;
        }
+#if 0
+       /* use this to put packets directly into our recv function */
+       ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length);
+       talloc_free(q);
+#else
+       event_add_timed(ctdb->ev, q, timeval_zero(), queue_next_trigger, q);
+#endif
 }
 
-
 /*
   queue a packet or die
 */
 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 {
        struct ctdb_node *node;
+       ctdb->status.node_packets_sent++;
+
+       if (!ctdb_validate_vnn(ctdb, hdr->destnode)) {
+               DEBUG(0,(__location__ " cant send to node %u that does not exist\n", 
+                        hdr->destnode));
+               return;
+       }
+
        node = ctdb->nodes[hdr->destnode];
-       if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
+
+       if (hdr->destnode == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
+               ctdb_defer_packet(ctdb, hdr);
+       } else if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
                ctdb_fatal(ctdb, "Unable to queue packet\n");
        }
 }
@@ -331,6 +453,7 @@ struct ctdb_context *ctdb_init(struct event_context *ev)
 
        ctdb = talloc_zero(ev, struct ctdb_context);
        ctdb->ev = ev;
+       ctdb->recovery_mode = CTDB_RECOVERY_NORMAL;
        ctdb->upcalls = &ctdb_upcalls;
        ctdb->idr = idr_init(ctdb);
        ctdb->max_lacount = CTDB_DEFAULT_MAX_LACOUNT;
@@ -338,11 +461,3 @@ struct ctdb_context *ctdb_init(struct event_context *ev)
        return ctdb;
 }
 
-int ctdb_start(struct ctdb_context *ctdb)
-{
-       if (ctdb->flags&CTDB_FLAG_DAEMON_MODE) {
-               return ctdbd_start(ctdb);
-       }
-
-       return ctdb->methods->start(ctdb);
-}