pd.if.org Git - nbds/blob - txn/txn.c

   1 /*
   2  * Written by Josh Dybnis and released to the public domain, as explained at
   3  * http://creativecommons.org/licenses/publicdomain
   4  */
   5 #include "common.h"
   6 #include "txn.h"
   7 #include "mem.h"
   8 #include "rcu.h"
   9 #include "lwt.h"
  10 #include "skiplist.h"
  11
  12 #define UNDETERMINED_VERSION 0
  13 #define ABORTED_VERSION      TAG_VALUE(0, TAG1)
  14 #define INITIAL_WRITES_SIZE  4
  15 #define PTR_TO_VAL(x) ((size_t)(x) >> 2)
  16 #define VAL_TO_PTR(x) ((update_t *)((x) << 2))
  17
  18 typedef struct update_rec update_t;
  19 typedef map_key_t version_t;
  20
  21 struct update_rec {
  22     version_t version; // tagged versions are txn_t pointers, untagged are actual version numbers
  23     map_val_t value;
  24     map_val_t next; // an earlier update
  25 };
  26
  27 typedef struct write_rec {
  28     map_key_t key;
  29     update_t *rec;
  30 } write_rec_t;
  31
  32 struct txn {
  33     version_t rv;
  34     version_t wv;
  35     map_t *map;
  36     write_rec_t *writes;
  37     size_t writes_size;
  38     size_t writes_count;
  39     size_t validate_scan;
  40     txn_state_e state;
  41 };
  42
  43 static txn_state_e txn_validate (txn_t *txn);
  44
  45 static skiplist_t *active_ = NULL;
  46
  47 static version_t version_ = 1;
  48
  49 // Validate the updates for <key>. Validation fails if there is a write-write conflict. That is if after our
  50 // read version another transaction committed a change to an entry we are also trying to change.
  51 //
  52 // If we encounter a potential conflict with a transaction that is in the process of validating, we help it
  53 // complete validating. It must be finished before we can decide to rollback or commit.
  54 //
  55 static txn_state_e validate_key (txn_t *txn, map_key_t key) {
  56     assert(txn->state != TXN_RUNNING);
  57
  58     map_val_t val = map_get(txn->map, key);
  59     update_t *update = NULL;
  60     for (; val != DOES_NOT_EXIST; val = update->next) {
  61
  62         // If the update or its version is not tagged it means the update is committed.
  63         //
  64         // We can stop at the first committed record we find that is at least as old as our read version. All
  65         // the other committed records following it will be older. And all the uncommitted records following it
  66         // will eventually conflict with it and abort.
  67         if (!IS_TAGGED(val, TAG2))
  68             return TXN_VALIDATED;
  69         update = VAL_TO_PTR(val);
  70         if (!IS_TAGGED(update->version, TAG1))
  71             return (update->version <= txn->rv) ? TXN_VALIDATED : TXN_ABORTED;
  72
  73         // If the update's version is tagged then either the update was aborted or the the version number is
  74         // actually a pointer to a running transaction's txn_t.
  75
  76         // Skip aborted transactions.
  77         if (EXPECT_FALSE(update->version == ABORTED_VERSION))
  78             continue;
  79
  80         // The update's transaction is still in progress. Access its txn_t.
  81         txn_t *writer = (txn_t *)VAL_TO_PTR(update->version);
  82         if (writer == txn)
  83             continue; // Skip our own updates.
  84         txn_state_e writer_state = writer->state;
  85
  86         // Any running transaction will only be able to acquire a wv greater than ours. A transaction changes its
  87         // state to validating before aquiring a wv. We can ignore an unvalidated transaction if its version is
  88         // greater than ours. See the next comment below for the explination why.
  89         if (writer_state == TXN_RUNNING)
  90             continue;
  91
  92         // If <writer> has a later version than us we can safely ignore its updates. It will not commit until
  93         // we have completed validation (in order to remain non-blocking it will help us validate if necessary).
  94         // This protocol ensures a deterministic resolution to every conflict and avoids infinite ping-ponging
  95         // between validating two conflicting transactions.
  96         if (writer_state == TXN_VALIDATING) {
  97             if (writer->wv > txn->wv)
  98                 continue;
  99             // Help <writer> commit. We need to know if <writer> aborts or commits before we can decide what to
 100             // do. But we don't want to block, so we assist.
 101             writer_state = txn_validate(writer);
 102         }
 103
 104         // Skip updates from aborted transactions.
 105         if (writer_state == TXN_ABORTED)
 106             continue;
 107
 108         assert(writer_state == TXN_VALIDATED);
 109         return (writer->wv <= txn->rv) ? TXN_VALIDATED : TXN_ABORTED;
 110     }
 111
 112     return TXN_VALIDATED;
 113 }
 114
 115 static txn_state_e txn_validate (txn_t *txn) {
 116     assert(txn->state != TXN_RUNNING);
 117     switch (txn->state) {
 118
 119         case TXN_VALIDATING:
 120             if (txn->wv == UNDETERMINED_VERSION) {
 121                 version_t wv = SYNC_ADD(&version_, 1);
 122                 (void)SYNC_CAS(&txn->wv, UNDETERMINED_VERSION, wv);
 123             }
 124
 125             for (int i = 0; i < txn->writes_count; ++i) {
 126                 txn_state_e s = validate_key(txn, txn->writes[i].key);
 127                 if (s == TXN_ABORTED) {
 128                     txn->state = TXN_ABORTED;
 129                     break;
 130                 }
 131                 assert(s == TXN_VALIDATED);
 132             }
 133             if (txn->state == TXN_VALIDATING) {
 134                 txn->state =  TXN_VALIDATED;
 135             }
 136             break;
 137
 138         case TXN_VALIDATED:
 139         case TXN_ABORTED:
 140             break;
 141
 142         default:
 143             assert(FALSE);
 144     }
 145
 146     return txn->state;
 147 }
 148
 149 static update_t *alloc_update_rec (version_t ver, map_val_t val) {
 150     update_t *u = (update_t *)nbd_malloc(sizeof(update_t));
 151     u->version = ver;
 152     u->value = val;
 153     u->next = DOES_NOT_EXIST;
 154     return u;
 155 }
 156
 157 txn_t *txn_begin (map_t *map) {
 158     TRACE("x1", "txn_begin: map %p", map, 0);
 159     txn_t *txn = (txn_t *)nbd_malloc(sizeof(txn_t));
 160     memset(txn, 0, sizeof(txn_t));
 161     txn->wv = UNDETERMINED_VERSION;
 162     txn->state = TXN_RUNNING;
 163     txn->map = map;
 164     txn->writes = nbd_malloc(sizeof(*txn->writes) * INITIAL_WRITES_SIZE);
 165     txn->writes_size = INITIAL_WRITES_SIZE;
 166     if (EXPECT_FALSE(active_ == NULL)) {
 167         skiplist_t *a = sl_alloc(NULL);
 168         if (SYNC_CAS(&active_, NULL, a) != NULL) {
 169             sl_free(a);
 170         }
 171     }
 172
 173     // acquire the read version for txn. must be careful to avoid a race
 174     do {
 175         txn->rv = version_;
 176
 177         unsigned old_count;
 178         unsigned temp = 0;
 179         do {
 180             old_count = temp;
 181             temp = sl_cas(active_, txn->rv, old_count, old_count + 1);
 182         } while (temp != old_count);
 183
 184         if (txn->rv == version_)
 185             break;
 186
 187         temp = 1;
 188         do {
 189             old_count = temp;
 190             temp = sl_cas(active_, (map_key_t)txn->rv, old_count, old_count - 1);
 191         } while (temp != old_count);
 192     } while (1);
 193
 194     TRACE("x1", "txn_begin: returning new transaction %p (read version %p)", txn, txn->rv);
 195     return txn;
 196 }
 197
 198 void txn_abort (txn_t *txn) {
 199     if (txn->state != TXN_RUNNING)
 200         return;
 201
 202     int i;
 203     for (i = 0; i < txn->writes_count; ++i) {
 204         update_t *update = (update_t *)txn->writes[i].rec;
 205         update->version = ABORTED_VERSION;
 206     }
 207
 208     rcu_defer_free(txn->writes);
 209     rcu_defer_free(txn);
 210 }
 211
 212 txn_state_e txn_commit (txn_t *txn) {
 213     if (txn->state != TXN_RUNNING)
 214         return txn->state;
 215
 216     assert(txn->state == TXN_RUNNING);
 217     txn->state = TXN_VALIDATING;
 218     txn_state_e state = txn_validate(txn);
 219
 220     // Detach <txn> from its updates.
 221     version_t wv = (txn->state == TXN_ABORTED) ? ABORTED_VERSION : txn->wv;
 222     int i;
 223     for (i = 0; i < txn->writes_count; ++i) {
 224         update_t *update = txn->writes[i].rec;
 225         update->version = wv;
 226     }
 227
 228     // Lower the reference count for <txn>'s read version
 229     unsigned temp = 2;
 230     unsigned old_count;
 231     do {
 232         old_count = temp;
 233         temp = sl_cas(active_, (map_key_t)txn->rv, old_count, old_count - 1);
 234         if (temp == 1 && txn->rv != version_) {
 235             sl_remove(active_, (map_key_t)txn->rv);
 236             break;
 237         }
 238     } while (old_count != temp);
 239
 240     rcu_defer_free(txn->writes);
 241     rcu_defer_free(txn);
 242
 243     return state;
 244 }
 245
 246 // Get most recent committed version prior to our read version.
 247 map_val_t txn_map_get (txn_t *txn, map_key_t key) {
 248     TRACE("x1", "txn_map_get: txn %p map %p", txn, txn->map);
 249     TRACE("x1", "txn_map_get: key %p", key, 0);
 250
 251     if (txn->state != TXN_RUNNING) {
 252         TRACE("x1", "txn_map_get: error txn not running (state %p)", txn->state, 0);
 253         return ERROR_TXN_NOT_RUNNING;
 254     }
 255
 256     // Iterate through the update records to find the latest committed version prior to our read version.
 257     map_val_t newest_val = map_get(txn->map, key);
 258     map_val_t val = newest_val;
 259     update_t *update;
 260     for ( ; (update = VAL_TO_PTR(val)) != NULL ; val = update->next) {
 261
 262         // If TAG2 is set in <val> it indicates that <val> is an update record. Otherwise all the following are
 263         // true: <val> is a literal value, it is older than any currently active transaction, and it is the most
 264         // recently set value for its key. Therefore it is visible to <txn>.
 265         if (!IS_TAGGED(val, TAG2)) {
 266             TRACE("x1", "txn_map_get: found untagged value; returning %p", val, 0);
 267             return val;
 268         }
 269
 270         // If the update's version is not tagged it means the update is committed.
 271         if (!IS_TAGGED(update->version, TAG1)) {
 272             if (update->version <= txn->rv) {
 273                 TRACE("x2", "txn_map_get: found committed update %p (version %p)", update, update->version);
 274                 break; // success
 275             }
 276             TRACE("x2", "txn_map_get: skipping update %p (version %p)", update, update->version);
 277             continue;
 278         }
 279
 280         // If the update's version is tagged then either the update was aborted or the the version number is
 281         // actually a pointer to a running transaction's txn_t.
 282
 283         // Skip updates from aborted transactions.
 284         if (EXPECT_FALSE(update->version == ABORTED_VERSION)) {
 285             TRACE("x2", "txn_map_get: skipping aborted update %p", update, 0);
 286             continue;
 287         }
 288
 289         // The update's transaction is still in progress. Access its txn_t.
 290         txn_t *writer = (txn_t *)VAL_TO_PTR(update->version);
 291         if (writer == txn) {
 292             TRACE("x2", "txn_map_get: found txn's own update %p", update, 0);
 293             break; // success
 294         }
 295
 296         txn_state_e writer_state = writer->state;
 297         if (writer_state == TXN_RUNNING) {
 298             TRACE("x2", "txn_map_get: skipping update %p of in-progress transaction %p", update, writer);
 299             continue;
 300         }
 301
 302         if (writer_state == TXN_VALIDATING) {
 303             TRACE("x2", "txn_map_get: update %p transaction %p validating", update, writer);
 304             if (writer->wv > txn->rv)
 305                 continue;
 306             writer_state = txn_validate(writer);
 307         }
 308
 309         // Skip updates from aborted transactions.
 310         if (writer_state == TXN_ABORTED) {
 311             TRACE("x2", "txn_map_get: skipping aborted update %p", update, 0);
 312             continue;
 313         }
 314
 315         assert(writer_state == TXN_VALIDATED);
 316         if (writer->wv > txn->rv) {
 317             TRACE("x2", "txn_map_get: skipping update %p (version %p)", update, update->version);
 318             continue;
 319         }
 320         break; // success
 321     }
 322
 323     if (update == NULL) {
 324         TRACE("x1", "txn_map_get: key does not exist in map", key, 0);
 325         return DOES_NOT_EXIST;
 326     }
 327
 328     map_val_t value = update->value;
 329     TRACE("x1", "txn_map_get: key found returning value %p", value, 0);
 330
 331     // collect some garbage
 332     version_t min_active_version = UNDETERMINED_VERSION;
 333     update_t *next_update = NULL;
 334     if (IS_TAGGED(update->next, TAG2)) {
 335         next_update = VAL_TO_PTR(update->next);
 336
 337         // If <next_update> (and all update records following it [execpt if it is aborted]) is old enough
 338         // that it is not visible to any active transaction we can safely free it.
 339         min_active_version = (version_t)sl_min_key(active_);
 340         if (next_update->version < min_active_version) {
 341
 342             // If the <next_update> is aborted, skip over it to look for more recent ones that may follow
 343             update_t *temp = next_update;
 344             while (temp->version == ABORTED_VERSION) {
 345                 assert(!IS_TAGGED(temp->version, TAG1));
 346                 map_val_t next = temp->next;
 347                 if (!IS_TAGGED(next, TAG2))
 348                     break;
 349
 350                 // Bail out of garbage collection if we find a record that might still be accessed by an
 351                 // ongoing transaction.
 352                 if (VAL_TO_PTR(next)->version >= min_active_version)
 353                     return value;
 354
 355                 temp = VAL_TO_PTR(next);
 356             }
 357
 358             // free the next update record and all the ones following it
 359             temp = next_update;
 360             map_val_t next;
 361             do {
 362                 next = SYNC_SWAP(&temp->next, DOES_NOT_EXIST);
 363
 364                 // if we find ourself in a race just back off and let the other thread take care of it
 365                 if (next == DOES_NOT_EXIST)
 366                     return value;
 367
 368                 nbd_free(temp);
 369
 370                 temp = VAL_TO_PTR(next);
 371
 372             } while (IS_TAGGED(next, TAG2));
 373         }
 374     }
 375
 376     // If there is one item left and it is visible by all active transactions we can merge it into the map itself.
 377     // There is no need for an update record.
 378     if (next_update == NULL && val == newest_val) {
 379         if (min_active_version == UNDETERMINED_VERSION) {
 380             min_active_version = (version_t)sl_min_key(active_);
 381         }
 382         if (update->version <= min_active_version) {
 383             if (map_cas(txn->map, key, TAG_VALUE(val, TAG2), value) == TAG_VALUE(val, TAG2)) {
 384                 rcu_defer_free(update);
 385             }
 386         }
 387     }
 388
 389     return value;
 390 }
 391
 392 void txn_map_set (txn_t *txn, map_key_t key, map_val_t value) {
 393     TRACE("x1", "txn_map_set: txn %p map %p", txn, txn->map);
 394     TRACE("x1", "txn_map_set: key %p value %p", key, value);
 395     assert(!IS_TAGGED(value, TAG1) && !IS_TAGGED(value, TAG2));
 396
 397     if (txn->state != TXN_RUNNING) {
 398         TRACE("x1", "txn_map_set: error txn not running (state %p)", txn->state, 0);
 399         return;
 400     }
 401
 402     // create a new update record
 403     version_t ver = TAG_VALUE(PTR_TO_VAL(txn), TAG1); // tagged versions are txn_t pointers
 404     update_t *update = alloc_update_rec(ver, value);
 405
 406     // push the new update record onto <key>'s update list
 407     map_val_t old_update = map_get(txn->map, key);
 408     TRACE("x2", "txn_map_set: old update %p new update record %p", old_update, update);
 409     do {
 410         update->next = old_update;
 411         map_val_t temp = map_cas(txn->map, key, old_update, TAG_VALUE(PTR_TO_VAL(update), TAG2));
 412         if (temp == old_update)
 413             break;
 414
 415         TRACE("x1", "txn_map_set: cas failed; found %p expected %p", temp, old_update);
 416         old_update = temp;
 417     } while (1);
 418
 419     // add <key> to the write set for commit-time validation
 420     if (txn->writes_count == txn->writes_size) {
 421         write_rec_t *w = nbd_malloc(sizeof(write_rec_t) * txn->writes_size * 2);
 422         memcpy(w, txn->writes, txn->writes_size * sizeof(write_rec_t));
 423         txn->writes_size *= 2;
 424         nbd_free(txn->writes);
 425         txn->writes = w;
 426     }
 427     int i = txn->writes_count++;
 428     txn->writes[i].key = key;
 429     txn->writes[i].rec = update;
 430 }