ctdb/server/ctdb_persistent.c

   1 /*
   2    persistent store logic
   3
   4    Copyright (C) Andrew Tridgell  2007
   5    Copyright (C) Ronnie Sahlberg  2007
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program; if not, see <http://www.gnu.org/licenses/>.
  19 */
  20
  21 #include "includes.h"
  22 #include "system/filesys.h"
  23 #include "system/wait.h"
  24 #include "db_wrap.h"
  25 #include "tdb.h"
  26 #include "../include/ctdb_private.h"
  27
  28 struct ctdb_persistent_state {
  29         struct ctdb_context *ctdb;
  30         struct ctdb_db_context *ctdb_db; /* used by trans3_commit */
  31         struct ctdb_client *client; /* used by trans3_commit */
  32         struct ctdb_req_control *c;
  33         const char *errormsg;
  34         uint32_t num_pending;
  35         int32_t status;
  36         uint32_t num_failed, num_sent;
  37 };
  38
  39 /*
  40   1) all nodes fail, and all nodes reply
  41   2) some nodes fail, all nodes reply
  42   3) some nodes timeout
  43   4) all nodes succeed
  44  */
  45
  46 /*
  47   called when a node has acknowledged a ctdb_control_update_record call
  48  */
  49 static void ctdb_persistent_callback(struct ctdb_context *ctdb,
  50                                      int32_t status, TDB_DATA data,
  51                                      const char *errormsg,
  52                                      void *private_data)
  53 {
  54         struct ctdb_persistent_state *state = talloc_get_type(private_data,
  55                                                               struct ctdb_persistent_state);
  56
  57         if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
  58                 DEBUG(DEBUG_INFO, ("ctdb_persistent_callback: ignoring reply "
  59                                    "during recovery\n"));
  60                 return;
  61         }
  62
  63         if (status != 0) {
  64                 DEBUG(DEBUG_ERR,("ctdb_persistent_callback failed with status %d (%s)\n",
  65                          status, errormsg?errormsg:"no error message given"));
  66                 state->status = status;
  67                 state->errormsg = errormsg;
  68                 state->num_failed++;
  69
  70                 /*
  71                  * If a node failed to complete the update_record control,
  72                  * then either a recovery is already running or something
  73                  * bad is going on. So trigger a recovery and let the
  74                  * recovery finish the transaction, sending back the reply
  75                  * for the trans3_commit control to the client.
  76                  */
  77                 ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
  78                 return;
  79         }
  80
  81         state->num_pending--;
  82
  83         if (state->num_pending != 0) {
  84                 return;
  85         }
  86
  87         ctdb_request_control_reply(state->ctdb, state->c, NULL, 0, state->errormsg);
  88         talloc_free(state);
  89 }
  90
  91 /*
  92   called if persistent store times out
  93  */
  94 static void ctdb_persistent_store_timeout(struct event_context *ev, struct timed_event *te,
  95                                          struct timeval t, void *private_data)
  96 {
  97         struct ctdb_persistent_state *state = talloc_get_type(private_data, struct ctdb_persistent_state);
  98
  99         if (state->ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
 100                 DEBUG(DEBUG_INFO, ("ctdb_persistent_store_timeout: ignoring "
 101                                    "timeout during recovery\n"));
 102                 return;
 103         }
 104
 105         ctdb_request_control_reply(state->ctdb, state->c, NULL, 1,
 106                                    "timeout in ctdb_persistent_state");
 107
 108         talloc_free(state);
 109 }
 110
 111 /**
 112  * Finish pending trans3 commit controls, i.e. send
 113  * reply to the client. This is called by the end-recovery
 114  * control to fix the situation when a recovery interrupts
 115  * the usual progress of a transaction.
 116  */
 117 void ctdb_persistent_finish_trans3_commits(struct ctdb_context *ctdb)
 118 {
 119         struct ctdb_db_context *ctdb_db;
 120
 121         if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
 122                 DEBUG(DEBUG_INFO, ("ctdb_persistent_finish_trans3_commits: "
 123                                    "skipping execution when recovery is "
 124                                    "active\n"));
 125                 return;
 126         }
 127
 128         for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
 129                 struct ctdb_persistent_state *state;
 130
 131                 if (ctdb_db->persistent_state == NULL) {
 132                         continue;
 133                 }
 134
 135                 state = ctdb_db->persistent_state;
 136
 137                 ctdb_request_control_reply(ctdb, state->c, NULL, 2,
 138                                            "trans3 commit ended by recovery");
 139
 140                 /* The destructor sets ctdb_db->persistent_state to NULL. */
 141                 talloc_free(state);
 142         }
 143 }
 144
 145 static int ctdb_persistent_state_destructor(struct ctdb_persistent_state *state)
 146 {
 147         if (state->client != NULL) {
 148                 state->client->db_id = 0;
 149         }
 150
 151         if (state->ctdb_db != NULL) {
 152                 state->ctdb_db->persistent_state = NULL;
 153         }
 154
 155         return 0;
 156 }
 157
 158 /*
 159  * Store a set of persistent records.
 160  * This is used to roll out a transaction to all nodes.
 161  */
 162 int32_t ctdb_control_trans3_commit(struct ctdb_context *ctdb,
 163                                    struct ctdb_req_control *c,
 164                                    TDB_DATA recdata, bool *async_reply)
 165 {
 166         struct ctdb_client *client;
 167         struct ctdb_persistent_state *state;
 168         int i;
 169         struct ctdb_marshall_buffer *m = (struct ctdb_marshall_buffer *)recdata.dptr;
 170         struct ctdb_db_context *ctdb_db;
 171
 172         if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
 173                 DEBUG(DEBUG_INFO,("rejecting ctdb_control_trans3_commit when recovery active\n"));
 174                 return -1;
 175         }
 176
 177         client = ctdb_reqid_find(ctdb, c->client_id, struct ctdb_client);
 178         if (client == NULL) {
 179                 DEBUG(DEBUG_ERR,(__location__ " can not match persistent_store "
 180                                  "to a client. Returning error\n"));
 181                 return -1;
 182         }
 183
 184         if (client->db_id != 0) {
 185                 DEBUG(DEBUG_ERR,(__location__ " ERROR: trans3_commit: "
 186                                  "client-db_id[0x%08x] != 0 "
 187                                  "(client_id[0x%08x]): trans3_commit active?\n",
 188                                  client->db_id, client->client_id));
 189                 return -1;
 190         }
 191
 192         ctdb_db = find_ctdb_db(ctdb, m->db_id);
 193         if (ctdb_db == NULL) {
 194                 DEBUG(DEBUG_ERR,(__location__ " ctdb_control_trans3_commit: "
 195                                  "Unknown database db_id[0x%08x]\n", m->db_id));
 196                 return -1;
 197         }
 198
 199         if (ctdb_db->persistent_state != NULL) {
 200                 DEBUG(DEBUG_ERR, (__location__ " Error: "
 201                                   "ctdb_control_trans3_commit "
 202                                   "called while a transaction commit is "
 203                                   "active. db_id[0x%08x]\n", m->db_id));
 204                 return -1;
 205         }
 206
 207         ctdb_db->persistent_state = talloc_zero(ctdb_db,
 208                                                 struct ctdb_persistent_state);
 209         CTDB_NO_MEMORY(ctdb, ctdb_db->persistent_state);
 210
 211         client->db_id = m->db_id;
 212
 213         state = ctdb_db->persistent_state;
 214         state->ctdb = ctdb;
 215         state->ctdb_db = ctdb_db;
 216         state->c    = c;
 217         state->client = client;
 218
 219         talloc_set_destructor(state, ctdb_persistent_state_destructor);
 220
 221         for (i = 0; i < ctdb->vnn_map->size; i++) {
 222                 struct ctdb_node *node = ctdb->nodes[ctdb->vnn_map->map[i]];
 223                 int ret;
 224
 225                 /* only send to active nodes */
 226                 if (node->flags & NODE_FLAGS_INACTIVE) {
 227                         continue;
 228                 }
 229
 230                 ret = ctdb_daemon_send_control(ctdb, node->pnn, 0,
 231                                                CTDB_CONTROL_UPDATE_RECORD,
 232                                                c->client_id, 0, recdata,
 233                                                ctdb_persistent_callback,
 234                                                state);
 235                 if (ret == -1) {
 236                         DEBUG(DEBUG_ERR,("Unable to send "
 237                                          "CTDB_CONTROL_UPDATE_RECORD "
 238                                          "to pnn %u\n", node->pnn));
 239                         talloc_free(state);
 240                         return -1;
 241                 }
 242
 243                 state->num_pending++;
 244                 state->num_sent++;
 245         }
 246
 247         if (state->num_pending == 0) {
 248                 talloc_free(state);
 249                 return 0;
 250         }
 251
 252         /* we need to wait for the replies */
 253         *async_reply = true;
 254
 255         /* need to keep the control structure around */
 256         talloc_steal(state, c);
 257
 258         /* but we won't wait forever */
 259         event_add_timed(ctdb->ev, state,
 260                         timeval_current_ofs(ctdb->tunable.control_timeout, 0),
 261                         ctdb_persistent_store_timeout, state);
 262
 263         return 0;
 264 }
 265
 266
 267 /*
 268   backwards compatibility:
 269
 270   start a persistent store operation. passing both the key, header and
 271   data to the daemon. If the client disconnects before it has issued
 272   a persistent_update call to the daemon we trigger a full recovery
 273   to ensure the databases are brought back in sync.
 274   for now we ignore the recdata that the client has passed to us.
 275  */
 276 int32_t ctdb_control_start_persistent_update(struct ctdb_context *ctdb,
 277                                       struct ctdb_req_control *c,
 278                                       TDB_DATA recdata)
 279 {
 280         struct ctdb_client *client = ctdb_reqid_find(ctdb, c->client_id, struct ctdb_client);
 281
 282         if (client == NULL) {
 283                 DEBUG(DEBUG_ERR,(__location__ " can not match start_persistent_update to a client. Returning error\n"));
 284                 return -1;
 285         }
 286
 287         client->num_persistent_updates++;
 288
 289         return 0;
 290 }
 291
 292 /*
 293   backwards compatibility:
 294
 295   called to tell ctdbd that it is no longer doing a persistent update
 296 */
 297 int32_t ctdb_control_cancel_persistent_update(struct ctdb_context *ctdb,
 298                                               struct ctdb_req_control *c,
 299                                               TDB_DATA recdata)
 300 {
 301         struct ctdb_client *client = ctdb_reqid_find(ctdb, c->client_id, struct ctdb_client);
 302
 303         if (client == NULL) {
 304                 DEBUG(DEBUG_ERR,(__location__ " can not match cancel_persistent_update to a client. Returning error\n"));
 305                 return -1;
 306         }
 307
 308         if (client->num_persistent_updates > 0) {
 309                 client->num_persistent_updates--;
 310         }
 311
 312         return 0;
 313 }
 314
 315 static int32_t ctdb_get_db_seqnum(struct ctdb_context *ctdb,
 316                                   uint32_t db_id,
 317                                   uint64_t *seqnum)
 318 {
 319         int32_t ret;
 320         struct ctdb_db_context *ctdb_db;
 321         const char *keyname = CTDB_DB_SEQNUM_KEY;
 322         TDB_DATA key;
 323         TDB_DATA data;
 324         TALLOC_CTX *mem_ctx = talloc_new(ctdb);
 325         struct ctdb_ltdb_header header;
 326
 327         ctdb_db = find_ctdb_db(ctdb, db_id);
 328         if (!ctdb_db) {
 329                 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", db_id));
 330                 ret = -1;
 331                 goto done;
 332         }
 333
 334         key.dptr = (uint8_t *)discard_const(keyname);
 335         key.dsize = strlen(keyname) + 1;
 336
 337         ret = (int32_t)ctdb_ltdb_fetch(ctdb_db, key, &header, mem_ctx, &data);
 338         if (ret != 0) {
 339                 goto done;
 340         }
 341
 342         if (data.dsize != sizeof(uint64_t)) {
 343                 *seqnum = 0;
 344                 goto done;
 345         }
 346
 347         *seqnum = *(uint64_t *)data.dptr;
 348
 349 done:
 350         talloc_free(mem_ctx);
 351         return ret;
 352 }
 353
 354 /**
 355  * Get the sequence number of a persistent database.
 356  */
 357 int32_t ctdb_control_get_db_seqnum(struct ctdb_context *ctdb,
 358                                    TDB_DATA indata,
 359                                    TDB_DATA *outdata)
 360 {
 361         uint32_t db_id;
 362         int32_t ret;
 363         uint64_t seqnum;
 364
 365         db_id = *(uint32_t *)indata.dptr;
 366         ret = ctdb_get_db_seqnum(ctdb, db_id, &seqnum);
 367         if (ret != 0) {
 368                 goto done;
 369         }
 370
 371         outdata->dsize = sizeof(uint64_t);
 372         outdata->dptr = (uint8_t *)talloc_zero(outdata, uint64_t);
 373         if (outdata->dptr == NULL) {
 374                 ret = -1;
 375                 goto done;
 376         }
 377
 378         *(outdata->dptr) = seqnum;
 379
 380 done:
 381         return ret;
 382 }