first test of forced migration of records. compiles but not tested.

(This used to be ctdb commit ac6ac290e79446f52caf31f429b4c38668c27eda)

diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index 75355f7ae7edc9355d921f74f0e09730c22f74a5..b13e2704360290bd754a058113955555947e6ae5 100644 (file)
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -329,9 +329,12 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
        }
 
        /* if this nodes has done enough consecutive calls on the same record
-          then give them the record */
-       if (header.laccessor == c->hdr.srcnode &&
-           header.lacount >= ctdb->max_lacount) {
+          then give them the record
+          or if the node requested an immediate migration
+       */
+       if ( (header.laccessor == c->hdr.srcnode
+             && header.lacount >= ctdb->max_lacount)
+          || c->flags&CTDB_IMMEDIATE_MIGRATION ) {
                ctdb_call_send_dmaster(ctdb_db, c, &header, &call.key, &data);
                talloc_free(data.dptr);
                return;
@@ -373,6 +376,7 @@ struct ctdb_call_state {
        struct ctdb_call call;
        int redirect_count;
        struct ctdb_ltdb_header header;
+       void *fetch_private;
 };
 
 
@@ -578,6 +582,7 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct c
        state->c->hdr.srcnode   = ctdb->vnn;
        /* this limits us to 16k outstanding messages - not unreasonable */
        state->c->hdr.reqid     = idr_get_new(ctdb->idr, state, 0xFFFF);
+       state->c->flags         = call->flags;
        state->c->db_id         = ctdb_db->db_id;
        state->c->callid        = call->call_id;
        state->c->keylen        = call->key.dsize;
@@ -604,6 +609,13 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct c
 }
 
 
+
+struct ctdb_record_handle {
+       struct ctdb_db_context *ctdb_db;
+       TDB_DATA key;
+       TDB_DATA *data;
+};
+
 /*
   make a remote ctdb call - async recv. 
 
@@ -612,6 +624,17 @@ struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct c
 */
 int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 {
+       struct ctdb_record_handle *rec = state->fetch_private;
+
+       /* ugly hack to manage forced migration */
+       if (rec != NULL) {
+               rec->data->dptr = talloc_memdup(rec, state->call.reply_data.dptr,
+                                                    state->call.reply_data.dsize);
+               rec->data->dsize = state->call.reply_data.dsize;
+               talloc_free(state);
+               return 0;
+       }
+
        while (state->state < CTDB_CALL_DONE) {
                event_loop_once(state->node->ctdb->ev);
        }
@@ -643,3 +666,58 @@ int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
        state = ctdb_call_send(ctdb_db, call);
        return ctdb_call_recv(state, call);
 }
+
+
+
+
+
+
+struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
+                                          TDB_DATA key, TDB_DATA *data)
+{
+       struct ctdb_call call;
+       struct ctdb_record_handle *rec;
+       struct ctdb_call_state *state;
+       int ret;
+
+       ZERO_STRUCT(call);
+       call.call_id = CTDB_FETCH_FUNC;
+       call.key = key;
+       call.flags = CTDB_IMMEDIATE_MIGRATION;
+
+       rec = talloc(mem_ctx, struct ctdb_record_handle);
+       CTDB_NO_MEMORY_NULL(ctdb_db->ctdb, rec);
+
+       rec->ctdb_db = ctdb_db;
+       rec->key = key;
+       rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize);
+
+       state = ctdb_call_send(ctdb_db, &call);
+       state->fetch_private = rec;
+
+       ret = ctdb_call_recv(state, &call);
+       if (ret != 0) {
+               talloc_free(rec);
+               return NULL;
+       }
+
+       return rec;
+}
+
+
+int ctdb_record_store(struct ctdb_record_handle *rec, TDB_DATA data)
+{
+       int ret;
+       struct ctdb_ltdb_header header;
+
+       /* should be avoided if possible    hang header off rec ? */
+       ret = ctdb_ltdb_fetch(rec->ctdb_db, rec->key, &header, NULL);
+       if (ret) {
+               ctdb_set_error(rec->ctdb_db->ctdb, "Fetch of locally held record failed");
+               return ret;
+       }
+
+       ret = ctdb_ltdb_store(rec->ctdb_db, rec->key, &header, data);
+               
+       return ret;
+}

diff --git a/ctdb/common/ctdb_ltdb.c b/ctdb/common/ctdb_ltdb.c
index 189816229f08be72acfe16522606ff533ae0a237..ceedb6c5ced3e5642a3d7200d820d9068d616521 100644 (file)
--- a/ctdb/common/ctdb_ltdb.c
+++ b/ctdb/common/ctdb_ltdb.c
@@ -41,6 +41,17 @@ struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *na
        return NULL;
 }
 
+
+/*
+  this is the dummy null procedure that all databases support
+*/
+static int ctdb_fetch_func(struct ctdb_call_info *call)
+{
+       call->reply_data = &call->record_data;
+       return 0;
+}
+
+
 /*
   attach to a specific database
 */
@@ -49,6 +60,7 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name,
 {
        struct ctdb_db_context *ctdb_db, *tmp_db;
        TDB_DATA data;
+       int ret;
 
        ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
        CTDB_NO_MEMORY_NULL(ctdb, ctdb_db);
@@ -80,7 +92,18 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name,
                return NULL;
        }
 
+
+       /* 
+         all databases support the "fetch" function. we need this in order to do forced migration of records
+        */
+       ret = ctdb_set_call(ctdb_db, ctdb_fetch_func, CTDB_FETCH_FUNC);
+       if (ret != 0) {
+               talloc_free(ctdb_db);
+               return NULL;
+       }
+
        DLIST_ADD(ctdb->db_list, ctdb_db);
+
        return ctdb_db;
 }
 
@@ -124,18 +147,25 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
                /* return an initial header */
                free(rec.dptr);
                ltdb_initial_header(ctdb_db, key, header);
-               data->dptr = NULL;
-               data->dsize = 0;
+               if (data) {
+                       data->dptr = NULL;
+                       data->dsize = 0;
+               }
                return 0;
        }
 
        *header = *(struct ctdb_ltdb_header *)rec.dptr;
 
-       data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
-       data->dptr = talloc_memdup(ctdb_db, sizeof(struct ctdb_ltdb_header)+rec.dptr,
-                                  data->dsize);
+       if (data) {
+               data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
+               data->dptr = talloc_memdup(ctdb_db, sizeof(struct ctdb_ltdb_header)+rec.dptr,
+                                          data->dsize);
+       }
+
        free(rec.dptr);
-       CTDB_NO_MEMORY(ctdb, data->dptr);
+       if (data) {
+               CTDB_NO_MEMORY(ctdb, data->dptr);
+       }
 
        return 0;
 }

diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index c5b6dbc7824fae43fb45b505e82d75036ec4fcf3..9049314401d8823e4343c12e642392f3418ab7b6 100644 (file)
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -21,12 +21,14 @@
 #ifndef _CTDB_H
 #define _CTDB_H
 
+#define CTDB_IMMEDIATE_MIGRATION       0x00000001
 struct ctdb_call {
        int call_id;
        TDB_DATA key;
        TDB_DATA call_data;
        TDB_DATA reply_data;
        uint32_t status;
+       uint32_t flags;
 };
 
 /*
@@ -148,4 +150,21 @@ int ctdb_set_message_handler(struct ctdb_context *ctdb, ctdb_message_fn_t handle
 int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn,
                      uint32_t srvid, TDB_DATA data);
 
+
+/* 
+   fetch and lock a ctdb record. Underneath this will force the
+   dmaster for the record to be moved to the local node. 
+
+   The lock is released when is talloc_free() is called on the
+   returned ctdb_record_handle. 
+*/
+struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data);
+
+/*
+  change the data in a record held with a ctdb_record_handle
+  if the new data is zero length, this implies a delete of the record
+ */
+int ctdb_record_store(struct ctdb_record_handle *rec, TDB_DATA data);
+
+
 #endif

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 1e2244ff7de49d49d10126a01cace70cb2fdbb7a..66a9a6e2444f20f44d572a64fe51f61b48521279 100644 (file)
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -23,6 +23,8 @@
 
 #include "ctdb.h"
 
+
+#define CTDB_FETCH_FUNC 0xf0000001
 /*
   an installed ctdb remote call
 */
@@ -166,6 +168,7 @@ struct ctdb_req_header {
 
 struct ctdb_req_call {
        struct ctdb_req_header hdr;
+       uint32_t flags;
        uint32_t db_id;
        uint32_t callid;
        uint32_t keylen;