You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by ji...@apache.org on 2017/12/21 22:28:29 UTC
[3/8] mesos git commit: Added tombstone flag to NOP log action.
Added tombstone flag to NOP log action.
NOP action is used as a filler for holes in the "fill" procedure and as
the action in a response to a promise request for a truncated position.
The tombstone flag is introduced so that replicas are able to
distinguish the truncated action from a real NOP.
Review: https://reviews.apache.org/r/62285/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/6d95a2f0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/6d95a2f0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/6d95a2f0
Branch: refs/heads/master
Commit: 6d95a2f02b55e7b957f686f2ff94548fe9b0ffc9
Parents: 9ac50f3
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:45 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:45 2017 -0800
----------------------------------------------------------------------
src/log/leveldb.cpp | 36 +++++++++++++++++++++-----
src/log/replica.cpp | 16 ++++++++++++
src/messages/log.proto | 4 ++-
src/tests/log_tests.cpp | 60 ++++++++++++++++++++++++++++++++++++++++++++
4 files changed, 109 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/6d95a2f0/src/log/leveldb.cpp
----------------------------------------------------------------------
diff --git a/src/log/leveldb.cpp b/src/log/leveldb.cpp
index 5310a12..50d7fe5 100644
--- a/src/log/leveldb.cpp
+++ b/src/log/leveldb.cpp
@@ -243,6 +243,12 @@ Try<Storage::State> LevelDBStorage::restore(const string& path)
state.unlearned.erase(action.position());
if (action.has_type() && action.type() == Action::TRUNCATE) {
state.begin = std::max(state.begin, action.truncate().to());
+ } else if (action.has_type() && action.type() == Action::NOP &&
+ action.nop().has_tombstone() && action.nop().tombstone()) {
+ // If we see a tombstone, this position was truncated.
+ // There must exist at least 1 position (TRUNCATE) in the
+ // log after it.
+ state.begin = std::max(state.begin, action.position() + 1);
}
} else {
state.learned.erase(action.position());
@@ -341,13 +347,31 @@ Try<Nothing> LevelDBStorage::persist(const Action& action)
VLOG(1) << "Persisting action (" << value.size()
<< " bytes) to leveldb took " << stopwatch.elapsed();
- // Delete positions if a truncate action has been *learned*. Note
- // that we do this in a best-effort fashion (i.e., we ignore any
- // failures to the database since we can always try again).
+ Option<uint64_t> truncateTo;
+
+ // Delete positions if a truncate action has been *learned*.
if (action.has_type() && action.type() == Action::TRUNCATE &&
action.has_learned() && action.learned()) {
CHECK(action.has_truncate());
+ truncateTo = action.truncate().to();
+ }
+
+ // Delete positions if a tombstone NOP action has been *learned*.
+ if (action.has_type() && action.type() == Action::NOP &&
+ action.nop().has_tombstone() && action.nop().tombstone() &&
+ action.has_learned() && action.learned()) {
+ // We truncate the log up to the tombstone position instead of the
+ // next one to allow the recovery code to see the tombstone and
+ // learn about the truncation. It's OK to persist a tombstone NOP,
+ // because eventually we'll remove it once we see the actual
+ // TRUNCATE action.
+ truncateTo = action.position();
+ }
+ // Delete truncated positions. Note that we do this in a best-effort
+ // fashion (i.e., we ignore any failures to the database since we
+ // can always try again).
+ if (truncateTo.isSome()) {
stopwatch.start(); // Restart the stopwatch.
// To actually perform the truncation in leveldb we need to remove
@@ -378,7 +402,7 @@ Try<Nothing> LevelDBStorage::persist(const Action& action)
// out of order) bulk catch-up and the truncate operation is
// caught up first.
uint64_t index = 0;
- while ((first.get() + index) < action.truncate().to()) {
+ while ((first.get() + index) < truncateTo.get()) {
batch.Delete(encode(first.get() + index));
index++;
}
@@ -393,8 +417,8 @@ Try<Nothing> LevelDBStorage::persist(const Action& action)
<< status.ToString();
} else {
// Save the new first position!
- CHECK_LT(first.get(), action.truncate().to());
- first = action.truncate().to();
+ CHECK_LT(first.get(), truncateTo.get());
+ first = truncateTo.get();
VLOG(1) << "Deleting ~" << index
<< " keys from leveldb took " << stopwatch.elapsed();
http://git-wip-us.apache.org/repos/asf/mesos/blob/6d95a2f0/src/log/replica.cpp
----------------------------------------------------------------------
diff --git a/src/log/replica.cpp b/src/log/replica.cpp
index 39f2879..c86a609 100644
--- a/src/log/replica.cpp
+++ b/src/log/replica.cpp
@@ -413,6 +413,7 @@ void ReplicaProcess::promise(const UPID& from, const PromiseRequest& request)
action.set_learned(true);
action.set_type(Action::NOP);
action.mutable_nop()->MergeFrom(Action::Nop());
+ action.mutable_nop()->set_tombstone(true);
PromiseResponse response;
response.set_type(PromiseResponse::ACCEPT);
@@ -731,6 +732,21 @@ bool ReplicaProcess::persist(const Action& action)
// And update the beginning position.
begin = std::max(begin, action.truncate().to());
+ } else if (action.has_type() && action.type() == Action::NOP &&
+ action.nop().has_tombstone() && action.nop().tombstone()) {
+ // No longer consider truncated positions as holes (so that a
+ // coordinator doesn't try and fill them).
+ holes -= (Bound<uint64_t>::open(0),
+ Bound<uint64_t>::open(action.position()));
+
+ // No longer consider truncated positions as unlearned (so that
+ // a coordinator doesn't try and fill them).
+ unlearned -= (Bound<uint64_t>::open(0),
+ Bound<uint64_t>::open(action.position()));
+
+ // And update the beginning position. There must exist at least
+ // 1 position (TRUNCATE) in the log after the tombstone.
+ begin = std::max(begin, action.position() + 1);
}
} else {
// We just introduced an unlearned position.
http://git-wip-us.apache.org/repos/asf/mesos/blob/6d95a2f0/src/messages/log.proto
----------------------------------------------------------------------
diff --git a/src/messages/log.proto b/src/messages/log.proto
index 95f1d2e..76b06bb 100644
--- a/src/messages/log.proto
+++ b/src/messages/log.proto
@@ -54,7 +54,9 @@ message Action {
TRUNCATE = 3;
}
- message Nop {}
+ message Nop {
+ optional bool tombstone = 1; // True if the position was truncated.
+ }
message Append {
required bytes bytes = 1;
http://git-wip-us.apache.org/repos/asf/mesos/blob/6d95a2f0/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index f9f9400..e11be88 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -1992,6 +1992,66 @@ TEST_F(RecoverTest, AutoInitializationRetry)
}
+TEST_F(RecoverTest, CatchupTruncated)
+{
+ const string path1 = path::join(os::getcwd(), ".log1");
+ initializer.flags.path = path1;
+ ASSERT_SOME(initializer.execute());
+
+ const string path2 = path::join(os::getcwd(), ".log2");
+ initializer.flags.path = path2;
+ ASSERT_SOME(initializer.execute());
+
+ const string path3 = path::join(os::getcwd(), ".log3");
+
+ Shared<Replica> replica1(new Replica(path1));
+ Shared<Replica> replica2(new Replica(path2));
+
+ set<UPID> pids{replica1->pid(), replica2->pid()};
+ Shared<Network> network1(new Network(pids));
+
+ Coordinator coord(2, replica1, network1);
+ Future<Option<uint64_t>> electing = coord.elect();
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
+
+ // Add some positions to the log.
+ IntervalSet<uint64_t> positions;
+ for (uint64_t position = 1; position <= 10; position++) {
+ Future<Option<uint64_t>> appending = coord.append(stringify(position));
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
+ positions += position;
+ }
+
+ // Truncate the log.
+ Future<Option<uint64_t>> truncating = coord.truncate(5);
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
+
+ Shared<Replica> replica3(new Replica(path3));
+
+ pids.insert(replica3->pid());
+ Shared<Network> network2(new Network(pids));
+
+ // Pretend we recovered stale 'begin' position of the log before
+ // truncation has happened.
+ Future<Nothing> catching = catchup(
+ 2, replica3, network2, None(), positions, Seconds(10));
+ AWAIT_READY(catching);
+
+ AWAIT_EXPECT_EQ(5u, replica3->beginning());
+ AWAIT_EXPECT_EQ(10u, replica3->ending());
+
+ // Recreate the replica to verify that storage recovery succeeds.
+ replica3.reset();
+ replica3.reset(new Replica(path3));
+
+ AWAIT_EXPECT_EQ(5u, replica3->beginning());
+ AWAIT_EXPECT_EQ(10u, replica3->ending());
+}
+
+
class LogTest : public TemporaryDirectoryTest
{
protected: