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

diff --git a/include/ctdb_private.h b/include/ctdb_private.h
index 142bbd5c716b79159d5e64b17d50c6d97e5ead93..0c611c287dcd646b2f1d468f321560842b76d471 100644 (file)
--- a/include/ctdb_private.h
+++ b/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/server/ctdb_control.c b/server/ctdb_control.c
index fcffca31ceccb382ec081c0c436126fd42aed758..9cc5591d3d75f861c4e3aa08de174c9ac94db92b 100644 (file)
--- a/server/ctdb_control.c
+++ b/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/server/ctdb_persistent.c b/server/ctdb_persistent.c
index 7fc45877f3bd0c5b09c79a182c5890a410f03f0f..c075a5d86a8bc9ebfcba3635eae374399d5e1f09 100644 (file)
--- a/server/ctdb_persistent.c
+++ b/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;