mirrors/scylladb
1
0
Fork 0
mirror of https://github.com/scylladb/scylladb.git synced 2026-04-22 17:40:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

streaming: Swith to use shared_ptr from std::shared_ptr

Browse Source 
Since our shared_ptr works with incomplete types now, switch to it.
This commit is contained in:
Asias He
2015-07-10 13:28:43 +08:00
parent 8fd8f39d63
commit e82bdf2995
14 changed files with 44 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
streaming/stream_coordinator.cc
View File

@@ -37,8 +37,8 @@ bool stream_coordinator::has_active_sessions() {
return false;
}
std::vector<std::shared_ptr<stream_session>> stream_coordinator::get_all_stream_sessions() {
std::vector<std::shared_ptr<stream_session>> results;
std::vector<shared_ptr<stream_session>> stream_coordinator::get_all_stream_sessions() {
std::vector<shared_ptr<stream_session>> results;
for (auto& x : _peer_sessions) {
auto s = x.second.get_all_stream_sessions();
std::move(s.begin(), s.end(), std::back_inserter(results));
@@ -114,11 +114,11 @@ bool stream_coordinator::host_streaming_data::has_active_sessions() {
return false;
}
std::shared_ptr<stream_session> stream_coordinator::host_streaming_data::get_or_create_next_session(inet_address peer, inet_address connecting) {
shared_ptr<stream_session> stream_coordinator::host_streaming_data::get_or_create_next_session(inet_address peer, inet_address connecting) {
// create
int size = _stream_sessions.size();
if (size < _connections_per_host) {
auto session = std::make_shared<stream_session>(peer, connecting, size, _keep_ss_table_level);
auto session = make_shared<stream_session>(peer, connecting, size, _keep_ss_table_level);
_stream_sessions.emplace(++_last_returned, session);
return _stream_sessions[_last_returned];
// get
@@ -130,19 +130,19 @@ std::shared_ptr<stream_session> stream_coordinator::host_streaming_data::get_or_
}
}
std::vector<std::shared_ptr<stream_session>> stream_coordinator::host_streaming_data::get_all_stream_sessions() {
std::vector<std::shared_ptr<stream_session>> sessions;
std::vector<shared_ptr<stream_session>> stream_coordinator::host_streaming_data::get_all_stream_sessions() {
std::vector<shared_ptr<stream_session>> sessions;
for (auto& x : _stream_sessions) {
sessions.push_back(x.second);
}
return sessions;
}
std::shared_ptr<stream_session> stream_coordinator::host_streaming_data::get_or_create_session_by_id(inet_address peer,
shared_ptr<stream_session> stream_coordinator::host_streaming_data::get_or_create_session_by_id(inet_address peer,
int id, inet_address connecting) {
auto it = _stream_sessions.find(id);
if (it == _stream_sessions.end()) {
it = _stream_sessions.emplace(id, std::make_shared<stream_session>(peer, connecting, id, _keep_ss_table_level)).first;
it = _stream_sessions.emplace(id, make_shared<stream_session>(peer, connecting, id, _keep_ss_table_level)).first;
}
return it->second;
}

14
streaming/stream_coordinator.hh
View File

@@ -63,7 +63,7 @@ public:
*/
bool has_active_sessions();
std::vector<std::shared_ptr<stream_session>> get_all_stream_sessions();
std::vector<shared_ptr<stream_session>> get_all_stream_sessions();
bool is_receiving();
@@ -76,11 +76,11 @@ public:
std::set<inet_address> get_peers();
public:
std::shared_ptr<stream_session> get_or_create_next_session(inet_address peer, inet_address connecting) {
shared_ptr<stream_session> get_or_create_next_session(inet_address peer, inet_address connecting) {
return get_or_create_host_data(peer).get_or_create_next_session(peer, connecting);
}
std::shared_ptr<stream_session> get_or_create_session_by_id(inet_address peer, int id, inet_address connecting) {
shared_ptr<stream_session> get_or_create_session_by_id(inet_address peer, int id, inet_address connecting) {
return get_or_create_host_data(peer).get_or_create_session_by_id(peer, id, connecting);
}
@@ -132,7 +132,7 @@ private:
class host_streaming_data {
using inet_address = gms::inet_address;
private:
std::map<int, std::shared_ptr<stream_session>> _stream_sessions;
std::map<int, shared_ptr<stream_session>> _stream_sessions;
std::map<int, session_info> _session_infos;
int _last_returned = -1;
int _connections_per_host;
@@ -148,7 +148,7 @@ private:
bool has_active_sessions();
std::shared_ptr<stream_session> get_or_create_next_session(inet_address peer, inet_address connecting);
shared_ptr<stream_session> get_or_create_next_session(inet_address peer, inet_address connecting);
void connect_all_stream_sessions() {
for (auto& x : _stream_sessions) {
@@ -158,9 +158,9 @@ private:
}
}
std::vector<std::shared_ptr<stream_session>> get_all_stream_sessions();
std::vector<shared_ptr<stream_session>> get_all_stream_sessions();
std::shared_ptr<stream_session> get_or_create_session_by_id(inet_address peer, int id, inet_address connecting);
shared_ptr<stream_session> get_or_create_session_by_id(inet_address peer, int id, inet_address connecting);
void update_progress(progress_info info);

2
streaming/stream_event.hh
View File

@@ -53,7 +53,7 @@ struct session_complete_event : public stream_event {
bool success;
int session_index;
session_complete_event(std::shared_ptr<stream_session> session)
session_complete_event(shared_ptr<stream_session> session)
: stream_event(stream_event::type::STREAM_COMPLETE, session->plan_id())
, peer(session->peer)
, success(session->is_success())

13
streaming/stream_reader.cc
View File

@@ -22,6 +22,19 @@
namespace streaming {
stream_reader::stream_reader(file_message_header header, shared_ptr<stream_session> session_)
: cf_id(header.cf_id)
, estimated_keys(header.estimated_keys)
, sections(header.sections)
, session(session_)
// input_version = header.format.info.getVersion(header.version)
, repaired_at(header.repaired_at)
, format(header.format)
, sstable_level(header.sstable_level) {
}
stream_reader::~stream_reader() = default;
int64_t stream_reader::total_size() {
int64_t size = 0;
for (auto section : sections)

14
streaming/stream_reader.hh
View File

@@ -40,7 +40,7 @@ protected:
UUID cf_id;
int64_t estimated_keys;
std::map<int64_t, int64_t> sections;
std::shared_ptr<stream_session> session;
shared_ptr<stream_session> session;
// FIXME: Version
version_types input_version;
int64_t repaired_at;
@@ -49,16 +49,8 @@ protected:
// FIXME: Descriptor
//Descriptor desc;
public:
stream_reader(file_message_header header, std::shared_ptr<stream_session> session_)
: cf_id(header.cf_id)
, estimated_keys(header.estimated_keys)
, sections(header.sections)
, session(session_)
// input_version = header.format.info.getVersion(header.version)
, repaired_at(header.repaired_at)
, format(header.format)
, sstable_level(header.sstable_level) {
}
stream_reader(file_message_header header, shared_ptr<stream_session> session_);
~stream_reader();
#if 0
/**

2
streaming/stream_receive_task.cc
View File

@@ -24,7 +24,7 @@
namespace streaming {
stream_receive_task::stream_receive_task(std::shared_ptr<stream_session> _session, UUID _cf_id, int _total_files, long _total_size)
stream_receive_task::stream_receive_task(shared_ptr<stream_session> _session, UUID _cf_id, int _total_files, long _total_size)
: stream_task(_session, _cf_id)
, total_files(_total_files)
, total_size(_total_size) {

2
streaming/stream_receive_task.hh
View File

@@ -46,7 +46,7 @@ private:
// holds references to SSTables received
// protected Collection<SSTableWriter> sstables;
public:
stream_receive_task(std::shared_ptr<stream_session> _session, UUID _cf_id, int _total_files, long _total_size);
stream_receive_task(shared_ptr<stream_session> _session, UUID _cf_id, int _total_files, long _total_size);
~stream_receive_task();
/**

2
streaming/stream_session.cc
View File

@@ -389,7 +389,7 @@ std::vector<column_family*> stream_session::get_column_family_stores(const sstri
abort();
// FIXME: stores.addAll(Keyspace.open(keyspace).getColumnFamilyStores());
} else {
// TODO: We can move this to database class and use std::shared_ptr<column_family> instead
// TODO: We can move this to database class and use shared_ptr<column_family> instead
for (auto& cf_name : column_families) {
auto& x = db.find_column_family(keyspace, cf_name);
stores.push_back(&x);

2
streaming/stream_session.hh
View File

@@ -115,7 +115,7 @@ void streaming_debug(const char* fmt, Args&&... args) {
* session is done is is closed (closeSession()). Otherwise, the node switch to the WAIT_COMPLETE state and
* send a CompleteMessage to the other side.
*/
class stream_session : public gms::i_endpoint_state_change_subscriber, public std::enable_shared_from_this<stream_session> {
class stream_session : public gms::i_endpoint_state_change_subscriber, public enable_shared_from_this<stream_session> {
private:
using messaging_verb = net::messaging_verb;
using messaging_service = net::messaging_service;

2
streaming/stream_task.cc
View File

@@ -24,7 +24,7 @@
namespace streaming {
stream_task::stream_task(std::shared_ptr<stream_session> _session, UUID _cf_id)
stream_task::stream_task(shared_ptr<stream_session> _session, UUID _cf_id)
: session(_session)
, cf_id(std::move(_cf_id)) {
}

5
streaming/stream_task.hh
View File

@@ -24,6 +24,7 @@
#include "utils/UUID.hh"
#include "streaming/stream_summary.hh"
#include <memory>
#include "core/shared_ptr.hh"
namespace streaming {
@@ -36,11 +37,11 @@ class stream_task {
public:
using UUID = utils::UUID;
/** StreamSession that this task belongs */
std::shared_ptr<stream_session> session;
shared_ptr<stream_session> session;
UUID cf_id;
stream_task(std::shared_ptr<stream_session> _session, UUID _cf_id);
stream_task(shared_ptr<stream_session> _session, UUID _cf_id);
~stream_task();
public:

2
streaming/stream_transfer_task.cc
View File

@@ -30,7 +30,7 @@
namespace streaming {
stream_transfer_task::stream_transfer_task(std::shared_ptr<stream_session> session, UUID cf_id)
stream_transfer_task::stream_transfer_task(shared_ptr<stream_session> session, UUID cf_id)
: stream_task(session, cf_id) {
}

2
streaming/stream_transfer_task.hh
View File

@@ -46,7 +46,7 @@ private:
long total_size;
public:
using UUID = utils::UUID;
stream_transfer_task(std::shared_ptr<stream_session> session, UUID cf_id);
stream_transfer_task(shared_ptr<stream_session> session, UUID cf_id);
~stream_transfer_task();
void add_transfer_file(stream_detail detail);

4
streaming/stream_writer.hh
View File

@@ -36,7 +36,7 @@ protected:
sstables::sstable& sstable;
std::map<int64_t, int64_t> sections;
//StreamRateLimiter limiter;
std::shared_ptr<stream_session> session;
shared_ptr<stream_session> session;
#if 0
private OutputStream compressedOutput;
// allocate buffer to use for transfers only once
@@ -44,7 +44,7 @@ protected:
#endif
public:
stream_writer(sstables::sstable& sstable_, std::map<int64_t, int64_t> sections_, std::shared_ptr<stream_session> session_)
stream_writer(sstables::sstable& sstable_, std::map<int64_t, int64_t> sections_, shared_ptr<stream_session> session_)
: sstable(sstable_)
, sections(std::move(sections_))
, session(session_) {