X-Git-Url: https://pd.if.org/git/?p=nbds;a=blobdiff_plain;f=txn%2Ftxn.c;h=1cc8e2f84fdeff975d719bfada0608bed62c42bb;hp=2893d35dbcfb40ec5152775146f59224cb96034d;hb=f3eb4799a11ceaeb47ab02034595b5d641c2f1c9;hpb=2cce67f0002cdb6dcdc2ab8ccf837e3d2b3336de diff --git a/txn/txn.c b/txn/txn.c index 2893d35..1cc8e2f 100644 --- a/txn/txn.c +++ b/txn/txn.c @@ -5,6 +5,7 @@ #include "common.h" #include "txn.h" #include "mem.h" +#include "rcu.h" #include "skiplist.h" #define UNDETERMINED_VERSION 0 @@ -12,9 +13,10 @@ #define INITIAL_WRITES_SIZE 4 typedef struct update_rec update_t; +typedef map_key_t version_t; struct update_rec { - uint64_t version; + version_t version; map_val_t value; map_val_t next; // an earlier update }; @@ -25,19 +27,19 @@ typedef struct write_rec { } write_rec_t; struct txn { - uint64_t rv; - uint64_t wv; + version_t rv; + version_t wv; map_t *map; write_rec_t *writes; - uint64_t writes_size; - uint64_t writes_count; - uint64_t writes_scan; + size_t writes_size; + size_t writes_count; + size_t writes_scan; txn_state_e state; }; static txn_state_e txn_validate (txn_t *txn); -static uint64_t version_ = 1; +static version_t version_ = 1; static skiplist_t *active_ = NULL; @@ -65,7 +67,7 @@ static txn_state_e validate_key (txn_t *txn, map_key_t key) { // will eventually conflict with it and abort. if (!IS_TAGGED(val, TAG2)) return TXN_VALIDATED; - update = (update_t *)(size_t)STRIP_TAG(val, TAG2); + update = (update_t *)STRIP_TAG(val, TAG2); if (!IS_TAGGED(update->version, TAG1)) return (update->version <= txn->rv) ? TXN_VALIDATED : TXN_ABORTED; @@ -77,7 +79,7 @@ static txn_state_e validate_key (txn_t *txn, map_key_t key) { continue; // The update's transaction is still in progress. Access its txn_t. - txn_t *writer = (txn_t *)(size_t)STRIP_TAG(update->version, TAG1); + txn_t *writer = (txn_t *)STRIP_TAG(update->version, TAG1); if (writer == txn) continue; // Skip our own updates. txn_state_e writer_state = writer->state; @@ -118,7 +120,7 @@ static txn_state_e txn_validate (txn_t *txn) { case TXN_VALIDATING: if (txn->wv == UNDETERMINED_VERSION) { - uint64_t wv = SYNC_ADD(&version_, 1); + version_t wv = SYNC_ADD(&version_, 1); SYNC_CAS(&txn->wv, UNDETERMINED_VERSION, wv); } @@ -165,11 +167,11 @@ txn_t *txn_begin (map_t *map) { do { txn->rv = version_; - uint64_t old_count; - uint64_t temp = 0; + unsigned old_count; + unsigned temp = 0; do { old_count = temp; - temp = (uint64_t)sl_cas(active_, (map_key_t)txn->rv, old_count, old_count + 1); + temp = sl_cas(active_, txn->rv, old_count, old_count + 1); } while (temp != old_count); if (txn->rv == version_) @@ -195,8 +197,8 @@ void txn_abort (txn_t *txn) { update->version = ABORTED_VERSION; } - nbd_defer_free(txn->writes); - nbd_defer_free(txn); + rcu_defer_free(txn->writes); + rcu_defer_free(txn); } txn_state_e txn_commit (txn_t *txn) { @@ -208,7 +210,7 @@ txn_state_e txn_commit (txn_t *txn) { txn_state_e state = txn_validate(txn); // Detach from its updates. - uint64_t wv = (txn->state == TXN_ABORTED) ? ABORTED_VERSION : txn->wv; + version_t wv = (txn->state == TXN_ABORTED) ? ABORTED_VERSION : txn->wv; int i; for (i = 0; i < txn->writes_count; ++i) { update_t *update = (update_t *)txn->writes[i].rec; @@ -216,8 +218,8 @@ txn_state_e txn_commit (txn_t *txn) { } // Lower the reference count for 's read version - uint64_t temp = 2; - uint64_t old_count; + unsigned temp = 2; + unsigned old_count; do { old_count = temp; temp = sl_cas(active_, (map_key_t)txn->rv, old_count, old_count - 1); @@ -227,27 +229,27 @@ txn_state_e txn_commit (txn_t *txn) { } } while (old_count != temp); - nbd_defer_free(txn->writes); - nbd_defer_free(txn); + rcu_defer_free(txn->writes); + rcu_defer_free(txn); return state; } // Get most recent committed version prior to our read version. -uint64_t txn_map_get (txn_t *txn, map_key_t key) { +map_val_t txn_map_get (txn_t *txn, map_key_t key) { if (txn->state != TXN_RUNNING) return ERROR_TXN_NOT_RUNNING; // Iterate through the update records to find the latest committed version prior to our read version. - uint64_t newest_val = map_get(txn->map, key); - uint64_t val = newest_val; + map_val_t newest_val = map_get(txn->map, key); + map_val_t val = newest_val; update_t *update = NULL; for ( ; ; val = update->next) { if (!IS_TAGGED(val, TAG2)) return val; - update = (update_t *)(size_t)STRIP_TAG(val, TAG2); + update = (update_t *)STRIP_TAG(val, TAG2); assert(update != NULL); // If the update's version is not tagged it means the update is committed. @@ -265,7 +267,7 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { continue; // The update's transaction is still in progress. Access its txn_t. - txn_t *writer = (txn_t *)(size_t)STRIP_TAG(update->version, TAG1); + txn_t *writer = (txn_t *)STRIP_TAG(update->version, TAG1); if (writer == txn) // found our own update break; // success @@ -289,14 +291,14 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { break; // success } - uint64_t value = update->value; + map_val_t value = update->value; // collect some garbage - uint64_t min_active_version = UNDETERMINED_VERSION; + version_t min_active_version = UNDETERMINED_VERSION; update_t *next_update = NULL; if (IS_TAGGED(update->next, TAG2)) { - next_update = (update_t *)(size_t)STRIP_TAG(update->next, TAG2); - min_active_version = (uint64_t)sl_min_key(active_); + next_update = (update_t *)STRIP_TAG(update->next, TAG2); + min_active_version = (version_t)sl_min_key(active_); if (next_update->version < min_active_version) { // (and all update records following it [execpt if it is aborted]) is old enough that it is // not visible to any active transaction. We can safely free it. @@ -305,11 +307,11 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { update_t *temp = next_update; while (temp->version == ABORTED_VERSION) { assert(!IS_TAGGED(temp->version, TAG1)); - uint64_t next = next_update->next; + map_val_t next = next_update->next; if (!IS_TAGGED(next, TAG2)) break; - temp = (update_t *)(size_t)STRIP_TAG(next, TAG2); + temp = (update_t *)STRIP_TAG(next, TAG2); if (temp->version >= min_active_version) return value; } @@ -317,7 +319,7 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { // free and all the update records following it temp = next_update; while (1) { - uint64_t next = SYNC_SWAP(&temp->next, DOES_NOT_EXIST); + map_val_t next = SYNC_SWAP(&temp->next, DOES_NOT_EXIST); // if we find ourself in a race just back off and let the other thread take care of it if (next == DOES_NOT_EXIST) @@ -326,7 +328,7 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { if (!IS_TAGGED(next, TAG2)) break; - temp = (update_t *)(size_t)STRIP_TAG(next, TAG2); + temp = (update_t *)STRIP_TAG(next, TAG2); nbd_free(update); } } @@ -336,11 +338,11 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { // There is no need for an update record. if (next_update == NULL && val == newest_val) { if (min_active_version == UNDETERMINED_VERSION) { - min_active_version = (uint64_t)sl_min_key(active_); + min_active_version = (version_t)sl_min_key(active_); } if (update->version <= min_active_version) { if (map_cas(txn->map, key, TAG_VALUE(val, TAG2), value) == TAG_VALUE(val, TAG2)) { - nbd_defer_free(update); + rcu_defer_free(update); } } } @@ -348,21 +350,21 @@ uint64_t txn_map_get (txn_t *txn, map_key_t key) { return value; } -void txn_map_set (txn_t *txn, map_key_t key, uint64_t value) { +void txn_map_set (txn_t *txn, map_key_t key, map_val_t value) { if (txn->state != TXN_RUNNING) return; // TODO: return some sort of error code // create a new update record update_t *update = alloc_update_rec(); update->value = value; - update->version = TAG_VALUE((uint64_t)(size_t)txn, TAG1); + update->version = TAG_VALUE((version_t)txn, TAG1); // push the new update record onto 's update list - uint64_t old_update; + map_val_t old_update; do { old_update = map_get(txn->map, key); update->next = old_update; - } while (map_cas(txn->map, key, old_update, TAG_VALUE((uint64_t)(size_t)update, TAG2)) != old_update); + } while (map_cas(txn->map, key, old_update, TAG_VALUE((map_val_t)update, TAG2)) != old_update); // add to the write set for commit-time validation if (txn->writes_count == txn->writes_size) {