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:27 UTC
[1/8] mesos git commit: Fixed a comment typo.
Repository: mesos
Updated Branches:
refs/heads/master 63f116a54 -> 310c2c612
Fixed a comment typo.
Review: https://reviews.apache.org/r/62283/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2fec43f7
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2fec43f7
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2fec43f7
Branch: refs/heads/master
Commit: 2fec43f7d6d639666be8f43adaf690fabd32c9d4
Parents: 63f116a
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:38 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:38 2017 -0800
----------------------------------------------------------------------
src/messages/log.proto | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2fec43f7/src/messages/log.proto
----------------------------------------------------------------------
diff --git a/src/messages/log.proto b/src/messages/log.proto
index 88a489c..95f1d2e 100644
--- a/src/messages/log.proto
+++ b/src/messages/log.proto
@@ -65,7 +65,7 @@ message Action {
required uint64 to = 1; // All positions before and exclusive of 'to'.
}
- optional Type type = 5; // Set iff performed is set.
+ optional Type type = 5; // Set if performed is set.
optional Nop nop = 6;
optional Append append = 7;
optional Truncate truncate = 8;
[7/8] mesos git commit: Added Log::Reader::catchup() method.
Posted by ji...@apache.org.
Added Log::Reader::catchup() method.
The new method catches-up positions missing in the local non-leading
replica to allow safe eventually consistent reads from it.
Review: https://reviews.apache.org/r/62287/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/49c91614
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/49c91614
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/49c91614
Branch: refs/heads/master
Commit: 49c91614da2a85dd49d1fdc33ef83f9c816de2d2
Parents: aa4d95a
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:50:00 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:50:00 2017 -0800
----------------------------------------------------------------------
include/mesos/log/log.hpp | 4 +++
src/log/log.cpp | 24 ++++++++++++++++++
src/log/log.hpp | 7 ++++++
src/tests/log_tests.cpp | 57 ++++++++++++++++++++++++++++++++++++++++++
4 files changed, 92 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/49c91614/include/mesos/log/log.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/log/log.hpp b/include/mesos/log/log.hpp
index d92d7a0..ba35533 100644
--- a/include/mesos/log/log.hpp
+++ b/include/mesos/log/log.hpp
@@ -148,6 +148,10 @@ public:
// partitioned).
process::Future<Position> ending();
+ // Launches the catch-up process. Returns the ending position of
+ // the caught-up range.
+ process::Future<Position> catchup();
+
private:
internal::log::LogReaderProcess* process;
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/49c91614/src/log/log.cpp
----------------------------------------------------------------------
diff --git a/src/log/log.cpp b/src/log/log.cpp
index dcb66f7..6a7080a 100644
--- a/src/log/log.cpp
+++ b/src/log/log.cpp
@@ -36,6 +36,7 @@
#include <stout/nothing.hpp>
#include <stout/set.hpp>
+#include "log/catchup.hpp"
#include "log/coordinator.hpp"
#include "log/network.hpp"
#include "log/recover.hpp"
@@ -287,6 +288,8 @@ void LogProcess::discarded()
LogReaderProcess::LogReaderProcess(Log* log)
: ProcessBase(ID::generate("log-reader")),
+ quorum(log->process->quorum),
+ network(log->process->network),
recovering(dispatch(log->process, &LogProcess::recover)) {}
@@ -431,6 +434,21 @@ Future<list<Log::Entry>> LogReaderProcess::__read(
}
+Future<Log::Position> LogReaderProcess::catchup()
+{
+ return recover().then(defer(self(), &Self::_catchup));
+}
+
+
+Future<Log::Position> LogReaderProcess::_catchup()
+{
+ CHECK_READY(recovering);
+
+ return log::catchup(quorum, recovering.get(), network)
+ .then([](uint64_t end) { return Log::Position(end); });
+}
+
+
Log::Position LogReaderProcess::position(uint64_t value)
{
return Log::Position(value);
@@ -714,6 +732,12 @@ Future<Log::Position> Log::Reader::ending()
}
+Future<Log::Position> Log::Reader::catchup()
+{
+ return dispatch(process, &LogReaderProcess::catchup);
+}
+
+
/////////////////////////////////////////////////
// Public interfaces for Log::Writer.
/////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/mesos/blob/49c91614/src/log/log.hpp
----------------------------------------------------------------------
diff --git a/src/log/log.hpp b/src/log/log.hpp
index 24481aa..be038d1 100644
--- a/src/log/log.hpp
+++ b/src/log/log.hpp
@@ -128,6 +128,8 @@ public:
const mesos::log::Log::Position& from,
const mesos::log::Log::Position& to);
+ process::Future<mesos::log::Log::Position> catchup();
+
protected:
virtual void initialize();
virtual void finalize();
@@ -155,6 +157,11 @@ private:
const mesos::log::Log::Position& to,
const std::list<Action>& actions);
+ process::Future<mesos::log::Log::Position> _catchup();
+
+ const size_t quorum;
+ const process::Shared<Network> network;
+
process::Future<process::Shared<Replica>> recovering;
std::list<process::Promise<Nothing>*> promises;
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/49c91614/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index aada363..5dcb693 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -2384,6 +2384,63 @@ TEST_F(LogTest, Metrics)
}
+TEST_F(LogTest, ReaderCatchup)
+{
+ const string path1 = os::getcwd() + "/.log1";
+ initializer.flags.path = path1;
+ ASSERT_SOME(initializer.execute());
+
+ const string path2 = os::getcwd() + "/.log2";
+ initializer.flags.path = path2;
+ ASSERT_SOME(initializer.execute());
+
+ const string path3 = os::getcwd() + "/.log3";
+ initializer.flags.path = path3;
+ ASSERT_SOME(initializer.execute());
+
+ Shared<Replica> replica1(new Replica(path1));
+ Shared<Replica> replica2(new Replica(path2));
+
+ set<UPID> pids{replica1->pid(), replica2->pid()};
+ Shared<Network> network(new Network(pids));
+
+ Coordinator coord(2, replica2, network);
+
+ Future<Option<uint64_t>> electing = coord.elect();
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
+
+ // Add some entries to the log.
+ 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());
+ }
+
+ Log log3(2, path3, pids);
+ Log::Reader reader(&log3);
+
+ // Catch-up the replica that missed positions adding.
+ Future<Log::Position> end = reader.catchup();
+ AWAIT_READY(end);
+
+ Future<Log::Position> begin = reader.beginning();
+ AWAIT_READY(begin);
+
+ // We expect to read 9 entries instead of 10, because the catch-up
+ // procedure doesn't catch-up the last recovered position. See
+ // comments for RecoverMissingProcess.
+ Future<list<Log::Entry>> entries = reader.read(begin.get(), end.get());
+ AWAIT_READY(entries);
+ ASSERT_EQ(9u, entries->size());
+
+ uint64_t position = 1;
+ foreach (const Log::Entry& entry, entries.get()) {
+ EXPECT_EQ(stringify(position++), entry.data);
+ }
+}
+
+
#ifdef MESOS_HAS_JAVA
// TODO(jieyu): We copy the code from TemporaryDirectoryTest here
// because we cannot inherit from two test fixtures. In this future,
[3/8] mesos git commit: Added tombstone flag to NOP log action.
Posted by ji...@apache.org.
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:
[2/8] mesos git commit: Fixed code style issues.
Posted by ji...@apache.org.
Fixed code style issues.
Review: https://reviews.apache.org/r/62284/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/9ac50f31
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/9ac50f31
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/9ac50f31
Branch: refs/heads/master
Commit: 9ac50f31886b63046439ae4d127aa26360f1d3d6
Parents: 2fec43f
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:41 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:41 2017 -0800
----------------------------------------------------------------------
src/log/consensus.cpp | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/9ac50f31/src/log/consensus.cpp
----------------------------------------------------------------------
diff --git a/src/log/consensus.cpp b/src/log/consensus.cpp
index 9b7463a..f766015 100644
--- a/src/log/consensus.cpp
+++ b/src/log/consensus.cpp
@@ -42,7 +42,8 @@ namespace mesos {
namespace internal {
namespace log {
-static bool isRejectedPromise(const PromiseResponse& response) {
+static bool isRejectedPromise(const PromiseResponse& response)
+{
if (response.has_type()) {
// New format (Mesos >= 0.26).
return response.type() == PromiseResponse::REJECT;
@@ -53,7 +54,8 @@ static bool isRejectedPromise(const PromiseResponse& response) {
}
-static bool isRejectedWrite(const WriteResponse& response) {
+static bool isRejectedWrite(const WriteResponse& response)
+{
if (response.has_type()) {
// New format (Mesos >= 0.26).
return response.type() == WriteResponse::REJECT;
[5/8] mesos git commit: Moved retry logic from RecoverProtocolProcess
to RecoverProcess.
Posted by ji...@apache.org.
Moved retry logic from RecoverProtocolProcess to RecoverProcess.
Review: https://reviews.apache.org/r/62761/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/9c25771d
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/9c25771d
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/9c25771d
Branch: refs/heads/master
Commit: 9c25771dcd1ed8dcc7696c0ceaad1d5f08033f88
Parents: 826277d
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:53 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:53 2017 -0800
----------------------------------------------------------------------
src/log/recover.cpp | 86 ++++++++++++++++++++++++--------------------
src/log/recover.hpp | 2 +-
src/tests/log_tests.cpp | 50 ++++++++++++++++++++++++++
3 files changed, 99 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/9c25771d/src/log/recover.cpp
----------------------------------------------------------------------
diff --git a/src/log/recover.cpp b/src/log/recover.cpp
index 8dab577..2167d41 100644
--- a/src/log/recover.cpp
+++ b/src/log/recover.cpp
@@ -91,7 +91,7 @@ public:
timeout(_timeout),
terminating(false) {}
- Future<RecoverResponse> future() { return promise.future(); }
+ Future<Option<RecoverResponse>> future() { return promise.future(); }
protected:
virtual void initialize()
@@ -329,20 +329,8 @@ private:
} else if (future.isFailed()) {
promise.fail(future.failure());
terminate(self());
- } else if (future.get().isNone()) {
- // Re-run the protocol. We add a random delay before each retry
- // because we do not want to saturate the network/disk IO in
- // some cases. The delay is chosen randomly to reduce the
- // likelihood of conflicts (i.e., a replica receives a recover
- // request while it is changing its status).
- static const Duration T = Milliseconds(500);
- Duration d = T * (1.0 + (double) os::random() / RAND_MAX);
- VLOG(2) << "Didn't receive enough responses for recovery, retrying "
- << "in " << stringify(d);
-
- delay(d, self(), &Self::start);
} else {
- promise.set(future.get().get());
+ promise.set(future.get());
terminate(self());
}
}
@@ -360,11 +348,11 @@ private:
Future<Option<RecoverResponse>> chain;
bool terminating;
- process::Promise<RecoverResponse> promise;
+ process::Promise<Option<RecoverResponse>> promise;
};
-Future<RecoverResponse> runRecoverProtocol(
+Future<Option<RecoverResponse>> runRecoverProtocol(
size_t quorum,
const Shared<Network>& network,
const Metadata::Status& status,
@@ -379,7 +367,7 @@ Future<RecoverResponse> runRecoverProtocol(
autoInitialize,
timeout);
- Future<RecoverResponse> future = process->future();
+ Future<Option<RecoverResponse>> future = process->future();
spawn(process, true);
return future;
}
@@ -451,12 +439,7 @@ protected:
// Register a callback to handle user initiated discard.
promise.future().onDiscard(defer(self(), &Self::discard));
- // Check the current status of the local replica and decide if
- // recovery is needed. Recovery is needed only if the local
- // replica is not in VOTING status.
- chain = replica->status()
- .then(defer(self(), &Self::recover, lambda::_1))
- .onAny(defer(self(), &Self::finished, lambda::_1));
+ start();
}
virtual void finalize()
@@ -470,22 +453,37 @@ private:
chain.discard();
}
- Future<Nothing> recover(const Metadata::Status& status)
+ void start()
+ {
+ // Check the current status of the local replica and decide if
+ // recovery is needed. Recovery is needed only if the local
+ // replica is not in VOTING status.
+ chain = replica->status()
+ .then(defer(self(), &Self::recover, lambda::_1))
+ .onAny(defer(self(), &Self::finished, lambda::_1));
+ }
+
+ Future<bool> recover(const Metadata::Status& status)
{
LOG(INFO) << "Replica is in " << status << " status";
if (status == Metadata::VOTING) {
// No need to do recovery.
- return Nothing();
+ return true;
} else {
return runRecoverProtocol(quorum, network, status, autoInitialize)
.then(defer(self(), &Self::_recover, lambda::_1));
}
}
- Future<Nothing> _recover(const RecoverResponse& result)
+ Future<bool> _recover(const Option<RecoverResponse>& result)
{
- switch (result.status()) {
+ if (result.isNone()) {
+ // Re-run the recover protocol.
+ return false;
+ }
+
+ switch (result->status()) {
case Metadata::STARTING:
// This is the auto-initialization case. As mentioned above, we
// use a two-phase protocol to bootstrap. When the control
@@ -504,17 +502,17 @@ private:
return updateReplicaStatus(Metadata::VOTING);
case Metadata::RECOVERING:
- CHECK(result.has_begin() && result.has_end());
+ CHECK(result->has_begin() && result->has_end());
return updateReplicaStatus(Metadata::RECOVERING)
- .then(defer(self(), &Self::catchup, result.begin(), result.end()));
+ .then(defer(self(), &Self::catchup, result->begin(), result->end()));
default:
return Failure("Unexpected status returned from the recover protocol");
}
}
- Future<Nothing> catchup(uint64_t begin, uint64_t end)
+ Future<bool> catchup(uint64_t begin, uint64_t end)
{
// We reach here either because the log is empty (uninitialized),
// or the log is not empty but a previous unfinished catch-up
@@ -561,7 +559,7 @@ private:
.then(defer(self(), &Self::updateReplicaStatus, Metadata::VOTING));
}
- Future<Nothing> updateReplicaStatus(const Metadata::Status& status)
+ Future<bool> updateReplicaStatus(const Metadata::Status& status)
{
LOG(INFO) << "Updating replica status to " << status;
@@ -569,7 +567,7 @@ private:
.then(defer(self(), &Self::_updateReplicaStatus, lambda::_1, status));
}
- Future<Nothing> _updateReplicaStatus(
+ Future<bool> _updateReplicaStatus(
bool updated, const Metadata::Status& status)
{
if (!updated) {
@@ -580,24 +578,24 @@ private:
LOG(INFO) << "Successfully joined the Paxos group";
}
- return Nothing();
+ return true;
}
- Future<Nothing> getReplicaOwnership(Shared<Replica> shared)
+ Future<bool> getReplicaOwnership(Shared<Replica> shared)
{
// Try to re-gain the ownership of the replica.
return shared.own()
.then(defer(self(), &Self::_getReplicaOwnership, lambda::_1));
}
- Future<Nothing> _getReplicaOwnership(Owned<Replica> owned)
+ Future<bool> _getReplicaOwnership(Owned<Replica> owned)
{
replica = owned;
- return Nothing();
+ return true;
}
- void finished(const Future<Nothing>& future)
+ void finished(const Future<bool>& future)
{
if (future.isDiscarded()) {
promise.discard();
@@ -605,6 +603,16 @@ private:
} else if (future.isFailed()) {
promise.fail(future.failure());
terminate(self());
+ } else if (!future.get()) {
+ // We add a random delay before each retry because we do not
+ // want to saturate the network/disk IO in some cases. The delay
+ // is chosen randomly to reduce the likelyhood of conflicts
+ // (i.e., a replica receives a recover request while it is
+ // changing its status).
+ static const Duration T = Milliseconds(500);
+ Duration d = T * (1.0 + (double) os::random() / RAND_MAX);
+ VLOG(2) << "Retrying recovery in " << stringify(d);
+ delay(d, self(), &Self::start);
} else {
promise.set(replica);
terminate(self());
@@ -616,7 +624,9 @@ private:
const Shared<Network> network;
const bool autoInitialize;
- Future<Nothing> chain;
+ // The value in this future speficies if the recovery was
+ // successfull or we need to retry it.
+ Future<bool> chain;
process::Promise<Owned<Replica>> promise;
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/9c25771d/src/log/recover.hpp
----------------------------------------------------------------------
diff --git a/src/log/recover.hpp b/src/log/recover.hpp
index da216e4..ba80535 100644
--- a/src/log/recover.hpp
+++ b/src/log/recover.hpp
@@ -34,7 +34,7 @@ namespace log {
// Runs the recover protocol. We will re-run the recover protocol if
// it cannot be finished within 'timeout'.
-process::Future<RecoverResponse> runRecoverProtocol(
+process::Future<Option<RecoverResponse>> runRecoverProtocol(
size_t quorum,
const process::Shared<Network>& network,
const Metadata::Status& status,
http://git-wip-us.apache.org/repos/asf/mesos/blob/9c25771d/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index e11be88..e13d054 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -1844,6 +1844,56 @@ TEST_F(RecoverTest, CatchupRetry)
}
+TEST_F(RecoverTest, RecoverProtocolRetry)
+{
+ const string path1 = path::join(os::getcwd(), ".log1");
+ initializer.flags.path = path1;
+ ASSERT_SOME(initializer.execute());
+
+ const string path2 = path::join(os::getcwd(), ".log2");
+ const string path3 = path::join(os::getcwd(), ".log3");
+
+ Owned<Replica> replica1(new Replica(path1));
+ Owned<Replica> replica2(new Replica(path2));
+ Owned<Replica> replica3(new Replica(path3));
+
+ set<UPID> pids{replica1->pid(), replica2->pid(), replica3->pid()};
+ Shared<Network> network(new Network(pids));
+
+ Future<Owned<Replica>> recovering = recover(2, replica3, network);
+
+ Clock::pause();
+
+ // Wait for the retry timer to be setup.
+ Clock::settle();
+ ASSERT_TRUE(recovering.isPending());
+
+ // Wait for recover process to retry.
+ Clock::advance(Seconds(10));
+ Clock::settle();
+ ASSERT_TRUE(recovering.isPending());
+
+ // Remove replica 2 from the network to be initialized. It is safe
+ // to have non-const access to shared Network here, because all
+ // Network operations are serialized through a Process.
+ const_cast<Network&>(*network).remove(replica2->pid());
+ replica2.reset();
+
+ initializer.flags.path = path2;
+ ASSERT_SOME(initializer.execute());
+
+ replica2.reset(new Replica(path2));
+ const_cast<Network&>(*network).add(replica2->pid());
+
+ // Wait for recover process to retry again, now with 2 VOTING
+ // replicas. It should successfully finish now.
+ Clock::advance(Seconds(10));
+ Clock::resume();
+
+ AWAIT_READY(recovering);
+}
+
+
TEST_F(RecoverTest, AutoInitialization)
{
const string path1 = os::getcwd() + "/.log1";
[6/8] mesos git commit: Added CatchupMissing log process.
Posted by ji...@apache.org.
Added CatchupMissing log process.
The process is used to catch-up a non-leading VOTING replica by running
the recovery protocol to find current begin and end positions of the log
and catching-up positions that are missing on the replica. This allows
following replicas to serve eventually consistent reads.
Review: https://reviews.apache.org/r/62286/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/aa4d95a0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/aa4d95a0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/aa4d95a0
Branch: refs/heads/master
Commit: aa4d95a0a509394a6616b92a1f170846f713c250
Parents: 9c25771
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:57 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:57 2017 -0800
----------------------------------------------------------------------
src/log/catchup.cpp | 171 +++++++++++++++++++++++++++++++++++++++++++
src/log/catchup.hpp | 14 ++++
src/tests/log_tests.cpp | 166 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 351 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/aa4d95a0/src/log/catchup.cpp
----------------------------------------------------------------------
diff --git a/src/log/catchup.cpp b/src/log/catchup.cpp
index 94e1b00..79cc18c 100644
--- a/src/log/catchup.cpp
+++ b/src/log/catchup.cpp
@@ -28,6 +28,7 @@
#include "log/catchup.hpp"
#include "log/consensus.hpp"
+#include "log/recover.hpp"
#include "messages/log.hpp"
@@ -308,6 +309,156 @@ static Future<Nothing> catchup(
}
+// This process is used to catch-up missing positions in the local
+// replica. We first check the status of the local replica. It if is
+// not in VOTING status, the recover process will terminate
+// immediately. Next we will run the log recover protocol to determine
+// the log's beginning and ending positions. After that we will start
+// doing catch-up.
+class CatchupMissingProcess : public Process<CatchupMissingProcess>
+{
+public:
+ CatchupMissingProcess(
+ size_t _quorum,
+ const Shared<Replica>& _replica,
+ const Shared<Network>& _network,
+ const Option<uint64_t>& _proposal,
+ const Duration& _timeout)
+ : ProcessBase(ID::generate("log-recover-missing")),
+ quorum(_quorum),
+ replica(_replica),
+ network(_network),
+ proposal(_proposal),
+ timeout(_timeout) {}
+
+ Future<uint64_t> future()
+ {
+ return promise.future();
+ }
+
+protected:
+ virtual void initialize()
+ {
+ LOG(INFO) << "Starting missing positions recovery";
+
+ // Register a callback to handle user initiated discard.
+ promise.future().onDiscard(defer(self(), &Self::discard));
+
+ // Check the current status of the local replica and decide if we
+ // proceed with recovery. We do it only if the local replica is in
+ // VOTING status.
+ chain = replica->status()
+ .then(defer(self(), &Self::recover, lambda::_1))
+ .onAny(defer(self(), &Self::finished, lambda::_1));
+ }
+
+ virtual void finalize()
+ {
+ VLOG(1) << "Recover process terminated";
+ }
+
+private:
+ void discard()
+ {
+ chain.discard();
+ }
+
+ Future<Nothing> recover(const Metadata::Status& status)
+ {
+ LOG(INFO) << "Replica is in " << status << " status";
+
+ if (status != Metadata::VOTING) {
+ return Nothing();
+ }
+
+ return runRecoverProtocol(quorum, network, status, false)
+ .then(defer(self(), &Self::_recover, lambda::_1));
+ }
+
+ Future<Nothing> _recover(const Option<RecoverResponse>& response)
+ {
+ if (response.isNone()) {
+ return Failure("Failed to recover begin and end positions of the log");
+ }
+
+ if (response->status() != Metadata::RECOVERING) {
+ return Failure("Unexpected status returned from the recover protocol");
+ }
+
+ CHECK(response->has_begin() && response->has_end());
+
+ if (response->begin() == response->end()) {
+ // This may happen if all replicas know only about position
+ // 0 (just initialized).
+ return Failure("Recovered only 1 position, cannot catch-up");
+ }
+
+ // We do not catchup the last recovered position in order to
+ // prevent coordinator demotion.
+ end = response->end() - 1;
+
+ return replica->beginning()
+ .then(defer(self(), [this, response](uint64_t begin) {
+ // Ideally we would only need to catch-up positions from
+ // the recovered range and then adjust local replica's
+ // 'begin'. However, the replica needs to persist the
+ // truncation indicator (TRUNCATE or a tombstone NOP) to
+ // be able to recover the same 'begin' after a restart. If
+ // we only catch-up positions from the recovered range, it
+ // is possible that the replica sees neither TRUNCATE
+ // (e.g. it is the last position in the log, which we
+ // don't catch-up), nor a tombstone. We catch-up positions
+ // starting with the lowest known begin position so that
+ // the replica either retains the same 'begin', or sees a
+ // truncation indicator.
+ begin = std::min(begin, response->begin());
+ return catchup(begin, end);
+ }));
+ }
+
+ Future<Nothing> catchup(uint64_t begin, uint64_t end)
+ {
+ CHECK_LE(begin, end);
+
+ LOG(INFO) << "Starting catch-up from position " << begin << " to " << end;
+
+ IntervalSet<uint64_t> positions(
+ Bound<uint64_t>::closed(begin),
+ Bound<uint64_t>::closed(end));
+
+ // TODO(ipronin): Consider using 'proposed' field from the local
+ // replica.
+ return log::catchup(quorum, replica, network, proposal, positions, timeout);
+ }
+
+ void finished(const Future<Nothing>& future)
+ {
+ if (future.isDiscarded()) {
+ promise.discard();
+ terminate(self());
+ } else if (future.isFailed()) {
+ promise.fail(future.failure());
+ terminate(self());
+ } else {
+ promise.set(end);
+ terminate(self());
+ }
+ }
+
+ const size_t quorum;
+ Shared<Replica> replica;
+ const Shared<Network> network;
+ const Option<uint64_t> proposal;
+ const Duration timeout;
+
+ Future<Nothing> chain;
+
+ uint64_t end;
+
+ process::Promise<uint64_t> promise;
+};
+
+
/////////////////////////////////////////////////
// Public interfaces below.
/////////////////////////////////////////////////
@@ -347,6 +498,26 @@ Future<Nothing> catchup(
return future;
}
+Future<uint64_t> catchup(
+ size_t quorum,
+ const process::Shared<Replica>& replica,
+ const process::Shared<Network>& network,
+ const Option<uint64_t>& proposal,
+ const Duration& timeout)
+{
+ CatchupMissingProcess* process =
+ new CatchupMissingProcess(
+ quorum,
+ replica,
+ network,
+ proposal,
+ timeout);
+
+ Future<uint64_t> future = process->future();
+ spawn(process, true);
+ return future;
+}
+
} // namespace log {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/aa4d95a0/src/log/catchup.hpp
----------------------------------------------------------------------
diff --git a/src/log/catchup.hpp b/src/log/catchup.hpp
index 123bc7a..2b5329e 100644
--- a/src/log/catchup.hpp
+++ b/src/log/catchup.hpp
@@ -50,6 +50,20 @@ extern process::Future<Nothing> catchup(
const IntervalSet<uint64_t>& positions,
const Duration& timeout = Seconds(10));
+
+// Catches-up missing log positions in the local replica. Returns the
+// highest position that was caught-up and now is safe to read. The
+// user of this function can provide a hint on the proposal number
+// that will be used for Paxos. We also allow the user to specify a
+// timeout for the catch-up operation on each position and retry the
+// operation if timeout happens.
+process::Future<uint64_t> catchup(
+ size_t quorum,
+ const process::Shared<Replica>& replica,
+ const process::Shared<Network>& network,
+ const Option<uint64_t>& proposal = None(),
+ const Duration& timeout = Seconds(10));
+
} // namespace log {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/aa4d95a0/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index e13d054..aada363 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -2102,6 +2102,172 @@ TEST_F(RecoverTest, CatchupTruncated)
}
+// Verifiy that we can catch-up a following VOTING replica.
+TEST_F(RecoverTest, CatchupVoting)
+{
+ 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");
+ initializer.flags.path = path3;
+ ASSERT_SOME(initializer.execute());
+
+ 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 entries to the log.
+ 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());
+ }
+
+ // Truncate the log.
+ Future<Option<uint64_t>> truncating = coord.truncate(5);
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
+
+ // Create one more replica. It is in VOTING status, but it missed
+ // positions adding and truncation.
+ Shared<Replica> replica3(new Replica(path3));
+
+ pids.insert(replica3->pid());
+ Shared<Network> network2(new Network(pids));
+
+ // Catch-up the VOTING replica for reading. We're using 3 as the
+ // quorum size here to simulate recovering a stale lowest position
+ // (from the local replica).
+ Future<uint64_t> catching = catchup(3, replica3, network2);
+ AWAIT_EXPECT_EQ(10u, catching);
+
+ Future<uint64_t> begin = replica3->beginning();
+ AWAIT_EXPECT_EQ(5u, begin);
+
+ Future<uint64_t> end = replica3->ending();
+ AWAIT_EXPECT_EQ(catching.get(), end);
+
+ Future<list<Action>> actions = replica3->read(begin.get(), end.get());
+ AWAIT_READY(actions);
+ EXPECT_EQ(end.get() - begin.get() + 1, actions->size());
+ foreach (const Action& action, actions.get()) {
+ ASSERT_TRUE(action.has_type());
+ ASSERT_EQ(Action::APPEND, action.type());
+ EXPECT_EQ(stringify(action.position()), action.append().bytes());
+ }
+}
+
+
+// Verifiy that we can catch-up a following VOTING replica.
+TEST_F(RecoverTest, CatchupVotingWithGap)
+{
+ 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");
+ initializer.flags.path = path3;
+ ASSERT_SOME(initializer.execute());
+
+ 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 entries to the log.
+ 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());
+ }
+
+ // Truncate the log.
+ Future<Option<uint64_t>> truncating = coord.truncate(5);
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
+
+ // Create one more replica. It is in VOTING status, but it missed
+ // positions adding and truncation.
+ Shared<Replica> replica3(new Replica(path3));
+
+ pids.insert(replica3->pid());
+ Shared<Network> network2(new Network(pids));
+
+ // Make sure replica3 doesn't receive recover request, so we won't
+ // recover stale 'begin' position.
+ DROP_PROTOBUFS(RecoverRequest(), _, Eq(replica3->pid()));
+
+ // Catch-up the VOTING replica for reading.
+ Future<uint64_t> catching = catchup(2, replica3, network2);
+ AWAIT_EXPECT_EQ(10u, catching);
+
+ Future<uint64_t> begin = replica3->beginning();
+ AWAIT_EXPECT_EQ(5u, begin);
+
+ Future<uint64_t> end = replica3->ending();
+ AWAIT_EXPECT_EQ(catching.get(), end);
+
+ Future<list<Action>> actions = replica3->read(begin.get(), end.get());
+ AWAIT_READY(actions);
+ EXPECT_EQ(end.get() - begin.get() + 1, actions->size());
+ foreach (const Action& action, actions.get()) {
+ ASSERT_TRUE(action.has_type());
+ ASSERT_EQ(Action::APPEND, action.type());
+ EXPECT_EQ(stringify(action.position()), action.append().bytes());
+ }
+}
+
+
+// Verifiy that catch-up fails if we recover only 1 position.
+TEST_F(RecoverTest, CatchupVotingOnePosition)
+{
+ 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");
+ initializer.flags.path = path3;
+ ASSERT_SOME(initializer.execute());
+
+ Shared<Replica> replica1(new Replica(path1));
+ Shared<Replica> replica2(new Replica(path2));
+ Shared<Replica> replica3(new Replica(path3));
+
+ set<UPID> pids{replica1->pid(), replica2->pid(), replica3->pid()};
+ Shared<Network> network(new Network(pids));
+
+ AWAIT_FAILED(catchup(2, replica3, network));
+ AWAIT_EXPECT_EQ(0u, replica3->beginning());
+ AWAIT_EXPECT_EQ(0u, replica3->ending());
+}
+
+
class LogTest : public TemporaryDirectoryTest
{
protected:
[8/8] mesos git commit: Added Log.Reader.catchup() method to Java
bindings.
Posted by ji...@apache.org.
Added Log.Reader.catchup() method to Java bindings.
The new method is a wrapper for Log::Reader::catchup() method of the C++
Replicated Log API.
Review: https://reviews.apache.org/r/62288/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/310c2c61
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/310c2c61
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/310c2c61
Branch: refs/heads/master
Commit: 310c2c6129e09097f5d31431f754e2285b3c3f58
Parents: 49c9161
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:50:04 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:50:04 2017 -0800
----------------------------------------------------------------------
src/java/jni/org_apache_mesos_Log.cpp | 41 +++++++++++++++++++++++++++++
src/java/src/org/apache/mesos/Log.java | 17 ++++++++++++
2 files changed, 58 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/310c2c61/src/java/jni/org_apache_mesos_Log.cpp
----------------------------------------------------------------------
diff --git a/src/java/jni/org_apache_mesos_Log.cpp b/src/java/jni/org_apache_mesos_Log.cpp
index eb0f54b..e5e632c 100644
--- a/src/java/jni/org_apache_mesos_Log.cpp
+++ b/src/java/jni/org_apache_mesos_Log.cpp
@@ -234,6 +234,47 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_ending
/*
* Class: org_apache_mesos_Log_Reader
+ * Method: catchup
+ * Signature: (JLjava/util/concurrent/TimeUnit;)Lorg/apache/mesos/Log/Position;
+ */
+JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_catchup
+ (JNIEnv* env, jobject thiz, jlong jtimeout, jobject junit)
+{
+ // Read out __reader.
+ jclass clazz = env->GetObjectClass(thiz);
+ jfieldID __reader = env->GetFieldID(clazz, "__reader", "J");
+ Log::Reader* reader = (Log::Reader*)env->GetLongField(thiz, __reader);
+
+ // long seconds = unit.toSeconds(time);
+ clazz = env->GetObjectClass(junit);
+ jmethodID toSeconds = env->GetMethodID(clazz, "toSeconds", "(J)J");
+ jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
+
+ Future<Log::Position> position = reader->catchup();
+
+ if (!position.await(Seconds(jseconds))) {
+ // Timed out while trying to catchup the log.
+ position.discard();
+ clazz = env->FindClass("java/util/concurrent/TimeoutException");
+ env->ThrowNew(clazz, "Timed out while attempting to catchup");
+ return nullptr;
+ } else if (!position.isReady()) {
+ // Failed to catchup the log.
+ clazz = env->FindClass("org/apache/mesos/Log$OperationFailedException");
+ env->ThrowNew(
+ clazz,
+ (position.isFailed()
+ ? position.failure().c_str() : "Discarded future"));
+ return nullptr;
+ }
+
+ jobject jposition = convert<Log::Position>(env, position.get());
+ return jposition;
+}
+
+
+/*
+ * Class: org_apache_mesos_Log_Reader
* Method: initialize
* Signature: (Lorg/apache/mesos/Log;)V
*/
http://git-wip-us.apache.org/repos/asf/mesos/blob/310c2c61/src/java/src/org/apache/mesos/Log.java
----------------------------------------------------------------------
diff --git a/src/java/src/org/apache/mesos/Log.java b/src/java/src/org/apache/mesos/Log.java
index f53d9eb..f00f457 100644
--- a/src/java/src/org/apache/mesos/Log.java
+++ b/src/java/src/org/apache/mesos/Log.java
@@ -238,6 +238,23 @@ public class Log {
*/
public native Position ending();
+ /**
+ * Attempts to catch-up positions from the log for reading.
+ *
+ * @param timeout Max number of time units to wait before a
+ * {@link TimeoutException}.
+ * @param unit Type of time units used for the timeout, e.g. seconds,
+ * minutes, etc.
+ *
+ * @return The ending position of the caught-up range.
+ *
+ * @throws TimeoutException If the catch-up doesn't happen before
+ * the timeout.
+ * @throws OperationFailedException If the catch-up fails.
+ */
+ public native Position catchup(long timeout, TimeUnit unit)
+ throws TimeoutException, OperationFailedException;
+
protected native void initialize(Log log);
protected native void finalize();
[4/8] mesos git commit: Exposed runRecoverProtocol() function in
recover.hpp.
Posted by ji...@apache.org.
Exposed runRecoverProtocol() function in recover.hpp.
This function is needed catch-up missing positions of a non-leading
VOTING replica.
Review: https://reviews.apache.org/r/62760/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/826277d0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/826277d0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/826277d0
Branch: refs/heads/master
Commit: 826277d0b19bfe74062f0f6facff0293133262c3
Parents: 6d95a2f
Author: Ilya Pronin <ip...@twopensource.com>
Authored: Thu Dec 21 12:49:50 2017 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Thu Dec 21 12:49:50 2017 -0800
----------------------------------------------------------------------
src/log/recover.cpp | 6 ++----
src/log/recover.hpp | 10 ++++++++++
2 files changed, 12 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/826277d0/src/log/recover.cpp
----------------------------------------------------------------------
diff --git a/src/log/recover.cpp b/src/log/recover.cpp
index b366818..8dab577 100644
--- a/src/log/recover.cpp
+++ b/src/log/recover.cpp
@@ -364,14 +364,12 @@ private:
};
-// The wrapper for running the recover protocol. We will re-run the
-// recover protocol if it cannot be finished within 'timeout'.
-static Future<RecoverResponse> runRecoverProtocol(
+Future<RecoverResponse> runRecoverProtocol(
size_t quorum,
const Shared<Network>& network,
const Metadata::Status& status,
bool autoInitialize,
- const Duration& timeout = Seconds(10))
+ const Duration& timeout)
{
RecoverProtocolProcess* process =
new RecoverProtocolProcess(
http://git-wip-us.apache.org/repos/asf/mesos/blob/826277d0/src/log/recover.hpp
----------------------------------------------------------------------
diff --git a/src/log/recover.hpp b/src/log/recover.hpp
index 94d6bb2..da216e4 100644
--- a/src/log/recover.hpp
+++ b/src/log/recover.hpp
@@ -32,6 +32,16 @@ namespace mesos {
namespace internal {
namespace log {
+// Runs the recover protocol. We will re-run the recover protocol if
+// it cannot be finished within 'timeout'.
+process::Future<RecoverResponse> runRecoverProtocol(
+ size_t quorum,
+ const process::Shared<Network>& network,
+ const Metadata::Status& status,
+ bool autoInitialize,
+ const Duration& timeout = Seconds(10));
+
+
// Recovers a replica by catching up enough missing positions. A
// replica starts with an empty log (e.g., in the case of a disk
// failure) should not be allowed to vote. Otherwise, the new votes it