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>

5 files changed, 160 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, "" },
 	};