filejournal: fix replay of non-idempotent ops

- start sync thread prior to replay, so that we can commit as we replay
  operations
- keep applied_seq accurate
- pass seq (not old op_seq) to do_transactions
- carry open_ops ref so that commit blocks until we have finished applying
  the full transaction

Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 2787110e58d52eee6e2c4b31d038dd53ec316809..e135fa32b9702e47dd51d65bad87fc90fe5d510f 100644 (file)
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -1683,6 +1683,8 @@ int FileStore::mount()
     }
   }
 
+  sync_thread.create();
+
   ret = journal_replay(initial_op_seq);
   if (ret < 0) {
     derr << "mount failed to open journal " << journalpath << ": "
@@ -1691,12 +1693,19 @@ int FileStore::mount()
       derr << "maybe journal is not pointing to a block device and its size "
           << "wasn't configured?" << dendl;
     }
+
+    // stop sync thread
+    lock.Lock();
+    stop = true;
+    sync_cond.Signal();
+    lock.Unlock();
+    sync_thread.join();
+
     goto close_current_fd;
   }
 
   journal_start();
 
-  sync_thread.create();
   op_tp.start();
   flusher_thread.create();
   op_finisher.start();

diff --git a/src/os/JournalingObjectStore.cc b/src/os/JournalingObjectStore.cc
index 66257bf5d9a215be13b44957273bd224104caa1d..095f2856d437dc596062fefd67ef429fc172b544 100644 (file)
--- a/src/os/JournalingObjectStore.cc
+++ b/src/os/JournalingObjectStore.cc
@@ -72,29 +72,33 @@ int JournalingObjectStore::journal_replay(uint64_t fs_op_seq)
     }
     assert(op_seq == seq-1);
     
-    dout(3) << "journal_replay: applying op seq " << seq << " (op_seq " << op_seq << ")" << dendl;
+    dout(3) << "journal_replay: applying op seq " << seq << dendl;
     bufferlist::iterator p = bl.begin();
     list<Transaction*> tls;
     while (!p.end()) {
       Transaction *t = new Transaction(p);
       tls.push_back(t);
     }
-    int r = do_transactions(tls, op_seq);
-    op_seq++;
+
+    open_ops++;
+    int r = do_transactions(tls, seq);
+    open_ops--;
+    cond.Signal();
+
+    op_seq = applied_seq = seq;
+
     while (!tls.empty()) {
       delete tls.front(); 
       tls.pop_front();
     }
 
-    dout(3) << "journal_replay: r = " << r << ", op now seq " << op_seq << dendl;
+    dout(3) << "journal_replay: r = " << r << ", op_seq now " << op_seq << dendl;
     assert(op_seq == seq);
     seq++;  // we expect the next op
   }
 
   replaying = false;
 
-  applied_seq = op_seq;
-
   // done reading, make writeable.
   journal->make_writeable();