server: add a new control CTDB_CONTROL_TRANS3_COMMIT

This is a simplified version of the trans2 commit control:
It just rolls out the marshall buffer to all active nodes.

It is the main ctdbd part of the re-implementation of the
persistent transactions. The client code is changed to
take a global lock to start a transactions and store into
the marshal buffer instead of writing to the local tdb
under a local transaction.

The old transaction implementation is going to be
removed in a later commit.

Michael

(This used to be ctdb commit f66428f9d2013080a414404c1ba6117888352fd6)

3 files changed, 93 insertions, 0 deletions

diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 142bbd5c71..0c611c287d 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -625,6 +625,7 @@ enum ctdb_controls {CTDB_CONTROL_PROCESS_EXISTS          = 0,
 		    CTDB_CONTROL_TRANS2_ACTIVE           = 116,
 		    CTDB_CONTROL_GET_LOG		 = 117,
 		    CTDB_CONTROL_CLEAR_LOG		 = 118,
+		    CTDB_CONTROL_TRANS3_COMMIT           = 119,
 };	
 
 /*
@@ -1426,6 +1427,10 @@ int32_t ctdb_control_trans2_commit(struct ctdb_context *ctdb,
 				   struct ctdb_req_control *c, 
 				   TDB_DATA recdata, bool *async_reply);
 
+int32_t ctdb_control_trans3_commit(struct ctdb_context *ctdb,
+				   struct ctdb_req_control *c,
+				   TDB_DATA recdata, bool *async_reply);
+
 int32_t ctdb_control_transaction_start(struct ctdb_context *ctdb, uint32_t id);
 int32_t ctdb_control_transaction_commit(struct ctdb_context *ctdb, uint32_t id);
 int32_t ctdb_control_transaction_cancel(struct ctdb_context *ctdb);
diff --git a/ctdb/server/ctdb_control.c b/ctdb/server/ctdb_control.c
index fcffca31ce..9cc5591d3d 100644
--- a/ctdb/server/ctdb_control.c
+++ b/ctdb/server/ctdb_control.c
@@ -428,6 +428,9 @@ static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb,
 		CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
 		return ctdb_control_trans2_active(ctdb, c, *(uint32_t *)indata.dptr);
 
+	case CTDB_CONTROL_TRANS3_COMMIT:
+		return ctdb_control_trans3_commit(ctdb, c, indata, async_reply);
+
 	case CTDB_CONTROL_RECD_PING:
 		CHECK_CONTROL_DATA_SIZE(0);
 		return ctdb_control_recd_ping(ctdb);
diff --git a/ctdb/server/ctdb_persistent.c b/ctdb/server/ctdb_persistent.c
index 7fc45877f3..c075a5d86a 100644
--- a/ctdb/server/ctdb_persistent.c
+++ b/ctdb/server/ctdb_persistent.c
@@ -242,6 +242,91 @@ int32_t ctdb_control_trans2_commit(struct ctdb_context *ctdb,
 }
 
 
+/*
+ * Store a set of persistent records.
+ * This is used to roll out a transaction to all nodes.
+ */
+int32_t ctdb_control_trans3_commit(struct ctdb_context *ctdb,
+				   struct ctdb_req_control *c,
+				   TDB_DATA recdata, bool *async_reply)
+{
+	struct ctdb_client *client;
+	struct ctdb_persistent_state *state;
+	int i;
+	struct ctdb_marshall_buffer *m = (struct ctdb_marshall_buffer *)recdata.dptr;
+	struct ctdb_db_context *ctdb_db;
+
+	if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
+		DEBUG(DEBUG_INFO,("rejecting ctdb_control_trans3_commit when recovery active\n"));
+		return -1;
+	}
+
+	ctdb_db = find_ctdb_db(ctdb, m->db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,(__location__ " ctdb_control_trans3_commit: "
+				 "Unknown database db_id[0x%08x]\n", m->db_id));
+		return -1;
+	}
+
+	client = ctdb_reqid_find(ctdb, c->client_id, struct ctdb_client);
+	if (client == NULL) {
+		DEBUG(DEBUG_ERR,(__location__ " can not match persistent_store "
+				 "to a client. Returning error\n"));
+		return -1;
+	}
+
+	state = talloc_zero(ctdb, struct ctdb_persistent_state);
+	CTDB_NO_MEMORY(ctdb, state);
+
+	state->ctdb = ctdb;
+	state->c    = c;
+
+	for (i = 0; i < ctdb->vnn_map->size; i++) {
+		struct ctdb_node *node = ctdb->nodes[ctdb->vnn_map->map[i]];
+		int ret;
+
+		/* only send to active nodes */
+		if (node->flags & NODE_FLAGS_INACTIVE) {
+			continue;
+		}
+
+		ret = ctdb_daemon_send_control(ctdb, node->pnn, 0,
+					       CTDB_CONTROL_UPDATE_RECORD,
+					       c->client_id, 0, recdata,
+					       ctdb_persistent_callback,
+					       state);
+		if (ret == -1) {
+			DEBUG(DEBUG_ERR,("Unable to send "
+					 "CTDB_CONTROL_UPDATE_RECORD "
+					 "to pnn %u\n", node->pnn));
+			talloc_free(state);
+			return -1;
+		}
+
+		state->num_pending++;
+		state->num_sent++;
+	}
+
+	if (state->num_pending == 0) {
+		talloc_free(state);
+		return 0;
+	}
+
+	/* we need to wait for the replies */
+	*async_reply = true;
+
+	/* need to keep the control structure around */
+	talloc_steal(state, c);
+
+	/* but we won't wait forever */
+	event_add_timed(ctdb->ev, state,
+			timeval_current_ofs(ctdb->tunable.control_timeout, 0),
+			ctdb_persistent_store_timeout, state);
+
+	return 0;
+}
+
+
 struct ctdb_persistent_write_state {
 	struct ctdb_db_context *ctdb_db;
 	struct ctdb_marshall_buffer *m;