Use cb::couchstore::seek(End) to detect concurrent write

As part of the catch up phase while running concurrent compaction
we want to detect if the flusher wrote data to the database while
the compactor was copying data without holding the lock.

Instead of iterating over _all_ 4k blocks on disk from the current
header until the end of the file to find the last header block it
is more efficient to just try to locate the last header on the
file (which should be at the end of the file).

Improve the logging messages to include if the flusher is being
paused or not while copying the data.

Change-Id: I515a513f6606d1b4bb8d21fa41f69f5e3c41a987
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/138071
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: trondn

engines/ep/src/couch-kvstore/couch-kvstore.cc

@@ -1449,22 +1449,16 @@ bool CouchKVStore::tryToCatchUpDbFile(Db& source,
 
     auto start = cb::couchstore::getHeader(source);
 
-    couchstore_error_t err;
-    while ((err = cb::couchstore::seek(source,
-                                       cb::couchstore::Direction::Forward)) ==
-           COUCHSTORE_SUCCESS) {
-        // nothing (I should probably add one seek END and SEEK_START)
-    }
-
-    if (err != COUCHSTORE_ERROR_NO_HEADER) {
+    auto err = cb::couchstore::seek(source, cb::couchstore::Direction::End);
+    if (err != COUCHSTORE_SUCCESS) {
         // seek failed - deal with it
         throw std::runtime_error("Failed to seek in the database file: " +
                                  couchkvstore_strerrno(&source, err));
     }
 
     auto end = cb::couchstore::getHeader(source);
     if (end.headerPosition == start.headerPosition) {
-        // at the end!
+        // Catchup complete
         return true;
     }
 
@@ -1484,10 +1478,11 @@ bool CouchKVStore::tryToCatchUpDbFile(Db& source,
                         .count();
     }
 
-    EP_LOG_INFO("Try to catch up {}: from {} to {}",
+    EP_LOG_INFO("Try to catch up {}: from {} to {} {} stopping flusher",
                 vbid,
                 start.updateSeqNum,
-                end.updateSeqNum);
+                end.updateSeqNum,
+                copyWithoutLock ? "without" : "while");
     err = cb::couchstore::replay(
             source,
             destination,
@@ -1506,14 +1501,19 @@ bool CouchKVStore::tryToCatchUpDbFile(Db& source,
         throw std::runtime_error("Failed to replay data!");
     }
 
+    bool ret = true;
     if (copyWithoutLock) {
         lock.lock();
         // the database may have changed.. just return false and let the
         // caller call us again to check..
-        return false;
+        ret = false;
     }
 
-    return true;
+    logger.info("Catch up of {} to {} complete{}",
+                vbid,
+                end.updateSeqNum,
+                copyWithoutLock ? ", stopping flusher" : "");
+    return ret;
 }
 
 vbucket_state* CouchKVStore::getVBucketState(Vbid vbucketId) {