You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2017/07/13 06:33:11 UTC
[2/3] kudu git commit: TabletReplica: Move Init() logic to Start()
TabletReplica: Move Init() logic to Start()
This is a cleanup / refactor that will make it easier to implement
tombstoned voting. In this patch we merge Init() and Start() since they
aren't really logically different right now.
An unrelated change in this patch is a minor API cleanup in
TSTabletManager, where we now pass TabletReplica directly to
TSTabletManager::OpenTablet() instead of registering it in a map first
and then looking it up again later.
Change-Id: Ib762db8cfaac628325feee445490713b5c555c5a
Reviewed-on: http://gerrit.cloudera.org:8080/7194
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/bbea22e1
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/bbea22e1
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/bbea22e1
Branch: refs/heads/master
Commit: bbea22e1950868de1c3cf3e9dab0eef03979b53a
Parents: 4f16320
Author: Mike Percy <mp...@apache.org>
Authored: Mon Jun 12 20:07:22 2017 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Thu Jul 13 06:32:32 2017 +0000
----------------------------------------------------------------------
src/kudu/master/sys_catalog.cc | 29 ++--
src/kudu/tablet/tablet.h | 1 +
src/kudu/tablet/tablet_replica-test.cc | 42 +++---
src/kudu/tablet/tablet_replica.cc | 145 ++++++++++---------
src/kudu/tablet/tablet_replica.h | 50 +++----
.../tserver/tablet_copy_source_session-test.cc | 18 +--
src/kudu/tserver/ts_tablet_manager.cc | 52 +++----
src/kudu/tserver/ts_tablet_manager.h | 4 +-
8 files changed, 166 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/master/sys_catalog.cc
----------------------------------------------------------------------
diff --git a/src/kudu/master/sys_catalog.cc b/src/kudu/master/sys_catalog.cc
index d9c37e2..5816484 100644
--- a/src/kudu/master/sys_catalog.cc
+++ b/src/kudu/master/sys_catalog.cc
@@ -303,8 +303,8 @@ Status SysCatalogTable::SetupTablet(const scoped_refptr<tablet::TabletMetadata>&
InitLocalRaftPeerPB();
- // TODO: handle crash mid-creation of tablet? do we ever end up with a
- // partially created tablet here?
+ // TODO(matteo): handle crash mid-creation of tablet? do we ever end up with
+ // a partially created tablet here?
tablet_replica_.reset(new TabletReplica(
metadata,
cmeta_manager_,
@@ -326,20 +326,17 @@ Status SysCatalogTable::SetupTablet(const scoped_refptr<tablet::TabletMetadata>&
tablet_replica_->log_anchor_registry(),
&consensus_info));
- // TODO: Do we have a setSplittable(false) or something from the outside is
- // handling split in the TS?
-
- RETURN_NOT_OK_PREPEND(tablet_replica_->Init(tablet,
- scoped_refptr<server::Clock>(master_->clock()),
- master_->messenger(),
- scoped_refptr<rpc::ResultTracker>(),
- log,
- tablet->GetMetricEntity(),
- master_->raft_pool(),
- master_->tablet_prepare_pool()),
- "Failed to Init() TabletReplica");
-
- RETURN_NOT_OK_PREPEND(tablet_replica_->Start(consensus_info),
+ // TODO(matteo): Do we have a setSplittable(false) or something from the
+ // outside is handling split in the TS?
+
+ RETURN_NOT_OK_PREPEND(tablet_replica_->Start(consensus_info,
+ tablet,
+ scoped_refptr<server::Clock>(master_->clock()),
+ master_->messenger(),
+ scoped_refptr<rpc::ResultTracker>(),
+ log,
+ master_->raft_pool(),
+ master_->tablet_prepare_pool()),
"Failed to Start() TabletReplica");
tablet_replica_->RegisterMaintenanceOps(master_->maintenance_manager());
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tablet/tablet.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index 67e854e..0c18bdf 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -331,6 +331,7 @@ class Tablet {
const TabletMetadata *metadata() const { return metadata_.get(); }
TabletMetadata *metadata() { return metadata_.get(); }
+ scoped_refptr<TabletMetadata> shared_metadata() const { return metadata_; }
void SetCompactionHooksForTests(const std::shared_ptr<CompactionFaultHooks> &hooks);
void SetFlushHooksForTests(const std::shared_ptr<FlushFaultHooks> &hooks);
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tablet/tablet_replica-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_replica-test.cc b/src/kudu/tablet/tablet_replica-test.cc
index 27bb052..55f71dc 100644
--- a/src/kudu/tablet/tablet_replica-test.cc
+++ b/src/kudu/tablet/tablet_replica-test.cc
@@ -104,7 +104,7 @@ class TabletReplicaTest : public KuduTabletTest {
// "Bootstrap" and start the TabletReplica.
tablet_replica_.reset(
- new TabletReplica(make_scoped_refptr(tablet()->metadata()),
+ new TabletReplica(tablet()->shared_metadata(),
cmeta_manager,
config_peer,
apply_pool_.get(),
@@ -125,28 +125,28 @@ class TabletReplicaTest : public KuduTabletTest {
scoped_refptr<ConsensusMetadata> cmeta;
ASSERT_OK(cmeta_manager->Create(tablet()->tablet_id(), config, consensus::kMinimumTerm,
&cmeta));
+ }
+ Status StartReplica(const ConsensusBootstrapInfo& info) {
scoped_refptr<Log> log;
- ASSERT_OK(Log::Open(LogOptions(), fs_manager(), tablet()->tablet_id(),
- *tablet()->schema(), tablet()->metadata()->schema_version(),
- metric_entity_.get(), &log));
-
+ RETURN_NOT_OK(Log::Open(LogOptions(), fs_manager(), tablet()->tablet_id(),
+ *tablet()->schema(), tablet()->metadata()->schema_version(),
+ metric_entity_.get(), &log));
tablet_replica_->SetBootstrapping();
- ASSERT_OK(tablet_replica_->Init(tablet(),
- clock(),
- messenger_,
- scoped_refptr<rpc::ResultTracker>(),
- log,
- metric_entity_,
- raft_pool_.get(),
- prepare_pool_.get()));
+ return tablet_replica_->Start(info,
+ tablet(),
+ clock(),
+ messenger_,
+ scoped_refptr<rpc::ResultTracker>(),
+ log,
+ raft_pool_.get(),
+ prepare_pool_.get());
}
- Status StartPeer(const ConsensusBootstrapInfo& info) {
+ Status StartReplicaAndWaitUntilLeader(const ConsensusBootstrapInfo& info) {
+ RETURN_NOT_OK(StartReplica(info));
const MonoDelta kTimeout = MonoDelta::FromSeconds(10);
- RETURN_NOT_OK(tablet_replica_->Start(info));
- RETURN_NOT_OK(tablet_replica_->consensus()->WaitUntilLeaderForTests(kTimeout));
- return Status::OK();
+ return tablet_replica_->consensus()->WaitUntilLeaderForTests(kTimeout);
}
void TabletReplicaStateChangedCallback(const string& tablet_id, const string& reason) {
@@ -299,7 +299,7 @@ class DelayedApplyTransaction : public WriteTransaction {
// Ensure that Log::GC() doesn't delete logs when the MRS has an anchor.
TEST_F(TabletReplicaTest, TestMRSAnchorPreventsLogGC) {
ConsensusBootstrapInfo info;
- ASSERT_OK(StartPeer(info));
+ ASSERT_OK(StartReplicaAndWaitUntilLeader(info));
Log* log = tablet_replica_->log_.get();
int32_t num_gced;
@@ -339,7 +339,7 @@ TEST_F(TabletReplicaTest, TestMRSAnchorPreventsLogGC) {
// Ensure that Log::GC() doesn't delete logs when the DMS has an anchor.
TEST_F(TabletReplicaTest, TestDMSAnchorPreventsLogGC) {
ConsensusBootstrapInfo info;
- ASSERT_OK(StartPeer(info));
+ ASSERT_OK(StartReplicaAndWaitUntilLeader(info));
Log* log = tablet_replica_->log_.get();
int32_t num_gced;
@@ -419,7 +419,7 @@ TEST_F(TabletReplicaTest, TestDMSAnchorPreventsLogGC) {
// Ensure that Log::GC() doesn't compact logs with OpIds of active transactions.
TEST_F(TabletReplicaTest, TestActiveTransactionPreventsLogGC) {
ConsensusBootstrapInfo info;
- ASSERT_OK(StartPeer(info));
+ ASSERT_OK(StartReplicaAndWaitUntilLeader(info));
Log* log = tablet_replica_->log_.get();
int32_t num_gced;
@@ -529,7 +529,7 @@ TEST_F(TabletReplicaTest, TestActiveTransactionPreventsLogGC) {
TEST_F(TabletReplicaTest, TestGCEmptyLog) {
ConsensusBootstrapInfo info;
- tablet_replica_->Start(info);
+ ASSERT_OK(StartReplica(info));
// We don't wait on consensus on purpose.
ASSERT_OK(tablet_replica_->RunLogGC());
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tablet/tablet_replica.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_replica.cc b/src/kudu/tablet/tablet_replica.cc
index 8c1be7e..07679d1 100644
--- a/src/kudu/tablet/tablet_replica.cc
+++ b/src/kudu/tablet/tablet_replica.cc
@@ -103,23 +103,23 @@ using std::unique_ptr;
using strings::Substitute;
TabletReplica::TabletReplica(
- const scoped_refptr<TabletMetadata>& meta,
- const scoped_refptr<consensus::ConsensusMetadataManager>& cmeta_manager,
+ scoped_refptr<TabletMetadata> meta,
+ scoped_refptr<consensus::ConsensusMetadataManager> cmeta_manager,
consensus::RaftPeerPB local_peer_pb,
ThreadPool* apply_pool,
Callback<void(const std::string& reason)> mark_dirty_clbk)
- : meta_(meta),
- cmeta_manager_(cmeta_manager),
- tablet_id_(meta->tablet_id()),
+ : meta_(DCHECK_NOTNULL(std::move(meta))),
+ cmeta_manager_(DCHECK_NOTNULL(std::move(cmeta_manager))),
+ tablet_id_(meta_->tablet_id()),
local_peer_pb_(std::move(local_peer_pb)),
- state_(NOT_STARTED),
- last_status_("Tablet initializing..."),
- apply_pool_(apply_pool),
log_anchor_registry_(new LogAnchorRegistry()),
- mark_dirty_clbk_(std::move(mark_dirty_clbk)) {}
+ apply_pool_(apply_pool),
+ mark_dirty_clbk_(std::move(mark_dirty_clbk)),
+ state_(NOT_STARTED),
+ last_status_("Tablet initializing...") {
+}
TabletReplica::~TabletReplica() {
- std::lock_guard<simple_spinlock> lock(lock_);
// We should either have called Shutdown(), or we should have never called
// Init().
CHECK(!tablet_)
@@ -127,76 +127,85 @@ TabletReplica::~TabletReplica() {
<< TabletStatePB_Name(state_);
}
-Status TabletReplica::Init(const shared_ptr<Tablet>& tablet,
- const scoped_refptr<server::Clock>& clock,
- const shared_ptr<Messenger>& messenger,
- const scoped_refptr<ResultTracker>& result_tracker,
- const scoped_refptr<Log>& log,
- const scoped_refptr<MetricEntity>& metric_entity,
- ThreadPool* raft_pool,
- ThreadPool* prepare_pool) {
-
+Status TabletReplica::Start(const ConsensusBootstrapInfo& bootstrap_info,
+ shared_ptr<Tablet> tablet,
+ scoped_refptr<server::Clock> clock,
+ shared_ptr<Messenger> messenger,
+ scoped_refptr<ResultTracker> result_tracker,
+ scoped_refptr<Log> log,
+ ThreadPool* raft_pool,
+ ThreadPool* prepare_pool) {
DCHECK(tablet) << "A TabletReplica must be provided with a Tablet";
DCHECK(log) << "A TabletReplica must be provided with a Log";
- prepare_pool_token_ = prepare_pool->NewTokenWithMetrics(
- ThreadPool::ExecutionMode::SERIAL,
- {
- METRIC_op_prepare_queue_length.Instantiate(metric_entity),
- METRIC_op_prepare_queue_time.Instantiate(metric_entity),
- METRIC_op_prepare_run_time.Instantiate(metric_entity)
- });
{
- std::lock_guard<simple_spinlock> lock(lock_);
- CHECK_EQ(BOOTSTRAPPING, state_);
- tablet_ = tablet;
- clock_ = clock;
- messenger_ = messenger;
- log_ = log;
- result_tracker_ = result_tracker;
+ std::lock_guard<simple_spinlock> state_change_guard(state_change_lock_);
- TRACE("Creating consensus instance");
- ConsensusOptions options;
- options.tablet_id = meta_->tablet_id();
- consensus_.reset(new RaftConsensus(options, local_peer_pb_, cmeta_manager_, raft_pool));
- RETURN_NOT_OK(consensus_->Init());
- }
-
- if (tablet_->metrics() != nullptr) {
- TRACE("Starting instrumentation");
- txn_tracker_.StartInstrumentation(tablet_->GetMetricEntity());
- }
- txn_tracker_.StartMemoryTracking(tablet_->mem_tracker());
+ {
+ std::lock_guard<simple_spinlock> l(lock_);
- TRACE("TabletReplica::Init() finished");
- VLOG(2) << "T " << tablet_id() << " P " << consensus_->peer_uuid() << ": Peer Initted";
- return Status::OK();
-}
+ CHECK_EQ(BOOTSTRAPPING, state_);
-Status TabletReplica::Start(const ConsensusBootstrapInfo& bootstrap_info) {
- std::lock_guard<simple_spinlock> l(state_change_lock_);
- TRACE("Starting consensus");
+ tablet_ = DCHECK_NOTNULL(std::move(tablet));
+ clock_ = DCHECK_NOTNULL(std::move(clock));
+ messenger_ = DCHECK_NOTNULL(std::move(messenger));
+ result_tracker_ = std::move(result_tracker); // Passed null in tablet_replica-test
+ log_ = DCHECK_NOTNULL(log); // Not moved because it's passed to RaftConsensus::Start() below.
+ }
- VLOG(2) << "T " << tablet_id() << " P " << consensus_->peer_uuid() << ": Peer starting";
+ // Unlock while we initialize RaftConsensus, which involves I/O.
+ TRACE("Creating consensus instance");
+ ConsensusOptions options;
+ options.tablet_id = meta_->tablet_id();
+ scoped_refptr<RaftConsensus> consensus(new RaftConsensus(std::move(options), local_peer_pb_,
+ cmeta_manager_, raft_pool));
+ RETURN_NOT_OK(consensus->Init());
- VLOG(2) << "RaftConfig before starting: " << SecureDebugString(consensus_->CommittedConfig());
+ scoped_refptr<MetricEntity> metric_entity;
+ gscoped_ptr<PeerProxyFactory> peer_proxy_factory;
+ scoped_refptr<TimeManager> time_manager;
+ {
+ std::lock_guard<simple_spinlock> l(lock_);
+ consensus_ = consensus;
+ metric_entity = tablet_->GetMetricEntity();
+ prepare_pool_token_ = prepare_pool->NewTokenWithMetrics(
+ ThreadPool::ExecutionMode::SERIAL,
+ {
+ METRIC_op_prepare_queue_length.Instantiate(metric_entity),
+ METRIC_op_prepare_queue_time.Instantiate(metric_entity),
+ METRIC_op_prepare_run_time.Instantiate(metric_entity)
+ });
+
+ if (tablet_->metrics()) {
+ TRACE("Starting instrumentation");
+ txn_tracker_.StartInstrumentation(tablet_->GetMetricEntity());
+ }
+ txn_tracker_.StartMemoryTracking(tablet_->mem_tracker());
- gscoped_ptr<PeerProxyFactory> peer_proxy_factory(new RpcPeerProxyFactory(messenger_));
- scoped_refptr<TimeManager> time_manager(new TimeManager(
- clock_, tablet_->mvcc_manager()->GetCleanTimestamp()));
+ TRACE("Starting consensus");
+ VLOG(2) << "T " << tablet_id() << " P " << consensus_->peer_uuid() << ": Peer starting";
+ VLOG(2) << "RaftConfig before starting: " << SecureDebugString(consensus_->CommittedConfig());
- RETURN_NOT_OK(consensus_->Start(
- bootstrap_info,
- std::move(peer_proxy_factory),
- log_,
- std::move(time_manager),
- this,
- tablet_->GetMetricEntity(),
- mark_dirty_clbk_));
+ peer_proxy_factory.reset(new RpcPeerProxyFactory(messenger_));
+ time_manager.reset(new TimeManager(clock_, tablet_->mvcc_manager()->GetCleanTimestamp()));
+ }
- {
- std::lock_guard<simple_spinlock> lock(lock_);
- CHECK_EQ(state_, BOOTSTRAPPING);
+ // We cannot hold 'lock_' while we call RaftConsensus::Start() because it
+ // may invoke TabletReplica::StartReplicaTransaction() during startup, causing
+ // a self-deadlock. We take a ref to members protected by 'lock_' before
+ // unlocking.
+ RETURN_NOT_OK(consensus->Start(
+ bootstrap_info,
+ std::move(peer_proxy_factory),
+ log,
+ std::move(time_manager),
+ this,
+ metric_entity,
+ mark_dirty_clbk_));
+
+ // Re-acquire 'lock_' to update our state variable.
+ std::lock_guard<simple_spinlock> l(lock_);
+ CHECK_EQ(BOOTSTRAPPING, state_); // We are still protected by 'state_change_lock_'.
state_ = RUNNING;
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tablet/tablet_replica.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_replica.h b/src/kudu/tablet/tablet_replica.h
index a969ba4..e813ba4 100644
--- a/src/kudu/tablet/tablet_replica.h
+++ b/src/kudu/tablet/tablet_replica.h
@@ -72,27 +72,23 @@ class TransactionDriver;
class TabletReplica : public RefCountedThreadSafe<TabletReplica>,
public consensus::ReplicaTransactionFactory {
public:
- TabletReplica(const scoped_refptr<TabletMetadata>& meta,
- const scoped_refptr<consensus::ConsensusMetadataManager>& cmeta_manager,
+ TabletReplica(scoped_refptr<TabletMetadata> meta,
+ scoped_refptr<consensus::ConsensusMetadataManager> cmeta_manager,
consensus::RaftPeerPB local_peer_pb,
ThreadPool* apply_pool,
Callback<void(const std::string& reason)> mark_dirty_clbk);
- // Initializes the TabletReplica, namely creating the Log and initializing
- // RaftConsensus.
- Status Init(const std::shared_ptr<tablet::Tablet>& tablet,
- const scoped_refptr<server::Clock>& clock,
- const std::shared_ptr<rpc::Messenger>& messenger,
- const scoped_refptr<rpc::ResultTracker>& result_tracker,
- const scoped_refptr<log::Log>& log,
- const scoped_refptr<MetricEntity>& metric_entity,
- ThreadPool* raft_pool,
- ThreadPool* prepare_pool);
-
// Starts the TabletReplica, making it available for Write()s. If this
// TabletReplica is part of a consensus configuration this will connect it to other replicas
// in the consensus configuration.
- Status Start(const consensus::ConsensusBootstrapInfo& info);
+ Status Start(const consensus::ConsensusBootstrapInfo& bootstrap_info,
+ std::shared_ptr<tablet::Tablet> tablet,
+ scoped_refptr<server::Clock> clock,
+ std::shared_ptr<rpc::Messenger> messenger,
+ scoped_refptr<rpc::ResultTracker> result_tracker,
+ scoped_refptr<log::Log> log,
+ ThreadPool* raft_pool,
+ ThreadPool* prepare_pool);
// Shutdown this tablet replica.
// If a shutdown is already in progress, blocks until that shutdown is complete.
@@ -287,8 +283,19 @@ class TabletReplica : public RefCountedThreadSafe<TabletReplica>,
const scoped_refptr<consensus::ConsensusMetadataManager> cmeta_manager_;
const std::string tablet_id_;
-
const consensus::RaftPeerPB local_peer_pb_;
+ scoped_refptr<log::LogAnchorRegistry> log_anchor_registry_; // Assigned in tablet_replica-test
+
+ // Pool that executes apply tasks for transactions. This is a multi-threaded
+ // pool, constructor-injected by either the Master (for system tables) or
+ // the Tablet server.
+ ThreadPool* const apply_pool_;
+
+ // Function to mark this TabletReplica's tablet as dirty in the TSTabletManager.
+ //
+ // Must be called whenever cluster membership or leadership changes, or when
+ // the tablet's schema changes.
+ const Callback<void(const std::string& reason)> mark_dirty_clbk_;
TabletStatePB state_;
Status error_;
@@ -319,21 +326,8 @@ class TabletReplica : public RefCountedThreadSafe<TabletReplica>,
// Token for serial task submission to the server-wide transaction prepare pool.
std::unique_ptr<ThreadPoolToken> prepare_pool_token_;
- // Pool that executes apply tasks for transactions. This is a multi-threaded
- // pool, constructor-injected by either the Master (for system tables) or
- // the Tablet server.
- ThreadPool* apply_pool_;
-
scoped_refptr<server::Clock> clock_;
- scoped_refptr<log::LogAnchorRegistry> log_anchor_registry_;
-
- // Function to mark this TabletReplica's tablet as dirty in the TSTabletManager.
- //
- // Must be called whenever cluster membership or leadership changes, or when
- // the tablet's schema changes.
- Callback<void(const std::string& reason)> mark_dirty_clbk_;
-
// List of maintenance operations for the tablet that need information that only the peer
// can provide.
std::vector<MaintenanceOp*> maintenance_ops_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tserver/tablet_copy_source_session-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_source_session-test.cc b/src/kudu/tserver/tablet_copy_source_session-test.cc
index 4ff1200..e8ff1c6 100644
--- a/src/kudu/tserver/tablet_copy_source_session-test.cc
+++ b/src/kudu/tserver/tablet_copy_source_session-test.cc
@@ -121,7 +121,6 @@ class TabletCopyTest : public KuduTabletTest {
Bind(&TabletCopyTest::TabletReplicaStateChangedCallback,
Unretained(this),
tablet()->tablet_id())));
-
// TODO(dralves) similar to code in tablet_replica-test, consider refactor.
RaftConfigPB config;
config.add_peers()->CopyFrom(config_peer);
@@ -137,16 +136,15 @@ class TabletCopyTest : public KuduTabletTest {
log_anchor_registry_.reset(new LogAnchorRegistry());
tablet_replica_->SetBootstrapping();
- ASSERT_OK(tablet_replica_->Init(tablet(),
- clock(),
- messenger,
- scoped_refptr<rpc::ResultTracker>(),
- log,
- metric_entity,
- raft_pool_.get(),
- prepare_pool_.get()));
consensus::ConsensusBootstrapInfo boot_info;
- ASSERT_OK(tablet_replica_->Start(boot_info));
+ ASSERT_OK(tablet_replica_->Start(boot_info,
+ tablet(),
+ clock(),
+ messenger,
+ scoped_refptr<rpc::ResultTracker>(),
+ log,
+ raft_pool_.get(),
+ prepare_pool_.get()));
ASSERT_OK(tablet_replica_->WaitUntilConsensusRunning(MonoDelta::FromSeconds(10)));
ASSERT_OK(tablet_replica_->consensus()->WaitUntilLeaderForTests(MonoDelta::FromSeconds(10)));
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tserver/ts_tablet_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/ts_tablet_manager.cc b/src/kudu/tserver/ts_tablet_manager.cc
index 979e23f..b564de3 100644
--- a/src/kudu/tserver/ts_tablet_manager.cc
+++ b/src/kudu/tserver/ts_tablet_manager.cc
@@ -200,7 +200,7 @@ Status TSTabletManager::Init() {
scoped_refptr<TabletReplica> replica = CreateAndRegisterTabletReplica(meta, NEW_REPLICA);
RETURN_NOT_OK(open_tablet_pool_->SubmitFunc(boost::bind(&TSTabletManager::OpenTablet,
- this, meta, deleter)));
+ this, replica, deleter)));
}
{
@@ -281,7 +281,7 @@ Status TSTabletManager::CreateNewTablet(const string& table_id,
// We can run this synchronously since there is nothing to bootstrap.
RETURN_NOT_OK(open_tablet_pool_->SubmitFunc(boost::bind(&TSTabletManager::OpenTablet,
- this, meta, deleter)));
+ this, new_replica, deleter)));
if (replica) {
*replica = new_replica;
@@ -554,21 +554,23 @@ void TSTabletManager::RunTabletCopy(
}
// startup it's still in a valid, fully-copied state.
- OpenTablet(meta, deleter);
+ OpenTablet(replica, deleter);
}
// Create and register a new TabletReplica, given tablet metadata.
scoped_refptr<TabletReplica> TSTabletManager::CreateAndRegisterTabletReplica(
- const scoped_refptr<TabletMetadata>& meta, RegisterTabletReplicaMode mode) {
+ scoped_refptr<TabletMetadata> meta,
+ RegisterTabletReplicaMode mode) {
+ const string& tablet_id = meta->tablet_id();
scoped_refptr<TabletReplica> replica(
- new TabletReplica(meta,
+ new TabletReplica(std::move(meta),
cmeta_manager_,
local_peer_pb_,
server_->tablet_apply_pool(),
Bind(&TSTabletManager::MarkTabletDirty,
Unretained(this),
- meta->tablet_id())));
- RegisterTablet(meta->tablet_id(), replica, mode);
+ tablet_id)));
+ RegisterTablet(tablet_id, replica, mode);
return replica;
}
@@ -724,16 +726,15 @@ Status TSTabletManager::OpenTabletMeta(const string& tablet_id,
return Status::OK();
}
-void TSTabletManager::OpenTablet(const scoped_refptr<TabletMetadata>& meta,
- const scoped_refptr<TransitionInProgressDeleter>& deleter) {
- string tablet_id = meta->tablet_id();
+// Note: 'deleter' is not used in the body of OpenTablet(), but is required
+// anyway because its destructor performs cleanup that should only happen when
+// OpenTablet() completes.
+void TSTabletManager::OpenTablet(const scoped_refptr<TabletReplica>& replica,
+ const scoped_refptr<TransitionInProgressDeleter>& /*deleter*/) {
+ const string& tablet_id = replica->tablet_id();
TRACE_EVENT1("tserver", "TSTabletManager::OpenTablet",
"tablet_id", tablet_id);
- scoped_refptr<TabletReplica> replica;
- CHECK(LookupTablet(tablet_id, &replica))
- << "Tablet not registered prior to OpenTabletAsync call: " << tablet_id;
-
shared_ptr<Tablet> tablet;
scoped_refptr<Log> log;
@@ -747,10 +748,11 @@ void TSTabletManager::OpenTablet(const scoped_refptr<TabletMetadata>& meta,
// potentially millions of transaction traces being attached to the
// TabletCopy trace.
ADOPT_TRACE(nullptr);
- // TODO: handle crash mid-creation of tablet? do we ever end up with a
- // partially created tablet here?
+
+ // TODO(mpercy): Handle crash mid-creation of tablet? Do we ever end up
+ // with a partially created tablet here?
replica->SetBootstrapping();
- s = BootstrapTablet(meta,
+ s = BootstrapTablet(replica->tablet_metadata(),
cmeta_manager_,
scoped_refptr<server::Clock>(server_->clock()),
server_->mem_tracker(),
@@ -771,25 +773,15 @@ void TSTabletManager::OpenTablet(const scoped_refptr<TabletMetadata>& meta,
MonoTime start(MonoTime::Now());
LOG_TIMING_PREFIX(INFO, LogPrefix(tablet_id), "starting tablet") {
- TRACE("Initializing tablet replica");
- s = replica->Init(tablet,
+ TRACE("Starting tablet replica");
+ s = replica->Start(bootstrap_info,
+ tablet,
scoped_refptr<server::Clock>(server_->clock()),
server_->messenger(),
server_->result_tracker(),
log,
- tablet->GetMetricEntity(),
server_->raft_pool(),
server_->tablet_prepare_pool());
-
- if (!s.ok()) {
- LOG(ERROR) << LogPrefix(tablet_id) << "Tablet failed to init: "
- << s.ToString();
- replica->SetFailed(s);
- return;
- }
-
- TRACE("Starting tablet replica");
- s = replica->Start(bootstrap_info);
if (!s.ok()) {
LOG(ERROR) << LogPrefix(tablet_id) << "Tablet failed to start: "
<< s.ToString();
http://git-wip-us.apache.org/repos/asf/kudu/blob/bbea22e1/src/kudu/tserver/ts_tablet_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/ts_tablet_manager.h b/src/kudu/tserver/ts_tablet_manager.h
index 53dbe3c..3b2ce45 100644
--- a/src/kudu/tserver/ts_tablet_manager.h
+++ b/src/kudu/tserver/ts_tablet_manager.h
@@ -229,7 +229,7 @@ class TSTabletManager : public tserver::TabletReplicaLookupIf {
// method. A TransitionInProgressDeleter must be passed as 'deleter' into
// this method in order to remove that transition-in-progress entry when
// opening the tablet is complete (in either a success or a failure case).
- void OpenTablet(const scoped_refptr<tablet::TabletMetadata>& meta,
+ void OpenTablet(const scoped_refptr<tablet::TabletReplica>& replica,
const scoped_refptr<TransitionInProgressDeleter>& deleter);
// Open a tablet whose metadata has already been loaded.
@@ -252,7 +252,7 @@ class TSTabletManager : public tserver::TabletReplicaLookupIf {
// the TablerPeer object. See RegisterTablet() for details about the
// semantics of 'mode' and the locking requirements.
scoped_refptr<tablet::TabletReplica> CreateAndRegisterTabletReplica(
- const scoped_refptr<tablet::TabletMetadata>& meta,
+ scoped_refptr<tablet::TabletMetadata> meta,
RegisterTabletReplicaMode mode);
// Helper to generate the report for a single tablet.