added async pull, push and rsn handling functions

(This used to be ctdb commit 05d30180f64aaff13411b92586ac554d84a35d9a)

diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
index 1935f5b7b15ec4c013f39a48e86884b1feabe852..70ccff617a1d7ac36bf37fc098ace913b990b05e 100644 (file)
--- a/ctdb/client/ctdb_client.c
+++ b/ctdb/client/ctdb_client.c
@@ -1278,67 +1278,118 @@ int ctdb_ctrl_setvnnmap(struct ctdb_context *ctdb, struct timeval timeout, uint3
        return 0;
 }
 
+
 /*
-  get all keys and records for a specific database
+  async send for pull database
  */
-int ctdb_ctrl_pulldb(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid, uint32_t lmaster, 
-                    TALLOC_CTX *mem_ctx, struct ctdb_key_list *keys)
+struct ctdb_client_control_state *ctdb_ctrl_pulldb_send(
+       struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid,
+       uint32_t lmaster, TALLOC_CTX *mem_ctx, struct timeval timeout)
 {
-       int i, ret;
-       TDB_DATA indata, outdata;
-       struct ctdb_control_pulldb pull;
-       struct ctdb_control_pulldb_reply *reply;
-       struct ctdb_rec_data *rec;
-       int32_t res;
+       TDB_DATA indata;
+       struct ctdb_control_pulldb *pull;
+       struct ctdb_client_control_state *state;
+
+       pull = talloc(mem_ctx, struct ctdb_control_pulldb);
+       CTDB_NO_MEMORY_NULL(ctdb, pull);
 
-       pull.db_id   = dbid;
-       pull.lmaster = lmaster;
+       pull->db_id   = dbid;
+       pull->lmaster = lmaster;
 
        indata.dsize = sizeof(struct ctdb_control_pulldb);
-       indata.dptr  = (unsigned char *)&pull;
+       indata.dptr  = (unsigned char *)pull;
 
-       ret = ctdb_control(ctdb, destnode, 0, 
-                          CTDB_CONTROL_PULL_DB, 0, indata, 
-                          mem_ctx, &outdata, &res, NULL, NULL);
-       if (ret != 0 || res != 0) {
-               DEBUG(0,(__location__ " ctdb_control for pulldb failed\n"));
+       state = ctdb_control_send(ctdb, destnode, 0, 
+                                 CTDB_CONTROL_PULL_DB, 0, indata, 
+                                 mem_ctx, NULL, &timeout, NULL);
+       talloc_free(pull);
+
+       return state;
+}
+
+/*
+  async recv for pull database
+ */
+int ctdb_ctrl_pulldb_recv(
+       struct ctdb_context *ctdb, 
+       TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, 
+       TDB_DATA *outdata)
+{
+       int ret;
+       int32_t res;
+
+       ret = ctdb_control_recv(ctdb, state, mem_ctx, outdata, &res, NULL);
+       if ( (ret != 0) || (res != 0) ){
+               DEBUG(0,(__location__ " ctdb_ctrl_pulldb_recv failed\n"));
                return -1;
        }
 
+       return 0;
+}
 
-       reply = (struct ctdb_control_pulldb_reply *)outdata.dptr;
-       keys->dbid     = reply->db_id;
-       keys->num      = reply->count;
+/*
+  pull all keys and records for a specific database on a node
+ */
+int ctdb_ctrl_pulldb(struct ctdb_context *ctdb, uint32_t destnode, 
+               uint32_t dbid, uint32_t lmaster, 
+               TALLOC_CTX *mem_ctx, struct timeval timeout,
+               TDB_DATA *outdata)
+{
+       struct ctdb_client_control_state *state;
+
+       state = ctdb_ctrl_pulldb_send(ctdb, destnode, dbid, lmaster, mem_ctx,
+                                     timeout);
        
-       keys->keys     = talloc_array(mem_ctx, TDB_DATA, keys->num);
-       keys->headers  = talloc_array(mem_ctx, struct ctdb_ltdb_header, keys->num);
-       keys->data     = talloc_array(mem_ctx, TDB_DATA, keys->num);
+       return ctdb_ctrl_pulldb_recv(ctdb, mem_ctx, state, outdata);
+}
 
-       rec = (struct ctdb_rec_data *)&reply->data[0];
 
-       for (i=0;i<reply->count;i++) {
-               keys->keys[i].dptr = talloc_memdup(mem_ctx, &rec->data[0], rec->keylen);
-               keys->keys[i].dsize = rec->keylen;
-               
-               keys->data[i].dptr = talloc_memdup(mem_ctx, &rec->data[keys->keys[i].dsize], rec->datalen);
-               keys->data[i].dsize = rec->datalen;
 
-               if (keys->data[i].dsize < sizeof(struct ctdb_ltdb_header)) {
-                       DEBUG(0,(__location__ " bad ltdb record\n"));
-                       return -1;
-               }
-               memcpy(&keys->headers[i], keys->data[i].dptr, sizeof(struct ctdb_ltdb_header));
-               keys->data[i].dptr += sizeof(struct ctdb_ltdb_header);
-               keys->data[i].dsize -= sizeof(struct ctdb_ltdb_header);
+/*
+  async send for pushdb
+ */
+struct ctdb_client_control_state *ctdb_ctrl_pushdb_send(
+       struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid,
+       TALLOC_CTX *mem_ctx, struct timeval timeout, TDB_DATA indata)
+{
+       return ctdb_control_send(ctdb, destnode, 0, 
+                          CTDB_CONTROL_PUSH_DB, 0, indata, 
+                          mem_ctx, NULL, &timeout, NULL);
+}
 
-               rec = (struct ctdb_rec_data *)(rec->length + (uint8_t *)rec);
-       }           
+int ctdb_ctrl_pushdb_recv(
+       struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, 
+       struct ctdb_client_control_state *state)
+{
+       int ret;
+       int32_t res;
 
-       talloc_free(outdata.dptr);
+       ret = ctdb_control_recv(ctdb, state, mem_ctx, NULL, &res, NULL);
+       if ( (ret != 0) || (res != 0) ){
+               DEBUG(0,(__location__ " ctdb_ctrl_pushdb_recv failed\n"));
+               return -1;
+       }
 
        return 0;
 }
 
+/*
+  push all records to a specific database on a node
+ */
+int ctdb_ctrl_pushdb(struct ctdb_context *ctdb, uint32_t destnode, 
+               uint32_t dbid,
+               TALLOC_CTX *mem_ctx, struct timeval timeout,
+               TDB_DATA indata)
+{
+       struct ctdb_client_control_state *state;
+
+       state = ctdb_ctrl_pushdb_send(ctdb, destnode, dbid, mem_ctx,
+                       timeout, indata);
+       
+       return ctdb_ctrl_pushdb_recv(ctdb, mem_ctx, state);
+}
+
+
 /*
   copy a tdb from one node to another node
  */
@@ -1346,32 +1397,22 @@ int ctdb_ctrl_copydb(struct ctdb_context *ctdb, struct timeval timeout, uint32_t
                     uint32_t destnode, uint32_t dbid, uint32_t lmaster, TALLOC_CTX *mem_ctx)
 {
        int ret;
-       TDB_DATA indata, outdata;
-       int32_t res;
-
-       indata.dsize = 2*sizeof(uint32_t);
-       indata.dptr  = (unsigned char *)talloc_array(mem_ctx, uint32_t, 2);
-
-       ((uint32_t *)(&indata.dptr[0]))[0] = dbid;
-       ((uint32_t *)(&indata.dptr[0]))[1] = lmaster;
+       TDB_DATA outdata;
 
        DEBUG(3,("pulling dbid 0x%x from %u\n", dbid, sourcenode));
 
-       ret = ctdb_control(ctdb, sourcenode, 0, 
-                          CTDB_CONTROL_PULL_DB, 0, indata, 
-                          mem_ctx, &outdata, &res, &timeout, NULL);
-       if (ret != 0 || res != 0) {
+       ret = ctdb_ctrl_pulldb(ctdb, sourcenode, dbid, lmaster, mem_ctx,
+                               timeout, &outdata);
+       if (ret != 0) {
                DEBUG(0,(__location__ " ctdb_control for pulldb failed\n"));
                return -1;
        }
 
        DEBUG(3,("pushing dbid 0x%x to %u\n", dbid, destnode));
 
-       ret = ctdb_control(ctdb, destnode, 0, 
-                          CTDB_CONTROL_PUSH_DB, 0, outdata, 
-                          mem_ctx, NULL, &res, &timeout, NULL);
+       ret = ctdb_ctrl_pushdb(ctdb, destnode, dbid, mem_ctx, timeout, outdata);
        talloc_free(outdata.dptr);
-       if (ret != 0 || res != 0) {
+       if (ret != 0) {
                DEBUG(0,(__location__ " ctdb_control for pushdb failed\n"));
                return -1;
        }
@@ -1982,12 +2023,11 @@ int ctdb_ctrl_get_max_rsn(struct ctdb_context *ctdb, struct timeval timeout,
 /*
   set the rsn on non-empty records to the given rsn
  */
-int ctdb_ctrl_set_rsn_nonempty(struct ctdb_context *ctdb, struct timeval timeout, 
-                              uint32_t destnode, uint32_t db_id, uint64_t rsn)
+struct ctdb_client_control_state *ctdb_ctrl_set_rsn_nonempty_send(
+       struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, 
+       uint32_t destnode, uint32_t db_id, uint64_t rsn)
 {
        TDB_DATA data;
-       int ret;
-       int32_t res;
        struct ctdb_control_set_rsn_nonempty p;
 
        memset(&p, 0, sizeof(p));
@@ -1997,43 +2037,89 @@ int ctdb_ctrl_set_rsn_nonempty(struct ctdb_context *ctdb, struct timeval timeout
        data.dptr = (uint8_t *)&p;
        data.dsize = sizeof(p);
 
-       ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_SET_RSN_NONEMPTY, 0, data, NULL,
-                          NULL, &res, &timeout, NULL);
+       return ctdb_control_send(ctdb, destnode, 0, CTDB_CONTROL_SET_RSN_NONEMPTY, 0, data, mem_ctx,
+                                NULL, &timeout, NULL);
+}
+
+/*
+  set the rsn on non-empty records to the given rsn
+ */
+int ctdb_ctrl_set_rsn_nonempty_recv(struct ctdb_context *ctdb, 
+                                   struct ctdb_client_control_state *state)
+{
+       int32_t res;
+       int ret;
+
+       ret = ctdb_control_recv(ctdb, state, NULL, NULL, &res, NULL);
        if (ret != 0 || res != 0) {
                DEBUG(0,(__location__ " ctdb_control for set_rsn_nonempty failed\n"));
                return -1;
        }
+       return 0;
+}
 
-       return 0;       
+/*
+  set the rsn on non-empty records to the given rsn
+ */
+int ctdb_ctrl_set_rsn_nonempty(struct ctdb_context *ctdb, struct timeval timeout, 
+                              uint32_t destnode, uint32_t db_id, uint64_t rsn)
+{
+       struct ctdb_client_control_state *state;
+       state = ctdb_ctrl_set_rsn_nonempty_send(ctdb, ctdb, timeout, destnode, db_id, rsn);
+       return ctdb_ctrl_set_rsn_nonempty_recv(ctdb, state);
 }
 
+
 /*
-  delete records which have a rsn below the given rsn
+  set the rsn on non-empty records to the given rsn
  */
-int ctdb_ctrl_delete_low_rsn(struct ctdb_context *ctdb, struct timeval timeout, 
-                            uint32_t destnode, uint32_t db_id, uint64_t rsn)
+struct ctdb_client_control_state *ctdb_ctrl_delete_low_rsn_send(
+       struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, 
+       uint32_t destnode, uint32_t db_id, uint64_t rsn)
 {
        TDB_DATA data;
-       int ret;
-       int32_t res;
        struct ctdb_control_delete_low_rsn p;
 
+       memset(&p, 0, sizeof(p));
        p.db_id = db_id;
        p.rsn = rsn;
 
        data.dptr = (uint8_t *)&p;
        data.dsize = sizeof(p);
 
-       ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_DELETE_LOW_RSN, 0, data, NULL,
-                          NULL, &res, &timeout, NULL);
+       return ctdb_control_send(ctdb, destnode, 0, CTDB_CONTROL_DELETE_LOW_RSN, 0, data, mem_ctx,
+                                NULL, &timeout, NULL);
+}
+
+/*
+  set the rsn on non-empty records to the given rsn
+ */
+int ctdb_ctrl_delete_low_rsn_recv(struct ctdb_context *ctdb, 
+                                   struct ctdb_client_control_state *state)
+{
+       int32_t res;
+       int ret;
+
+       ret = ctdb_control_recv(ctdb, state, NULL, NULL, &res, NULL);
        if (ret != 0 || res != 0) {
                DEBUG(0,(__location__ " ctdb_control for delete_low_rsn failed\n"));
                return -1;
        }
+       return 0;
+}
 
-       return 0;       
+/*
+  set the rsn on non-empty records to the given rsn
+ */
+int ctdb_ctrl_delete_low_rsn(struct ctdb_context *ctdb, struct timeval timeout, 
+                              uint32_t destnode, uint32_t db_id, uint64_t rsn)
+{
+       struct ctdb_client_control_state *state;
+       state = ctdb_ctrl_delete_low_rsn_send(ctdb, ctdb, timeout, destnode, db_id, rsn);
+       return ctdb_ctrl_delete_low_rsn_recv(ctdb, state);
 }
 
+
 /* 
   sent to a node to make it take over an ip address
 */