You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2014/03/10 07:38:57 UTC
[1/6] git commit: Used EXPECT_SOME_EQ in log tests.
Repository: mesos
Updated Branches:
refs/heads/master 8f4516f05 -> 8a09ecff0
Used EXPECT_SOME_EQ in log tests.
Review: https://reviews.apache.org/r/18263
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/8a09ecff
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/8a09ecff
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/8a09ecff
Branch: refs/heads/master
Commit: 8a09ecff08565f6b9a665ea80cbe2f716d171514
Parents: f8c3957
Author: Benjamin Hindman <be...@gmail.com>
Authored: Mon Feb 17 22:34:42 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/tests/log_tests.cpp | 221 ++++++++++++++++++++-----------------------
1 file changed, 103 insertions(+), 118 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/8a09ecff/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index e49610b..b368cde 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -595,9 +595,8 @@ TEST_F(CoordinatorTest, Elect)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
{
@@ -634,16 +633,15 @@ TEST_F(CoordinatorTest, AppendRead)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -684,16 +682,15 @@ TEST_F(CoordinatorTest, AppendReadError)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -731,7 +728,7 @@ TEST_F(CoordinatorTest, AppendDiscarded)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_SOME(electing.get());
EXPECT_EQ(0u, electing.get().get());
}
@@ -750,7 +747,7 @@ TEST_F(CoordinatorTest, AppendDiscarded)
{
Future<Option<uint64_t> > appending = coord.append("hello moto");
- AWAIT_SOME(appending);
+ AWAIT_READY(appending);
EXPECT_NONE(appending.get());
}
@@ -808,9 +805,8 @@ TEST_F(CoordinatorTest, AppendNoQuorum)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
process::terminate(replica2->pid());
@@ -853,16 +849,15 @@ TEST_F(CoordinatorTest, Failover)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord1.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -874,9 +869,8 @@ TEST_F(CoordinatorTest, Failover)
{
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(position, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(position, electing.get());
}
{
@@ -914,16 +908,15 @@ TEST_F(CoordinatorTest, Demoted)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord1.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -935,20 +928,19 @@ TEST_F(CoordinatorTest, Demoted)
{
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(position, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(position, electing.get());
}
{
Future<Option<uint64_t> > appending = coord1.append("hello moto");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
EXPECT_NONE(appending.get());
}
{
Future<Option<uint64_t> > appending = coord2.append("hello hello");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(2u, position);
@@ -993,16 +985,15 @@ TEST_F(CoordinatorTest, Fill)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord1.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -1019,14 +1010,16 @@ TEST_F(CoordinatorTest, Fill)
Coordinator coord2(2, replica3, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(position, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(position, electing.get());
}
{
@@ -1072,16 +1065,15 @@ TEST_F(CoordinatorTest, NotLearnedFill)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
uint64_t position;
{
Future<Option<uint64_t> > appending = coord1.append("hello world");
- AWAIT_READY_FOR(appending, Seconds(10));
+ AWAIT_READY(appending);
ASSERT_SOME(appending.get());
position = appending.get().get();
EXPECT_EQ(1u, position);
@@ -1098,14 +1090,16 @@ TEST_F(CoordinatorTest, NotLearnedFill)
Coordinator coord2(2, replica3, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(position, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(position, electing.get());
}
{
@@ -1143,16 +1137,14 @@ TEST_F(CoordinatorTest, MultipleAppends)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
{
@@ -1199,16 +1191,14 @@ TEST_F(CoordinatorTest, MultipleAppendsNotLearnedFill)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord1.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1222,14 +1212,16 @@ TEST_F(CoordinatorTest, MultipleAppendsNotLearnedFill)
Coordinator coord2(2, replica3, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(10u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(10u, electing.get());
}
{
@@ -1268,23 +1260,20 @@ TEST_F(CoordinatorTest, Truncate)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
{
Future<Option<uint64_t> > truncating = coord.truncate(7);
- AWAIT_READY_FOR(truncating, Seconds(10));
- ASSERT_SOME(truncating.get());
- EXPECT_EQ(11u, truncating.get().get());
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
}
{
@@ -1337,23 +1326,20 @@ TEST_F(CoordinatorTest, TruncateNotLearnedFill)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord1.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
{
Future<Option<uint64_t> > truncating = coord1.truncate(7);
- AWAIT_READY_FOR(truncating, Seconds(10));
- ASSERT_SOME(truncating.get());
- EXPECT_EQ(11u, truncating.get().get());
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1367,14 +1353,16 @@ TEST_F(CoordinatorTest, TruncateNotLearnedFill)
Coordinator coord2(2, replica3, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(11u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(11u, electing.get());
}
{
@@ -1423,23 +1411,20 @@ TEST_F(CoordinatorTest, TruncateLearnedFill)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord1.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
{
Future<Option<uint64_t> > truncating = coord1.truncate(7);
- AWAIT_READY_FOR(truncating, Seconds(10));
- ASSERT_SOME(truncating.get());
- EXPECT_EQ(11u, truncating.get().get());
+ AWAIT_READY(truncating);
+ EXPECT_SOME_EQ(11u, truncating.get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1453,14 +1438,16 @@ TEST_F(CoordinatorTest, TruncateLearnedFill)
Coordinator coord2(2, replica3, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(11u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(11u, electing.get());
}
{
@@ -1523,16 +1510,14 @@ TEST_F(RecoverTest, RacingCatchup)
{
Future<Option<uint64_t> > electing = coord1.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
for (uint64_t position = 1; position <= 10; position++) {
Future<Option<uint64_t> > appending = coord1.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(position, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
}
// Two replicas both want to recover.
@@ -1556,14 +1541,16 @@ TEST_F(RecoverTest, RacingCatchup)
Coordinator coord2(3, shared4, network2);
{
+ // Note that the first election should fail because 'coord2' get's
+ // it's proposal number from 'replica3' which is any empty log and
+ // thus a second attempt will need to be made.
Future<Option<uint64_t> > electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
+ AWAIT_READY(electing);
ASSERT_NONE(electing.get());
electing = coord2.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(10u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(10u, electing.get());
}
{
@@ -1579,9 +1566,8 @@ TEST_F(RecoverTest, RacingCatchup)
{
Future<Option<uint64_t> > appending = coord2.append("hello hello");
- AWAIT_READY_FOR(appending, Seconds(10));
- ASSERT_SOME(appending.get());
- EXPECT_EQ(11u, appending.get().get());
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(11u, appending.get());
}
{
@@ -1624,17 +1610,16 @@ TEST_F(RecoverTest, CatchupRetry)
{
Future<Option<uint64_t> > electing = coord.elect();
- AWAIT_READY_FOR(electing, Seconds(10));
- ASSERT_SOME(electing.get());
- EXPECT_EQ(0u, electing.get().get());
+ AWAIT_READY(electing);
+ EXPECT_SOME_EQ(0u, electing.get());
}
IntervalSet<uint64_t> positions;
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord.append(stringify(position));
- AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ Future<Option<uint64_t> > appending = coord.append(stringify(position));
+ AWAIT_READY(appending);
+ EXPECT_SOME_EQ(position, appending.get());
positions += position;
}
[2/6] git commit: Replaced Log::Writer constructor with explicit
Log::Writer::start.
Posted by be...@apache.org.
Replaced Log::Writer constructor with explicit Log::Writer::start.
Review: https://reviews.apache.org/r/17345
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/0d3bf14a
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/0d3bf14a
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/0d3bf14a
Branch: refs/heads/master
Commit: 0d3bf14a4658630b723408ccf560615e913fa0ee
Parents: 2c12b26
Author: Benjamin Hindman <be...@gmail.com>
Authored: Wed Jan 22 17:10:21 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/java/jni/org_apache_mesos_Log.cpp | 19 ++++++-
src/log/log.cpp | 82 ++++++++++--------------------
src/log/log.hpp | 16 ++++--
src/log/tool/benchmark.cpp | 32 +++++++++---
src/tests/log_tests.cpp | 21 ++++++--
5 files changed, 100 insertions(+), 70 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0d3bf14a/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 ed904c1..2b4b76a 100644
--- a/src/java/jni/org_apache_mesos_Log.cpp
+++ b/src/java/jni/org_apache_mesos_Log.cpp
@@ -428,12 +428,29 @@ JNIEXPORT void JNICALL Java_org_apache_mesos_Log_00024Writer_initialize
int retries = jretries;
// Create the C++ Log::Writer and initialize the __writer variable.
- Log::Writer* writer = new Log::Writer(log, seconds, retries);
+ Log::Writer* writer = new Log::Writer(log);
clazz = env->GetObjectClass(thiz);
jfieldID __writer = env->GetFieldID(clazz, "__writer", "J");
env->SetLongField(thiz, __writer, (jlong) writer);
+
+ // Try to start the writer.
+ while (retries-- > 0) {
+ Future<Option<Log::Position> > position = writer->start();
+
+ if (!position.await(seconds)) {
+ // Cancel the pending start. It is likely that we'll retry right
+ // away but that is safe.
+ position.discard();
+ } else if (position.isReady() && position.get().isSome()) {
+ // Started!
+ break;
+ }
+
+ // Either starting failed or was discarded, either way, retry!
+ // TODO(benh): Do we really want to retry on a failure?
+ }
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0d3bf14a/src/log/log.cpp
----------------------------------------------------------------------
diff --git a/src/log/log.cpp b/src/log/log.cpp
index 5938d57..f5b6670 100644
--- a/src/log/log.cpp
+++ b/src/log/log.cpp
@@ -153,7 +153,7 @@ class LogWriterProcess : public Process<LogWriterProcess>
public:
LogWriterProcess(Log* log);
- Future<Option<Log::Position> > elect();
+ Future<Option<Log::Position> > start();
Future<Log::Position> append(const string& bytes);
Future<Log::Position> truncate(const Log::Position& to);
@@ -169,13 +169,13 @@ private:
// Continuations.
void _recover();
- Future<Option<Log::Position> > _elect();
- Option<Log::Position> __elect(const Option<uint64_t>& position);
+ Future<Option<Log::Position> > _start();
+ Option<Log::Position> __start(const Option<uint64_t>& position);
static Future<Log::Position> _append(const Option<uint64_t>& position);
static Future<Log::Position> _truncate(const Option<uint64_t>& position);
- void failed(const string& message);
+ void failed(const string& message, const string& reason);
const size_t quorum;
const Shared<Network> network;
@@ -616,16 +616,17 @@ void LogWriterProcess::_recover()
}
-Future<Option<Log::Position> > LogWriterProcess::elect()
+Future<Option<Log::Position> > LogWriterProcess::start()
{
- return recover().then(defer(self(), &Self::_elect));
+ return recover().then(defer(self(), &Self::_start));
}
-Future<Option<Log::Position> > LogWriterProcess::_elect()
+Future<Option<Log::Position> > LogWriterProcess::_start()
{
// We delete the existing coordinator (if exists) and create a new
- // coordinator each time 'elect' is called.
+ // coordinator each time 'start' is called.
+ // TODO(benh): We shouldn't need to delete the coordinator everytime.
delete coordinator;
error = None();
@@ -633,19 +634,24 @@ Future<Option<Log::Position> > LogWriterProcess::_elect()
coordinator = new Coordinator(quorum, recovering.get(), network);
+ LOG(INFO) << "Attempting to start the writer";
+
return coordinator->elect()
- .then(defer(self(), &Self::__elect, lambda::_1))
- .onFailed(defer(self(), &Self::failed, lambda::_1));
+ .then(defer(self(), &Self::__start, lambda::_1))
+ .onFailed(defer(self(), &Self::failed, "Failed to start", lambda::_1));
}
-Option<Log::Position> LogWriterProcess::__elect(
+Option<Log::Position> LogWriterProcess::__start(
const Option<uint64_t>& position)
{
if (position.isNone()) {
+ LOG(INFO) << "Could not start the writer, but can be retried";
return None();
}
+ LOG(INFO) << "Writer started with ending position " << position.get();
+
return Log::Position(position.get());
}
@@ -664,7 +670,7 @@ Future<Log::Position> LogWriterProcess::append(const string& bytes)
return coordinator->append(bytes)
.then(lambda::bind(&Self::_append, lambda::_1))
- .onFailed(defer(self(), &Self::failed, lambda::_1));
+ .onFailed(defer(self(), &Self::failed, "Failed to append", lambda::_1));
}
@@ -692,7 +698,7 @@ Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
return coordinator->truncate(to.value)
.then(lambda::bind(&Self::_truncate, lambda::_1))
- .onFailed(defer(self(), &Self::failed, lambda::_1));
+ .onFailed(defer(self(), &Self::failed, "Failed to truncate", lambda::_1));
}
@@ -707,9 +713,9 @@ Future<Log::Position> LogWriterProcess::_truncate(
}
-void LogWriterProcess::failed(const string& message)
+void LogWriterProcess::failed(const string& message, const string& reason)
{
- error = message;
+ error = message + ": " + reason;
}
@@ -797,48 +803,10 @@ Future<Log::Position> Log::Reader::ending()
/////////////////////////////////////////////////
-Log::Writer::Writer(Log* log, const Duration& timeout, int retries)
+Log::Writer::Writer(Log* log)
{
process = new LogWriterProcess(log);
spawn(process);
-
- // Trying to get elected.
- for (;;) {
- LOG(INFO) << "Attempting to get elected within " << timeout;
-
- Future<Option<Log::Position> > future =
- dispatch(process, &LogWriterProcess::elect);
-
- if (!future.await(timeout)) {
- LOG(INFO) << "Timed out while trying to get elected";
-
- // Cancel the election. It is likely that the election is done
- // right after the timeout has been reached. In that case, we
- // may unnecessarily rerun the election, but it is safe.
- future.discard();
- } else {
- if (!future.isReady()) {
- string failure =
- future.isFailed() ?
- future.failure() :
- "Not expecting discarded future";
-
- LOG(ERROR) << "Failed to get elected: " << failure;
- break;
- } else if (future.get().isNone()) {
- LOG(INFO) << "Lost an election, but can be retried";
- } else {
- LOG(INFO) << "Elected with current position "
- << future.get().get().value;
- return;
- }
- }
-
- if (--retries < 0) {
- LOG(ERROR) << "Retry limit has been reached during election";
- break;
- }
- }
}
@@ -850,6 +818,12 @@ Log::Writer::~Writer()
}
+Future<Option<Log::Position> > Log::Writer::start()
+{
+ return dispatch(process, &LogWriterProcess::start);
+}
+
+
Future<Log::Position> Log::Writer::append(const string& data)
{
return dispatch(process, &LogWriterProcess::append, data);
http://git-wip-us.apache.org/repos/asf/mesos/blob/0d3bf14a/src/log/log.hpp
----------------------------------------------------------------------
diff --git a/src/log/log.hpp b/src/log/log.hpp
index d0dd9e8..c2cb6cf 100644
--- a/src/log/log.hpp
+++ b/src/log/log.hpp
@@ -154,12 +154,20 @@ public:
{
public:
// Creates a new writer associated with the specified log. Only
- // one writer (local and remote) is valid at a time. A writer
- // becomes invalid if any operation returns an error, and a new
- // writer must be created in order perform subsequent operations.
- Writer(Log* log, const Duration& timeout, int retries = 3);
+ // one writer (local or remote) can be valid at any point in
+ // time. A writer becomes invalid if either Writer::append or
+ // Writer::truncate return None, in which case, the writer (or
+ // another writer) must be restarted.
+ Writer(Log* log);
~Writer();
+ // Attempts to get a promise (from the log's replicas) for
+ // exclusive writes, i.e., no other writer's will be able to
+ // perform append and truncate operations. Returns the ending
+ // position of the log or none if the promise to exclusively write
+ // could not be attained but may be retried.
+ process::Future<Option<Position> > start();
+
// Attempts to append the specified data to the log. Returns the
// new ending position of the log or 'none' if this writer has
// lost it's promise to exclusively write (which can be reacquired
http://git-wip-us.apache.org/repos/asf/mesos/blob/0d3bf14a/src/log/tool/benchmark.cpp
----------------------------------------------------------------------
diff --git a/src/log/tool/benchmark.cpp b/src/log/tool/benchmark.cpp
index 7d5a032..54d7898 100644
--- a/src/log/tool/benchmark.cpp
+++ b/src/log/tool/benchmark.cpp
@@ -24,6 +24,7 @@
#include <sstream>
#include <process/clock.hpp>
+#include <process/future.hpp>
#include <process/process.hpp>
#include <process/time.hpp>
@@ -182,7 +183,18 @@ Try<Nothing> Benchmark::execute(int argc, char** argv)
flags.znode.get());
// Create the log writer.
- Log::Writer writer(&log, Seconds(15), 20);
+ Log::Writer writer(&log);
+
+ Future<Option<Log::Position> > position = writer.start();
+
+ if (!position.await(Seconds(15))) {
+ return Error("Failed to start a log writer: timed out");
+ } else if (!position.isReady()) {
+ return Error("Failed to start a log writer: " +
+ (position.isFailed()
+ ? position.failure()
+ : "Discarded future"));
+ }
// Statistics to output.
vector<Bytes> sizes;
@@ -227,13 +239,17 @@ Try<Nothing> Benchmark::execute(int argc, char** argv)
Stopwatch stopwatch;
stopwatch.start();
- Result<Log::Position> position =
- writer.append(data[i], Timeout::in(Seconds(10)));
-
- if (position.isError()) {
- return Error("Failed to append: " + position.error());
- } else if (position.isNone()) {
- return Error("Timed out while appending");
+ position = writer.append(data[i]);
+
+ if (position.await(Seconds(10))) {
+ return Error("Failed to append: timed out");
+ } else if (!position.isReady()) {
+ return Error("Failed to append: " +
+ (position.isFailed()
+ ? position.failure()
+ : "Discarded future"));
+ } else if (position.get().isNone()) {
+ return Error("Failed to append: exclusive write promise lost");
}
durations.push_back(stopwatch.elapsed());
http://git-wip-us.apache.org/repos/asf/mesos/blob/0d3bf14a/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index 295fdec..669179b 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -1703,7 +1703,12 @@ TEST_F(LogTest, WriteRead)
Log log(2, path2, pids);
- Log::Writer writer(&log, Seconds(10));
+ Log::Writer writer(&log);
+
+ Future<Option<Log::Position> > start = writer.start();
+
+ AWAIT_READY(start);
+ ASSERT_SOME(start.get());
Future<Log::Position> position = writer.append("hello world");
@@ -1739,7 +1744,12 @@ TEST_F(LogTest, Position)
Log log(2, path2, pids);
- Log::Writer writer(&log, Seconds(10));
+ Log::Writer writer(&log);
+
+ Future<Option<Log::Position> > start = writer.start();
+
+ AWAIT_READY(start);
+ ASSERT_SOME(start.get());
Future<Log::Position> position = writer.append("hello world");
@@ -1811,7 +1821,12 @@ TEST_F(LogZooKeeperTest, WriteRead)
Log log1(2, path1, servers, NO_TIMEOUT, "/log/", None());
Log log2(2, path2, servers, NO_TIMEOUT, "/log/", None());
- Log::Writer writer(&log2, Seconds(10));
+ Log::Writer writer(&log2);
+
+ Future<Option<Log::Position> > start = writer.start();
+
+ AWAIT_READY(start);
+ ASSERT_SOME(start.get());
Future<Log::Position> position = writer.append("hello world");
[5/6] git commit: Exposed coordinator demotion in Log.
Posted by be...@apache.org.
Exposed coordinator demotion in Log.
Review: https://reviews.apache.org/r/17346
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f8c3957a
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f8c3957a
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f8c3957a
Branch: refs/heads/master
Commit: f8c3957a74036ef1830a82e9617d97a18680429e
Parents: 0d3bf14
Author: Benjamin Hindman <be...@gmail.com>
Authored: Thu Jan 23 22:21:10 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/java/jni/org_apache_mesos_Log.cpp | 25 ++++++++++++++++----
src/log/log.cpp | 38 ++++++++++++------------------
src/tests/log_tests.cpp | 21 ++++++++++-------
3 files changed, 48 insertions(+), 36 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f8c3957a/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 2b4b76a..0548913 100644
--- a/src/java/jni/org_apache_mesos_Log.cpp
+++ b/src/java/jni/org_apache_mesos_Log.cpp
@@ -230,7 +230,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_ending
Future<Log::Position> position = reader->ending();
// TODO(benh): Don't wait forever for 'position'!
- return convert(env, position.get());
+ return convert<Log::Position>(env, position.get());
}
@@ -308,7 +308,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_append
jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
- Future<Log::Position> position = writer->append(data);
+ Future<Option<Log::Position> > position = writer->append(data);
if (!position.await(Seconds(jseconds))) {
// Timed out while trying to append the log.
@@ -327,11 +327,19 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_append
? position.failure().c_str()
: "Discarded future"));
return NULL;
+ } else if (position.get().isNone()) {
+ // Lost exclusive write promise.
+ env->ReleaseByteArrayElements(jdata, temp, 0);
+ clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
+ env->ThrowNew(
+ clazz,
+ "Exclusive write promise lost");
+ return NULL;
}
env->ReleaseByteArrayElements(jdata, temp, 0);
- jobject jposition = convert(env, position.get());
+ jobject jposition = convert<Log::Position>(env, position.get().get());
return jposition;
}
@@ -366,7 +374,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_truncate
jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
- Future<Log::Position> position = writer->truncate(to);
+ Future<Option<Log::Position> > position = writer->truncate(to);
if (!position.await(Seconds(jseconds))) {
// Timed out while trying to truncate the log.
@@ -383,9 +391,16 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_truncate
? position.failure().c_str()
: "Discarded future"));
return NULL;
+ } else if (position.get().isNone()) {
+ // Lost exclusive write promise.
+ clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
+ env->ThrowNew(
+ clazz,
+ "Exclusive write promise lost");
+ return NULL;
}
- jobject jposition = convert(env, position.get());
+ jobject jposition = convert<Log::Position>(env, position.get().get());
return jposition;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/f8c3957a/src/log/log.cpp
----------------------------------------------------------------------
diff --git a/src/log/log.cpp b/src/log/log.cpp
index f5b6670..6992518 100644
--- a/src/log/log.cpp
+++ b/src/log/log.cpp
@@ -154,14 +154,18 @@ public:
LogWriterProcess(Log* log);
Future<Option<Log::Position> > start();
- Future<Log::Position> append(const string& bytes);
- Future<Log::Position> truncate(const Log::Position& to);
+ Future<Option<Log::Position> > append(const string& bytes);
+ Future<Option<Log::Position> > truncate(const Log::Position& to);
protected:
virtual void initialize();
virtual void finalize();
private:
+ // Helper for converting an optional position returned from the
+ // coordinator into a Log::Position.
+ static Option<Log::Position> position(const Option<uint64_t>& position);
+
// Returns a future which gets set when the log recovery has
// finished (either succeeded or failed).
Future<Nothing> recover();
@@ -172,9 +176,6 @@ private:
Future<Option<Log::Position> > _start();
Option<Log::Position> __start(const Option<uint64_t>& position);
- static Future<Log::Position> _append(const Option<uint64_t>& position);
- static Future<Log::Position> _truncate(const Option<uint64_t>& position);
-
void failed(const string& message, const string& reason);
const size_t quorum;
@@ -656,7 +657,7 @@ Option<Log::Position> LogWriterProcess::__start(
}
-Future<Log::Position> LogWriterProcess::append(const string& bytes)
+Future<Option<Log::Position> > LogWriterProcess::append(const string& bytes)
{
LOG(INFO) << "Attempting to append " << bytes.size() << " bytes to the log";
@@ -669,22 +670,13 @@ Future<Log::Position> LogWriterProcess::append(const string& bytes)
}
return coordinator->append(bytes)
- .then(lambda::bind(&Self::_append, lambda::_1))
+ .then(lambda::bind(&Self::position, lambda::_1))
.onFailed(defer(self(), &Self::failed, "Failed to append", lambda::_1));
}
-Future<Log::Position> LogWriterProcess::_append(
- const Option<uint64_t>& position)
-{
- if (position.isNone()) {
- return Failure("Lost exclusive write access while appending");
- }
-
- return Log::Position(position.get());
-}
-
-Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
+Future<Option<Log::Position> > LogWriterProcess::truncate(
+ const Log::Position& to)
{
LOG(INFO) << "Attempting to truncate the log to " << to.value;
@@ -697,16 +689,16 @@ Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
}
return coordinator->truncate(to.value)
- .then(lambda::bind(&Self::_truncate, lambda::_1))
+ .then(lambda::bind(&Self::position, lambda::_1))
.onFailed(defer(self(), &Self::failed, "Failed to truncate", lambda::_1));
}
-Future<Log::Position> LogWriterProcess::_truncate(
+Option<Log::Position> LogWriterProcess::position(
const Option<uint64_t>& position)
{
if (position.isNone()) {
- return Failure("Lost exclusive write access while truncating");
+ return None();
}
return Log::Position(position.get());
@@ -824,13 +816,13 @@ Future<Option<Log::Position> > Log::Writer::start()
}
-Future<Log::Position> Log::Writer::append(const string& data)
+Future<Option<Log::Position> > Log::Writer::append(const string& data)
{
return dispatch(process, &LogWriterProcess::append, data);
}
-Future<Log::Position> Log::Writer::truncate(const Log::Position& to)
+Future<Option<Log::Position> > Log::Writer::truncate(const Log::Position& to)
{
return dispatch(process, &LogWriterProcess::truncate, to);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/f8c3957a/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index 669179b..e49610b 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -1710,19 +1710,20 @@ TEST_F(LogTest, WriteRead)
AWAIT_READY(start);
ASSERT_SOME(start.get());
- Future<Log::Position> position = writer.append("hello world");
+ Future<Option<Log::Position> > position = writer.append("hello world");
AWAIT_READY(position);
+ ASSERT_SOME(position.get());
Log::Reader reader(&log);
Future<list<Log::Entry> > entries =
- reader.read(position.get(), position.get());
+ reader.read(position.get().get(), position.get().get());
AWAIT_READY(entries);
ASSERT_EQ(1u, entries.get().size());
- EXPECT_EQ(position.get(), entries.get().front().position);
+ EXPECT_EQ(position.get().get(), entries.get().front().position);
EXPECT_EQ("hello world", entries.get().front().data);
}
@@ -1751,11 +1752,14 @@ TEST_F(LogTest, Position)
AWAIT_READY(start);
ASSERT_SOME(start.get());
- Future<Log::Position> position = writer.append("hello world");
+ Future<Option<Log::Position> > position = writer.append("hello world");
AWAIT_READY(position);
+ ASSERT_SOME(position.get());
- ASSERT_EQ(position.get(), log.position(position.get().identity()));
+ ASSERT_EQ(
+ position.get().get(),
+ log.position(position.get().get().identity()));
}
@@ -1828,19 +1832,20 @@ TEST_F(LogZooKeeperTest, WriteRead)
AWAIT_READY(start);
ASSERT_SOME(start.get());
- Future<Log::Position> position = writer.append("hello world");
+ Future<Option<Log::Position> > position = writer.append("hello world");
AWAIT_READY(position);
+ ASSERT_SOME(position.get());
Log::Reader reader(&log2);
Future<list<Log::Entry> > entries =
- reader.read(position.get(), position.get());
+ reader.read(position.get().get(), position.get().get());
AWAIT_READY(entries);
ASSERT_EQ(1u, entries.get().size());
- EXPECT_EQ(position.get(), entries.get().front().position);
+ EXPECT_EQ(position.get().get(), entries.get().front().position);
EXPECT_EQ("hello world", entries.get().front().data);
}
#endif // MESOS_HAS_JAVA
[4/6] git commit: Exposed coordinator demotion.
Posted by be...@apache.org.
Exposed coordinator demotion.
A demotion was captured as a failed future which makes it hard to
distinguish from a non-retryable failure. By returning an option for
Coordinator::append and Coordinator::truncate we can distinguish
demotion from a failure.
Review: https://reviews.apache.org/r/17343
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/93c4d4d2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/93c4d4d2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/93c4d4d2
Branch: refs/heads/master
Commit: 93c4d4d21f4ebed2ab7cfced3e09b16492f55edb
Parents: bce0b26
Author: Benjamin Hindman <be...@gmail.com>
Authored: Mon Jan 20 19:45:49 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/log/coordinator.cpp | 49 +++++++++++----------
src/log/coordinator.hpp | 10 +++--
src/log/log.cpp | 44 +++++++++++++------
src/log/log.hpp | 18 ++++----
src/state/zookeeper.hpp | 3 +-
src/tests/log_tests.cpp | 101 +++++++++++++++++++++++++------------------
6 files changed, 131 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/log/coordinator.cpp
----------------------------------------------------------------------
diff --git a/src/log/coordinator.cpp b/src/log/coordinator.cpp
index 9a2faba..aacbcba 100644
--- a/src/log/coordinator.cpp
+++ b/src/log/coordinator.cpp
@@ -63,8 +63,8 @@ public:
// See comments in 'coordinator.hpp'.
Future<Option<uint64_t> > elect();
Future<uint64_t> demote();
- Future<uint64_t> append(const string& bytes);
- Future<uint64_t> truncate(uint64_t to);
+ Future<Option<uint64_t> > append(const string& bytes);
+ Future<Option<uint64_t> > truncate(uint64_t to);
protected:
virtual void finalize()
@@ -93,12 +93,14 @@ private:
// Writing related functions. //
/////////////////////////////////
- Future<uint64_t> write(const Action& action);
+ Future<Option<uint64_t> > write(const Action& action);
Future<WriteResponse> runWritePhase(const Action& action);
- Future<Nothing> checkWritePhase(const WriteResponse& response);
+ Future<Option<uint64_t> > checkWritePhase(
+ const Action& action,
+ const WriteResponse& response);
Future<Nothing> runLearnPhase(const Action& action);
Future<bool> checkLearnPhase(const Action& action);
- Future<uint64_t> updateIndexAfterWritten(bool missing);
+ Future<Option<uint64_t> > updateIndexAfterWritten(bool missing);
void writingFinished();
void writingFailed();
void writingAborted();
@@ -129,7 +131,7 @@ private:
uint64_t index;
Future<Option<uint64_t> > electing;
- Future<uint64_t> writing;
+ Future<Option<uint64_t> > writing;
};
@@ -284,10 +286,10 @@ Future<uint64_t> CoordinatorProcess::demote()
/////////////////////////////////////////////////
-Future<uint64_t> CoordinatorProcess::append(const string& bytes)
+Future<Option<uint64_t> > CoordinatorProcess::append(const string& bytes)
{
if (state == INITIAL) {
- return Failure("Coordinator is not elected");
+ return None();
} else if (state == ELECTING) {
return Failure("Coordinator is being elected");
} else if (state == WRITING) {
@@ -306,10 +308,10 @@ Future<uint64_t> CoordinatorProcess::append(const string& bytes)
}
-Future<uint64_t> CoordinatorProcess::truncate(uint64_t to)
+Future<Option<uint64_t> > CoordinatorProcess::truncate(uint64_t to)
{
if (state == INITIAL) {
- return Failure("Coordinator is not elected");
+ return None();
} else if (state == ELECTING) {
return Failure("Coordinator is being elected");
} else if (state == WRITING) {
@@ -328,7 +330,7 @@ Future<uint64_t> CoordinatorProcess::truncate(uint64_t to)
}
-Future<uint64_t> CoordinatorProcess::write(const Action& action)
+Future<Option<uint64_t> > CoordinatorProcess::write(const Action& action)
{
LOG(INFO) << "Coordinator attempting to write " << action.type()
<< " action at position " << action.position();
@@ -339,10 +341,7 @@ Future<uint64_t> CoordinatorProcess::write(const Action& action)
state = WRITING;
writing = runWritePhase(action)
- .then(defer(self(), &Self::checkWritePhase, lambda::_1))
- .then(defer(self(), &Self::runLearnPhase, action))
- .then(defer(self(), &Self::checkLearnPhase, action))
- .then(defer(self(), &Self::updateIndexAfterWritten, lambda::_1))
+ .then(defer(self(), &Self::checkWritePhase, action, lambda::_1))
.onReady(defer(self(), &Self::writingFinished))
.onFailed(defer(self(), &Self::writingFailed))
.onDiscarded(defer(self(), &Self::writingAborted));
@@ -357,18 +356,21 @@ Future<WriteResponse> CoordinatorProcess::runWritePhase(const Action& action)
}
-Future<Nothing> CoordinatorProcess::checkWritePhase(
+Future<Option<uint64_t> > CoordinatorProcess::checkWritePhase(
+ const Action& action,
const WriteResponse& response)
{
- if (!response.okay()) {
+ if (!response.okay()) {
// Received a NACK. Save the proposal number.
CHECK_LE(proposal, response.proposal());
proposal = response.proposal();
- return Failure("Coordinator demoted");
- } else {
- return Nothing();
+ return None();
}
+
+ return runLearnPhase(action)
+ .then(defer(self(), &Self::checkLearnPhase, action))
+ .then(defer(self(), &Self::updateIndexAfterWritten, lambda::_1));
}
@@ -387,7 +389,8 @@ Future<bool> CoordinatorProcess::checkLearnPhase(const Action& action)
}
-Future<uint64_t> CoordinatorProcess::updateIndexAfterWritten(bool missing)
+Future<Option<uint64_t> > CoordinatorProcess::updateIndexAfterWritten(
+ bool missing)
{
CHECK(!missing) << "Not expecting local replica to be missing position "
<< index << " after the writing is done";
@@ -457,13 +460,13 @@ Future<uint64_t> Coordinator::demote()
}
-Future<uint64_t> Coordinator::append(const string& bytes)
+Future<Option<uint64_t> > Coordinator::append(const string& bytes)
{
return dispatch(process, &CoordinatorProcess::append, bytes);
}
-Future<uint64_t> Coordinator::truncate(uint64_t to)
+Future<Option<uint64_t> > Coordinator::truncate(uint64_t to)
{
return dispatch(process, &CoordinatorProcess::truncate, to);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/log/coordinator.hpp
----------------------------------------------------------------------
diff --git a/src/log/coordinator.hpp b/src/log/coordinator.hpp
index 35b68e9..b8a475f 100644
--- a/src/log/coordinator.hpp
+++ b/src/log/coordinator.hpp
@@ -61,13 +61,15 @@ public:
process::Future<uint64_t> demote();
// Appends the specified bytes to the end of the log. Returns the
- // position of the appended entry if the operation succeeds.
- process::Future<uint64_t> append(const std::string& bytes);
+ // position of the appended entry if the operation succeeds or none
+ // if the coordinator was demoted.
+ process::Future<Option<uint64_t> > append(const std::string& bytes);
// Removes all log entries preceding the log entry at the given
// position (to). Returns the position at which the truncate
- // operation is written if the operation succeeds.
- process::Future<uint64_t> truncate(uint64_t to);
+ // operation is written if the operation succeeds or none if the
+ // coordinator was demoted.
+ process::Future<Option<uint64_t> > truncate(uint64_t to);
private:
CoordinatorProcess* process;
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/log/log.cpp
----------------------------------------------------------------------
diff --git a/src/log/log.cpp b/src/log/log.cpp
index a41edd3..5938d57 100644
--- a/src/log/log.cpp
+++ b/src/log/log.cpp
@@ -162,9 +162,6 @@ protected:
virtual void finalize();
private:
- // Returns a position from a raw value.
- static Log::Position position(uint64_t value);
-
// Returns a future which gets set when the log recovery has
// finished (either succeeded or failed).
Future<Nothing> recover();
@@ -173,7 +170,10 @@ private:
void _recover();
Future<Option<Log::Position> > _elect();
- Option<Log::Position> __elect(const Option<uint64_t>& result);
+ Option<Log::Position> __elect(const Option<uint64_t>& position);
+
+ static Future<Log::Position> _append(const Option<uint64_t>& position);
+ static Future<Log::Position> _truncate(const Option<uint64_t>& position);
void failed(const string& message);
@@ -639,13 +639,14 @@ Future<Option<Log::Position> > LogWriterProcess::_elect()
}
-Option<Log::Position> LogWriterProcess::__elect(const Option<uint64_t>& result)
+Option<Log::Position> LogWriterProcess::__elect(
+ const Option<uint64_t>& position)
{
- if (result.isNone()) {
+ if (position.isNone()) {
return None();
- } else {
- return position(result.get());
}
+
+ return Log::Position(position.get());
}
@@ -662,11 +663,21 @@ Future<Log::Position> LogWriterProcess::append(const string& bytes)
}
return coordinator->append(bytes)
- .then(lambda::bind(&Self::position, lambda::_1))
+ .then(lambda::bind(&Self::_append, lambda::_1))
.onFailed(defer(self(), &Self::failed, lambda::_1));
}
+Future<Log::Position> LogWriterProcess::_append(
+ const Option<uint64_t>& position)
+{
+ if (position.isNone()) {
+ return Failure("Lost exclusive write access while appending");
+ }
+
+ return Log::Position(position.get());
+}
+
Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
{
LOG(INFO) << "Attempting to truncate the log to " << to.value;
@@ -680,20 +691,25 @@ Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
}
return coordinator->truncate(to.value)
- .then(lambda::bind(&Self::position, lambda::_1))
+ .then(lambda::bind(&Self::_truncate, lambda::_1))
.onFailed(defer(self(), &Self::failed, lambda::_1));
}
-void LogWriterProcess::failed(const string& message)
+Future<Log::Position> LogWriterProcess::_truncate(
+ const Option<uint64_t>& position)
{
- error = message;
+ if (position.isNone()) {
+ return Failure("Lost exclusive write access while truncating");
+ }
+
+ return Log::Position(position.get());
}
-Log::Position LogWriterProcess::position(uint64_t value)
+void LogWriterProcess::failed(const string& message)
{
- return Log::Position(value);
+ error = message;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/log/log.hpp
----------------------------------------------------------------------
diff --git a/src/log/log.hpp b/src/log/log.hpp
index a8de0b1..d0dd9e8 100644
--- a/src/log/log.hpp
+++ b/src/log/log.hpp
@@ -160,17 +160,17 @@ public:
Writer(Log* log, const Duration& timeout, int retries = 3);
~Writer();
- // Attempts to append the specified data to the log. A none result
- // means the operation timed out, otherwise the new ending
- // position of the log is returned or an error. Upon error a new
- // Writer must be created.
- process::Future<Position> append(const std::string& data);
+ // Attempts to append the specified data to the log. Returns the
+ // new ending position of the log or 'none' if this writer has
+ // lost it's promise to exclusively write (which can be reacquired
+ // by invoking Writer::start).
+ process::Future<Option<Position> > append(const std::string& data);
// Attempts to truncate the log up to but not including the
- // specificed position. A none result means the operation timed
- // out, otherwise the new ending position of the log is returned
- // or an error. Upon error a new Writer must be created.
- process::Future<Position> truncate(const Position& to);
+ // specificed position. Returns the new ending position of the log
+ // or 'none' if this writer has lost it's promise to exclusively
+ // write (which can be reacquired by invoking Writer::start).
+ process::Future<Option<Position> > truncate(const Position& to);
private:
LogWriterProcess* process;
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/state/zookeeper.hpp
----------------------------------------------------------------------
diff --git a/src/state/zookeeper.hpp b/src/state/zookeeper.hpp
index d1d1fed..cfc8831 100644
--- a/src/state/zookeeper.hpp
+++ b/src/state/zookeeper.hpp
@@ -39,8 +39,7 @@ public:
const std::string& servers,
const Duration& timeout,
const std::string& znode,
- const Option<zookeeper::Authentication>& auth =
- Option<zookeeper::Authentication>());
+ const Option<zookeeper::Authentication>& auth = None());
virtual ~ZooKeeperStorage();
// Storage implementation.
http://git-wip-us.apache.org/repos/asf/mesos/blob/93c4d4d2/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index e15cb76..295fdec 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -642,9 +642,10 @@ TEST_F(CoordinatorTest, AppendRead)
uint64_t position;
{
- Future<uint64_t> appending = coord.append("hello world");
+ Future<Option<uint64_t> > appending = coord.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -691,9 +692,10 @@ TEST_F(CoordinatorTest, AppendReadError)
uint64_t position;
{
- Future<uint64_t> appending = coord.append("hello world");
+ Future<Option<uint64_t> > appending = coord.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -739,7 +741,7 @@ TEST_F(CoordinatorTest, AppendDiscarded)
replica2.reset();
{
- Future<uint64_t> appending = coord.append("hello world");
+ Future<Option<uint64_t> > appending = coord.append("hello world");
ASSERT_TRUE(appending.isPending());
appending.discard();
@@ -747,10 +749,10 @@ TEST_F(CoordinatorTest, AppendDiscarded)
}
{
- Future<uint64_t> appending = coord.append("hello moto");
- AWAIT_FAILED(appending);
+ Future<Option<uint64_t> > appending = coord.append("hello moto");
+ AWAIT_SOME(appending);
- EXPECT_EQ("Coordinator is not elected", appending.failure());
+ EXPECT_NONE(appending.get());
}
}
@@ -817,7 +819,7 @@ TEST_F(CoordinatorTest, AppendNoQuorum)
Clock::pause();
- Future<uint64_t> appending = coord.append("hello world");
+ Future<Option<uint64_t> > appending = coord.append("hello world");
Clock::advance(Seconds(10));
Clock::settle();
@@ -859,9 +861,10 @@ TEST_F(CoordinatorTest, Failover)
uint64_t position;
{
- Future<uint64_t> appending = coord1.append("hello world");
+ Future<Option<uint64_t> > appending = coord1.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -919,9 +922,10 @@ TEST_F(CoordinatorTest, Demoted)
uint64_t position;
{
- Future<uint64_t> appending = coord1.append("hello world");
+ Future<Option<uint64_t> > appending = coord1.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -937,15 +941,16 @@ TEST_F(CoordinatorTest, Demoted)
}
{
- Future<uint64_t> appending = coord1.append("hello moto");
- AWAIT_FAILED(appending);
- EXPECT_EQ("Coordinator demoted", appending.failure());
+ Future<Option<uint64_t> > appending = coord1.append("hello moto");
+ AWAIT_READY_FOR(appending, Seconds(10));
+ EXPECT_NONE(appending.get());
}
{
- Future<uint64_t> appending = coord2.append("hello hello");
+ Future<Option<uint64_t> > appending = coord2.append("hello hello");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(2u, position);
}
@@ -996,9 +1001,10 @@ TEST_F(CoordinatorTest, Fill)
uint64_t position;
{
- Future<uint64_t> appending = coord1.append("hello world");
+ Future<Option<uint64_t> > appending = coord1.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -1074,9 +1080,10 @@ TEST_F(CoordinatorTest, NotLearnedFill)
uint64_t position;
{
- Future<uint64_t> appending = coord1.append("hello world");
+ Future<Option<uint64_t> > appending = coord1.append("hello world");
AWAIT_READY_FOR(appending, Seconds(10));
- position = appending.get();
+ ASSERT_SOME(appending.get());
+ position = appending.get().get();
EXPECT_EQ(1u, position);
}
@@ -1142,9 +1149,10 @@ TEST_F(CoordinatorTest, MultipleAppends)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
{
@@ -1197,9 +1205,10 @@ TEST_F(CoordinatorTest, MultipleAppendsNotLearnedFill)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord1.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord1.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1265,15 +1274,17 @@ TEST_F(CoordinatorTest, Truncate)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
{
- Future<uint64_t> truncating = coord.truncate(7);
+ Future<Option<uint64_t> > truncating = coord.truncate(7);
AWAIT_READY_FOR(truncating, Seconds(10));
- EXPECT_EQ(11u, truncating.get());
+ ASSERT_SOME(truncating.get());
+ EXPECT_EQ(11u, truncating.get().get());
}
{
@@ -1332,15 +1343,17 @@ TEST_F(CoordinatorTest, TruncateNotLearnedFill)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord1.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord1.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
{
- Future<uint64_t> truncating = coord1.truncate(7);
+ Future<Option<uint64_t> > truncating = coord1.truncate(7);
AWAIT_READY_FOR(truncating, Seconds(10));
- EXPECT_EQ(11u, truncating.get());
+ ASSERT_SOME(truncating.get());
+ EXPECT_EQ(11u, truncating.get().get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1416,15 +1429,17 @@ TEST_F(CoordinatorTest, TruncateLearnedFill)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord1.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord1.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
{
- Future<uint64_t> truncating = coord1.truncate(7);
+ Future<Option<uint64_t> > truncating = coord1.truncate(7);
AWAIT_READY_FOR(truncating, Seconds(10));
- EXPECT_EQ(11u, truncating.get());
+ ASSERT_SOME(truncating.get());
+ EXPECT_EQ(11u, truncating.get().get());
}
Shared<Replica> replica3(new Replica(path3));
@@ -1514,9 +1529,10 @@ TEST_F(RecoverTest, RacingCatchup)
}
for (uint64_t position = 1; position <= 10; position++) {
- Future<uint64_t> appending = coord1.append(stringify(position));
+ Future<Option<uint64_t> > appending = coord1.append(stringify(position));
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(position, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(position, appending.get().get());
}
// Two replicas both want to recover.
@@ -1562,9 +1578,10 @@ TEST_F(RecoverTest, RacingCatchup)
}
{
- Future<uint64_t> appending = coord2.append("hello hello");
+ Future<Option<uint64_t> > appending = coord2.append("hello hello");
AWAIT_READY_FOR(appending, Seconds(10));
- EXPECT_EQ(11u, appending.get());
+ ASSERT_SOME(appending.get());
+ EXPECT_EQ(11u, appending.get().get());
}
{
[6/6] git commit: Simplified coordinator election semantics.
Posted by be...@apache.org.
Simplified coordinator election semantics.
Review: https://reviews.apache.org/r/17344
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2c12b260
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2c12b260
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2c12b260
Branch: refs/heads/master
Commit: 2c12b260a71bca2e06f26971cfce9cde85c4b114
Parents: 93c4d4d
Author: Benjamin Hindman <be...@gmail.com>
Authored: Mon Jan 20 20:06:14 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/log/coordinator.cpp | 12 ++++--------
1 file changed, 4 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2c12b260/src/log/coordinator.cpp
----------------------------------------------------------------------
diff --git a/src/log/coordinator.cpp b/src/log/coordinator.cpp
index aacbcba..caa4b85 100644
--- a/src/log/coordinator.cpp
+++ b/src/log/coordinator.cpp
@@ -143,9 +143,9 @@ private:
Future<Option<uint64_t> > CoordinatorProcess::elect()
{
if (state == ELECTING) {
- return Failure("Coordinator already being elected");
+ return electing;
} else if (state == ELECTED) {
- return Failure("Coordinator already elected");
+ return index - 1; // The last learned position!
} else if (state == WRITING) {
return Failure("Coordinator already elected, and is currently writing");
}
@@ -288,10 +288,8 @@ Future<uint64_t> CoordinatorProcess::demote()
Future<Option<uint64_t> > CoordinatorProcess::append(const string& bytes)
{
- if (state == INITIAL) {
+ if (state == INITIAL || state == ELECTING) {
return None();
- } else if (state == ELECTING) {
- return Failure("Coordinator is being elected");
} else if (state == WRITING) {
return Failure("Coordinator is currently writing");
}
@@ -310,10 +308,8 @@ Future<Option<uint64_t> > CoordinatorProcess::append(const string& bytes)
Future<Option<uint64_t> > CoordinatorProcess::truncate(uint64_t to)
{
- if (state == INITIAL) {
+ if (state == INITIAL || state == ELECTING) {
return None();
- } else if (state == ELECTING) {
- return Failure("Coordinator is being elected");
} else if (state == WRITING) {
return Failure("Coordinator is currently writing");
}
[3/6] git commit: Updated Log::Reader and Log::Writer to be
asynchronous.
Posted by be...@apache.org.
Updated Log::Reader and Log::Writer to be asynchronous.
Review: https://reviews.apache.org/r/17342
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/bce0b265
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/bce0b265
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/bce0b265
Branch: refs/heads/master
Commit: bce0b265c481b1da88c0c204cdc2ea2cb0c70939
Parents: 8f4516f
Author: Benjamin Hindman <be...@gmail.com>
Authored: Sat Jan 18 13:48:41 2014 -0800
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Sun Mar 9 23:35:10 2014 -0700
----------------------------------------------------------------------
src/java/jni/org_apache_mesos_Log.cpp | 82 +++++++++++++----------
src/log/log.cpp | 101 +++++------------------------
src/log/log.hpp | 19 ++----
src/tests/log_tests.cpp | 29 ++++-----
4 files changed, 86 insertions(+), 145 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/bce0b265/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 13096fe..ed904c1 100644
--- a/src/java/jni/org_apache_mesos_Log.cpp
+++ b/src/java/jni/org_apache_mesos_Log.cpp
@@ -19,7 +19,7 @@
#include <jni.h>
#include <stdint.h>
-#include <process/timeout.hpp>
+#include <process/future.hpp>
#include <stout/check.hpp>
#include <stout/duration.hpp>
@@ -153,21 +153,25 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_read
jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
- Result<std::list<Log::Entry> > entries =
- reader->read(from, to, Timeout::in(Seconds(jseconds)));
+ Future<std::list<Log::Entry> > entries = reader->read(from, to);
- if (entries.isError()) {
- clazz = env->FindClass("org/apache/mesos/Log$OperationFailedException");
- env->ThrowNew(clazz, entries.error().c_str());
- return NULL;
- } else if (entries.isNone()) {
+ if (!entries.await(Seconds(jseconds))) {
+ // Timed out while trying to read the log.
+ entries.discard();
clazz = env->FindClass("java/util/concurrent/TimeoutException");
env->ThrowNew(clazz, "Timed out while attempting to read");
return NULL;
+ } else if (!entries.isReady()) {
+ // Failed to read the log.
+ clazz = env->FindClass("org/apache/mesos/Log$OperationFailedException");
+ env->ThrowNew(
+ clazz,
+ (entries.isFailed()
+ ? entries.failure().c_str()
+ : "Discarded future"));
+ return NULL;
}
- CHECK_SOME(entries);
-
// List entries = new ArrayList();
clazz = env->FindClass("java/util/ArrayList");
@@ -201,9 +205,10 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_beginning
Log::Reader* reader = (Log::Reader*) env->GetLongField(thiz, __reader);
- Log::Position position = reader->beginning();
+ Future<Log::Position> position = reader->beginning();
- return convert<Log::Position>(env, position);
+ // TODO(benh): Don't wait forever for 'position'!
+ return convert<Log::Position>(env, position.get());
}
@@ -222,9 +227,10 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Reader_ending
Log::Reader* reader = (Log::Reader*) env->GetLongField(thiz, __reader);
- Log::Position position = reader->ending();
+ Future<Log::Position> position = reader->ending();
- return convert(env, position);
+ // TODO(benh): Don't wait forever for 'position'!
+ return convert(env, position.get());
}
@@ -302,22 +308,28 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_append
jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
- Result<Log::Position> position =
- writer->append(data, Timeout::in(Seconds(jseconds)));
+ Future<Log::Position> position = writer->append(data);
- env->ReleaseByteArrayElements(jdata, temp, 0);
-
- if (position.isError()) {
- clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
- env->ThrowNew(clazz, position.error().c_str());
- return NULL;
- } else if (position.isNone()) {
+ if (!position.await(Seconds(jseconds))) {
+ // Timed out while trying to append the log.
+ position.discard();
+ env->ReleaseByteArrayElements(jdata, temp, 0);
clazz = env->FindClass("java/util/concurrent/TimeoutException");
env->ThrowNew(clazz, "Timed out while attempting to append");
return NULL;
+ } else if (!position.isReady()) {
+ // Failed to append the log.
+ env->ReleaseByteArrayElements(jdata, temp, 0);
+ clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
+ env->ThrowNew(
+ clazz,
+ (position.isFailed()
+ ? position.failure().c_str()
+ : "Discarded future"));
+ return NULL;
}
- CHECK_SOME(position);
+ env->ReleaseByteArrayElements(jdata, temp, 0);
jobject jposition = convert(env, position.get());
@@ -354,21 +366,25 @@ JNIEXPORT jobject JNICALL Java_org_apache_mesos_Log_00024Writer_truncate
jlong jseconds = env->CallLongMethod(junit, toSeconds, jtimeout);
- Result<Log::Position> position =
- writer->truncate(to, Timeout::in(Seconds(jseconds)));
+ Future<Log::Position> position = writer->truncate(to);
- if (position.isError()) {
- clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
- env->ThrowNew(clazz, position.error().c_str());
- return NULL;
- } else if (position.isNone()) {
+ if (!position.await(Seconds(jseconds))) {
+ // Timed out while trying to truncate the log.
+ position.discard();
clazz = env->FindClass("java/util/concurrent/TimeoutException");
env->ThrowNew(clazz, "Timed out while attempting to truncate");
return NULL;
+ } else if (!position.isReady()) {
+ // Failed to truncate the log.
+ clazz = env->FindClass("org/apache/mesos/Log$WriterFailedException");
+ env->ThrowNew(
+ clazz,
+ (position.isFailed()
+ ? position.failure().c_str()
+ : "Discarded future"));
+ return NULL;
}
- CHECK_SOME(position);
-
jobject jposition = convert(env, position.get());
return jposition;
http://git-wip-us.apache.org/repos/asf/mesos/blob/bce0b265/src/log/log.cpp
----------------------------------------------------------------------
diff --git a/src/log/log.cpp b/src/log/log.cpp
index c427920..a41edd3 100644
--- a/src/log/log.cpp
+++ b/src/log/log.cpp
@@ -20,6 +20,7 @@
#include <process/defer.hpp>
#include <process/dispatch.hpp>
+#include <process/future.hpp>
#include <process/id.hpp>
#include <process/owned.hpp>
#include <process/process.hpp>
@@ -650,6 +651,8 @@ Option<Log::Position> LogWriterProcess::__elect(const Option<uint64_t>& result)
Future<Log::Position> LogWriterProcess::append(const string& bytes)
{
+ LOG(INFO) << "Attempting to append " << bytes.size() << " bytes to the log";
+
if (coordinator == NULL) {
return Failure("No election has been performed");
}
@@ -666,6 +669,8 @@ Future<Log::Position> LogWriterProcess::append(const string& bytes)
Future<Log::Position> LogWriterProcess::truncate(const Log::Position& to)
{
+ LOG(INFO) << "Attempting to truncate the log to " << to.value;
+
if (coordinator == NULL) {
return Failure("No election has been performed");
}
@@ -751,47 +756,23 @@ Log::Reader::~Reader()
}
-Result<list<Log::Entry> > Log::Reader::read(
+Future<list<Log::Entry> > Log::Reader::read(
const Log::Position& from,
- const Log::Position& to,
- const Timeout& timeout)
+ const Log::Position& to)
{
- Future<list<Log::Entry> > future =
- dispatch(process, &LogReaderProcess::read, from, to);
-
- if (!future.await(timeout.remaining())) {
- LOG(INFO) << "Timed out while trying to read the log";
-
- future.discard();
- return None();
- } else {
- if (!future.isReady()) {
- string failure =
- future.isFailed() ?
- future.failure() :
- "Not expecting discarded future";
-
- LOG(ERROR) << "Failed to read the log: " << failure;
-
- return Error(failure);
- } else {
- return future.get();
- }
- }
+ return dispatch(process, &LogReaderProcess::read, from, to);
}
-Log::Position Log::Reader::beginning()
+Future<Log::Position> Log::Reader::beginning()
{
- // TODO(benh): Take a timeout and return an Option.
- return dispatch(process, &LogReaderProcess::beginning).get();
+ return dispatch(process, &LogReaderProcess::beginning);
}
-Log::Position Log::Reader::ending()
+Future<Log::Position> Log::Reader::ending()
{
- // TODO(benh): Take a timeout and return an Option.
- return dispatch(process, &LogReaderProcess::ending).get();
+ return dispatch(process, &LogReaderProcess::ending);
}
@@ -853,65 +834,15 @@ Log::Writer::~Writer()
}
-Result<Log::Position> Log::Writer::append(
- const string& data,
- const Timeout& timeout)
+Future<Log::Position> Log::Writer::append(const string& data)
{
- LOG(INFO) << "Attempting to append " << data.size() << " bytes to the log";
-
- Future<Log::Position> future =
- dispatch(process, &LogWriterProcess::append, data);
-
- if (!future.await(timeout.remaining())) {
- LOG(INFO) << "Timed out while trying to append the log";
-
- future.discard();
- return None();
- } else {
- if (!future.isReady()) {
- string failure =
- future.isFailed() ?
- future.failure() :
- "Not expecting discarded future";
-
- LOG(ERROR) << "Failed to append the log: " << failure;
-
- return Error(failure);
- } else {
- return future.get();
- }
- }
+ return dispatch(process, &LogWriterProcess::append, data);
}
-Result<Log::Position> Log::Writer::truncate(
- const Log::Position& to,
- const Timeout& timeout)
+Future<Log::Position> Log::Writer::truncate(const Log::Position& to)
{
- LOG(INFO) << "Attempting to truncate the log to " << to.value;
-
- Future<Log::Position> future =
- dispatch(process, &LogWriterProcess::truncate, to);
-
- if (!future.await(timeout.remaining())) {
- LOG(INFO) << "Timed out while trying to truncate the log";
-
- future.discard();
- return None();
- } else {
- if (!future.isReady()) {
- string failure =
- future.isFailed() ?
- future.failure() :
- "Not expecting discarded future";
-
- LOG(ERROR) << "Failed to truncate the log: " << failure;
-
- return Error(failure);
- } else {
- return future.get();
- }
- }
+ return dispatch(process, &LogWriterProcess::truncate, to);
}
} // namespace log {
http://git-wip-us.apache.org/repos/asf/mesos/blob/bce0b265/src/log/log.hpp
----------------------------------------------------------------------
diff --git a/src/log/log.hpp b/src/log/log.hpp
index 628a4b8..a8de0b1 100644
--- a/src/log/log.hpp
+++ b/src/log/log.hpp
@@ -25,6 +25,7 @@
#include <set>
#include <string>
+#include <process/future.hpp>
#include <process/owned.hpp>
#include <process/process.hpp>
#include <process/shared.hpp>
@@ -33,7 +34,6 @@
#include <stout/duration.hpp>
#include <stout/none.hpp>
#include <stout/option.hpp>
-#include <stout/result.hpp>
#include "zookeeper/group.hpp"
@@ -131,21 +131,20 @@ public:
// Returns all entries between the specified positions, unless
// those positions are invalid, in which case returns an error.
- Result<std::list<Entry> > read(
+ process::Future<std::list<Entry> > read(
const Position& from,
- const Position& to,
- const process::Timeout& timeout);
+ const Position& to);
// Returns the beginning position of the log from the perspective
// of the local replica (which may be out of date if the log has
// been opened and truncated while this replica was partitioned).
- Position beginning();
+ process::Future<Position> beginning();
// Returns the ending (i.e., last) position of the log from the
// perspective of the local replica (which may be out of date if
// the log has been opened and appended to while this replica was
// partitioned).
- Position ending();
+ process::Future<Position> ending();
private:
LogReaderProcess* process;
@@ -165,17 +164,13 @@ public:
// means the operation timed out, otherwise the new ending
// position of the log is returned or an error. Upon error a new
// Writer must be created.
- Result<Position> append(
- const std::string& data,
- const process::Timeout& timeout);
+ process::Future<Position> append(const std::string& data);
// Attempts to truncate the log up to but not including the
// specificed position. A none result means the operation timed
// out, otherwise the new ending position of the log is returned
// or an error. Upon error a new Writer must be created.
- Result<Position> truncate(
- const Position& to,
- const process::Timeout& timeout);
+ process::Future<Position> truncate(const Position& to);
private:
LogWriterProcess* process;
http://git-wip-us.apache.org/repos/asf/mesos/blob/bce0b265/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index da827a1..e15cb76 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -1688,17 +1688,17 @@ TEST_F(LogTest, WriteRead)
Log::Writer writer(&log, Seconds(10));
- Result<Log::Position> position =
- writer.append("hello world", Timeout::in(Seconds(10)));
+ Future<Log::Position> position = writer.append("hello world");
- ASSERT_SOME(position);
+ AWAIT_READY(position);
Log::Reader reader(&log);
- Result<list<Log::Entry> > entries =
- reader.read(position.get(), position.get(), Timeout::in(Seconds(10)));
+ Future<list<Log::Entry> > entries =
+ reader.read(position.get(), position.get());
+
+ AWAIT_READY(entries);
- ASSERT_SOME(entries);
ASSERT_EQ(1u, entries.get().size());
EXPECT_EQ(position.get(), entries.get().front().position);
EXPECT_EQ("hello world", entries.get().front().data);
@@ -1724,10 +1724,9 @@ TEST_F(LogTest, Position)
Log::Writer writer(&log, Seconds(10));
- Result<Log::Position> position =
- writer.append("hello world", Timeout::in(Seconds(10)));
+ Future<Log::Position> position = writer.append("hello world");
- ASSERT_SOME(position);
+ AWAIT_READY(position);
ASSERT_EQ(position.get(), log.position(position.get().identity()));
}
@@ -1797,17 +1796,17 @@ TEST_F(LogZooKeeperTest, WriteRead)
Log::Writer writer(&log2, Seconds(10));
- Result<Log::Position> position =
- writer.append("hello world", Timeout::in(Seconds(10)));
+ Future<Log::Position> position = writer.append("hello world");
- ASSERT_SOME(position);
+ AWAIT_READY(position);
Log::Reader reader(&log2);
- Result<list<Log::Entry> > entries =
- reader.read(position.get(), position.get(), Timeout::in(Seconds(10)));
+ Future<list<Log::Entry> > entries =
+ reader.read(position.get(), position.get());
+
+ AWAIT_READY(entries);
- ASSERT_SOME(entries);
ASSERT_EQ(1u, entries.get().size());
EXPECT_EQ(position.get(), entries.get().front().position);
EXPECT_EQ("hello world", entries.get().front().data);