-static txn_state_t txn_validate (txn_t *txn);
-
-update_rec_t *alloc_update_rec (void) {
- update_rec_t *u = (update_rec_t *)nbd_malloc(sizeof(update_rec_t));
- memset(u, 0, sizeof(update_rec_t));
- return u;
-}
-
-txn_t *txn_begin (txn_access_t access, txn_isolation_t isolation, hashtable_t *ht) {
- txn_t *txn = (txn_t *)nbd_malloc(sizeof(txn_t));
- memset(txn, 0, sizeof(txn_t));
- txn->access = access;
- txn->isolation = isolation;
- txn->rv = GlobalVersion;
- txn->wv = UNDETERMINED_VERSION;
- txn->state = TXN_RUNNING;
- txn->ht = ht;
- if (isolation != TXN_READ_ONLY) {
- txn->writes = nbd_malloc(sizeof(*txn->writes) * INITIAL_WRITES_SIZE);
- txn->writes_size = INITIAL_WRITES_SIZE;
- }
- return txn;
-}
-
-// Get most recent committed version prior to our read version.
-int64_t txn_ht_get (txn_t *txn, const char *key, uint32_t key_len) {
-
- // Iterate through update records associated with <key> to find the latest committed version.
- // We can use the first matching version. Older updates always come later in the list.
- update_rec_t *update = (update_rec_t *) ht_get(txn->ht, key, key_len);
- for (; update != NULL; update = update->prev) {
- uint64_t writer_version = update->version;
- if (writer_version < txn->rv)
- return update->value;
-
- // If the version is tagged, it means that it is not a version number, but a pointer to an
- // in progress transaction.
- if (IS_TAGGED(update->version)) {
- txn_t *writer = (txn_t *)STRIP_TAG(writer_version);
-
- if (writer == txn)
- return update->type == UPDATE_TYPE_DELETE ? DOES_NOT_EXIST : update->value;
-
- // Skip updates from aborted transactions.
- txn_state_t writer_state = writer->state;
- if (EXPECT_FALSE(writer_state == TXN_ABORTED))
- continue;
-
- if (writer_state == TXN_VALIDATING) {
- writer_state = txn_validate(writer);
- }
-
- if (writer_state == TXN_VALIDATED && writer->wv <= txn->rv && writer->wv != UNDETERMINED_VERSION)
- return update->type == UPDATE_TYPE_DELETE ? DOES_NOT_EXIST : update->value;
- }
- }
- return DOES_NOT_EXIST;
-}