You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2014/02/13 23:53:45 UTC
[2/3] git commit: Used interval set to store holes and unlearned
positions in replica.
Used interval set to store holes and unlearned positions in replica.
Review: https://reviews.apache.org/r/18062
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/6eb9c50f
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/6eb9c50f
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/6eb9c50f
Branch: refs/heads/master
Commit: 6eb9c50fde4ebfb0a3047535bfcbe5bc9c349dc6
Parents: db6974b
Author: Jie Yu <yu...@gmail.com>
Authored: Thu Feb 13 14:43:03 2014 -0800
Committer: Vinod Kone <vi...@twitter.com>
Committed: Thu Feb 13 14:43:03 2014 -0800
----------------------------------------------------------------------
src/log/catchup.cpp | 12 +++---
src/log/catchup.hpp | 4 +-
src/log/coordinator.cpp | 10 ++---
src/log/recover.cpp | 6 ++-
src/log/replica.cpp | 100 +++++++++++++++++++------------------------
src/log/replica.hpp | 4 +-
src/log/storage.hpp | 11 +++--
src/tests/log_tests.cpp | 5 ++-
8 files changed, 75 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/catchup.cpp
----------------------------------------------------------------------
diff --git a/src/log/catchup.cpp b/src/log/catchup.cpp
index 4ee32f2..59facbf 100644
--- a/src/log/catchup.cpp
+++ b/src/log/catchup.cpp
@@ -34,7 +34,7 @@
using namespace process;
using std::list;
-using std::set;
+using std::vector;
namespace mesos {
namespace internal {
@@ -174,7 +174,7 @@ public:
const Shared<Replica>& _replica,
const Shared<Network>& _network,
uint64_t _proposal,
- const set<uint64_t>& _positions,
+ const vector<uint64_t>& _positions,
const Duration& _timeout)
: ProcessBase(ID::generate("log-bulk-catch-up")),
quorum(_quorum),
@@ -195,7 +195,7 @@ protected:
promise.future().onDiscarded(lambda::bind(
static_cast<void(*)(const UPID&, bool)>(terminate), self(), true));
- // Catch-up each position in the set sequentially.
+ // Catch-up sequentially.
it = positions.begin();
catchup();
@@ -264,11 +264,11 @@ private:
const size_t quorum;
const Shared<Replica> replica;
const Shared<Network> network;
- const set<uint64_t> positions;
+ const vector<uint64_t> positions;
const Duration timeout;
uint64_t proposal;
- set<uint64_t>::iterator it;
+ vector<uint64_t>::const_iterator it;
process::Promise<Nothing> promise;
Future<uint64_t> catching;
@@ -285,7 +285,7 @@ Future<Nothing> catchup(
const Shared<Replica>& replica,
const Shared<Network>& network,
const Option<uint64_t>& proposal,
- const set<uint64_t>& positions,
+ const vector<uint64_t>& positions,
const Duration& timeout)
{
BulkCatchUpProcess* process =
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/catchup.hpp
----------------------------------------------------------------------
diff --git a/src/log/catchup.hpp b/src/log/catchup.hpp
index c72ed81..45dc016 100644
--- a/src/log/catchup.hpp
+++ b/src/log/catchup.hpp
@@ -21,7 +21,7 @@
#include <stdint.h>
-#include <set>
+#include <vector>
#include <process/future.hpp>
#include <process/shared.hpp>
@@ -50,7 +50,7 @@ extern process::Future<Nothing> catchup(
const process::Shared<Replica>& replica,
const process::Shared<Network>& network,
const Option<uint64_t>& proposal,
- const std::set<uint64_t>& positions,
+ const std::vector<uint64_t>& positions,
const Duration& timeout = Seconds(10));
} // namespace log {
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/coordinator.cpp
----------------------------------------------------------------------
diff --git a/src/log/coordinator.cpp b/src/log/coordinator.cpp
index bc85e66..daa1ca8 100644
--- a/src/log/coordinator.cpp
+++ b/src/log/coordinator.cpp
@@ -35,8 +35,8 @@
using namespace process;
-using std::set;
using std::string;
+using std::vector;
namespace mesos {
namespace internal {
@@ -81,8 +81,8 @@ private:
Future<Nothing> updateProposal(uint64_t promised);
Future<PromiseResponse> runPromisePhase();
Future<Option<uint64_t> > checkPromisePhase(const PromiseResponse& response);
- Future<set<uint64_t> > getMissingPositions();
- Future<Nothing> catchupMissingPositions(const set<uint64_t>& positions);
+ Future<vector<uint64_t> > getMissingPositions();
+ Future<Nothing> catchupMissingPositions(const vector<uint64_t>& positions);
Future<Option<uint64_t> > updateIndexAfterElected();
void electingFinished(const Option<uint64_t>& position);
void electingFailed();
@@ -217,14 +217,14 @@ Future<Option<uint64_t> > CoordinatorProcess::checkPromisePhase(
}
-Future<set<uint64_t> > CoordinatorProcess::getMissingPositions()
+Future<vector<uint64_t> > CoordinatorProcess::getMissingPositions()
{
return replica->missing(0, index);
}
Future<Nothing> CoordinatorProcess::catchupMissingPositions(
- const set<uint64_t>& positions)
+ const vector<uint64_t>& positions)
{
LOG(INFO) << "Coordinator attemping to fill missing position";
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/recover.cpp
----------------------------------------------------------------------
diff --git a/src/log/recover.cpp b/src/log/recover.cpp
index bb32e51..d06e5ad 100644
--- a/src/log/recover.cpp
+++ b/src/log/recover.cpp
@@ -19,6 +19,7 @@
#include <stdlib.h>
#include <set>
+#include <vector>
#include <process/defer.hpp>
#include <process/delay.hpp>
@@ -41,6 +42,7 @@
using namespace process;
using std::set;
+using std::vector;
namespace mesos {
namespace internal {
@@ -336,9 +338,9 @@ private:
<< lowestBeginPosition.get() << " to "
<< highestEndPosition.get();
- set<uint64_t> positions;
+ vector<uint64_t> positions;
for (uint64_t p = begin; p <= end; ++p) {
- positions.insert(p);
+ positions.push_back(p);
}
// Share the ownership of the replica. From this point until the
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/replica.cpp
----------------------------------------------------------------------
diff --git a/src/log/replica.cpp b/src/log/replica.cpp
index ec6e38c..746d6c3 100644
--- a/src/log/replica.cpp
+++ b/src/log/replica.cpp
@@ -18,6 +18,9 @@
#include <algorithm>
+#include <boost/icl/interval.hpp>
+#include <boost/icl/interval_set.hpp>
+
#include <process/dispatch.hpp>
#include <process/id.hpp>
@@ -36,11 +39,13 @@
#include "log/replica.hpp"
#include "log/storage.hpp"
+using namespace boost::icl;
+
using namespace process;
using std::list;
-using std::set;
using std::string;
+using std::vector;
namespace mesos {
namespace internal {
@@ -83,7 +88,7 @@ public:
// Returns missing positions in the log (i.e., unlearned or holes)
// within the specified range [from, to].
- set<uint64_t> missing(uint64_t from, uint64_t to);
+ vector<uint64_t> missing(uint64_t from, uint64_t to);
// Returns the beginning position of the log.
uint64_t beginning();
@@ -141,10 +146,10 @@ private:
uint64_t end;
// Holes in the log.
- set<uint64_t> holes;
+ interval_set<uint64_t> holes;
// Unlearned positions in the log.
- set<uint64_t> unlearned;
+ interval_set<uint64_t> unlearned;
};
@@ -186,7 +191,7 @@ Result<Action> ReplicaProcess::read(uint64_t position)
return Error("Attempted to read truncated position");
} else if (end < position) {
return None(); // These semantics are assumed above!
- } else if (holes.count(position) > 0) {
+ } else if (contains(holes, position)) {
return None();
}
@@ -246,7 +251,7 @@ bool ReplicaProcess::missing(uint64_t position)
} else if (position > end) {
return true;
} else {
- if (unlearned.count(position) != 0 || holes.count(position) != 0) {
+ if (contains(unlearned, position) || contains(holes, position)) {
return true;
} else {
return false;
@@ -255,31 +260,30 @@ bool ReplicaProcess::missing(uint64_t position)
}
-set<uint64_t> ReplicaProcess::missing(uint64_t from, uint64_t to)
+vector<uint64_t> ReplicaProcess::missing(uint64_t from, uint64_t to)
{
- // TODO(jieyu): Optimize the performence for the common case.
- set<uint64_t> positions;
+ if (from > to) {
+ return vector<uint64_t>();
+ }
+
+ interval_set<uint64_t> positions;
// Add unlearned positions.
- foreach (uint64_t p, unlearned) {
- if (p >= from && p <= to) {
- positions.insert(p);
- }
- }
+ positions += unlearned;
// Add holes.
- foreach (uint64_t p, holes) {
- if (p >= from && p <= to) {
- positions.insert(p);
- }
- }
+ positions += holes;
// Add all the unknown positions beyond our end.
- for (; to > end; to--) {
- positions.insert(to);
+ if (to > end) {
+ positions += interval<uint64_t>::closed(end + 1, to);
}
- return positions;
+ // Do not consider positions outside [from, to].
+ positions &= interval<uint64_t>::closed(from, to);
+
+ // Generate the resultant vector.
+ return vector<uint64_t>(elements_begin(positions), elements_end(positions));
}
@@ -667,39 +671,31 @@ bool ReplicaProcess::persist(const Action& action)
LOG(INFO) << "Persisted action at " << action.position();
// No longer a hole here (if there even was one).
- holes.erase(action.position());
+ holes -= action.position();
// Update unlearned positions and deal with truncation actions.
if (action.has_learned() && action.learned()) {
- unlearned.erase(action.position());
+ unlearned -= action.position();
if (action.has_type() && action.type() == Action::TRUNCATE) {
// No longer consider truncated positions as holes (so that a
// coordinator doesn't try and fill them).
- foreach (uint64_t position, utils::copy(holes)) {
- if (position < action.truncate().to()) {
- holes.erase(position);
- }
- }
+ holes -= interval<uint64_t>::open(0, action.truncate().to());
// No longer consider truncated positions as unlearned (so that
// a coordinator doesn't try and fill them).
- foreach (uint64_t position, utils::copy(unlearned)) {
- if (position < action.truncate().to()) {
- unlearned.erase(position);
- }
- }
+ unlearned -= interval<uint64_t>::open(0, action.truncate().to());
// And update the beginning position.
begin = std::max(begin, action.truncate().to());
}
} else {
// We just introduced an unlearned position.
- unlearned.insert(action.position());
+ unlearned += action.position();
}
// Update holes if we just wrote many positions past the last end.
- for (uint64_t position = end + 1; position < action.position(); position++) {
- holes.insert(position);
+ if (action.position() > end) {
+ holes += interval<uint64_t>::open(end, action.position());
}
// And update the end position.
@@ -722,27 +718,21 @@ void ReplicaProcess::restore(const string& path)
unlearned = state.get().unlearned;
// Only use the learned positions to help determine the holes.
- const set<uint64_t>& learned = state.get().learned;
-
- // We need to assume that position 0 is a hole for a brand new log
- // (a coordinator will simply fill it with a no-op when it first
- // gets elected), unless the position was found during recovery or
- // it has been truncated.
- if (learned.count(0) == 0 && unlearned.count(0) == 0 && begin == 0) {
- holes.insert(0);
- }
+ const interval_set<uint64_t>& learned = state.get().learned;
- // Now determine the rest of the holes.
- for (uint64_t position = begin; position < end; position++) {
- if (learned.count(position) == 0 && unlearned.count(position) == 0) {
- holes.insert(position);
- }
- }
+ // Holes are those positions in [begin, end] that are not in both
+ // learned and unlearned sets. In the case of a brand new log (begin
+ // and end are 0, and learned and unlearned are empty), we assume
+ // position 0 is a hole, and a coordinator will simply fill it with
+ // a no-op when it first gets elected.
+ holes += interval<uint64_t>::closed(begin, end);
+ holes -= learned;
+ holes -= unlearned;
LOG(INFO) << "Replica recovered with log positions "
<< begin << " -> " << end
- << " and holes " << stringify(holes)
- << " and unlearned " << stringify(unlearned);
+ << " with " << holes.size() << " holes"
+ << " and " << unlearned.size() << " unlearned";
}
@@ -773,7 +763,7 @@ Future<bool> Replica::missing(uint64_t position) const
}
-Future<set<uint64_t> > Replica::missing(uint64_t from, uint64_t to) const
+Future<vector<uint64_t> > Replica::missing(uint64_t from, uint64_t to) const
{
return dispatch(process, &ReplicaProcess::missing, from, to);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/replica.hpp
----------------------------------------------------------------------
diff --git a/src/log/replica.hpp b/src/log/replica.hpp
index 467d0d9..08ddcb1 100644
--- a/src/log/replica.hpp
+++ b/src/log/replica.hpp
@@ -22,8 +22,8 @@
#include <stdint.h>
#include <list>
-#include <set>
#include <string>
+#include <vector>
#include <process/future.hpp>
#include <process/pid.hpp>
@@ -73,7 +73,7 @@ public:
// Returns missing positions in the log (i.e., unlearned or holes)
// within the specified range [from, to].
- process::Future<std::set<uint64_t> > missing(
+ process::Future<std::vector<uint64_t> > missing(
uint64_t from,
uint64_t to) const;
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/log/storage.hpp
----------------------------------------------------------------------
diff --git a/src/log/storage.hpp b/src/log/storage.hpp
index 663146f..c0eba1b 100644
--- a/src/log/storage.hpp
+++ b/src/log/storage.hpp
@@ -21,9 +21,10 @@
#include <stdint.h>
-#include <set>
#include <string>
+#include <boost/icl/interval_set.hpp>
+
#include <stout/nothing.hpp>
#include <stout/try.hpp>
@@ -42,8 +43,12 @@ public:
Metadata metadata; // The metadata for the replica.
uint64_t begin; // Beginning position of the log.
uint64_t end; // Ending position of the log.
- std::set<uint64_t> learned; // Positions present and learned
- std::set<uint64_t> unlearned; // Positions present but unlearned.
+
+ // Note that here we use boost interval set to store learned and
+ // unlearned positions in a more compact way. Adjacent positions
+ // will be merged and represented using an interval.
+ boost::icl::interval_set<uint64_t> learned;
+ boost::icl::interval_set<uint64_t> unlearned;
};
virtual ~Storage() {}
http://git-wip-us.apache.org/repos/asf/mesos/blob/6eb9c50f/src/tests/log_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/log_tests.cpp b/src/tests/log_tests.cpp
index b1fb200..7e6dbde 100644
--- a/src/tests/log_tests.cpp
+++ b/src/tests/log_tests.cpp
@@ -64,6 +64,7 @@ using namespace process;
using std::list;
using std::set;
using std::string;
+using std::vector;
using testing::_;
using testing::Eq;
@@ -1365,13 +1366,13 @@ TEST_F(RecoverTest, CatchupRetry)
EXPECT_EQ(0u, electing.get().get());
}
- set<uint64_t> positions;
+ vector<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());
- positions.insert(position);
+ positions.push_back(position);
}
Shared<Replica> replica3(new Replica(path3));