ctdb-protocol: Add new controls DB_PULL and DB_PUSH_START/DB_PUSH_CONFIRM

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

diff --git a/ctdb/protocol/protocol.h b/ctdb/protocol/protocol.h
index 1b48c3920fa74c38f3affd9e8849648b5a0a13dd..e4bc3b9a0d4a0cd74fd69c3906af7034372d798b 100644 (file)
--- a/ctdb/protocol/protocol.h
+++ b/ctdb/protocol/protocol.h
@@ -355,6 +355,9 @@ enum ctdb_controls {CTDB_CONTROL_PROCESS_EXISTS          = 0,
                    CTDB_CONTROL_DB_TRANSACTION_START    = 143,
                    CTDB_CONTROL_DB_TRANSACTION_COMMIT   = 144,
                    CTDB_CONTROL_DB_TRANSACTION_CANCEL   = 145,
+                   CTDB_CONTROL_DB_PULL                 = 146,
+                   CTDB_CONTROL_DB_PUSH_START           = 147,
+                   CTDB_CONTROL_DB_PUSH_CONFIRM         = 148,
 };
 
 #define CTDB_MONITORING_ACTIVE         0
@@ -861,6 +864,7 @@ struct ctdb_req_control_data {
                struct ctdb_vnn_map *vnnmap;
                uint32_t loglevel;
                struct ctdb_pulldb *pulldb;
+               struct ctdb_pulldb_ext *pulldb_ext;
                struct ctdb_rec_buffer *recbuf;
                uint32_t recmode;
                const char *db_name;
@@ -929,6 +933,7 @@ struct ctdb_reply_control_data {
                struct ctdb_uint8_array *u8_array;
                struct ctdb_db_statistics *dbstats;
                enum ctdb_runstate runstate;
+               uint32_t num_records;
        } data;
 };
 

diff --git a/ctdb/protocol/protocol_api.h b/ctdb/protocol/protocol_api.h
index 8fa74e164268e426219be1ca22f904a43b4be853..d07b4198d0a086a699394bf85cfde137759d2502 100644 (file)
--- a/ctdb/protocol/protocol_api.h
+++ b/ctdb/protocol/protocol_api.h
@@ -630,6 +630,20 @@ void ctdb_req_control_db_transaction_cancel(struct ctdb_req_control *request,
                                            uint32_t db_id);
 int ctdb_reply_control_db_transaction_cancel(struct ctdb_reply_control *reply);
 
+void ctdb_req_control_db_pull(struct ctdb_req_control *request,
+                             struct ctdb_pulldb_ext *pulldb_ext);
+int ctdb_reply_control_db_pull(struct ctdb_reply_control *reply,
+                              uint32_t *num_records);
+
+void ctdb_req_control_db_push_start(struct ctdb_req_control *request,
+                                   struct ctdb_pulldb_ext *pulldb_ext);
+int ctdb_reply_control_db_push_start(struct ctdb_reply_control *reply);
+
+void ctdb_req_control_db_push_confirm(struct ctdb_req_control *request,
+                                     uint32_t db_id);
+int ctdb_reply_control_db_push_confirm(struct ctdb_reply_control *reply,
+                                      uint32_t *num_records);
+
 /* From protocol/protocol_debug.c */
 
 void ctdb_packet_print(uint8_t *buf, size_t buflen, FILE *fp);

diff --git a/ctdb/protocol/protocol_client.c b/ctdb/protocol/protocol_client.c
index 7c983b80001f40229f36aa5bc6126459a917cd28..6466fcaf901be8a27242058a5a3b4aaa1d902dc5 100644 (file)
--- a/ctdb/protocol/protocol_client.c
+++ b/ctdb/protocol/protocol_client.c
@@ -2448,3 +2448,73 @@ int ctdb_reply_control_db_transaction_cancel(struct ctdb_reply_control *reply)
 {
        return ctdb_reply_control_generic(reply);
 }
+
+/* CTDB_CONTROL_DB_PULL */
+
+void ctdb_req_control_db_pull(struct ctdb_req_control *request,
+                             struct ctdb_pulldb_ext *pulldb_ext)
+{
+       request->opcode = CTDB_CONTROL_DB_PULL;
+       request->pad = 0;
+       request->srvid = 0;
+       request->client_id = 0;
+       request->flags = 0;
+
+       request->rdata.opcode = CTDB_CONTROL_DB_PULL;
+       request->rdata.data.pulldb_ext = pulldb_ext;
+}
+
+int ctdb_reply_control_db_pull(struct ctdb_reply_control *reply,
+                              uint32_t *num_records)
+{
+       if (reply->status == 0 &&
+           reply->rdata.opcode == CTDB_CONTROL_DB_PULL) {
+               *num_records = reply->rdata.data.num_records;
+       }
+       return reply->status;
+}
+
+/* CTDB_CONTROL_DB_PUSH_START */
+
+void ctdb_req_control_db_push_start(struct ctdb_req_control *request,
+                                   struct ctdb_pulldb_ext *pulldb_ext)
+{
+       request->opcode = CTDB_CONTROL_DB_PUSH_START;
+       request->pad = 0;
+       request->srvid = 0;
+       request->client_id = 0;
+       request->flags = 0;
+
+       request->rdata.opcode = CTDB_CONTROL_DB_PUSH_START;
+       request->rdata.data.pulldb_ext = pulldb_ext;
+}
+
+int ctdb_reply_control_db_push_start(struct ctdb_reply_control *reply)
+{
+       return ctdb_reply_control_generic(reply);
+}
+
+/* CTDB_CONTROL_DB_PUSH_CONFIRM */
+
+void ctdb_req_control_db_push_confirm(struct ctdb_req_control *request,
+                                     uint32_t db_id)
+{
+       request->opcode = CTDB_CONTROL_DB_PUSH_CONFIRM;
+       request->pad = 0;
+       request->srvid = 0;
+       request->client_id = 0;
+       request->flags = 0;
+
+       request->rdata.opcode = CTDB_CONTROL_DB_PUSH_CONFIRM;
+       request->rdata.data.db_id = db_id;
+}
+
+int ctdb_reply_control_db_push_confirm(struct ctdb_reply_control *reply,
+                                      uint32_t *num_records)
+{
+       if (reply->status == 0 &&
+           reply->rdata.opcode == CTDB_CONTROL_DB_PUSH_CONFIRM) {
+               *num_records = reply->rdata.data.num_records;
+       }
+       return reply->status;
+}

diff --git a/ctdb/protocol/protocol_control.c b/ctdb/protocol/protocol_control.c
index d57167b64ff0a7a1159ce086c45ca5357c88c440..57737ab678fcdc014f7fc0108c20fe850bf299cc 100644 (file)
--- a/ctdb/protocol/protocol_control.c
+++ b/ctdb/protocol/protocol_control.c
@@ -477,6 +477,18 @@ static size_t ctdb_req_control_data_len(struct ctdb_req_control_data *cd)
        case CTDB_CONTROL_DB_TRANSACTION_CANCEL:
                len = ctdb_uint32_len(cd->data.db_id);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               len = ctdb_pulldb_ext_len(cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               len = ctdb_pulldb_ext_len(cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               len = ctdb_uint32_len(cd->data.db_id);
+               break;
        }
 
        return len;
@@ -783,6 +795,18 @@ static void ctdb_req_control_data_push(struct ctdb_req_control_data *cd,
        case CTDB_CONTROL_DB_TRANSACTION_CANCEL:
                ctdb_uint32_push(cd->data.db_id, buf);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               ctdb_pulldb_ext_push(cd->data.pulldb_ext, buf);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               ctdb_pulldb_ext_push(cd->data.pulldb_ext, buf);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               ctdb_uint32_push(cd->data.db_id, buf);
+               break;
        }
 }
 
@@ -1164,6 +1188,21 @@ static int ctdb_req_control_data_pull(uint8_t *buf, size_t buflen,
                ret = ctdb_uint32_pull(buf, buflen, mem_ctx,
                                        &cd->data.db_id);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               ret = ctdb_pulldb_ext_pull(buf, buflen, mem_ctx,
+                                          &cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               ret = ctdb_pulldb_ext_pull(buf, buflen, mem_ctx,
+                                          &cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               ret = ctdb_uint32_pull(buf, buflen, mem_ctx,
+                                      &cd->data.db_id);
+               break;
        }
 
        return ret;
@@ -1558,6 +1597,17 @@ static size_t ctdb_reply_control_data_len(struct ctdb_reply_control_data *cd)
 
        case CTDB_CONTROL_DB_TRANSACTION_CANCEL:
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               len = ctdb_uint32_len(cd->data.num_records);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               len = ctdb_uint32_len(cd->data.num_records);
+               break;
        }
 
        return len;
@@ -1710,6 +1760,14 @@ static void ctdb_reply_control_data_push(struct ctdb_reply_control_data *cd,
        case CTDB_CONTROL_GET_NODES_FILE:
                ctdb_node_map_push(cd->data.nodemap, buf);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               ctdb_uint32_push(cd->data.num_records, buf);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               ctdb_uint32_push(cd->data.num_records, buf);
+               break;
        }
 }
 
@@ -1897,6 +1955,16 @@ static int ctdb_reply_control_data_pull(uint8_t *buf, size_t buflen,
                ret = ctdb_node_map_pull(buf, buflen, mem_ctx,
                                         &cd->data.nodemap);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               ret = ctdb_uint32_pull(buf, buflen, mem_ctx,
+                                      &cd->data.num_records);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               ret = ctdb_uint32_pull(buf, buflen, mem_ctx,
+                                      &cd->data.num_records);
+               break;
        }
 
        return ret;

diff --git a/ctdb/protocol/protocol_debug.c b/ctdb/protocol/protocol_debug.c
index 7d4fe3b24ddeb8ab65f113a975dcf4da050c0f25..47cf855f52bf9fb3d65898e04a50f37f76c111a6 100644 (file)
--- a/ctdb/protocol/protocol_debug.c
+++ b/ctdb/protocol/protocol_debug.c
@@ -235,6 +235,9 @@ static void ctdb_opcode_print(uint32_t opcode, FILE *fp)
                { CTDB_CONTROL_DB_TRANSACTION_START, "DB_TRANSACTION_START" },
                { CTDB_CONTROL_DB_TRANSACTION_COMMIT, "DB_TRANSACTION_COMMIT" },
                { CTDB_CONTROL_DB_TRANSACTION_CANCEL, "DB_TRANSACTION_CANCEL" },
+               { CTDB_CONTROL_DB_PULL, "DB_PULL" },
+               { CTDB_CONTROL_DB_PUSH_START, "DB_PUSH_START" },
+               { CTDB_CONTROL_DB_PUSH_CONFIRM, "DB_PUSH_CONFIRM" },
                { MAP_END, "" },
        };
 

diff --git a/ctdb/tests/src/protocol_client_test.c b/ctdb/tests/src/protocol_client_test.c
index 4719eac8be7544eebdd7e77adfab89428dc2d062..659594b417179b80e09f58884a9fec1b1755cf88 100644 (file)
--- a/ctdb/tests/src/protocol_client_test.c
+++ b/ctdb/tests/src/protocol_client_test.c
@@ -652,6 +652,23 @@ static void fill_ctdb_req_control_data(TALLOC_CTX *mem_ctx,
        case CTDB_CONTROL_DB_TRANSACTION_CANCEL:
                cd->data.db_id = rand32();
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               cd->data.pulldb_ext = talloc(mem_ctx, struct ctdb_pulldb_ext);
+               assert(cd->data.pulldb_ext != NULL);
+               fill_ctdb_pulldb_ext(mem_ctx, cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               cd->data.pulldb_ext = talloc(mem_ctx, struct ctdb_pulldb_ext);
+               assert(cd->data.pulldb_ext != NULL);
+               fill_ctdb_pulldb_ext(mem_ctx, cd->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               cd->data.db_id = rand32();
+               break;
+
        }
 }
 
@@ -1095,6 +1112,21 @@ static void verify_ctdb_req_control_data(struct ctdb_req_control_data *cd,
        case CTDB_CONTROL_DB_TRANSACTION_CANCEL:
                assert(cd->data.db_id == cd2->data.db_id);
                break;
+
+       case CTDB_CONTROL_DB_PULL:
+               verify_ctdb_pulldb_ext(cd->data.pulldb_ext,
+                                      cd2->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_START:
+               verify_ctdb_pulldb_ext(cd->data.pulldb_ext,
+                                      cd2->data.pulldb_ext);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               assert(cd->data.db_id == cd2->data.db_id);
+               break;
+
        }
 }
 
@@ -1548,6 +1580,14 @@ static void fill_ctdb_reply_control_data(TALLOC_CTX *mem_ctx,
                fill_ctdb_node_map(mem_ctx, cd->data.nodemap);
                break;
 
+       case CTDB_CONTROL_DB_PULL:
+               cd->data.num_records = rand32();
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               cd->data.num_records = rand32();
+               break;
+
        }
 }
 
@@ -1928,6 +1968,14 @@ static void verify_ctdb_reply_control_data(struct ctdb_reply_control_data *cd,
                verify_ctdb_node_map(cd->data.nodemap, cd2->data.nodemap);
                break;
 
+       case CTDB_CONTROL_DB_PULL:
+               assert(cd->data.num_records == cd2->data.num_records);
+               break;
+
+       case CTDB_CONTROL_DB_PUSH_CONFIRM:
+               assert(cd->data.num_records == cd2->data.num_records);
+               break;
+
        }
 }
 

diff --git a/ctdb/tests/src/protocol_types_test.c b/ctdb/tests/src/protocol_types_test.c
index bab104c583d669b5faf941dcb30d3ae5cfb473b5..55ee743def57111e7f9f693ff5d611ce6e070b90 100644 (file)
--- a/ctdb/tests/src/protocol_types_test.c
+++ b/ctdb/tests/src/protocol_types_test.c
@@ -182,6 +182,22 @@ static void verify_ctdb_pulldb(struct ctdb_pulldb *p1, struct ctdb_pulldb *p2)
        assert(p1->lmaster == p2->lmaster);
 }
 
+static void fill_ctdb_pulldb_ext(TALLOC_CTX *mem_ctx,
+                                struct ctdb_pulldb_ext *p)
+{
+       p->db_id = rand32();
+       p->lmaster = rand32();
+       p->srvid = rand64();
+}
+
+static void verify_ctdb_pulldb_ext(struct ctdb_pulldb_ext *p1,
+                                  struct ctdb_pulldb_ext *p2)
+{
+       assert(p1->db_id == p2->db_id);
+       assert(p1->lmaster == p2->lmaster);
+       assert(p1->srvid == p2->srvid);
+}
+
 static void fill_ctdb_ltdb_header(TALLOC_CTX *mem_ctx,
                                  struct ctdb_ltdb_header *p)
 {
@@ -1177,6 +1193,7 @@ DEFINE_TEST(struct ctdb_statistics, ctdb_statistics);
 DEFINE_TEST(struct ctdb_vnn_map, ctdb_vnn_map);
 DEFINE_TEST(struct ctdb_dbid_map, ctdb_dbid_map);
 DEFINE_TEST(struct ctdb_pulldb, ctdb_pulldb);
+DEFINE_TEST(struct ctdb_pulldb_ext, ctdb_pulldb_ext);
 DEFINE_TEST(struct ctdb_rec_data, ctdb_rec_data);
 DEFINE_TEST(struct ctdb_rec_buffer, ctdb_rec_buffer);
 DEFINE_TEST(struct ctdb_traverse_start, ctdb_traverse_start);
@@ -1240,6 +1257,7 @@ int main(int argc, char *argv[])
        TEST_FUNC(ctdb_vnn_map)();
        TEST_FUNC(ctdb_dbid_map)();
        TEST_FUNC(ctdb_pulldb)();
+       TEST_FUNC(ctdb_pulldb_ext)();
        TEST_FUNC(ctdb_rec_data)();
        TEST_FUNC(ctdb_rec_buffer)();
        TEST_FUNC(ctdb_traverse_start)();