Rev 34: added redirect handling in http://samba.org/~tridge/ctdb

[tridge at samba.org]

------------------------------------------------------------
revno: 34
revision-id: tridge at samba.org-20061218034406-oo5cyvgxltp9osop
parent: tridge at samba.org-20061218032720-i855uc7pif1s4tbq
committer: Andrew Tridgell <tridge at samba.org>
branch nick: tridge
timestamp: Mon 2006-12-18 14:44:06 +1100
message:
  added redirect handling
modified:
  common/ctdb.c                  ctdb.c-20061127094323-t50f58d65iaao5of-2
  common/ctdb_call.c             ctdb_call.c-20061128065342-to93h6eejj5kon81-1
  common/ctdb_ltdb.c             ctdb_ltdb.c-20061128065342-to93h6eejj5kon81-2
  include/ctdb_private.h         ctdb_private.h-20061117234101-o3qt14umlg9en8z0-13
=== modified file 'common/ctdb.c'
--- a/common/ctdb.c	2006-12-18 03:27:20 +0000
+++ b/common/ctdb.c	2006-12-18 03:44:06 +0000
@@ -181,6 +181,10 @@
 		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);

=== modified file 'common/ctdb_call.c'
--- a/common/ctdb_call.c	2006-12-18 03:27:20 +0000
+++ b/common/ctdb_call.c	2006-12-18 03:44:06 +0000
@@ -123,6 +123,7 @@
 	talloc_free(r);
 }
 
+
 /*
   send a redirect reply
 */
@@ -130,7 +131,22 @@
 				    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_node *node;
 	const char *errmsg;
 	TDB_DATA reply_data;
+	TDB_DATA key;
+	int redirect_count;
 };
 
 
@@ -240,6 +258,34 @@
 	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 @@
 	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;

=== modified file 'common/ctdb_ltdb.c'
--- a/common/ctdb_ltdb.c	2006-12-18 03:05:49 +0000
+++ b/common/ctdb_ltdb.c	2006-12-18 03:44:06 +0000
@@ -41,6 +41,14 @@
 	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 @@
 {
 	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;
 }

=== modified file 'include/ctdb_private.h'
--- a/include/ctdb_private.h	2006-12-18 03:27:20 +0000
+++ b/include/ctdb_private.h	2006-12-18 03:44:06 +0000
@@ -101,6 +101,9 @@
 /* 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 @@
 	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_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,