mirrors/scylladb
1
0
Fork 0
mirror of https://github.com/scylladb/scylladb.git synced 2026-04-27 11:55:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'shared_ptr'

Browse Source 
Split shared_ptr into a traditional polymorphic shared_ptr implementation,
and a lighter-weight, non polymorphic lw_shared_ptr.
This commit is contained in:
Avi Kivity
2015-01-05 10:40:09 +02:00
parent b7a83903b5 c7d7494a65
commit f2e25b8442
13 changed files with 321 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
apps/memcached/memcache.cc
View File

@@ -63,13 +63,13 @@ public:
};
class subdevice {
foreign_ptr<shared_ptr<flashcache::devfile>> _dev;
foreign_ptr<lw_shared_ptr<flashcache::devfile>> _dev;
uint64_t _offset;
uint64_t _end;
std::queue<block> _free_blocks;
semaphore _par = { 1000 };
public:
subdevice(foreign_ptr<shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length)
subdevice(foreign_ptr<lw_shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length)
: _dev(std::move(dev))
, _offset(offset)
, _end(offset + length)
@@ -417,7 +417,7 @@ private:
item_lru_list _lru;
cache_stats _stats;
public:
void do_setup(foreign_ptr<shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {}
void do_setup(foreign_ptr<lw_shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {}
void do_erase(item_type& item_ref) {
_lru.erase(_lru.iterator_to(item_ref));
@@ -462,7 +462,7 @@ public:
return subdev_ref;
}
void do_setup(foreign_ptr<shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {
void do_setup(foreign_ptr<lw_shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {
_subdev = std::make_unique<flashcache::subdevice>(std::move(dev), offset, length);
}
@@ -593,7 +593,7 @@ future<> flashcache_cache_base::load_item_data(boost::intrusive_ptr<item_type> i
assert(item->get_state() == item_state::DISK);
flashcache::subdevice& subdev = this->get_subdevice();
auto sem = make_shared<semaphore>({ 0 });
auto sem = make_lw_shared<semaphore>({ 0 });
auto& item_data = item->data();
auto item_size = item->size();
auto blocks_to_load = item->used_blocks_size();
@@ -659,7 +659,7 @@ future<> flashcache_cache_base::store_item_data(boost::intrusive_ptr<item_type>
assert(item->get_state() == item_state::TO_MEM_DISK);
flashcache::subdevice& subdev = this->get_subdevice();
auto sem = make_shared<semaphore>({ 0 });
auto sem = make_lw_shared<semaphore>({ 0 });
auto& item_data = item->data();
auto item_size = item->size();
auto blocks_to_store = (item_size + (flashcache::block_size - 1)) / flashcache::block_size;
@@ -878,7 +878,7 @@ public:
_flush_timer.set_callback([this] { flush_all(); });
}
future<> setup(foreign_ptr<shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {
future<> setup(foreign_ptr<lw_shared_ptr<flashcache::devfile>> dev, uint64_t offset, uint64_t length) {
this->do_setup(std::move(dev), offset, length);
return make_ready_future<>();
}
@@ -1058,7 +1058,7 @@ public:
return decr<remote_origin_tag>(key, delta);
}
std::pair<unsigned, foreign_ptr<shared_ptr<std::string>>> print_hash_stats() {
std::pair<unsigned, foreign_ptr<lw_shared_ptr<std::string>>> print_hash_stats() {
static constexpr unsigned bits = sizeof(size_t) * 8;
size_t histo[bits + 1] {};
size_t max_size = 0;
@@ -1096,7 +1096,7 @@ public:
}
ss << histo[i] << "\n";
}
return {engine.cpu_id(), make_foreign(make_shared(ss.str()))};
return {engine.cpu_id(), make_foreign(make_lw_shared<std::string>(ss.str()))};
}
future<> stop() { return make_ready_future<>(); }
@@ -1195,7 +1195,7 @@ public:
}
future<> print_hash_stats(output_stream<char>& out) {
return _peers.map_reduce([&out] (std::pair<unsigned, foreign_ptr<shared_ptr<std::string>>> data) mutable {
return _peers.map_reduce([&out] (std::pair<unsigned, foreign_ptr<lw_shared_ptr<std::string>>> data) mutable {
return out.write("=== CPU " + std::to_string(data.first) + " ===\r\n")
.then([&out, str = std::move(data.second)] {
return out.write(*str);
@@ -1667,7 +1667,7 @@ public:
auto request_id = hdr._request_id;
auto in = as_input_stream(std::move(p));
auto conn = make_shared<connection>(dgram.get_src(), request_id, std::move(in),
auto conn = make_lw_shared<connection>(dgram.get_src(), request_id, std::move(in),
_max_datagram_size - sizeof(header), _cache, _system_stats);
if (hdr._n != 1 || hdr._sequence_number != 0) {
@@ -1693,7 +1693,7 @@ public:
template <bool WithFlashCache>
class tcp_server {
private:
shared_ptr<server_socket> _listener;
lw_shared_ptr<server_socket> _listener;
sharded_cache<WithFlashCache>& _cache;
distributed<system_stats>& _system_stats;
uint16_t _port;
@@ -1732,7 +1732,7 @@ public:
_listener = engine.listen(make_ipv4_address({_port}), lo);
keep_doing([this] {
return _listener->accept().then([this] (connected_socket fd, socket_address addr) mutable {
auto conn = make_shared<connection>(std::move(fd), addr, _cache, _system_stats);
auto conn = make_lw_shared<connection>(std::move(fd), addr, _cache, _system_stats);
do_until([conn] { return conn->_in.eof(); }, [this, conn] {
return conn->_proto.handle(conn->_in, conn->_out).then([conn] {
return conn->_out.flush();
@@ -1818,7 +1818,7 @@ int start_instance(int ac, char** av) {
if (WithFlashCache) {
auto device_path = config["device"].as<std::string>();
return engine.open_file_dma(device_path).then([&] (file f) {
auto dev = make_shared<flashcache::devfile>({std::move(f)});
auto dev = make_lw_shared<flashcache::devfile>({std::move(f)});
return dev->f().stat().then([&, dev] (struct stat st) mutable {
assert(S_ISBLK(st.st_mode));
return dev->f().size().then([&, dev] (size_t device_size) mutable {

6
core/future-util.hh
View File

@@ -156,7 +156,7 @@ template <typename T>
struct reducer_with_get_traits {
using result_type = decltype(std::declval<T>().get());
using future_type = future<result_type>;
static future_type maybe_call_get(future<> f, shared_ptr<T> r) {
static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
return f.then([r = std::move(r)] () mutable {
return make_ready_future<result_type>(std::move(*r).get());
});
@@ -166,7 +166,7 @@ struct reducer_with_get_traits {
template <typename T, typename V = void>
struct reducer_traits {
using future_type = future<>;
static future_type maybe_call_get(future<> f, shared_ptr<T> r) {
static future_type maybe_call_get(future<> f, lw_shared_ptr<T> r) {
return f.then([r = std::move(r)] {});
}
};
@@ -188,7 +188,7 @@ auto
map_reduce(Iterator begin, Iterator end, Mapper&& mapper, Reducer&& r)
-> typename reducer_traits<Reducer>::future_type
{
auto r_ptr = make_shared(std::forward<Reducer>(r));
auto r_ptr = make_lw_shared(std::forward<Reducer>(r));
future<> ret = make_ready_future<>();
while (begin != end) {
ret = mapper(*begin++).then([ret = std::move(ret), r_ptr] (auto value) mutable {

2
core/memory.cc
View File

@@ -742,7 +742,7 @@ void configure(std::vector<resource::memory> m,
if (hugetlbfs_path) {
// std::function is copyable, but file_desc is not, so we must use
// a shared_ptr to allow sys_alloc to be copied around
auto fdp = make_shared<file_desc>(file_desc::temporary(*hugetlbfs_path));
auto fdp = make_lw_shared<file_desc>(file_desc::temporary(*hugetlbfs_path));
sys_alloc = [fdp] (optional<void*> where, size_t how_much) {
return allocate_hugetlbfs_memory(*fdp, where, how_much);
};

2
core/scollectd.cc
View File

@@ -286,7 +286,7 @@ private:
typedef value_list_map::iterator iterator;
typedef std::tuple<iterator, cpwriter> context;
auto ctxt = make_shared<context>();
auto ctxt = make_lw_shared<context>();
// note we're doing this unsynced since we assume
// all registrations to this instance will be done on the

320
core/shared_ptr.hh
View File

@@ -8,20 +8,57 @@
#include <utility>
#include <type_traits>
// This header defines two shared pointer facilities, lw_shared_ptr<> and
// shared_ptr<>, both modeled after std::shared_ptr<>.
//
// Unlike std::shared_ptr<>, neither of these implementations are thread
// safe, and two pointers sharing the same object must not be used in
// different threads.
//
// lw_shared_ptr<> is the more lightweight variant, with a lw_shared_ptr<>
// occupying just one machine word, and adding just one word to the shared
// object. However, it does not support polymorphism.
//
// shared_ptr<> is more expensive, with a pointer occupying two machine
// words, and with two words of overhead in the shared object. In return,
// it does support polymorphism.
//
// Both variants support shared_from_this() via enable_shared_from_this<>
// and lw_enable_shared_from_this<>().
//
template <typename T>
class lw_shared_ptr;
template <typename T>
class shared_ptr;
template <typename T>
class enable_lw_shared_from_this;
template <typename T>
class enable_shared_from_this;
template <typename T, typename... A>
lw_shared_ptr<T> make_lw_shared(A&&... a);
template <typename T>
lw_shared_ptr<T> make_lw_shared(T&& a);
template <typename T>
lw_shared_ptr<T> make_lw_shared(T& a);
template <typename T, typename... A>
shared_ptr<T> make_shared(A&&... a);
template <typename T>
shared_ptr<T> make_shared(T&& a);
template <typename T, typename U>
shared_ptr<T> static_pointer_cast(const shared_ptr<U>& p);
template <typename T>
shared_ptr<T> make_shared(T& a);
template <typename T, typename U>
shared_ptr<T> dynamic_pointer_cast(const shared_ptr<U>& p);
template <typename T, typename U>
shared_ptr<T> const_pointer_cast(const shared_ptr<U>& p);
// We want to support two use cases for shared_ptr<T>:
//
@@ -43,18 +80,18 @@ shared_ptr<T> make_shared(T& a);
// CRTP from this to enable shared_from_this:
template <typename T>
class enable_shared_from_this {
class enable_lw_shared_from_this {
long _count = 0;
using ctor = T;
T* to_value() { return static_cast<T*>(this); }
T* to_internal_object() { return static_cast<T*>(this); }
protected:
enable_shared_from_this& operator=(const enable_shared_from_this&) { return *this; }
enable_shared_from_this& operator=(enable_shared_from_this&&) { return *this; }
enable_lw_shared_from_this& operator=(const enable_lw_shared_from_this&) { return *this; }
enable_lw_shared_from_this& operator=(enable_lw_shared_from_this&&) { return *this; }
public:
shared_ptr<T> shared_from_this();
lw_shared_ptr<T> shared_from_this();
template <typename X>
friend class shared_ptr;
friend class lw_shared_ptr;
};
template <typename T>
@@ -71,64 +108,64 @@ struct shared_ptr_no_esft {
template <typename... A>
shared_ptr_no_esft(A&&... a) : _value(std::forward<A>(a)...) {}
template <typename X>
friend class shared_ptr;
friend class lw_shared_ptr;
};
template <typename T>
using shared_ptr_impl
= std::conditional_t<
std::is_base_of<enable_shared_from_this<T>, T>::value,
enable_shared_from_this<T>,
std::is_base_of<enable_lw_shared_from_this<T>, T>::value,
enable_lw_shared_from_this<T>,
shared_ptr_no_esft<T>
>;
template <typename T>
class shared_ptr {
class lw_shared_ptr {
mutable shared_ptr_impl<T>* _p = nullptr;
private:
shared_ptr(shared_ptr_impl<T>* p) : _p(p) {
lw_shared_ptr(shared_ptr_impl<T>* p) : _p(p) {
if (_p) {
++_p->_count;
}
}
template <typename... A>
static shared_ptr make(A&&... a) {
return shared_ptr(new typename shared_ptr_impl<T>::ctor(std::forward<A>(a)...));
static lw_shared_ptr make(A&&... a) {
return lw_shared_ptr(new typename shared_ptr_impl<T>::ctor(std::forward<A>(a)...));
}
public:
using element_type = T;
shared_ptr() = default;
shared_ptr(const shared_ptr& x) : _p(x._p) {
lw_shared_ptr() = default;
lw_shared_ptr(const lw_shared_ptr& x) : _p(x._p) {
if (_p) {
++_p->_count;
}
}
shared_ptr(shared_ptr&& x) : _p(x._p) {
lw_shared_ptr(lw_shared_ptr&& x) : _p(x._p) {
x._p = nullptr;
}
~shared_ptr() {
~lw_shared_ptr() {
if (_p && !--_p->_count) {
delete _p->to_internal_object();
}
}
shared_ptr& operator=(const shared_ptr& x) {
lw_shared_ptr& operator=(const lw_shared_ptr& x) {
if (_p != x._p) {
this->~shared_ptr();
new (this) shared_ptr(x);
this->~lw_shared_ptr();
new (this) lw_shared_ptr(x);
}
return *this;
}
shared_ptr& operator=(shared_ptr&& x) {
lw_shared_ptr& operator=(lw_shared_ptr&& x) {
if (_p != x._p) {
this->~shared_ptr();
new (this) shared_ptr(std::move(x));
this->~lw_shared_ptr();
new (this) lw_shared_ptr(std::move(x));
}
return *this;
}
shared_ptr& operator=(T&& x) {
this->~shared_ptr();
new (this) shared_ptr(make_shared<T>(std::move(x)));
lw_shared_ptr& operator=(T&& x) {
this->~lw_shared_ptr();
new (this) lw_shared_ptr(make_lw_shared<T>(std::move(x)));
return *this;
}
@@ -144,8 +181,8 @@ public:
}
}
operator shared_ptr<const T>() const {
return shared_ptr<const T>(_p);
operator lw_shared_ptr<const T>() const {
return lw_shared_ptr<const T>(_p);
}
explicit operator bool() const {
@@ -157,34 +194,235 @@ public:
}
template <typename X, typename... A>
friend shared_ptr<X> make_shared(A&&...);
friend lw_shared_ptr<X> make_lw_shared(A&&...);
template <typename U>
friend shared_ptr<U> make_shared(U&&);
friend lw_shared_ptr<U> make_lw_shared(U&&);
template <typename U>
friend shared_ptr<U> make_shared(U&);
friend lw_shared_ptr<U> make_lw_shared(U&);
template <typename U>
friend class enable_shared_from_this;
friend class enable_lw_shared_from_this;
};
template <typename T, typename... A>
inline
shared_ptr<T> make_shared(A&&... a) {
return shared_ptr<T>::make(std::forward<A>(a)...);
lw_shared_ptr<T> make_lw_shared(A&&... a) {
return lw_shared_ptr<T>::make(std::forward<A>(a)...);
}
template <typename T>
inline
shared_ptr<T> make_shared(T&& a) {
return shared_ptr<T>::make(std::move(a));
lw_shared_ptr<T> make_lw_shared(T&& a) {
return lw_shared_ptr<T>::make(std::move(a));
}
template <typename T>
inline
shared_ptr<T> make_shared(T& a) {
return shared_ptr<T>::make(a);
lw_shared_ptr<T> make_lw_shared(T& a) {
return lw_shared_ptr<T>::make(a);
}
template <typename T>
inline
lw_shared_ptr<T>
enable_lw_shared_from_this<T>::shared_from_this() {
return lw_shared_ptr<T>(this);
}
// Polymorphic shared pointer class
struct shared_ptr_count_base {
// destructor is responsible for fully-typed deletion
virtual ~shared_ptr_count_base() {}
long count = 0;
};
template <typename T>
struct shared_ptr_count_for : shared_ptr_count_base {
T data;
template <typename... A>
shared_ptr_count_for(A&&... a) : data(std::forward<A>(a)...) {}
};
template <typename T>
class enable_shared_from_this : private shared_ptr_count_base {
public:
shared_ptr<T> shared_from_this();
template <typename U>
friend class shared_ptr;
};
template <typename T>
class shared_ptr {
mutable shared_ptr_count_base* _b = nullptr;
mutable T* _p = nullptr;
private:
explicit shared_ptr(shared_ptr_count_for<T>* b) : _b(b), _p(&b->data) {
++_b->count;
}
shared_ptr(shared_ptr_count_for<T>* b, T* p) : _b(b), _p(p) {
// test _p, not _b, since dynamic_pointer_cast<>() can zero p but not b
if (_p) {
++_b->count;
}
}
explicit shared_ptr(enable_shared_from_this<T>* p) : _b(p), _p(static_cast<T*>(p)) {
if (_b) {
++_b->count;
}
}
public:
shared_ptr() = default;
shared_ptr(const shared_ptr& x)
: _b(x._b)
, _p(x._p) {
if (_b) {
++_b->count;
}
}
shared_ptr(shared_ptr&& x)
: _b(x._b)
, _p(x._p) {
x._b = nullptr;
x._p = nullptr;
}
template <typename U, typename = std::enable_if_t<std::is_base_of<T, U>::value>>
shared_ptr(const shared_ptr<U>& x)
: _b(x._b)
, _p(x._p) {
if (_b) {
++_b->count;
}
}
template <typename U, typename = std::enable_if_t<std::is_base_of<T, U>::value>>
shared_ptr(shared_ptr<U>&& x)
: _b(x._b)
, _p(x._p) {
x._b = nullptr;
x._p = nullptr;
}
~shared_ptr() {
if (_b && !--_b->count) {
delete _b;
}
}
shared_ptr& operator=(const shared_ptr& x) {
if (this != &x) {
this->~shared_ptr();
new (this) shared_ptr(x);
}
return *this;
}
shared_ptr& operator=(shared_ptr&& x) {
if (this != &x) {
this->~shared_ptr();
new (this) shared_ptr(std::move(x));
}
return *this;
}
template <typename U, typename = std::enable_if_t<std::is_base_of<T, U>::value>>
shared_ptr& operator=(const shared_ptr<U>& x) {
if (this != &x) {
this->~shared_ptr();
new (this) shared_ptr(x);
}
return *this;
}
template <typename U, typename = std::enable_if_t<std::is_base_of<T, U>::value>>
shared_ptr& operator=(shared_ptr<U>&& x) {
if (this != &x) {
this->~shared_ptr();
new (this) shared_ptr(std::move(x));
}
return *this;
}
explicit operator bool() const {
return _p;
}
T& operator*() const{
return *_p;
}
T* operator->() const {
return _p;
}
T* get() const {
return _p;
}
template <bool esft>
struct make_helper;
template <typename U, typename... A>
friend shared_ptr<U> make_shared(A&&... a);
template <typename V, typename U>
friend shared_ptr<V> static_pointer_cast(const shared_ptr<U>& p);
template <typename V, typename U>
friend shared_ptr<V> dynamic_pointer_cast(const shared_ptr<U>& p);
template <typename V, typename U>
friend shared_ptr<V> const_pointer_cast(const shared_ptr<U>& p);
template <bool esft, typename... A>
static shared_ptr make(A&&... a);
template <typename U>
friend class enable_shared_from_this;
template <typename U, bool esft>
friend struct shared_ptr_make_helper;
};
template <typename U, bool esft>
struct shared_ptr_make_helper;
template <typename T>
struct shared_ptr_make_helper<T, false> {
template <typename... A>
static shared_ptr<T> make(A&&... a) {
return shared_ptr<T>(new shared_ptr_count_for<T>(std::forward<A>(a)...));
}
};
template <typename T>
struct shared_ptr_make_helper<T, true> {
template <typename... A>
static shared_ptr<T> make(A&&... a) {
return shared_ptr<T>(new T(std::forward<A>(a)...));
}
};
template <typename T, typename... A>
inline
shared_ptr<T>
make_shared(A&&... a) {
using helper = shared_ptr_make_helper<T, std::is_base_of<enable_shared_from_this<T>, T>::value>;
return helper::make(std::forward<A>(a)...);
}
template <typename T, typename U>
inline
shared_ptr<T>
static_pointer_cast(const shared_ptr<U>& p) {
return shared_ptr<T>(p->_b, static_cast<T*>(p._p));
}
template <typename T, typename U>
inline
shared_ptr<T>
dynamic_pointer_cast(const shared_ptr<U>& p) {
return shared_ptr<T>(p->_b, dynamic_cast<T*>(p._p));
}
template <typename T, typename U>
inline
shared_ptr<T>
const_pointer_cast(const shared_ptr<U>& p) {
return shared_ptr<T>(p->_b, const_cast<T*>(p._p));
}
template <typename T>

2
net/ip.cc
View File

@@ -247,7 +247,7 @@ future<> ipv4::send(ipv4_address to, ip_protocol_num proto_num, packet p) {
uint16_t remaining;
uint16_t offset;
};
auto si = make_shared<send_info>({std::move(p), remaining, offset});
auto si = make_lw_shared<send_info>({std::move(p), remaining, offset});
auto stop = [si] { return si->remaining == 0; };
auto send_frag = [this, send_pkt, si] () mutable {
auto& remaining = si->remaining;

2
net/native-stack.cc
View File

@@ -184,7 +184,7 @@ native_network_stack::listen(socket_address sa, listen_options opts) {
using namespace std::chrono_literals;
future<> native_network_stack::run_dhcp(bool is_renew, const dhcp::lease& res) {
shared_ptr<dhcp> d = make_shared<dhcp>(_inet);
lw_shared_ptr<dhcp> d = make_lw_shared<dhcp>(_inet);
// Hijack the ip-stack.
for (unsigned i = 0; i < smp::count; i++) {

12
net/tcp.hh
View File

@@ -148,7 +148,7 @@ public:
private:
class tcb;
class tcb : public enable_shared_from_this<tcb> {
class tcb : public enable_lw_shared_from_this<tcb> {
using clock_type = lowres_clock;
// Instead of tracking state through an enum, track individual
// bits of the state. This reduces duplication in state handling.
@@ -298,13 +298,13 @@ private:
friend class connection;
};
inet_type& _inet;
std::unordered_map<connid, shared_ptr<tcb>, connid_hash> _tcbs;
std::unordered_map<connid, lw_shared_ptr<tcb>, connid_hash> _tcbs;
std::unordered_map<uint16_t, listener*> _listening;
public:
class connection {
shared_ptr<tcb> _tcb;
lw_shared_ptr<tcb> _tcb;
public:
explicit connection(shared_ptr<tcb> tcbp) : _tcb(std::move(tcbp)) { _tcb->_conn = this; }
explicit connection(lw_shared_ptr<tcb> tcbp) : _tcb(std::move(tcbp)) { _tcb->_conn = this; }
connection(const connection&) = delete;
connection(connection&& x) noexcept : _tcb(std::move(x._tcb)) {
_tcb->_conn = this;
@@ -406,14 +406,14 @@ void tcp<InetTraits>::received(packet p, ipaddr from, ipaddr to) {
auto h = ntoh(*th);
auto id = connid{to, from, h.dst_port, h.src_port};
auto tcbi = _tcbs.find(id);
shared_ptr<tcb> tcbp;
lw_shared_ptr<tcb> tcbp;
if (tcbi == _tcbs.end()) {
if (h.f_syn && !h.f_ack) {
auto listener = _listening.find(id.local_port);
if (listener == _listening.end() || listener->second->_q.full()) {
return respond_with_reset(&h, id.local_ip, id.foreign_ip);
}
tcbp = make_shared<tcb>(*this, id);
tcbp = make_lw_shared<tcb>(*this, id);
listener->second->_q.push(connection(tcbp));
_tcbs.insert({id, tcbp});
}

6
net/udp.cc
View File

@@ -46,9 +46,9 @@ private:
udp_v4& _proto;
udp_v4::registration _reg;
bool _closed;
shared_ptr<udp_channel_state> _state;
lw_shared_ptr<udp_channel_state> _state;
public:
native_channel(udp_v4 &proto, udp_v4::registration reg, shared_ptr<udp_channel_state> state)
native_channel(udp_v4 &proto, udp_v4::registration reg, lw_shared_ptr<udp_channel_state> state)
: _proto(proto)
, _reg(reg)
, _closed(false)
@@ -168,7 +168,7 @@ udp_v4::make_channel(ipv4_addr addr) {
_next_anonymous_port = next_port(_next_anonymous_port);
}
auto chan_state = make_shared<udp_channel_state>(_queue_size);
auto chan_state = make_lw_shared<udp_channel_state>(_queue_size);
_channels[bind_port] = chan_state;
return udp_channel(std::make_unique<native_channel>(*this, registration(*this, bind_port), chan_state));
}

2
net/udp.hh
View File

@@ -49,7 +49,7 @@ public:
private:
static const uint16_t min_anonymous_port = 32768;
ipv4 &_inet;
std::unordered_map<uint16_t, shared_ptr<udp_channel_state>> _channels;
std::unordered_map<uint16_t, lw_shared_ptr<udp_channel_state>> _channels;
int _queue_size = default_queue_size;
uint16_t _next_anonymous_port = min_anonymous_port;
private:

12
tests/memcached/test_ascii_parser.cc
View File

@@ -33,11 +33,11 @@ static auto make_input_stream(packet&& p) {
}
static auto parse(packet&& p) {
auto is = make_shared(make_input_stream(std::move(p)));
auto parser = make_shared<parser_type>();
auto is = make_lw_shared<input_stream<char>>(make_input_stream(std::move(p)));
auto parser = make_lw_shared<parser_type>();
parser->init();
return is->consume(*parser).then([is, parser] {
return make_ready_future<shared_ptr<parser_type>>(parser);
return make_ready_future<lw_shared_ptr<parser_type>>(parser);
});
}
@@ -259,7 +259,7 @@ SEASTAR_TEST_CASE(test_catches_errors_in_get) {
SEASTAR_TEST_CASE(test_parser_returns_eof_state_when_no_command_follows) {
return for_each_fragment_size([] (auto make_packet) {
auto p = make_shared<parser_type>();
auto is = make_shared(make_input_stream(make_packet({"get key\r\n"})));
auto is = make_shared<input_stream<char>>(make_input_stream(make_packet({"get key\r\n"})));
p->init();
return is->consume(*p).then([p] {
BOOST_REQUIRE(p->_state == parser_type::state::cmd_get);
@@ -275,7 +275,7 @@ SEASTAR_TEST_CASE(test_parser_returns_eof_state_when_no_command_follows) {
SEASTAR_TEST_CASE(test_incomplete_command_is_an_error) {
return for_each_fragment_size([] (auto make_packet) {
auto p = make_shared<parser_type>();
auto is = make_shared(make_input_stream(make_packet({"get"})));
auto is = make_shared<input_stream<char>>(make_input_stream(make_packet({"get"})));
p->init();
return is->consume(*p).then([p] {
BOOST_REQUIRE(p->_state == parser_type::state::error);
@@ -291,7 +291,7 @@ SEASTAR_TEST_CASE(test_incomplete_command_is_an_error) {
SEASTAR_TEST_CASE(test_multiple_requests_in_one_stream) {
return for_each_fragment_size([] (auto make_packet) {
auto p = make_shared<parser_type>();
auto is = make_shared(make_input_stream(make_packet({"set key1 1 1 5\r\ndata1\r\nset key2 2 2 6\r\ndata2+\r\n"})));
auto is = make_shared<input_stream<char>>(make_input_stream(make_packet({"set key1 1 1 5\r\ndata1\r\nset key2 2 2 6\r\ndata2+\r\n"})));
p->init();
return is->consume(*p).then([p] {
BOOST_REQUIRE(p->_state == parser_type::state::cmd_set);

8
tests/output_stream_test.cc
View File

@@ -37,8 +37,8 @@ struct stream_maker {
return std::move(*this);
}
shared_ptr<output_stream<char>> operator()(data_sink sink) {
return make_shared<output_stream<char>>(std::move(sink), _size, _trim);
lw_shared_ptr<output_stream<char>> operator()(data_sink sink) {
return make_lw_shared<output_stream<char>>(std::move(sink), _size, _trim);
}
};
@@ -47,8 +47,8 @@ future<> assert_split(StreamConstructor stream_maker, std::initializer_list<T> w
std::vector<std::string> expected_split) {
static int i = 0;
BOOST_TEST_MESSAGE("checking split: " << i++);
auto sh_write_calls = make_shared(std::move(write_calls));
auto sh_expected_splits = make_shared(std::move(expected_split));
auto sh_write_calls = make_lw_shared<std::initializer_list<T>>(std::move(write_calls));
auto sh_expected_splits = make_lw_shared<std::vector<std::string>>(std::move(expected_split));
auto v = make_shared<std::vector<packet>>();
auto out = stream_maker(data_sink(std::make_unique<vector_data_sink>(*v)));

2
tests/udp_zero_copy.cc
View File

@@ -85,7 +85,7 @@ public:
keep_doing([this] {
return _chan.receive().then([this] (udp_datagram dgram) {
auto chunk = next_chunk();
shared_ptr<sstring> item;
lw_shared_ptr<sstring> item;
if (_copy) {
_packets.clear();
_out->write(chunk, _chunk_size);