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>

7 files changed, 226 insertions, 0 deletions

diff --git a/ctdb/protocol/protocol.h b/ctdb/protocol/protocol.h
index 1b48c3920fa..e4bc3b9a0d4 100644
--- 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 8fa74e16426..d07b4198d0a 100644
--- 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 7c983b80001..6466fcaf901 100644
--- 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 d57167b64ff..57737ab678f 100644
--- 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 7d4fe3b24dd..47cf855f52b 100644
--- 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 4719eac8be7..659594b4171 100644
--- 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 bab104c583d..55ee743def5 100644
--- 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)();