added redirect handling

(This used to be ctdb commit 3c1dc8b98c8e843c44a172ac15e67f4ab8c47500)

diff --git a/ctdb/common/ctdb.c b/ctdb/common/ctdb.c
index 80e4649c4da6c89389a71511870964a51c0b349d..d50a1570ad785c44009f88034bef7dda2ebddfa8 100644 (file)
--- a/ctdb/common/ctdb.c
+++ b/ctdb/common/ctdb.c
@@ -181,6 +181,10 @@ static void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t len
                ctdb_reply_error(ctdb, hdr);
                break;
 
+       case CTDB_REPLY_REDIRECT:
+               ctdb_reply_redirect(ctdb, hdr);
+               break;
+
        default:
                printf("Packet with unknown operation %d\n", hdr->operation);
                talloc_free(hdr);

diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index 6349146b12d6c38a69b16e254ed20b875c3ae35c..fa67085737dd99091caf6a033a7ef6a61ca05c89 100644 (file)
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -123,6 +123,7 @@ static void ctdb_send_error(struct ctdb_context *ctdb,
        talloc_free(r);
 }
 
+
 /*
   send a redirect reply
 */
@@ -130,7 +131,22 @@ static void ctdb_call_send_redirect(struct ctdb_context *ctdb,
                                    struct ctdb_req_call *c, 
                                    struct ctdb_ltdb_header *header)
 {
-       printf("ctdb_call_send_redirect not implemented\n");
+       struct ctdb_reply_redirect *r;
+       struct ctdb_node *node;
+
+       r = talloc_size(ctdb, sizeof(*r));
+       r->hdr.length = sizeof(*r);
+       r->hdr.operation = CTDB_REPLY_REDIRECT;
+       r->hdr.destnode  = c->hdr.srcnode;
+       r->hdr.srcnode   = ctdb->vnn;
+       r->hdr.reqid     = c->hdr.reqid;
+       r->dmaster       = header->dmaster;
+
+       node = ctdb->nodes[r->hdr.destnode];
+
+       ctdb->methods->queue_pkt(node, (uint8_t *)r, r->hdr.length);
+
+       talloc_free(r);
 }
 
 /*
@@ -199,6 +215,8 @@ struct ctdb_call_state {
        struct ctdb_node *node;
        const char *errmsg;
        TDB_DATA reply_data;
+       TDB_DATA key;
+       int redirect_count;
 };
 
 
@@ -240,6 +258,34 @@ void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
        state->errmsg = (char *)c->msg;
 }
 
+
+/*
+  called when a CTDB_REPLY_REDIRECT packet comes in
+*/
+void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+       struct ctdb_reply_redirect *c = (struct ctdb_reply_redirect *)hdr;
+       struct ctdb_call_state *state;
+
+       state = idr_find(ctdb->idr, hdr->reqid);
+
+       talloc_steal(state, c);
+       
+       /* don't allow for too many redirects */
+       if (state->redirect_count++ == CTDB_MAX_REDIRECT) {
+               c->dmaster = ctdb_lmaster(ctdb, state->key);
+       }
+
+       /* send it off again */
+       state->node = ctdb->nodes[c->dmaster];
+
+       if (ctdb->methods->queue_pkt(state->node, (uint8_t *)state->c, 
+                                    state->c->hdr.length) != 0) {
+               state->state = CTDB_CALL_ERROR;
+               state->errmsg = "unable to queue in ctdb_reply_redirect";
+       }
+}
+
 /*
   destroy a ctdb_call
 */
@@ -331,6 +377,8 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb,
        if (call_data) {
                memcpy(&state->c->data[key.dsize], call_data->dptr, call_data->dsize);
        }
+       state->key.dptr         = &state->c->data[0];
+       state->key.dsize        = key.dsize;
 
        state->node = ctdb->nodes[header.dmaster];
        state->state = CTDB_CALL_WAIT;

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index 821aad5ee59aef32804f97035a29f4c4a703ccc4..1bd1bb54a9117a0cb3afa050884090bf0d2e7ef4 100644 (file)
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -41,6 +41,14 @@ int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags,
        return 0;
 }
 
+/*
+  return the lmaster given a key
+*/
+uint32_t ctdb_lmaster(struct ctdb_context *ctdb, TDB_DATA key)
+{
+       return ctdb_hash(&key) % ctdb->num_nodes;
+}
+
 
 /*
   construct an initial header for a record with no ltdb header yet
@@ -51,7 +59,7 @@ static void ltdb_initial_header(struct ctdb_context *ctdb,
 {
        header->rsn = 0;
        /* initial dmaster is the lmaster */
-       header->dmaster = ctdb_hash(&key) % ctdb->num_nodes;
+       header->dmaster = ctdb_lmaster(ctdb, key);
        header->laccessor = header->dmaster;
        header->lacount = 0;
 }

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 6270ac3132fa9d967aa82d6fb9ace76d78a0ceda..b58d55d8ac1d1fdb226e4bf1d70206a97cb96019 100644 (file)
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -101,6 +101,9 @@ struct ctdb_context {
 /* arbitrary maximum timeout for ctdb operations */
 #define CTDB_REQ_TIMEOUT 10
 
+/* max number of redirects before we ask the lmaster */
+#define CTDB_MAX_REDIRECT 2
+
 /*
   the extended header for records in the ltdb
 */
@@ -156,6 +159,11 @@ struct ctdb_reply_error {
        uint8_t  msg[0];
 };
 
+struct ctdb_reply_redirect {
+       struct ctdb_req_header hdr;
+       uint32_t dmaster;
+};
+
 /* internal prototypes */
 void ctdb_set_error(struct ctdb_context *ctdb, const char *fmt, ...);
 bool ctdb_same_address(struct ctdb_address *a1, struct ctdb_address *a2);
@@ -167,6 +175,7 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 
+uint32_t ctdb_lmaster(struct ctdb_context *ctdb, TDB_DATA key);
 int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
                    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data);
 int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key,