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 2016/11/29 20:04:26 UTC
[1/2] kudu git commit: tablet copy: Rename tablet copy session source
files
Repository: kudu
Updated Branches:
refs/heads/master 2f1a2a06d -> a0f425f76
tablet copy: Rename tablet copy session source files
This is just a quick cleanup to the class rename in the previous commit.
Change-Id: I23da316bc3c5d5de412d2d86b1bc7ee77ef60be5
Reviewed-on: http://gerrit.cloudera.org:8080/5042
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/f2dac5d3
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/f2dac5d3
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/f2dac5d3
Branch: refs/heads/master
Commit: f2dac5d3cebd0ab462ff16ff4c42fe680b0feeef
Parents: 2f1a2a0
Author: Mike Percy <mp...@apache.org>
Authored: Mon Nov 7 13:49:14 2016 +0000
Committer: Mike Percy <mp...@apache.org>
Committed: Tue Nov 29 20:03:47 2016 +0000
----------------------------------------------------------------------
src/kudu/tserver/CMakeLists.txt | 4 +-
src/kudu/tserver/tablet_copy_service.cc | 2 +-
src/kudu/tserver/tablet_copy_session-test.cc | 334 ----------------
src/kudu/tserver/tablet_copy_session.cc | 386 -------------------
src/kudu/tserver/tablet_copy_session.h | 197 ----------
.../tserver/tablet_copy_source_session-test.cc | 334 ++++++++++++++++
src/kudu/tserver/tablet_copy_source_session.cc | 386 +++++++++++++++++++
src/kudu/tserver/tablet_copy_source_session.h | 197 ++++++++++
8 files changed, 920 insertions(+), 920 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/CMakeLists.txt b/src/kudu/tserver/CMakeLists.txt
index 0245428..7cb4655 100644
--- a/src/kudu/tserver/CMakeLists.txt
+++ b/src/kudu/tserver/CMakeLists.txt
@@ -109,7 +109,7 @@ set(TSERVER_SRCS
scanners.cc
tablet_copy_client.cc
tablet_copy_service.cc
- tablet_copy_session.cc
+ tablet_copy_source_session.cc
tablet_server.cc
tablet_server_options.cc
tablet_service.cc
@@ -166,7 +166,7 @@ set(KUDU_TEST_LINK_LIBS
tserver_test_util
${KUDU_MIN_TEST_LIBS})
ADD_KUDU_TEST(tablet_copy_client-test)
-ADD_KUDU_TEST(tablet_copy_session-test)
+ADD_KUDU_TEST(tablet_copy_source_session-test)
ADD_KUDU_TEST(tablet_copy_service-test)
ADD_KUDU_TEST(tablet_server-test)
ADD_KUDU_TEST(tablet_server-stress-test RUN_SERIAL true)
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_service.cc b/src/kudu/tserver/tablet_copy_service.cc
index 502fd00..a8b54f6 100644
--- a/src/kudu/tserver/tablet_copy_service.cc
+++ b/src/kudu/tserver/tablet_copy_service.cc
@@ -28,7 +28,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/map-util.h"
#include "kudu/rpc/rpc_context.h"
-#include "kudu/tserver/tablet_copy_session.h"
+#include "kudu/tserver/tablet_copy_source_session.h"
#include "kudu/tserver/tablet_peer_lookup.h"
#include "kudu/tablet/tablet_peer.h"
#include "kudu/util/crc.h"
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_session-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_session-test.cc b/src/kudu/tserver/tablet_copy_session-test.cc
deleted file mode 100644
index 8b4ec0a..0000000
--- a/src/kudu/tserver/tablet_copy_session-test.cc
+++ /dev/null
@@ -1,334 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-#include "kudu/tablet/tablet-test-util.h"
-
-#include <glog/logging.h>
-#include <gtest/gtest.h>
-#include <memory>
-
-#include "kudu/common/partial_row.h"
-#include "kudu/common/row_operations.h"
-#include "kudu/common/schema.h"
-#include "kudu/consensus/consensus_meta.h"
-#include "kudu/consensus/log.h"
-#include "kudu/consensus/metadata.pb.h"
-#include "kudu/consensus/opid_util.h"
-#include "kudu/fs/block_id.h"
-#include "kudu/gutil/gscoped_ptr.h"
-#include "kudu/gutil/ref_counted.h"
-#include "kudu/gutil/strings/fastmem.h"
-#include "kudu/gutil/strings/substitute.h"
-#include "kudu/rpc/messenger.h"
-#include "kudu/tserver/tablet_copy_session.h"
-#include "kudu/tablet/tablet_peer.h"
-#include "kudu/util/crc.h"
-#include "kudu/util/metrics.h"
-#include "kudu/util/test_util.h"
-#include "kudu/util/threadpool.h"
-
-METRIC_DECLARE_entity(tablet);
-
-using std::shared_ptr;
-using std::string;
-using std::unique_ptr;
-
-namespace kudu {
-namespace tserver {
-
-using consensus::ConsensusMetadata;
-using consensus::OpId;
-using consensus::RaftConfigPB;
-using consensus::RaftPeerPB;
-using fs::ReadableBlock;
-using log::Log;
-using log::LogOptions;
-using log::LogAnchorRegistry;
-using rpc::Messenger;
-using rpc::MessengerBuilder;
-using strings::Substitute;
-using tablet::ColumnDataPB;
-using tablet::DeltaDataPB;
-using tablet::KuduTabletTest;
-using tablet::RowSetDataPB;
-using tablet::TabletPeer;
-using tablet::TabletSuperBlockPB;
-using tablet::WriteTransactionState;
-
-class TabletCopyTest : public KuduTabletTest {
- public:
- TabletCopyTest()
- : KuduTabletTest(Schema({ ColumnSchema("key", STRING),
- ColumnSchema("val", INT32) }, 1)) {
- CHECK_OK(ThreadPoolBuilder("test-exec").Build(&apply_pool_));
- }
-
- virtual void SetUp() OVERRIDE {
- NO_FATALS(KuduTabletTest::SetUp());
- NO_FATALS(SetUpTabletPeer());
- NO_FATALS(PopulateTablet());
- NO_FATALS(InitSession());
- }
-
- virtual void TearDown() OVERRIDE {
- session_.reset();
- tablet_peer_->Shutdown();
- KuduTabletTest::TearDown();
- }
-
- protected:
- void SetUpTabletPeer() {
- scoped_refptr<Log> log;
- ASSERT_OK(Log::Open(LogOptions(), fs_manager(), tablet()->tablet_id(),
- *tablet()->schema(),
- 0, // schema_version
- NULL, &log));
-
- scoped_refptr<MetricEntity> metric_entity =
- METRIC_ENTITY_tablet.Instantiate(&metric_registry_, CURRENT_TEST_NAME());
-
- RaftPeerPB config_peer;
- config_peer.set_permanent_uuid(fs_manager()->uuid());
- config_peer.mutable_last_known_addr()->set_host("0.0.0.0");
- config_peer.mutable_last_known_addr()->set_port(0);
- config_peer.set_member_type(RaftPeerPB::VOTER);
-
- tablet_peer_.reset(
- new TabletPeer(tablet()->metadata(),
- config_peer,
- apply_pool_.get(),
- Bind(&TabletCopyTest::TabletPeerStateChangedCallback,
- Unretained(this),
- tablet()->tablet_id())));
-
- // TODO similar to code in tablet_peer-test, consider refactor.
- RaftConfigPB config;
- config.add_peers()->CopyFrom(config_peer);
- config.set_opid_index(consensus::kInvalidOpIdIndex);
-
- unique_ptr<ConsensusMetadata> cmeta;
- ASSERT_OK(ConsensusMetadata::Create(tablet()->metadata()->fs_manager(),
- tablet()->tablet_id(), fs_manager()->uuid(),
- config, consensus::kMinimumTerm, &cmeta));
-
- shared_ptr<Messenger> messenger;
- MessengerBuilder mbuilder(CURRENT_TEST_NAME());
- mbuilder.Build(&messenger);
-
- log_anchor_registry_.reset(new LogAnchorRegistry());
- tablet_peer_->SetBootstrapping();
- ASSERT_OK(tablet_peer_->Init(tablet(),
- clock(),
- messenger,
- scoped_refptr<rpc::ResultTracker>(),
- log,
- metric_entity));
- consensus::ConsensusBootstrapInfo boot_info;
- ASSERT_OK(tablet_peer_->Start(boot_info));
- ASSERT_OK(tablet_peer_->WaitUntilConsensusRunning(MonoDelta::FromSeconds(10)));
- ASSERT_OK(tablet_peer_->consensus()->WaitUntilLeaderForTests(MonoDelta::FromSeconds(10)));
- }
-
- void TabletPeerStateChangedCallback(const string& tablet_id, const string& reason) {
- LOG(INFO) << "Tablet peer state changed for tablet " << tablet_id << ". Reason: " << reason;
- }
-
- void PopulateTablet() {
- for (int32_t i = 0; i < 1000; i++) {
- WriteRequestPB req;
- req.set_tablet_id(tablet_peer_->tablet_id());
- ASSERT_OK(SchemaToPB(client_schema_, req.mutable_schema()));
- RowOperationsPB* data = req.mutable_row_operations();
- RowOperationsPBEncoder enc(data);
- KuduPartialRow row(&client_schema_);
-
- string key = Substitute("key$0", i);
- ASSERT_OK(row.SetStringNoCopy(0, key));
- ASSERT_OK(row.SetInt32(1, i));
- enc.Add(RowOperationsPB::INSERT, row);
-
- WriteResponsePB resp;
- CountDownLatch latch(1);
-
- unique_ptr<tablet::WriteTransactionState> state(
- new tablet::WriteTransactionState(tablet_peer_.get(),
- &req,
- nullptr, // No RequestIdPB
- &resp));
- state->set_completion_callback(gscoped_ptr<tablet::TransactionCompletionCallback>(
- new tablet::LatchTransactionCompletionCallback<WriteResponsePB>(&latch, &resp)));
- ASSERT_OK(tablet_peer_->SubmitWrite(std::move(state)));
- latch.Wait();
- ASSERT_FALSE(resp.has_error()) << "Request failed: " << resp.error().ShortDebugString();
- ASSERT_EQ(0, resp.per_row_errors_size()) << "Insert error: " << resp.ShortDebugString();
- }
- ASSERT_OK(tablet()->Flush());
- }
-
- void InitSession() {
- session_.reset(new TabletCopySourceSession(tablet_peer_.get(), "TestSession", "FakeUUID",
- fs_manager()));
- ASSERT_OK(session_->Init());
- }
-
- // Read the specified BlockId, via the TabletCopySourceSession, into a file.
- // 'path' will be populated with the name of the file used.
- // 'file' will be set to point to the SequentialFile containing the data.
- void FetchBlockToFile(const BlockId& block_id,
- string* path,
- unique_ptr<SequentialFile>* file) {
- string data;
- int64_t block_file_size = 0;
- TabletCopyErrorPB::Code error_code;
- CHECK_OK(session_->GetBlockPiece(block_id, 0, 0, &data, &block_file_size, &error_code));
- if (block_file_size > 0) {
- CHECK_GT(data.size(), 0);
- }
-
- // Write the file to a temporary location.
- WritableFileOptions opts;
- string path_template = GetTestPath(Substitute("test_block_$0.tmp.XXXXXX", block_id.ToString()));
- unique_ptr<WritableFile> writable_file;
- CHECK_OK(Env::Default()->NewTempWritableFile(opts, path_template, path, &writable_file));
- CHECK_OK(writable_file->Append(Slice(data.data(), data.size())));
- CHECK_OK(writable_file->Close());
-
- CHECK_OK(Env::Default()->NewSequentialFile(*path, file));
- }
-
- MetricRegistry metric_registry_;
- scoped_refptr<LogAnchorRegistry> log_anchor_registry_;
- gscoped_ptr<ThreadPool> apply_pool_;
- scoped_refptr<TabletPeer> tablet_peer_;
- scoped_refptr<TabletCopySourceSession> session_;
-};
-
-// Ensure that the serialized SuperBlock included in the TabletCopySourceSession is
-// equal to the serialized live superblock (on a quiesced tablet).
-TEST_F(TabletCopyTest, TestSuperBlocksEqual) {
- // Compare content of superblocks.
- faststring session_buf;
- faststring tablet_buf;
-
- {
- const TabletSuperBlockPB& session_superblock = session_->tablet_superblock();
- int size = session_superblock.ByteSize();
- session_buf.resize(size);
- uint8_t* session_dst = session_buf.data();
- session_dst = session_superblock.SerializeWithCachedSizesToArray(session_dst);
- }
-
- {
- TabletSuperBlockPB tablet_superblock;
- ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
- int size = tablet_superblock.ByteSize();
- tablet_buf.resize(size);
- uint8_t* tablet_dst = tablet_buf.data();
- tablet_dst = tablet_superblock.SerializeWithCachedSizesToArray(tablet_dst);
- }
-
- ASSERT_EQ(session_buf.size(), tablet_buf.size());
- int size = tablet_buf.size();
- ASSERT_EQ(0, strings::fastmemcmp_inlined(session_buf.data(), tablet_buf.data(), size));
-}
-
-// Test fetching all files from tablet server, ensure the checksums for each
-// chunk and the total file sizes match.
-TEST_F(TabletCopyTest, TestBlocksEqual) {
- TabletSuperBlockPB tablet_superblock;
- ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
- for (int i = 0; i < tablet_superblock.rowsets_size(); i++) {
- const RowSetDataPB& rowset = tablet_superblock.rowsets(i);
- for (int j = 0; j < rowset.columns_size(); j++) {
- const ColumnDataPB& column = rowset.columns(j);
- const BlockIdPB& block_id_pb = column.block();
- BlockId block_id = BlockId::FromPB(block_id_pb);
-
- string path;
- unique_ptr<SequentialFile> file;
- FetchBlockToFile(block_id, &path, &file);
- uint64_t session_block_size = 0;
- ASSERT_OK(Env::Default()->GetFileSize(path, &session_block_size));
- faststring buf;
- buf.resize(session_block_size);
- Slice data;
- ASSERT_OK(file->Read(session_block_size, &data, buf.data()));
- uint32_t session_crc = crc::Crc32c(data.data(), data.size());
- LOG(INFO) << "session block file has size of " << session_block_size
- << " and CRC32C of " << session_crc << ": " << path;
-
- gscoped_ptr<ReadableBlock> tablet_block;
- ASSERT_OK(fs_manager()->OpenBlock(block_id, &tablet_block));
- uint64_t tablet_block_size = 0;
- ASSERT_OK(tablet_block->Size(&tablet_block_size));
- buf.resize(tablet_block_size);
- ASSERT_OK(tablet_block->Read(0, tablet_block_size, &data, buf.data()));
- uint32_t tablet_crc = crc::Crc32c(data.data(), data.size());
- LOG(INFO) << "tablet block file has size of " << tablet_block_size
- << " and CRC32C of " << tablet_crc
- << ": " << block_id;
-
- // Compare the blocks.
- ASSERT_EQ(tablet_block_size, session_block_size);
- ASSERT_EQ(tablet_crc, session_crc);
- }
- }
-}
-
-// Ensure that blocks are still readable through the open session even
-// after they've been deleted.
-TEST_F(TabletCopyTest, TestBlocksAreFetchableAfterBeingDeleted) {
- TabletSuperBlockPB tablet_superblock;
- ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
-
- // Gather all the blocks.
- vector<BlockId> data_blocks;
- for (const RowSetDataPB& rowset : tablet_superblock.rowsets()) {
- for (const DeltaDataPB& redo : rowset.redo_deltas()) {
- data_blocks.push_back(BlockId::FromPB(redo.block()));
- }
- for (const DeltaDataPB& undo : rowset.undo_deltas()) {
- data_blocks.push_back(BlockId::FromPB(undo.block()));
- }
- for (const ColumnDataPB& column : rowset.columns()) {
- data_blocks.push_back(BlockId::FromPB(column.block()));
- }
- if (rowset.has_bloom_block()) {
- data_blocks.push_back(BlockId::FromPB(rowset.bloom_block()));
- }
- if (rowset.has_adhoc_index_block()) {
- data_blocks.push_back(BlockId::FromPB(rowset.adhoc_index_block()));
- }
- }
-
- // Delete them.
- for (const BlockId& block_id : data_blocks) {
- ASSERT_OK(fs_manager()->DeleteBlock(block_id));
- }
-
- // Read them back.
- for (const BlockId& block_id : data_blocks) {
- ASSERT_TRUE(session_->IsBlockOpenForTests(block_id));
- string data;
- TabletCopyErrorPB::Code error_code;
- int64_t piece_size;
- ASSERT_OK(session_->GetBlockPiece(block_id, 0, 0,
- &data, &piece_size, &error_code));
- }
-}
-
-} // namespace tserver
-} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_session.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_session.cc b/src/kudu/tserver/tablet_copy_session.cc
deleted file mode 100644
index 100fbaa..0000000
--- a/src/kudu/tserver/tablet_copy_session.cc
+++ /dev/null
@@ -1,386 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-#include "kudu/tserver/tablet_copy_session.h"
-
-#include <algorithm>
-#include <mutex>
-
-#include "kudu/consensus/log.h"
-#include "kudu/consensus/log_reader.h"
-#include "kudu/consensus/metadata.pb.h"
-#include "kudu/fs/block_manager.h"
-#include "kudu/gutil/map-util.h"
-#include "kudu/gutil/strings/substitute.h"
-#include "kudu/gutil/type_traits.h"
-#include "kudu/rpc/transfer.h"
-#include "kudu/tablet/tablet_peer.h"
-#include "kudu/util/flag_tags.h"
-#include "kudu/util/mutex.h"
-#include "kudu/util/stopwatch.h"
-#include "kudu/util/trace.h"
-
-DEFINE_int32(tablet_copy_transfer_chunk_size_bytes, 4 * 1024 * 1024,
- "Size of chunks to transfer when copying tablets between "
- "tablet servers.");
-TAG_FLAG(tablet_copy_transfer_chunk_size_bytes, hidden);
-
-namespace kudu {
-namespace tserver {
-
-using consensus::MinimumOpId;
-using consensus::OpId;
-using fs::ReadableBlock;
-using log::LogAnchorRegistry;
-using log::ReadableLogSegment;
-using std::shared_ptr;
-using strings::Substitute;
-using tablet::ColumnDataPB;
-using tablet::DeltaDataPB;
-using tablet::RowSetDataPB;
-using tablet::TabletMetadata;
-using tablet::TabletPeer;
-using tablet::TabletSuperBlockPB;
-
-TabletCopySourceSession::TabletCopySourceSession(
- const scoped_refptr<TabletPeer>& tablet_peer, std::string session_id,
- std::string requestor_uuid, FsManager* fs_manager)
- : tablet_peer_(tablet_peer),
- session_id_(std::move(session_id)),
- requestor_uuid_(std::move(requestor_uuid)),
- fs_manager_(fs_manager),
- blocks_deleter_(&blocks_),
- logs_deleter_(&logs_) {}
-
-TabletCopySourceSession::~TabletCopySourceSession() {
- // No lock taken in the destructor, should only be 1 thread with access now.
- CHECK_OK(UnregisterAnchorIfNeededUnlocked());
-}
-
-Status TabletCopySourceSession::Init() {
- MutexLock l(session_lock_);
- CHECK(!initted_);
-
- const string& tablet_id = tablet_peer_->tablet_id();
-
- // Prevent log GC while we grab log segments and Tablet metadata.
- string anchor_owner_token = Substitute("TabletCopy-$0", session_id_);
- tablet_peer_->log_anchor_registry()->Register(
- MinimumOpId().index(), anchor_owner_token, &log_anchor_);
-
- // Read the SuperBlock from disk.
- const scoped_refptr<TabletMetadata>& metadata = tablet_peer_->tablet_metadata();
- RETURN_NOT_OK_PREPEND(metadata->ReadSuperBlockFromDisk(&tablet_superblock_),
- Substitute("Unable to access superblock for tablet $0",
- tablet_id));
-
- // Anchor the data blocks by opening them and adding them to the cache.
- //
- // All subsequent requests should reuse the opened blocks.
- vector<BlockIdPB> data_blocks;
- TabletMetadata::CollectBlockIdPBs(tablet_superblock_, &data_blocks);
- for (const BlockIdPB& block_id : data_blocks) {
- VLOG(1) << "Opening block " << block_id.DebugString();
- RETURN_NOT_OK(OpenBlockUnlocked(BlockId::FromPB(block_id)));
- }
-
- // Get the latest opid in the log at this point in time so we can re-anchor.
- OpId last_logged_opid;
- tablet_peer_->log()->GetLatestEntryOpId(&last_logged_opid);
-
- // Get the current segments from the log, including the active segment.
- // The Log doesn't add the active segment to the log reader's list until
- // a header has been written to it (but it will not have a footer).
- shared_ptr<log::LogReader> reader = tablet_peer_->log()->reader();
- if (!reader) {
- tablet::TabletStatePB tablet_state = tablet_peer_->state();
- return Status::IllegalState(Substitute(
- "Unable to initialize tablet copy session for tablet $0. "
- "Log reader is not available. Tablet state: $1 ($2)",
- tablet_id, tablet::TabletStatePB_Name(tablet_state), tablet_state));
- }
- reader->GetSegmentsSnapshot(&log_segments_);
- for (const scoped_refptr<ReadableLogSegment>& segment : log_segments_) {
- RETURN_NOT_OK(OpenLogSegmentUnlocked(segment->header().sequence_number()));
- }
-
- // Look up the committed consensus state.
- // We do this after snapshotting the log to avoid a scenario where the latest
- // entry in the log has a term higher than the term stored in the consensus
- // metadata, which will results in a CHECK failure on RaftConsensus init.
- scoped_refptr<consensus::Consensus> consensus = tablet_peer_->shared_consensus();
- if (!consensus) {
- tablet::TabletStatePB tablet_state = tablet_peer_->state();
- return Status::IllegalState(Substitute(
- "Unable to initialize tablet copy session for tablet $0. "
- "Consensus is not available. Tablet state: $1 ($2)",
- tablet_id, tablet::TabletStatePB_Name(tablet_state), tablet_state));
- }
- initial_committed_cstate_ = consensus->ConsensusState(consensus::CONSENSUS_CONFIG_COMMITTED);
-
- // Re-anchor on the highest OpId that was in the log right before we
- // snapshotted the log segments. This helps ensure that we don't end up in a
- // tablet copy loop due to a follower falling too far behind the
- // leader's log when tablet copy is slow. The remote controls when
- // this anchor is released by ending the tablet copy session.
- RETURN_NOT_OK(tablet_peer_->log_anchor_registry()->UpdateRegistration(
- last_logged_opid.index(), anchor_owner_token, &log_anchor_));
-
- LOG(INFO) << Substitute(
- "T $0 P $1: Tablet Copy: opened $2 blocks and $3 log segments",
- tablet_id, consensus->peer_uuid(), data_blocks.size(), log_segments_.size());
- initted_ = true;
- return Status::OK();
-}
-
-const std::string& TabletCopySourceSession::tablet_id() const {
- DCHECK(initted_);
- return tablet_peer_->tablet_id();
-}
-
-const std::string& TabletCopySourceSession::requestor_uuid() const {
- DCHECK(initted_);
- return requestor_uuid_;
-}
-
-// Determine the length of the data chunk to return to the client.
-static int64_t DetermineReadLength(int64_t bytes_remaining, int64_t requested_len) {
- // Overhead in the RPC for things like headers, protobuf data, etc.
- static const int kOverhead = 4096;
-
- if (requested_len <= 0) {
- requested_len = FLAGS_tablet_copy_transfer_chunk_size_bytes;
- } else {
- requested_len = std::min<int64_t>(requested_len,
- FLAGS_tablet_copy_transfer_chunk_size_bytes);
- }
- requested_len = std::min<int64_t>(requested_len, FLAGS_rpc_max_message_size - kOverhead);
- CHECK_GT(requested_len, 0) << "rpc_max_message_size is too low to transfer data: "
- << FLAGS_rpc_max_message_size;
- return std::min(bytes_remaining, requested_len);
-}
-
-// Calculate the size of the data to return given a maximum client message
-// length, the file itself, and the offset into the file to be read from.
-static Status GetResponseDataSize(int64_t total_size,
- uint64_t offset, int64_t client_maxlen,
- TabletCopyErrorPB::Code* error_code, int64_t* data_size) {
- // If requested offset is off the end of the data, bail.
- if (offset >= total_size) {
- *error_code = TabletCopyErrorPB::INVALID_TABLET_COPY_REQUEST;
- return Status::InvalidArgument(
- Substitute("Requested offset ($0) is beyond the data size ($1)",
- offset, total_size));
- }
-
- int64_t bytes_remaining = total_size - offset;
-
- *data_size = DetermineReadLength(bytes_remaining, client_maxlen);
- DCHECK_GT(*data_size, 0);
- if (client_maxlen > 0) {
- DCHECK_LE(*data_size, client_maxlen);
- }
-
- return Status::OK();
-}
-
-// Read a chunk of a file into a buffer.
-// data_name provides a string for the block/log to be used in error messages.
-template <class Info>
-static Status ReadFileChunkToBuf(const Info* info,
- uint64_t offset, int64_t client_maxlen,
- const string& data_name,
- string* data, int64_t* file_size,
- TabletCopyErrorPB::Code* error_code) {
- int64_t response_data_size = 0;
- RETURN_NOT_OK_PREPEND(GetResponseDataSize(info->size, offset, client_maxlen, error_code,
- &response_data_size),
- Substitute("Error reading $0", data_name));
-
- Stopwatch chunk_timer(Stopwatch::THIS_THREAD);
- chunk_timer.start();
-
- // Writing into a std::string buffer is basically guaranteed to work on C++11,
- // however any modern compiler should be compatible with it.
- // Violates the API contract, but avoids excessive copies.
- data->resize(response_data_size);
- uint8_t* buf = reinterpret_cast<uint8_t*>(const_cast<char*>(data->data()));
- Slice slice;
- Status s = info->ReadFully(offset, response_data_size, &slice, buf);
- if (PREDICT_FALSE(!s.ok())) {
- s = s.CloneAndPrepend(
- Substitute("Unable to read existing file for $0", data_name));
- LOG(WARNING) << s.ToString();
- *error_code = TabletCopyErrorPB::IO_ERROR;
- return s;
- }
- // Figure out if Slice points to buf or if Slice points to the mmap.
- // If it points to the mmap then copy into buf.
- if (slice.data() != buf) {
- memcpy(buf, slice.data(), slice.size());
- }
- chunk_timer.stop();
- TRACE("Tablet Copy: $0: $1 total bytes read. Total time elapsed: $2",
- data_name, response_data_size, chunk_timer.elapsed().ToString());
-
- *file_size = info->size;
- return Status::OK();
-}
-
-Status TabletCopySourceSession::GetBlockPiece(const BlockId& block_id,
- uint64_t offset, int64_t client_maxlen,
- string* data, int64_t* block_file_size,
- TabletCopyErrorPB::Code* error_code) {
- ImmutableReadableBlockInfo* block_info;
- RETURN_NOT_OK(FindBlock(block_id, &block_info, error_code));
-
- RETURN_NOT_OK(ReadFileChunkToBuf(block_info, offset, client_maxlen,
- Substitute("block $0", block_id.ToString()),
- data, block_file_size, error_code));
-
- // Note: We do not eagerly close the block, as doing so may delete the
- // underlying data if this was its last reader and it had been previously
- // marked for deletion. This would be a problem for parallel readers in
- // the same session; they would not be able to find the block.
-
- return Status::OK();
-}
-
-Status TabletCopySourceSession::GetLogSegmentPiece(uint64_t segment_seqno,
- uint64_t offset, int64_t client_maxlen,
- std::string* data, int64_t* block_file_size,
- TabletCopyErrorPB::Code* error_code) {
- ImmutableRandomAccessFileInfo* file_info;
- RETURN_NOT_OK(FindLogSegment(segment_seqno, &file_info, error_code));
- RETURN_NOT_OK(ReadFileChunkToBuf(file_info, offset, client_maxlen,
- Substitute("log segment $0", segment_seqno),
- data, block_file_size, error_code));
-
- // Note: We do not eagerly close log segment files, since we share ownership
- // of the LogSegment objects with the Log itself.
-
- return Status::OK();
-}
-
-bool TabletCopySourceSession::IsBlockOpenForTests(const BlockId& block_id) const {
- MutexLock l(session_lock_);
- return ContainsKey(blocks_, block_id);
-}
-
-// Add a file to the cache and populate the given ImmutableRandomAcccessFileInfo
-// object with the file ref and size.
-template <class Collection, class Key, class Readable>
-static Status AddImmutableFileToMap(Collection* const cache,
- const Key& key,
- const Readable& readable,
- uint64_t size) {
- // Sanity check for 0-length files.
- if (size == 0) {
- return Status::Corruption("Found 0-length object");
- }
-
- // Looks good, add it to the cache.
- typedef typename Collection::mapped_type InfoPtr;
- typedef typename base::remove_pointer<InfoPtr>::type Info;
- InsertOrDie(cache, key, new Info(readable, size));
-
- return Status::OK();
-}
-
-Status TabletCopySourceSession::OpenBlockUnlocked(const BlockId& block_id) {
- session_lock_.AssertAcquired();
-
- gscoped_ptr<ReadableBlock> block;
- Status s = fs_manager_->OpenBlock(block_id, &block);
- if (PREDICT_FALSE(!s.ok())) {
- LOG(WARNING) << "Unable to open requested (existing) block file: "
- << block_id.ToString() << ": " << s.ToString();
- return s.CloneAndPrepend(Substitute("Unable to open block file for block $0",
- block_id.ToString()));
- }
-
- uint64_t size;
- s = block->Size(&size);
- if (PREDICT_FALSE(!s.ok())) {
- return s.CloneAndPrepend("Unable to get size of block");
- }
-
- s = AddImmutableFileToMap(&blocks_, block_id, block.get(), size);
- if (!s.ok()) {
- s = s.CloneAndPrepend(Substitute("Error accessing data for block $0", block_id.ToString()));
- LOG(DFATAL) << "Data block disappeared: " << s.ToString();
- } else {
- ignore_result(block.release());
- }
- return s;
-}
-
-Status TabletCopySourceSession::FindBlock(const BlockId& block_id,
- ImmutableReadableBlockInfo** block_info,
- TabletCopyErrorPB::Code* error_code) {
- Status s;
- MutexLock l(session_lock_);
- if (!FindCopy(blocks_, block_id, block_info)) {
- *error_code = TabletCopyErrorPB::BLOCK_NOT_FOUND;
- s = Status::NotFound("Block not found", block_id.ToString());
- }
- return s;
-}
-
-Status TabletCopySourceSession::OpenLogSegmentUnlocked(uint64_t segment_seqno) {
- session_lock_.AssertAcquired();
-
- scoped_refptr<log::ReadableLogSegment> log_segment;
- int position = -1;
- if (!log_segments_.empty()) {
- position = segment_seqno - log_segments_[0]->header().sequence_number();
- }
- if (position < 0 || position >= log_segments_.size()) {
- return Status::NotFound(Substitute("Segment with sequence number $0 not found",
- segment_seqno));
- }
- log_segment = log_segments_[position];
- CHECK_EQ(log_segment->header().sequence_number(), segment_seqno);
-
- uint64_t size = log_segment->readable_up_to();
- Status s = AddImmutableFileToMap(&logs_, segment_seqno, log_segment->readable_file(), size);
- if (!s.ok()) {
- s = s.CloneAndPrepend(
- Substitute("Error accessing data for log segment with seqno $0",
- segment_seqno));
- LOG(INFO) << s.ToString();
- }
- return s;
-}
-
-Status TabletCopySourceSession::FindLogSegment(uint64_t segment_seqno,
- ImmutableRandomAccessFileInfo** file_info,
- TabletCopyErrorPB::Code* error_code) {
- MutexLock l(session_lock_);
- if (!FindCopy(logs_, segment_seqno, file_info)) {
- *error_code = TabletCopyErrorPB::WAL_SEGMENT_NOT_FOUND;
- return Status::NotFound(Substitute("Segment with sequence number $0 not found",
- segment_seqno));
- }
- return Status::OK();
-}
-
-Status TabletCopySourceSession::UnregisterAnchorIfNeededUnlocked() {
- return tablet_peer_->log_anchor_registry()->UnregisterIfAnchored(&log_anchor_);
-}
-
-} // namespace tserver
-} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_session.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_session.h b/src/kudu/tserver/tablet_copy_session.h
deleted file mode 100644
index aaf0043..0000000
--- a/src/kudu/tserver/tablet_copy_session.h
+++ /dev/null
@@ -1,197 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#pragma once
-
-#include <memory>
-#include <string>
-#include <unordered_map>
-#include <vector>
-
-#include "kudu/consensus/log_anchor_registry.h"
-#include "kudu/consensus/log_util.h"
-#include "kudu/consensus/metadata.pb.h"
-#include "kudu/consensus/opid_util.h"
-#include "kudu/fs/block_id.h"
-#include "kudu/fs/block_manager.h"
-#include "kudu/gutil/macros.h"
-#include "kudu/gutil/ref_counted.h"
-#include "kudu/gutil/stl_util.h"
-#include "kudu/tserver/tablet_copy.pb.h"
-#include "kudu/util/env_util.h"
-#include "kudu/util/locks.h"
-#include "kudu/util/status.h"
-
-namespace kudu {
-
-class FsManager;
-
-namespace tablet {
-class TabletPeer;
-} // namespace tablet
-
-namespace tserver {
-
-class TabletPeerLookupIf;
-
-// Caches file size and holds a shared_ptr reference to a RandomAccessFile.
-// Assumes that the file underlying the RandomAccessFile is immutable.
-struct ImmutableRandomAccessFileInfo {
- std::shared_ptr<RandomAccessFile> readable;
- int64_t size;
-
- ImmutableRandomAccessFileInfo(std::shared_ptr<RandomAccessFile> readable,
- int64_t size)
- : readable(std::move(readable)), size(size) {}
-
- Status ReadFully(uint64_t offset, int64_t size, Slice* data, uint8_t* scratch) const {
- return env_util::ReadFully(readable.get(), offset, size, data, scratch);
- }
-};
-
-// Caches block size and holds an exclusive reference to a ReadableBlock.
-// Assumes that the block underlying the ReadableBlock is immutable.
-struct ImmutableReadableBlockInfo {
- gscoped_ptr<fs::ReadableBlock> readable;
- int64_t size;
-
- ImmutableReadableBlockInfo(fs::ReadableBlock* readable,
- int64_t size)
- : readable(readable),
- size(size) {
- }
-
- Status ReadFully(uint64_t offset, int64_t size, Slice* data, uint8_t* scratch) const {
- return readable->Read(offset, size, data, scratch);
- }
-};
-
-// A potential Learner must establish a TabletCopySourceSession with the leader in order
-// to fetch the needed superblock, blocks, and log segments.
-// This class is refcounted to make it easy to remove it from the session map
-// on expiration while it is in use by another thread.
-class TabletCopySourceSession : public RefCountedThreadSafe<TabletCopySourceSession> {
- public:
- TabletCopySourceSession(const scoped_refptr<tablet::TabletPeer>& tablet_peer,
- std::string session_id, std::string requestor_uuid,
- FsManager* fs_manager);
-
- // Initialize the session, including anchoring files (TODO) and fetching the
- // tablet superblock and list of WAL segments.
- //
- // Must not be called more than once.
- Status Init();
-
- // Return ID of tablet corresponding to this session.
- const std::string& tablet_id() const;
-
- // Return UUID of the requestor that initiated this session.
- const std::string& requestor_uuid() const;
-
- // Open block for reading, if it's not already open, and read some of it.
- // If maxlen is 0, we use a system-selected length for the data piece.
- // *data is set to a std::string containing the data. Ownership of this object
- // is passed to the caller. A string is used because the RPC interface is
- // sending data serialized as protobuf and we want to minimize copying.
- // On error, Status is set to a non-OK value and error_code is filled in.
- //
- // This method is thread-safe.
- Status GetBlockPiece(const BlockId& block_id,
- uint64_t offset, int64_t client_maxlen,
- std::string* data, int64_t* block_file_size,
- TabletCopyErrorPB::Code* error_code);
-
- // Get a piece of a log segment.
- // The behavior and params are very similar to GetBlockPiece(), but this one
- // is only for sending WAL segment files.
- Status GetLogSegmentPiece(uint64_t segment_seqno,
- uint64_t offset, int64_t client_maxlen,
- std::string* data, int64_t* log_file_size,
- TabletCopyErrorPB::Code* error_code);
-
- const tablet::TabletSuperBlockPB& tablet_superblock() const {
- DCHECK(initted_);
- return tablet_superblock_;
- }
-
- const consensus::ConsensusStatePB& initial_committed_cstate() const {
- DCHECK(initted_);
- return initial_committed_cstate_;
- }
-
- const log::SegmentSequence& log_segments() const {
- DCHECK(initted_);
- return log_segments_;
- }
-
- // Check if a block is currently open.
- bool IsBlockOpenForTests(const BlockId& block_id) const;
-
- private:
- friend class RefCountedThreadSafe<TabletCopySourceSession>;
-
- typedef std::unordered_map<BlockId, ImmutableReadableBlockInfo*, BlockIdHash> BlockMap;
- typedef std::unordered_map<uint64_t, ImmutableRandomAccessFileInfo*> LogMap;
-
- ~TabletCopySourceSession();
-
- // Open the block and add it to the block map.
- Status OpenBlockUnlocked(const BlockId& block_id);
-
- // Look up cached block information.
- Status FindBlock(const BlockId& block_id,
- ImmutableReadableBlockInfo** block_info,
- TabletCopyErrorPB::Code* error_code);
-
- // Snapshot the log segment's length and put it into segment map.
- Status OpenLogSegmentUnlocked(uint64_t segment_seqno);
-
- // Look up log segment in cache or log segment map.
- Status FindLogSegment(uint64_t segment_seqno,
- ImmutableRandomAccessFileInfo** file_info,
- TabletCopyErrorPB::Code* error_code);
-
- // Unregister log anchor, if it's registered.
- Status UnregisterAnchorIfNeededUnlocked();
-
- scoped_refptr<tablet::TabletPeer> tablet_peer_;
- const std::string session_id_;
- const std::string requestor_uuid_;
- FsManager* const fs_manager_;
-
- mutable Mutex session_lock_;
- bool initted_ = false;
- BlockMap blocks_; // Protected by session_lock_.
- LogMap logs_; // Protected by session_lock_.
- ValueDeleter blocks_deleter_;
- ValueDeleter logs_deleter_;
-
- tablet::TabletSuperBlockPB tablet_superblock_;
-
- consensus::ConsensusStatePB initial_committed_cstate_;
-
- // The sequence of log segments that will be sent in the course of this
- // session.
- log::SegmentSequence log_segments_;
-
- log::LogAnchor log_anchor_;
-
- DISALLOW_COPY_AND_ASSIGN(TabletCopySourceSession);
-};
-
-} // namespace tserver
-} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/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
new file mode 100644
index 0000000..ec5194c
--- /dev/null
+++ b/src/kudu/tserver/tablet_copy_source_session-test.cc
@@ -0,0 +1,334 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+#include "kudu/tablet/tablet-test-util.h"
+
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+#include <memory>
+
+#include "kudu/common/partial_row.h"
+#include "kudu/common/row_operations.h"
+#include "kudu/common/schema.h"
+#include "kudu/consensus/consensus_meta.h"
+#include "kudu/consensus/log.h"
+#include "kudu/consensus/metadata.pb.h"
+#include "kudu/consensus/opid_util.h"
+#include "kudu/fs/block_id.h"
+#include "kudu/gutil/gscoped_ptr.h"
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/gutil/strings/fastmem.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/rpc/messenger.h"
+#include "kudu/tserver/tablet_copy_source_session.h"
+#include "kudu/tablet/tablet_peer.h"
+#include "kudu/util/crc.h"
+#include "kudu/util/metrics.h"
+#include "kudu/util/test_util.h"
+#include "kudu/util/threadpool.h"
+
+METRIC_DECLARE_entity(tablet);
+
+using std::shared_ptr;
+using std::string;
+using std::unique_ptr;
+
+namespace kudu {
+namespace tserver {
+
+using consensus::ConsensusMetadata;
+using consensus::OpId;
+using consensus::RaftConfigPB;
+using consensus::RaftPeerPB;
+using fs::ReadableBlock;
+using log::Log;
+using log::LogOptions;
+using log::LogAnchorRegistry;
+using rpc::Messenger;
+using rpc::MessengerBuilder;
+using strings::Substitute;
+using tablet::ColumnDataPB;
+using tablet::DeltaDataPB;
+using tablet::KuduTabletTest;
+using tablet::RowSetDataPB;
+using tablet::TabletPeer;
+using tablet::TabletSuperBlockPB;
+using tablet::WriteTransactionState;
+
+class TabletCopyTest : public KuduTabletTest {
+ public:
+ TabletCopyTest()
+ : KuduTabletTest(Schema({ ColumnSchema("key", STRING),
+ ColumnSchema("val", INT32) }, 1)) {
+ CHECK_OK(ThreadPoolBuilder("test-exec").Build(&apply_pool_));
+ }
+
+ virtual void SetUp() OVERRIDE {
+ NO_FATALS(KuduTabletTest::SetUp());
+ NO_FATALS(SetUpTabletPeer());
+ NO_FATALS(PopulateTablet());
+ NO_FATALS(InitSession());
+ }
+
+ virtual void TearDown() OVERRIDE {
+ session_.reset();
+ tablet_peer_->Shutdown();
+ KuduTabletTest::TearDown();
+ }
+
+ protected:
+ void SetUpTabletPeer() {
+ scoped_refptr<Log> log;
+ ASSERT_OK(Log::Open(LogOptions(), fs_manager(), tablet()->tablet_id(),
+ *tablet()->schema(),
+ 0, // schema_version
+ NULL, &log));
+
+ scoped_refptr<MetricEntity> metric_entity =
+ METRIC_ENTITY_tablet.Instantiate(&metric_registry_, CURRENT_TEST_NAME());
+
+ RaftPeerPB config_peer;
+ config_peer.set_permanent_uuid(fs_manager()->uuid());
+ config_peer.mutable_last_known_addr()->set_host("0.0.0.0");
+ config_peer.mutable_last_known_addr()->set_port(0);
+ config_peer.set_member_type(RaftPeerPB::VOTER);
+
+ tablet_peer_.reset(
+ new TabletPeer(tablet()->metadata(),
+ config_peer,
+ apply_pool_.get(),
+ Bind(&TabletCopyTest::TabletPeerStateChangedCallback,
+ Unretained(this),
+ tablet()->tablet_id())));
+
+ // TODO similar to code in tablet_peer-test, consider refactor.
+ RaftConfigPB config;
+ config.add_peers()->CopyFrom(config_peer);
+ config.set_opid_index(consensus::kInvalidOpIdIndex);
+
+ unique_ptr<ConsensusMetadata> cmeta;
+ ASSERT_OK(ConsensusMetadata::Create(tablet()->metadata()->fs_manager(),
+ tablet()->tablet_id(), fs_manager()->uuid(),
+ config, consensus::kMinimumTerm, &cmeta));
+
+ shared_ptr<Messenger> messenger;
+ MessengerBuilder mbuilder(CURRENT_TEST_NAME());
+ mbuilder.Build(&messenger);
+
+ log_anchor_registry_.reset(new LogAnchorRegistry());
+ tablet_peer_->SetBootstrapping();
+ ASSERT_OK(tablet_peer_->Init(tablet(),
+ clock(),
+ messenger,
+ scoped_refptr<rpc::ResultTracker>(),
+ log,
+ metric_entity));
+ consensus::ConsensusBootstrapInfo boot_info;
+ ASSERT_OK(tablet_peer_->Start(boot_info));
+ ASSERT_OK(tablet_peer_->WaitUntilConsensusRunning(MonoDelta::FromSeconds(10)));
+ ASSERT_OK(tablet_peer_->consensus()->WaitUntilLeaderForTests(MonoDelta::FromSeconds(10)));
+ }
+
+ void TabletPeerStateChangedCallback(const string& tablet_id, const string& reason) {
+ LOG(INFO) << "Tablet peer state changed for tablet " << tablet_id << ". Reason: " << reason;
+ }
+
+ void PopulateTablet() {
+ for (int32_t i = 0; i < 1000; i++) {
+ WriteRequestPB req;
+ req.set_tablet_id(tablet_peer_->tablet_id());
+ ASSERT_OK(SchemaToPB(client_schema_, req.mutable_schema()));
+ RowOperationsPB* data = req.mutable_row_operations();
+ RowOperationsPBEncoder enc(data);
+ KuduPartialRow row(&client_schema_);
+
+ string key = Substitute("key$0", i);
+ ASSERT_OK(row.SetStringNoCopy(0, key));
+ ASSERT_OK(row.SetInt32(1, i));
+ enc.Add(RowOperationsPB::INSERT, row);
+
+ WriteResponsePB resp;
+ CountDownLatch latch(1);
+
+ unique_ptr<tablet::WriteTransactionState> state(
+ new tablet::WriteTransactionState(tablet_peer_.get(),
+ &req,
+ nullptr, // No RequestIdPB
+ &resp));
+ state->set_completion_callback(gscoped_ptr<tablet::TransactionCompletionCallback>(
+ new tablet::LatchTransactionCompletionCallback<WriteResponsePB>(&latch, &resp)));
+ ASSERT_OK(tablet_peer_->SubmitWrite(std::move(state)));
+ latch.Wait();
+ ASSERT_FALSE(resp.has_error()) << "Request failed: " << resp.error().ShortDebugString();
+ ASSERT_EQ(0, resp.per_row_errors_size()) << "Insert error: " << resp.ShortDebugString();
+ }
+ ASSERT_OK(tablet()->Flush());
+ }
+
+ void InitSession() {
+ session_.reset(new TabletCopySourceSession(tablet_peer_.get(), "TestSession", "FakeUUID",
+ fs_manager()));
+ ASSERT_OK(session_->Init());
+ }
+
+ // Read the specified BlockId, via the TabletCopySourceSession, into a file.
+ // 'path' will be populated with the name of the file used.
+ // 'file' will be set to point to the SequentialFile containing the data.
+ void FetchBlockToFile(const BlockId& block_id,
+ string* path,
+ unique_ptr<SequentialFile>* file) {
+ string data;
+ int64_t block_file_size = 0;
+ TabletCopyErrorPB::Code error_code;
+ CHECK_OK(session_->GetBlockPiece(block_id, 0, 0, &data, &block_file_size, &error_code));
+ if (block_file_size > 0) {
+ CHECK_GT(data.size(), 0);
+ }
+
+ // Write the file to a temporary location.
+ WritableFileOptions opts;
+ string path_template = GetTestPath(Substitute("test_block_$0.tmp.XXXXXX", block_id.ToString()));
+ unique_ptr<WritableFile> writable_file;
+ CHECK_OK(Env::Default()->NewTempWritableFile(opts, path_template, path, &writable_file));
+ CHECK_OK(writable_file->Append(Slice(data.data(), data.size())));
+ CHECK_OK(writable_file->Close());
+
+ CHECK_OK(Env::Default()->NewSequentialFile(*path, file));
+ }
+
+ MetricRegistry metric_registry_;
+ scoped_refptr<LogAnchorRegistry> log_anchor_registry_;
+ gscoped_ptr<ThreadPool> apply_pool_;
+ scoped_refptr<TabletPeer> tablet_peer_;
+ scoped_refptr<TabletCopySourceSession> session_;
+};
+
+// Ensure that the serialized SuperBlock included in the TabletCopySourceSession is
+// equal to the serialized live superblock (on a quiesced tablet).
+TEST_F(TabletCopyTest, TestSuperBlocksEqual) {
+ // Compare content of superblocks.
+ faststring session_buf;
+ faststring tablet_buf;
+
+ {
+ const TabletSuperBlockPB& session_superblock = session_->tablet_superblock();
+ int size = session_superblock.ByteSize();
+ session_buf.resize(size);
+ uint8_t* session_dst = session_buf.data();
+ session_dst = session_superblock.SerializeWithCachedSizesToArray(session_dst);
+ }
+
+ {
+ TabletSuperBlockPB tablet_superblock;
+ ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
+ int size = tablet_superblock.ByteSize();
+ tablet_buf.resize(size);
+ uint8_t* tablet_dst = tablet_buf.data();
+ tablet_dst = tablet_superblock.SerializeWithCachedSizesToArray(tablet_dst);
+ }
+
+ ASSERT_EQ(session_buf.size(), tablet_buf.size());
+ int size = tablet_buf.size();
+ ASSERT_EQ(0, strings::fastmemcmp_inlined(session_buf.data(), tablet_buf.data(), size));
+}
+
+// Test fetching all files from tablet server, ensure the checksums for each
+// chunk and the total file sizes match.
+TEST_F(TabletCopyTest, TestBlocksEqual) {
+ TabletSuperBlockPB tablet_superblock;
+ ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
+ for (int i = 0; i < tablet_superblock.rowsets_size(); i++) {
+ const RowSetDataPB& rowset = tablet_superblock.rowsets(i);
+ for (int j = 0; j < rowset.columns_size(); j++) {
+ const ColumnDataPB& column = rowset.columns(j);
+ const BlockIdPB& block_id_pb = column.block();
+ BlockId block_id = BlockId::FromPB(block_id_pb);
+
+ string path;
+ unique_ptr<SequentialFile> file;
+ FetchBlockToFile(block_id, &path, &file);
+ uint64_t session_block_size = 0;
+ ASSERT_OK(Env::Default()->GetFileSize(path, &session_block_size));
+ faststring buf;
+ buf.resize(session_block_size);
+ Slice data;
+ ASSERT_OK(file->Read(session_block_size, &data, buf.data()));
+ uint32_t session_crc = crc::Crc32c(data.data(), data.size());
+ LOG(INFO) << "session block file has size of " << session_block_size
+ << " and CRC32C of " << session_crc << ": " << path;
+
+ gscoped_ptr<ReadableBlock> tablet_block;
+ ASSERT_OK(fs_manager()->OpenBlock(block_id, &tablet_block));
+ uint64_t tablet_block_size = 0;
+ ASSERT_OK(tablet_block->Size(&tablet_block_size));
+ buf.resize(tablet_block_size);
+ ASSERT_OK(tablet_block->Read(0, tablet_block_size, &data, buf.data()));
+ uint32_t tablet_crc = crc::Crc32c(data.data(), data.size());
+ LOG(INFO) << "tablet block file has size of " << tablet_block_size
+ << " and CRC32C of " << tablet_crc
+ << ": " << block_id;
+
+ // Compare the blocks.
+ ASSERT_EQ(tablet_block_size, session_block_size);
+ ASSERT_EQ(tablet_crc, session_crc);
+ }
+ }
+}
+
+// Ensure that blocks are still readable through the open session even
+// after they've been deleted.
+TEST_F(TabletCopyTest, TestBlocksAreFetchableAfterBeingDeleted) {
+ TabletSuperBlockPB tablet_superblock;
+ ASSERT_OK(tablet()->metadata()->ToSuperBlock(&tablet_superblock));
+
+ // Gather all the blocks.
+ vector<BlockId> data_blocks;
+ for (const RowSetDataPB& rowset : tablet_superblock.rowsets()) {
+ for (const DeltaDataPB& redo : rowset.redo_deltas()) {
+ data_blocks.push_back(BlockId::FromPB(redo.block()));
+ }
+ for (const DeltaDataPB& undo : rowset.undo_deltas()) {
+ data_blocks.push_back(BlockId::FromPB(undo.block()));
+ }
+ for (const ColumnDataPB& column : rowset.columns()) {
+ data_blocks.push_back(BlockId::FromPB(column.block()));
+ }
+ if (rowset.has_bloom_block()) {
+ data_blocks.push_back(BlockId::FromPB(rowset.bloom_block()));
+ }
+ if (rowset.has_adhoc_index_block()) {
+ data_blocks.push_back(BlockId::FromPB(rowset.adhoc_index_block()));
+ }
+ }
+
+ // Delete them.
+ for (const BlockId& block_id : data_blocks) {
+ ASSERT_OK(fs_manager()->DeleteBlock(block_id));
+ }
+
+ // Read them back.
+ for (const BlockId& block_id : data_blocks) {
+ ASSERT_TRUE(session_->IsBlockOpenForTests(block_id));
+ string data;
+ TabletCopyErrorPB::Code error_code;
+ int64_t piece_size;
+ ASSERT_OK(session_->GetBlockPiece(block_id, 0, 0,
+ &data, &piece_size, &error_code));
+ }
+}
+
+} // namespace tserver
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_source_session.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_source_session.cc b/src/kudu/tserver/tablet_copy_source_session.cc
new file mode 100644
index 0000000..d7c5d1c
--- /dev/null
+++ b/src/kudu/tserver/tablet_copy_source_session.cc
@@ -0,0 +1,386 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+#include "kudu/tserver/tablet_copy_source_session.h"
+
+#include <algorithm>
+#include <mutex>
+
+#include "kudu/consensus/log.h"
+#include "kudu/consensus/log_reader.h"
+#include "kudu/consensus/metadata.pb.h"
+#include "kudu/fs/block_manager.h"
+#include "kudu/gutil/map-util.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/gutil/type_traits.h"
+#include "kudu/rpc/transfer.h"
+#include "kudu/tablet/tablet_peer.h"
+#include "kudu/util/flag_tags.h"
+#include "kudu/util/mutex.h"
+#include "kudu/util/stopwatch.h"
+#include "kudu/util/trace.h"
+
+DEFINE_int32(tablet_copy_transfer_chunk_size_bytes, 4 * 1024 * 1024,
+ "Size of chunks to transfer when copying tablets between "
+ "tablet servers.");
+TAG_FLAG(tablet_copy_transfer_chunk_size_bytes, hidden);
+
+namespace kudu {
+namespace tserver {
+
+using consensus::MinimumOpId;
+using consensus::OpId;
+using fs::ReadableBlock;
+using log::LogAnchorRegistry;
+using log::ReadableLogSegment;
+using std::shared_ptr;
+using strings::Substitute;
+using tablet::ColumnDataPB;
+using tablet::DeltaDataPB;
+using tablet::RowSetDataPB;
+using tablet::TabletMetadata;
+using tablet::TabletPeer;
+using tablet::TabletSuperBlockPB;
+
+TabletCopySourceSession::TabletCopySourceSession(
+ const scoped_refptr<TabletPeer>& tablet_peer, std::string session_id,
+ std::string requestor_uuid, FsManager* fs_manager)
+ : tablet_peer_(tablet_peer),
+ session_id_(std::move(session_id)),
+ requestor_uuid_(std::move(requestor_uuid)),
+ fs_manager_(fs_manager),
+ blocks_deleter_(&blocks_),
+ logs_deleter_(&logs_) {}
+
+TabletCopySourceSession::~TabletCopySourceSession() {
+ // No lock taken in the destructor, should only be 1 thread with access now.
+ CHECK_OK(UnregisterAnchorIfNeededUnlocked());
+}
+
+Status TabletCopySourceSession::Init() {
+ MutexLock l(session_lock_);
+ CHECK(!initted_);
+
+ const string& tablet_id = tablet_peer_->tablet_id();
+
+ // Prevent log GC while we grab log segments and Tablet metadata.
+ string anchor_owner_token = Substitute("TabletCopy-$0", session_id_);
+ tablet_peer_->log_anchor_registry()->Register(
+ MinimumOpId().index(), anchor_owner_token, &log_anchor_);
+
+ // Read the SuperBlock from disk.
+ const scoped_refptr<TabletMetadata>& metadata = tablet_peer_->tablet_metadata();
+ RETURN_NOT_OK_PREPEND(metadata->ReadSuperBlockFromDisk(&tablet_superblock_),
+ Substitute("Unable to access superblock for tablet $0",
+ tablet_id));
+
+ // Anchor the data blocks by opening them and adding them to the cache.
+ //
+ // All subsequent requests should reuse the opened blocks.
+ vector<BlockIdPB> data_blocks;
+ TabletMetadata::CollectBlockIdPBs(tablet_superblock_, &data_blocks);
+ for (const BlockIdPB& block_id : data_blocks) {
+ VLOG(1) << "Opening block " << block_id.DebugString();
+ RETURN_NOT_OK(OpenBlockUnlocked(BlockId::FromPB(block_id)));
+ }
+
+ // Get the latest opid in the log at this point in time so we can re-anchor.
+ OpId last_logged_opid;
+ tablet_peer_->log()->GetLatestEntryOpId(&last_logged_opid);
+
+ // Get the current segments from the log, including the active segment.
+ // The Log doesn't add the active segment to the log reader's list until
+ // a header has been written to it (but it will not have a footer).
+ shared_ptr<log::LogReader> reader = tablet_peer_->log()->reader();
+ if (!reader) {
+ tablet::TabletStatePB tablet_state = tablet_peer_->state();
+ return Status::IllegalState(Substitute(
+ "Unable to initialize tablet copy session for tablet $0. "
+ "Log reader is not available. Tablet state: $1 ($2)",
+ tablet_id, tablet::TabletStatePB_Name(tablet_state), tablet_state));
+ }
+ reader->GetSegmentsSnapshot(&log_segments_);
+ for (const scoped_refptr<ReadableLogSegment>& segment : log_segments_) {
+ RETURN_NOT_OK(OpenLogSegmentUnlocked(segment->header().sequence_number()));
+ }
+
+ // Look up the committed consensus state.
+ // We do this after snapshotting the log to avoid a scenario where the latest
+ // entry in the log has a term higher than the term stored in the consensus
+ // metadata, which will results in a CHECK failure on RaftConsensus init.
+ scoped_refptr<consensus::Consensus> consensus = tablet_peer_->shared_consensus();
+ if (!consensus) {
+ tablet::TabletStatePB tablet_state = tablet_peer_->state();
+ return Status::IllegalState(Substitute(
+ "Unable to initialize tablet copy session for tablet $0. "
+ "Consensus is not available. Tablet state: $1 ($2)",
+ tablet_id, tablet::TabletStatePB_Name(tablet_state), tablet_state));
+ }
+ initial_committed_cstate_ = consensus->ConsensusState(consensus::CONSENSUS_CONFIG_COMMITTED);
+
+ // Re-anchor on the highest OpId that was in the log right before we
+ // snapshotted the log segments. This helps ensure that we don't end up in a
+ // tablet copy loop due to a follower falling too far behind the
+ // leader's log when tablet copy is slow. The remote controls when
+ // this anchor is released by ending the tablet copy session.
+ RETURN_NOT_OK(tablet_peer_->log_anchor_registry()->UpdateRegistration(
+ last_logged_opid.index(), anchor_owner_token, &log_anchor_));
+
+ LOG(INFO) << Substitute(
+ "T $0 P $1: Tablet Copy: opened $2 blocks and $3 log segments",
+ tablet_id, consensus->peer_uuid(), data_blocks.size(), log_segments_.size());
+ initted_ = true;
+ return Status::OK();
+}
+
+const std::string& TabletCopySourceSession::tablet_id() const {
+ DCHECK(initted_);
+ return tablet_peer_->tablet_id();
+}
+
+const std::string& TabletCopySourceSession::requestor_uuid() const {
+ DCHECK(initted_);
+ return requestor_uuid_;
+}
+
+// Determine the length of the data chunk to return to the client.
+static int64_t DetermineReadLength(int64_t bytes_remaining, int64_t requested_len) {
+ // Overhead in the RPC for things like headers, protobuf data, etc.
+ static const int kOverhead = 4096;
+
+ if (requested_len <= 0) {
+ requested_len = FLAGS_tablet_copy_transfer_chunk_size_bytes;
+ } else {
+ requested_len = std::min<int64_t>(requested_len,
+ FLAGS_tablet_copy_transfer_chunk_size_bytes);
+ }
+ requested_len = std::min<int64_t>(requested_len, FLAGS_rpc_max_message_size - kOverhead);
+ CHECK_GT(requested_len, 0) << "rpc_max_message_size is too low to transfer data: "
+ << FLAGS_rpc_max_message_size;
+ return std::min(bytes_remaining, requested_len);
+}
+
+// Calculate the size of the data to return given a maximum client message
+// length, the file itself, and the offset into the file to be read from.
+static Status GetResponseDataSize(int64_t total_size,
+ uint64_t offset, int64_t client_maxlen,
+ TabletCopyErrorPB::Code* error_code, int64_t* data_size) {
+ // If requested offset is off the end of the data, bail.
+ if (offset >= total_size) {
+ *error_code = TabletCopyErrorPB::INVALID_TABLET_COPY_REQUEST;
+ return Status::InvalidArgument(
+ Substitute("Requested offset ($0) is beyond the data size ($1)",
+ offset, total_size));
+ }
+
+ int64_t bytes_remaining = total_size - offset;
+
+ *data_size = DetermineReadLength(bytes_remaining, client_maxlen);
+ DCHECK_GT(*data_size, 0);
+ if (client_maxlen > 0) {
+ DCHECK_LE(*data_size, client_maxlen);
+ }
+
+ return Status::OK();
+}
+
+// Read a chunk of a file into a buffer.
+// data_name provides a string for the block/log to be used in error messages.
+template <class Info>
+static Status ReadFileChunkToBuf(const Info* info,
+ uint64_t offset, int64_t client_maxlen,
+ const string& data_name,
+ string* data, int64_t* file_size,
+ TabletCopyErrorPB::Code* error_code) {
+ int64_t response_data_size = 0;
+ RETURN_NOT_OK_PREPEND(GetResponseDataSize(info->size, offset, client_maxlen, error_code,
+ &response_data_size),
+ Substitute("Error reading $0", data_name));
+
+ Stopwatch chunk_timer(Stopwatch::THIS_THREAD);
+ chunk_timer.start();
+
+ // Writing into a std::string buffer is basically guaranteed to work on C++11,
+ // however any modern compiler should be compatible with it.
+ // Violates the API contract, but avoids excessive copies.
+ data->resize(response_data_size);
+ uint8_t* buf = reinterpret_cast<uint8_t*>(const_cast<char*>(data->data()));
+ Slice slice;
+ Status s = info->ReadFully(offset, response_data_size, &slice, buf);
+ if (PREDICT_FALSE(!s.ok())) {
+ s = s.CloneAndPrepend(
+ Substitute("Unable to read existing file for $0", data_name));
+ LOG(WARNING) << s.ToString();
+ *error_code = TabletCopyErrorPB::IO_ERROR;
+ return s;
+ }
+ // Figure out if Slice points to buf or if Slice points to the mmap.
+ // If it points to the mmap then copy into buf.
+ if (slice.data() != buf) {
+ memcpy(buf, slice.data(), slice.size());
+ }
+ chunk_timer.stop();
+ TRACE("Tablet Copy: $0: $1 total bytes read. Total time elapsed: $2",
+ data_name, response_data_size, chunk_timer.elapsed().ToString());
+
+ *file_size = info->size;
+ return Status::OK();
+}
+
+Status TabletCopySourceSession::GetBlockPiece(const BlockId& block_id,
+ uint64_t offset, int64_t client_maxlen,
+ string* data, int64_t* block_file_size,
+ TabletCopyErrorPB::Code* error_code) {
+ ImmutableReadableBlockInfo* block_info;
+ RETURN_NOT_OK(FindBlock(block_id, &block_info, error_code));
+
+ RETURN_NOT_OK(ReadFileChunkToBuf(block_info, offset, client_maxlen,
+ Substitute("block $0", block_id.ToString()),
+ data, block_file_size, error_code));
+
+ // Note: We do not eagerly close the block, as doing so may delete the
+ // underlying data if this was its last reader and it had been previously
+ // marked for deletion. This would be a problem for parallel readers in
+ // the same session; they would not be able to find the block.
+
+ return Status::OK();
+}
+
+Status TabletCopySourceSession::GetLogSegmentPiece(uint64_t segment_seqno,
+ uint64_t offset, int64_t client_maxlen,
+ std::string* data, int64_t* block_file_size,
+ TabletCopyErrorPB::Code* error_code) {
+ ImmutableRandomAccessFileInfo* file_info;
+ RETURN_NOT_OK(FindLogSegment(segment_seqno, &file_info, error_code));
+ RETURN_NOT_OK(ReadFileChunkToBuf(file_info, offset, client_maxlen,
+ Substitute("log segment $0", segment_seqno),
+ data, block_file_size, error_code));
+
+ // Note: We do not eagerly close log segment files, since we share ownership
+ // of the LogSegment objects with the Log itself.
+
+ return Status::OK();
+}
+
+bool TabletCopySourceSession::IsBlockOpenForTests(const BlockId& block_id) const {
+ MutexLock l(session_lock_);
+ return ContainsKey(blocks_, block_id);
+}
+
+// Add a file to the cache and populate the given ImmutableRandomAcccessFileInfo
+// object with the file ref and size.
+template <class Collection, class Key, class Readable>
+static Status AddImmutableFileToMap(Collection* const cache,
+ const Key& key,
+ const Readable& readable,
+ uint64_t size) {
+ // Sanity check for 0-length files.
+ if (size == 0) {
+ return Status::Corruption("Found 0-length object");
+ }
+
+ // Looks good, add it to the cache.
+ typedef typename Collection::mapped_type InfoPtr;
+ typedef typename base::remove_pointer<InfoPtr>::type Info;
+ InsertOrDie(cache, key, new Info(readable, size));
+
+ return Status::OK();
+}
+
+Status TabletCopySourceSession::OpenBlockUnlocked(const BlockId& block_id) {
+ session_lock_.AssertAcquired();
+
+ gscoped_ptr<ReadableBlock> block;
+ Status s = fs_manager_->OpenBlock(block_id, &block);
+ if (PREDICT_FALSE(!s.ok())) {
+ LOG(WARNING) << "Unable to open requested (existing) block file: "
+ << block_id.ToString() << ": " << s.ToString();
+ return s.CloneAndPrepend(Substitute("Unable to open block file for block $0",
+ block_id.ToString()));
+ }
+
+ uint64_t size;
+ s = block->Size(&size);
+ if (PREDICT_FALSE(!s.ok())) {
+ return s.CloneAndPrepend("Unable to get size of block");
+ }
+
+ s = AddImmutableFileToMap(&blocks_, block_id, block.get(), size);
+ if (!s.ok()) {
+ s = s.CloneAndPrepend(Substitute("Error accessing data for block $0", block_id.ToString()));
+ LOG(DFATAL) << "Data block disappeared: " << s.ToString();
+ } else {
+ ignore_result(block.release());
+ }
+ return s;
+}
+
+Status TabletCopySourceSession::FindBlock(const BlockId& block_id,
+ ImmutableReadableBlockInfo** block_info,
+ TabletCopyErrorPB::Code* error_code) {
+ Status s;
+ MutexLock l(session_lock_);
+ if (!FindCopy(blocks_, block_id, block_info)) {
+ *error_code = TabletCopyErrorPB::BLOCK_NOT_FOUND;
+ s = Status::NotFound("Block not found", block_id.ToString());
+ }
+ return s;
+}
+
+Status TabletCopySourceSession::OpenLogSegmentUnlocked(uint64_t segment_seqno) {
+ session_lock_.AssertAcquired();
+
+ scoped_refptr<log::ReadableLogSegment> log_segment;
+ int position = -1;
+ if (!log_segments_.empty()) {
+ position = segment_seqno - log_segments_[0]->header().sequence_number();
+ }
+ if (position < 0 || position >= log_segments_.size()) {
+ return Status::NotFound(Substitute("Segment with sequence number $0 not found",
+ segment_seqno));
+ }
+ log_segment = log_segments_[position];
+ CHECK_EQ(log_segment->header().sequence_number(), segment_seqno);
+
+ uint64_t size = log_segment->readable_up_to();
+ Status s = AddImmutableFileToMap(&logs_, segment_seqno, log_segment->readable_file(), size);
+ if (!s.ok()) {
+ s = s.CloneAndPrepend(
+ Substitute("Error accessing data for log segment with seqno $0",
+ segment_seqno));
+ LOG(INFO) << s.ToString();
+ }
+ return s;
+}
+
+Status TabletCopySourceSession::FindLogSegment(uint64_t segment_seqno,
+ ImmutableRandomAccessFileInfo** file_info,
+ TabletCopyErrorPB::Code* error_code) {
+ MutexLock l(session_lock_);
+ if (!FindCopy(logs_, segment_seqno, file_info)) {
+ *error_code = TabletCopyErrorPB::WAL_SEGMENT_NOT_FOUND;
+ return Status::NotFound(Substitute("Segment with sequence number $0 not found",
+ segment_seqno));
+ }
+ return Status::OK();
+}
+
+Status TabletCopySourceSession::UnregisterAnchorIfNeededUnlocked() {
+ return tablet_peer_->log_anchor_registry()->UnregisterIfAnchored(&log_anchor_);
+}
+
+} // namespace tserver
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/f2dac5d3/src/kudu/tserver/tablet_copy_source_session.h
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_source_session.h b/src/kudu/tserver/tablet_copy_source_session.h
new file mode 100644
index 0000000..aaf0043
--- /dev/null
+++ b/src/kudu/tserver/tablet_copy_source_session.h
@@ -0,0 +1,197 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "kudu/consensus/log_anchor_registry.h"
+#include "kudu/consensus/log_util.h"
+#include "kudu/consensus/metadata.pb.h"
+#include "kudu/consensus/opid_util.h"
+#include "kudu/fs/block_id.h"
+#include "kudu/fs/block_manager.h"
+#include "kudu/gutil/macros.h"
+#include "kudu/gutil/ref_counted.h"
+#include "kudu/gutil/stl_util.h"
+#include "kudu/tserver/tablet_copy.pb.h"
+#include "kudu/util/env_util.h"
+#include "kudu/util/locks.h"
+#include "kudu/util/status.h"
+
+namespace kudu {
+
+class FsManager;
+
+namespace tablet {
+class TabletPeer;
+} // namespace tablet
+
+namespace tserver {
+
+class TabletPeerLookupIf;
+
+// Caches file size and holds a shared_ptr reference to a RandomAccessFile.
+// Assumes that the file underlying the RandomAccessFile is immutable.
+struct ImmutableRandomAccessFileInfo {
+ std::shared_ptr<RandomAccessFile> readable;
+ int64_t size;
+
+ ImmutableRandomAccessFileInfo(std::shared_ptr<RandomAccessFile> readable,
+ int64_t size)
+ : readable(std::move(readable)), size(size) {}
+
+ Status ReadFully(uint64_t offset, int64_t size, Slice* data, uint8_t* scratch) const {
+ return env_util::ReadFully(readable.get(), offset, size, data, scratch);
+ }
+};
+
+// Caches block size and holds an exclusive reference to a ReadableBlock.
+// Assumes that the block underlying the ReadableBlock is immutable.
+struct ImmutableReadableBlockInfo {
+ gscoped_ptr<fs::ReadableBlock> readable;
+ int64_t size;
+
+ ImmutableReadableBlockInfo(fs::ReadableBlock* readable,
+ int64_t size)
+ : readable(readable),
+ size(size) {
+ }
+
+ Status ReadFully(uint64_t offset, int64_t size, Slice* data, uint8_t* scratch) const {
+ return readable->Read(offset, size, data, scratch);
+ }
+};
+
+// A potential Learner must establish a TabletCopySourceSession with the leader in order
+// to fetch the needed superblock, blocks, and log segments.
+// This class is refcounted to make it easy to remove it from the session map
+// on expiration while it is in use by another thread.
+class TabletCopySourceSession : public RefCountedThreadSafe<TabletCopySourceSession> {
+ public:
+ TabletCopySourceSession(const scoped_refptr<tablet::TabletPeer>& tablet_peer,
+ std::string session_id, std::string requestor_uuid,
+ FsManager* fs_manager);
+
+ // Initialize the session, including anchoring files (TODO) and fetching the
+ // tablet superblock and list of WAL segments.
+ //
+ // Must not be called more than once.
+ Status Init();
+
+ // Return ID of tablet corresponding to this session.
+ const std::string& tablet_id() const;
+
+ // Return UUID of the requestor that initiated this session.
+ const std::string& requestor_uuid() const;
+
+ // Open block for reading, if it's not already open, and read some of it.
+ // If maxlen is 0, we use a system-selected length for the data piece.
+ // *data is set to a std::string containing the data. Ownership of this object
+ // is passed to the caller. A string is used because the RPC interface is
+ // sending data serialized as protobuf and we want to minimize copying.
+ // On error, Status is set to a non-OK value and error_code is filled in.
+ //
+ // This method is thread-safe.
+ Status GetBlockPiece(const BlockId& block_id,
+ uint64_t offset, int64_t client_maxlen,
+ std::string* data, int64_t* block_file_size,
+ TabletCopyErrorPB::Code* error_code);
+
+ // Get a piece of a log segment.
+ // The behavior and params are very similar to GetBlockPiece(), but this one
+ // is only for sending WAL segment files.
+ Status GetLogSegmentPiece(uint64_t segment_seqno,
+ uint64_t offset, int64_t client_maxlen,
+ std::string* data, int64_t* log_file_size,
+ TabletCopyErrorPB::Code* error_code);
+
+ const tablet::TabletSuperBlockPB& tablet_superblock() const {
+ DCHECK(initted_);
+ return tablet_superblock_;
+ }
+
+ const consensus::ConsensusStatePB& initial_committed_cstate() const {
+ DCHECK(initted_);
+ return initial_committed_cstate_;
+ }
+
+ const log::SegmentSequence& log_segments() const {
+ DCHECK(initted_);
+ return log_segments_;
+ }
+
+ // Check if a block is currently open.
+ bool IsBlockOpenForTests(const BlockId& block_id) const;
+
+ private:
+ friend class RefCountedThreadSafe<TabletCopySourceSession>;
+
+ typedef std::unordered_map<BlockId, ImmutableReadableBlockInfo*, BlockIdHash> BlockMap;
+ typedef std::unordered_map<uint64_t, ImmutableRandomAccessFileInfo*> LogMap;
+
+ ~TabletCopySourceSession();
+
+ // Open the block and add it to the block map.
+ Status OpenBlockUnlocked(const BlockId& block_id);
+
+ // Look up cached block information.
+ Status FindBlock(const BlockId& block_id,
+ ImmutableReadableBlockInfo** block_info,
+ TabletCopyErrorPB::Code* error_code);
+
+ // Snapshot the log segment's length and put it into segment map.
+ Status OpenLogSegmentUnlocked(uint64_t segment_seqno);
+
+ // Look up log segment in cache or log segment map.
+ Status FindLogSegment(uint64_t segment_seqno,
+ ImmutableRandomAccessFileInfo** file_info,
+ TabletCopyErrorPB::Code* error_code);
+
+ // Unregister log anchor, if it's registered.
+ Status UnregisterAnchorIfNeededUnlocked();
+
+ scoped_refptr<tablet::TabletPeer> tablet_peer_;
+ const std::string session_id_;
+ const std::string requestor_uuid_;
+ FsManager* const fs_manager_;
+
+ mutable Mutex session_lock_;
+ bool initted_ = false;
+ BlockMap blocks_; // Protected by session_lock_.
+ LogMap logs_; // Protected by session_lock_.
+ ValueDeleter blocks_deleter_;
+ ValueDeleter logs_deleter_;
+
+ tablet::TabletSuperBlockPB tablet_superblock_;
+
+ consensus::ConsensusStatePB initial_committed_cstate_;
+
+ // The sequence of log segments that will be sent in the course of this
+ // session.
+ log::SegmentSequence log_segments_;
+
+ log::LogAnchor log_anchor_;
+
+ DISALLOW_COPY_AND_ASSIGN(TabletCopySourceSession);
+};
+
+} // namespace tserver
+} // namespace kudu
[2/2] kudu git commit: tablet copy: Fix tidy warnings in
TabletCopySourceSession{Test}
Posted by mp...@apache.org.
tablet copy: Fix tidy warnings in TabletCopySourceSession{Test}
Done as a follow-up commit to allow for a clean rename in the previous
commit.
Change-Id: I5b9dd173a8c0a7118225a5b4f24ae905c98e3c27
Reviewed-on: http://gerrit.cloudera.org:8080/5257
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/a0f425f7
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/a0f425f7
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/a0f425f7
Branch: refs/heads/master
Commit: a0f425f7636541422e97b39b89f9fe39e9dbd4f6
Parents: f2dac5d
Author: Mike Percy <mp...@apache.org>
Authored: Fri Nov 25 14:57:53 2016 +0000
Committer: Mike Percy <mp...@apache.org>
Committed: Tue Nov 29 20:04:03 2016 +0000
----------------------------------------------------------------------
src/kudu/tserver/tablet_copy_source_session-test.cc | 3 +--
src/kudu/tserver/tablet_copy_source_session.cc | 14 ++++----------
2 files changed, 5 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/a0f425f7/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 ec5194c..9be1c81 100644
--- a/src/kudu/tserver/tablet_copy_source_session-test.cc
+++ b/src/kudu/tserver/tablet_copy_source_session-test.cc
@@ -50,7 +50,6 @@ namespace kudu {
namespace tserver {
using consensus::ConsensusMetadata;
-using consensus::OpId;
using consensus::RaftConfigPB;
using consensus::RaftPeerPB;
using fs::ReadableBlock;
@@ -114,7 +113,7 @@ class TabletCopyTest : public KuduTabletTest {
Unretained(this),
tablet()->tablet_id())));
- // TODO similar to code in tablet_peer-test, consider refactor.
+ // TODO(dralves) similar to code in tablet_peer-test, consider refactor.
RaftConfigPB config;
config.add_peers()->CopyFrom(config_peer);
config.set_opid_index(consensus::kInvalidOpIdIndex);
http://git-wip-us.apache.org/repos/asf/kudu/blob/a0f425f7/src/kudu/tserver/tablet_copy_source_session.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_copy_source_session.cc b/src/kudu/tserver/tablet_copy_source_session.cc
index d7c5d1c..597714b 100644
--- a/src/kudu/tserver/tablet_copy_source_session.cc
+++ b/src/kudu/tserver/tablet_copy_source_session.cc
@@ -44,16 +44,11 @@ namespace tserver {
using consensus::MinimumOpId;
using consensus::OpId;
using fs::ReadableBlock;
-using log::LogAnchorRegistry;
using log::ReadableLogSegment;
using std::shared_ptr;
using strings::Substitute;
-using tablet::ColumnDataPB;
-using tablet::DeltaDataPB;
-using tablet::RowSetDataPB;
using tablet::TabletMetadata;
using tablet::TabletPeer;
-using tablet::TabletSuperBlockPB;
TabletCopySourceSession::TabletCopySourceSession(
const scoped_refptr<TabletPeer>& tablet_peer, std::string session_id,
@@ -260,18 +255,17 @@ Status TabletCopySourceSession::GetBlockPiece(const BlockId& block_id,
}
Status TabletCopySourceSession::GetLogSegmentPiece(uint64_t segment_seqno,
- uint64_t offset, int64_t client_maxlen,
- std::string* data, int64_t* block_file_size,
- TabletCopyErrorPB::Code* error_code) {
+ uint64_t offset, int64_t client_maxlen,
+ std::string* data, int64_t* log_file_size,
+ TabletCopyErrorPB::Code* error_code) {
ImmutableRandomAccessFileInfo* file_info;
RETURN_NOT_OK(FindLogSegment(segment_seqno, &file_info, error_code));
RETURN_NOT_OK(ReadFileChunkToBuf(file_info, offset, client_maxlen,
Substitute("log segment $0", segment_seqno),
- data, block_file_size, error_code));
+ data, log_file_size, error_code));
// Note: We do not eagerly close log segment files, since we share ownership
// of the LogSegment objects with the Log itself.
-
return Status::OK();
}