Cleaned up snapshot version assignment code. The old code was relying…

… on a lot of assumptions which were difficult to verify.

This *seems* to close #728 , but I'm doing more testing now.

src/buffer_cache/mirrored/mirrored.cc

@@ -533,23 +533,43 @@ mc_buf_lock_t::mc_buf_lock_t(mc_transaction_t *transaction,
         }
     }
 
-    initialize(transaction->snapshot_version, transaction->get_io_account(), call_when_in_line);
+    // The versions must be assigned in the same order as the transactions
+    // acquire their first block. We use this callback to guarantee
+    // this.
+    // (before this was introduced, maybe_finalize_version() was simply called
+    // after initialize returns. However that only works if certain
+    // assumptions can be made, and I found it difficult to verify those.
+    // ~daniel)
+    struct version_finalizer_t : public lock_in_line_callback_t {
+        void on_in_line() {
+            trx->maybe_finalize_version();
+            if (other_callback != NULL) {
+                other_callback->on_in_line();
+            }
+        }
+        mc_transaction_t *trx;
+        lock_in_line_callback_t *other_callback;
+    } on_in_line;
+    on_in_line.trx = transaction;
+    on_in_line.other_callback = call_when_in_line;
+
+    initialize(transaction->get_io_account(), &on_in_line);
 
     if (is_write_mode(mode)) {
         touch_recency(transaction->recency_timestamp);
     }
 
-    transaction->maybe_finalize_version();
     transaction->assert_thread();
 }
 
-void mc_buf_lock_t::initialize(mc_inner_buf_t::version_id_t version_to_access,
-                               file_account_t *io_account,
+void mc_buf_lock_t::initialize(file_account_t *io_account,
                                lock_in_line_callback_t *call_when_in_line)
 {
     inner_buf->cache->assert_thread();
     inner_buf->refcount++;
 
+    mc_inner_buf_t::version_id_t version_to_access = parent_transaction->snapshot_version;
+
     if (snapshotted) {
         rassert(is_read_mode(mode), "Only read access is allowed to block snapshots");
         // Our snapshotting-specific code can't handle weird read
@@ -566,27 +586,38 @@ void mc_buf_lock_t::initialize(mc_inner_buf_t::version_id_t version_to_access,
     // If the top version is less or equal to snapshot_version, then we need to acquire
     // a read lock first (otherwise we may get the data of the unfinished write on top).
     if (snapshotted && version_to_access != mc_inner_buf_t::faux_version_id && version_to_access < inner_buf->version_id) {
+        // we do not need to get in line, so just call the function straight-up to ensure it gets called.
+        // Note that we call it in the same order as in the case where we do acquire
+        // a lock, that is before the potentially blocking acquire_snapshot_data()
+        // and acquire_block() respectively.
+        if (call_when_in_line) call_when_in_line->on_in_line();
+        // Update version_to_access since the callback might have modified it:
+        version_to_access = parent_transaction->snapshot_version;
+
         // acquire the snapshotted block; no need to lock
         data = inner_buf->acquire_snapshot_data(version_to_access, io_account, &subtree_recency, &block_size);
         guarantee(data != NULL);
-
-        // we never needed to get in line, so just call the function straight-up to ensure it gets called.
-        if (call_when_in_line) call_when_in_line->on_in_line();
     } else {
         ticks_t lock_start_time;
         // the top version is the right one for us; acquire a lock of the appropriate type first
         inner_buf->cache->stats->pm_bufs_acquiring.begin(&lock_start_time);
         inner_buf->lock.co_lock(mode == rwi_read_outdated_ok ? rwi_read : mode, call_when_in_line);
         inner_buf->cache->stats->pm_bufs_acquiring.end(&lock_start_time);
 
+        // Update version_to_access since the callback might have modified it:
+        version_to_access = parent_transaction->snapshot_version;
+        // At this point we assume that maybe_finalize_version() has been called and
+        // initialized our version id.
+        rassert(version_to_access != mc_inner_buf_t::faux_version_id);
+
         // It's possible that, now that we've acquired the lock, that
         // the inner buf's version id has changed, and that we should
         // get a snapshotted version.
 
-        if (snapshotted && version_to_access != mc_inner_buf_t::faux_version_id && version_to_access < inner_buf->version_id) {
+        if (snapshotted && version_to_access < inner_buf->version_id) {
             inner_buf->lock.unlock();
 
-            // acquire the snapshotted block; no need to lock
+            // acquire the snapshotted block; no need to hold the lock any longer
             data = inner_buf->acquire_snapshot_data(version_to_access, io_account, &subtree_recency, &block_size);
             guarantee(data != NULL);
 
@@ -627,7 +658,7 @@ mc_buf_lock_t::mc_buf_lock_t(mc_transaction_t *transaction) THROWS_NOTHING :
     transaction->assert_thread();
 
     // This must pass since no one else holds references to this block.
-    initialize(transaction->snapshot_version, transaction->get_io_account(), 0);
+    initialize(transaction->get_io_account(), 0);
 
     transaction->assert_thread();
 }
@@ -647,14 +678,15 @@ void mc_buf_lock_t::release_if_acquired() {
 void mc_buf_lock_t::acquire_block(mc_inner_buf_t::version_id_t version_to_access) {
     inner_buf->cache->assert_thread();
     rassert(!inner_buf->do_delete);
+    rassert(version_to_access != mc_inner_buf_t::faux_version_id);
 
     subtree_recency = inner_buf->subtree_recency;
 
     switch (mode) {
         case rwi_read_sync:
         case rwi_read: {
             if (snapshotted) {
-                rassert(version_to_access == mc_inner_buf_t::faux_version_id || inner_buf->version_id <= version_to_access);
+                rassert(inner_buf->version_id <= version_to_access);
                 ++inner_buf->snap_refcount;
             }
             // TODO (sam): Obviously something's f'd up about this.
@@ -676,8 +708,6 @@ void mc_buf_lock_t::acquire_block(mc_inner_buf_t::version_id_t version_to_access
             break;
         }
         case rwi_write: {
-            if (version_to_access == mc_inner_buf_t::faux_version_id)
-                version_to_access = inner_buf->cache->get_current_version_id();
             rassert(inner_buf->version_id <= version_to_access);
 
             inner_buf->snapshot_if_needed(version_to_access, true);

src/buffer_cache/mirrored/mirrored.hpp

@@ -194,7 +194,7 @@ class mc_buf_lock_t : public home_thread_mixin_t {
     friend class writeback_t::buf_writer_t;
 
     // Internal functions used during construction
-    void initialize(mc_inner_buf_t::version_id_t version, file_account_t *io_account, lock_in_line_callback_t *call_when_in_line);
+    void initialize(file_account_t *io_account, lock_in_line_callback_t *call_when_in_line);
     void acquire_block(mc_inner_buf_t::version_id_t version_to_access);
 
     // True if this is an mc_buf_lock_t for a snapshotted view of the buf.