#define dout_subsys ceph_subsys_ms
#include "common/debug.h"
+using ceph::cref_t;
+using ceph::ref_t;
/*******************
* DispatchQueue
void DispatchQueue::discard_queue(uint64_t id) {
std::lock_guard l{lock};
- list<QueueItem> removed;
+ std::list<QueueItem> removed;
mqueue.remove_by_class(id, &removed);
- for (list<QueueItem>::iterator i = removed.begin();
- i != removed.end();
- ++i) {
+ for (auto i = removed.begin(); i != removed.end(); ++i) {
ceph_assert(!(i->is_code())); // We don't discard id 0, ever!
const ref_t<Message>& m = i->get_message();
remove_arrival(m);
class QueueItem {
int type;
ConnectionRef con;
- ref_t<Message> m;
+ ceph::ref_t<Message> m;
public:
- explicit QueueItem(const ref_t<Message>& m) : type(-1), con(0), m(m) {}
+ explicit QueueItem(const ceph::ref_t<Message>& m) : type(-1), con(0), m(m) {}
QueueItem(int type, Connection *con) : type(type), con(con), m(0) {}
bool is_code() const {
return type != -1;
ceph_assert(is_code());
return type;
}
- const ref_t<Message>& get_message() {
+ const ceph::ref_t<Message>& get_message() {
ceph_assert(!is_code());
return m;
}
return con.get();
}
};
-
+
CephContext *cct;
Messenger *msgr;
mutable ceph::mutex lock;
PrioritizedQueue<QueueItem, uint64_t> mqueue;
- std::set<pair<double, ref_t<Message>>> marrival;
- map<ref_t<Message>, decltype(marrival)::iterator> marrival_map;
- void add_arrival(const ref_t<Message>& m) {
+ std::set<std::pair<double, ceph::ref_t<Message>>> marrival;
+ std::map<ceph::ref_t<Message>, decltype(marrival)::iterator> marrival_map;
+ void add_arrival(const ceph::ref_t<Message>& m) {
marrival_map.insert(
make_pair(
m,
- marrival.insert(make_pair(m->get_recv_stamp(), m)).first
+ marrival.insert(std::make_pair(m->get_recv_stamp(), m)).first
)
);
}
- void remove_arrival(const ref_t<Message>& m) {
+ void remove_arrival(const ceph::ref_t<Message>& m) {
auto it = marrival_map.find(m);
ceph_assert(it != marrival_map.end());
marrival.erase(it->second);
}
std::atomic<uint64_t> next_id;
-
+
enum { D_CONNECT = 1, D_ACCEPT, D_BAD_REMOTE_RESET, D_BAD_RESET, D_CONN_REFUSED, D_NUM_CODES };
/**
ceph::mutex local_delivery_lock;
ceph::condition_variable local_delivery_cond;
bool stop_local_delivery;
- std::queue<pair<ref_t<Message>, int>> local_messages;
+ std::queue<std::pair<ceph::ref_t<Message>, int>> local_messages;
class LocalDeliveryThread : public Thread {
DispatchQueue *dq;
public:
}
} local_delivery_thread;
- uint64_t pre_dispatch(const ref_t<Message>& m);
- void post_dispatch(const ref_t<Message>& m, uint64_t msize);
+ uint64_t pre_dispatch(const ceph::ref_t<Message>& m);
+ void post_dispatch(const ceph::ref_t<Message>& m, uint64_t msize);
public:
Throttle dispatch_throttler;
bool stop;
- void local_delivery(const ref_t<Message>& m, int priority);
+ void local_delivery(const ceph::ref_t<Message>& m, int priority);
void local_delivery(Message* m, int priority) {
- return local_delivery(ref_t<Message>(m, false), priority); /* consume ref */
+ return local_delivery(ceph::ref_t<Message>(m, false), priority); /* consume ref */
}
void run_local_delivery();
cond.notify_all();
}
- bool can_fast_dispatch(const cref_t<Message> &m) const;
- void fast_dispatch(const ref_t<Message>& m);
+ bool can_fast_dispatch(const ceph::cref_t<Message> &m) const;
+ void fast_dispatch(const ceph::ref_t<Message>& m);
void fast_dispatch(Message* m) {
- return fast_dispatch(ref_t<Message>(m, false)); /* consume ref */
+ return fast_dispatch(ceph::ref_t<Message>(m, false)); /* consume ref */
}
- void fast_preprocess(const ref_t<Message>& m);
- void enqueue(const ref_t<Message>& m, int priority, uint64_t id);
+ void fast_preprocess(const ceph::ref_t<Message>& m);
+ void enqueue(const ceph::ref_t<Message>& m, int priority, uint64_t id);
void enqueue(Message* m, int priority, uint64_t id) {
- return enqueue(ref_t<Message>(m, false), priority, id); /* consume ref */
+ return enqueue(ceph::ref_t<Message>(m, false), priority, id); /* consume ref */
}
void discard_queue(uint64_t id);
void discard_local();
void shutdown();
bool is_started() const {return dispatch_thread.is_started();}
- DispatchQueue(CephContext *cct, Messenger *msgr, string &name)
+ DispatchQueue(CephContext *cct, Messenger *msgr, std::string &name)
: cct(cct), msgr(msgr),
lock(ceph::make_mutex("Messenger::DispatchQueue::lock" + name)),
mqueue(cct->_conf->ms_pq_max_tokens_per_priority,
local_delivery_lock(ceph::make_mutex("Messenger::DispatchQueue::local_delivery_lock" + name)),
stop_local_delivery(false),
local_delivery_thread(this),
- dispatch_throttler(cct, string("msgr_dispatch_throttler-") + name,
+ dispatch_throttler(cct, std::string("msgr_dispatch_throttler-") + name,
cct->_conf->ms_dispatch_throttle_bytes),
stop(false)
{}
}
}
-void Message::dump(Formatter *f) const
+void Message::dump(ceph::Formatter *f) const
{
- stringstream ss;
+ std::stringstream ss;
print(ss);
f->dump_string("summary", ss.str());
}
}
// make message
- ref_t<Message> m;
+ ceph::ref_t<Message> m;
int type = header.type;
switch (type) {
// -- with payload --
+ using ceph::make_message;
+
case MSG_PGSTATS:
m = make_message<MPGStats>();
break;
try {
m->decode_payload();
}
- catch (const buffer::error &e) {
+ catch (const ceph::buffer::error &e) {
if (cct) {
lderr(cct) << "failed to decode message of type " << type
<< " v" << header.version
return m.detach();
}
-void Message::encode_trace(bufferlist &bl, uint64_t features) const
+void Message::encode_trace(ceph::bufferlist &bl, uint64_t features) const
{
using ceph::encode;
auto p = trace.get_info();
encode(*p, bl);
}
-void Message::decode_trace(bufferlist::const_iterator &p, bool create)
+void Message::decode_trace(ceph::bufferlist::const_iterator &p, bool create)
{
blkin_trace_info info = {};
decode(info, p);
// problems, we currently always encode and decode using the old footer format that doesn't
// allow for message authentication. Eventually we should fix that. PLR
-void encode_message(Message *msg, uint64_t features, bufferlist& payload)
+void encode_message(Message *msg, uint64_t features, ceph::bufferlist& payload)
{
ceph_msg_footer_old old_footer;
msg->encode(features, MSG_CRC_ALL);
old_footer.flags = footer.flags;
encode(old_footer, payload);
+ using ceph::encode;
encode(msg->get_payload(), payload);
encode(msg->get_middle(), payload);
encode(msg->get_data(), payload);
// We've slipped in a 0 signature at this point, so any signature checking after this will
// fail. PLR
-Message *decode_message(CephContext *cct, int crcflags, bufferlist::const_iterator& p)
+Message *decode_message(CephContext *cct, int crcflags, ceph::bufferlist::const_iterator& p)
{
ceph_msg_header h;
ceph_msg_footer_old fo;
ceph_msg_footer f;
- bufferlist fr, mi, da;
+ ceph::bufferlist fr, mi, da;
decode(h, p);
decode(fo, p);
f.front_crc = fo.front_crc;
f.data_crc = fo.data_crc;
f.flags = fo.flags;
f.sig = 0;
+ using ceph::decode;
decode(fr, p);
decode(mi, p);
decode(da, p);
#include "msg/async/AsyncMessenger.h"
-Messenger *Messenger::create_client_messenger(CephContext *cct, string lname)
+Messenger *Messenger::create_client_messenger(CephContext *cct, std::string lname)
{
std::string public_msgr_type = cct->_conf->ms_public_type.empty() ? cct->_conf.get_val<std::string>("ms_type") : cct->_conf->ms_public_type;
auto nonce = get_random_nonce();
return ceph::util::generate_random_number<uint64_t>();
}
-Messenger *Messenger::create(CephContext *cct, const string &type,
- entity_name_t name, string lname,
+Messenger *Messenger::create(CephContext *cct, const std::string &type,
+ entity_name_t name, std::string lname,
uint64_t nonce, uint64_t cflags)
{
int r = -1;
*
* @param m The Message we are testing.
*/
- bool ms_can_fast_dispatch(const cref_t<Message>& m) {
+ bool ms_can_fast_dispatch(const ceph::cref_t<Message>& m) {
for (const auto &dispatcher : fast_dispatchers) {
if (dispatcher->ms_can_fast_dispatch2(m))
return true;
* @param m The Message we are fast dispatching.
* If none of our Dispatchers can handle it, ceph_abort().
*/
- void ms_fast_dispatch(const ref_t<Message> &m) {
+ void ms_fast_dispatch(const ceph::ref_t<Message> &m) {
m->set_dispatch_stamp(ceph_clock_now());
for (const auto &dispatcher : fast_dispatchers) {
if (dispatcher->ms_can_fast_dispatch2(m)) {
ceph_abort();
}
void ms_fast_dispatch(Message *m) {
- return ms_fast_dispatch(ref_t<Message>(m, false)); /* consume ref */
+ return ms_fast_dispatch(ceph::ref_t<Message>(m, false)); /* consume ref */
}
/**
*
*/
- void ms_fast_preprocess(const ref_t<Message> &m) {
+ void ms_fast_preprocess(const ceph::ref_t<Message> &m) {
for (const auto &dispatcher : fast_dispatchers) {
dispatcher->ms_fast_preprocess2(m);
}
*
* @param m The Message to deliver.
*/
- void ms_deliver_dispatch(const ref_t<Message> &m) {
+ void ms_deliver_dispatch(const ceph::ref_t<Message> &m) {
m->set_dispatch_stamp(ceph_clock_now());
for (const auto &dispatcher : dispatchers) {
if (dispatcher->ms_dispatch2(m))
ceph_assert(!cct->_conf->ms_die_on_unhandled_msg);
}
void ms_deliver_dispatch(Message *m) {
- return ms_deliver_dispatch(ref_t<Message>(m, false)); /* consume ref */
+ return ms_deliver_dispatch(ceph::ref_t<Message>(m, false)); /* consume ref */
}
/**
* Notify each Dispatcher of a new Connection. Call
void QueueStrategy::entry(QSThread *thrd)
{
for (;;) {
- ref_t<Message> m;
+ ceph::ref_t<Message> m;
std::unique_lock l{lock};
for (;;) {
if (! mqueue.empty()) {
- m = ref_t<Message>(&mqueue.front(), false);
+ m = ceph::ref_t<Message>(&mqueue.front(), false);
mqueue.pop_front();
break;
}
std::lock_guard l{lock};
threads.reserve(n_threads);
for (int ix = 0; ix < n_threads; ++ix) {
- string thread_name = "ms_qs_";
+ std::string thread_name = "ms_qs_";
thread_name.append(std::to_string(ix));
auto thrd = std::make_unique<QSThread>(this);
thrd->create(thread_name.c_str());
#define dout_subsys ceph_subsys_ms
#undef dout_prefix
#define dout_prefix _conn_prefix(_dout)
-ostream& AsyncConnection::_conn_prefix(std::ostream *_dout) {
+std::ostream& AsyncConnection::_conn_prefix(std::ostream *_dout) {
return *_dout << "-- " << async_msgr->get_myaddrs() << " >> "
<< *peer_addrs << " conn(" << this
<< (msgr2 ? " msgr2=" : " legacy=")
if (!delay_state) {
async_msgr->cct->_conf.with_val<std::string>(
"ms_inject_delay_type",
- [this](const string& s) {
- if (s.find(ceph_entity_type_name(peer_type)) != string::npos) {
+ [this](const std::string& s) {
+ if (s.find(ceph_entity_type_name(peer_type)) != std::string::npos) {
ldout(msgr->cct, 1) << __func__ << " setting up a delay queue"
<< dendl;
delay_state = new DelayedDelivery(async_msgr, center, dispatch_queue,
return nread;
}
-ssize_t AsyncConnection::write(bufferlist &bl,
+ssize_t AsyncConnection::write(ceph::buffer::list &bl,
std::function<void(ssize_t)> callback,
bool more) {
ssize_t read_until(unsigned needed, char *p);
ssize_t read_bulk(char *buf, unsigned len);
- ssize_t write(bufferlist &bl, std::function<void(ssize_t)> callback,
+ ssize_t write(ceph::buffer::list &bl, std::function<void(ssize_t)> callback,
bool more=false);
ssize_t _try_send(bool more=false);
public:
void maybe_start_delay_thread();
- ostream& _conn_prefix(std::ostream *_dout);
+ std::ostream& _conn_prefix(std::ostream *_dout);
bool is_connected() override;
DispatchQueue *dispatch_queue;
// lockfree, only used in own thread
- bufferlist outgoing_bl;
+ ceph::buffer::list outgoing_bl;
bool open_write = false;
std::mutex write_lock;
uint32_t recv_max_prefetch;
uint32_t recv_start;
uint32_t recv_end;
- set<uint64_t> register_time_events; // need to delete it if stop
+ std::set<uint64_t> register_time_events; // need to delete it if stop
ceph::coarse_mono_clock::time_point last_connect_started;
ceph::coarse_mono_clock::time_point last_active;
ceph::mono_clock::time_point recv_start_time;
#define dout_subsys ceph_subsys_ms
#undef dout_prefix
#define dout_prefix _prefix(_dout, this)
-static ostream& _prefix(std::ostream *_dout, AsyncMessenger *m) {
+static std::ostream& _prefix(std::ostream *_dout, AsyncMessenger *m) {
return *_dout << "-- " << m->get_myaddrs() << " ";
}
-static ostream& _prefix(std::ostream *_dout, Processor *p) {
+static std::ostream& _prefix(std::ostream *_dout, Processor *p) {
return *_dout << " Processor -- ";
}
listen_handler(new C_processor_accept(this)) {}
int Processor::bind(const entity_addrvec_t &bind_addrs,
- const set<int>& avoid_ports,
+ const std::set<int>& avoid_ports,
entity_addrvec_t* bound_addrs)
{
const auto& conf = msgr->cct->_conf;
*/
AsyncMessenger::AsyncMessenger(CephContext *cct, entity_name_t name,
- const std::string &type, string mname, uint64_t _nonce)
+ const std::string &type, std::string mname, uint64_t _nonce)
: SimplePolicyMessenger(cct, name),
dispatch_queue(cct, this, mname),
nonce(_nonce)
lock.unlock();
// bind to a socket
- set<int> avoid_ports;
+ std::set<int> avoid_ports;
entity_addrvec_t bound_addrs;
unsigned i = 0;
for (auto &&p : processors) {
return 0;
}
-int AsyncMessenger::rebind(const set<int>& avoid_ports)
+int AsyncMessenger::rebind(const std::set<int>& avoid_ports)
{
ldout(cct,1) << __func__ << " rebind avoid " << avoid_ports << dendl;
ceph_assert(did_bind);
entity_addrvec_t bound_addrs;
entity_addrvec_t bind_addrs = get_myaddrs();
- set<int> new_avoid(avoid_ports);
+ std::set<int> new_avoid(avoid_ports);
for (auto& a : bind_addrs.v) {
new_avoid.insert(a.get_port());
a.set_port(0);
*/
class Processor {
AsyncMessenger *msgr;
- NetHandler net;
+ ceph::NetHandler net;
Worker *worker;
- vector<ServerSocket> listen_sockets;
+ std::vector<ServerSocket> listen_sockets;
EventCallbackRef listen_handler;
class C_processor_accept;
void stop();
int bind(const entity_addrvec_t &bind_addrs,
- const set<int>& avoid_ports,
+ const std::set<int>& avoid_ports,
entity_addrvec_t* bound_addrs);
void start();
void accept();
* be a value that will be repeated if the daemon restarts.
*/
AsyncMessenger(CephContext *cct, entity_name_t name, const std::string &type,
- string mname, uint64_t _nonce);
+ std::string mname, uint64_t _nonce);
/**
* Destroy the AsyncMessenger. Pretty simple since all the work is done
}
int bind(const entity_addr_t& bind_addr) override;
- int rebind(const set<int>& avoid_ports) override;
+ int rebind(const std::set<int>& avoid_ports) override;
int client_bind(const entity_addr_t& bind_addr) override;
int bindv(const entity_addrvec_t& bind_addrs) override;
*
* These are not yet in the conns map.
*/
- set<AsyncConnectionRef> accepting_conns;
+ s
td::set<AsyncConnectionRef> accepting_conns;
/// anonymous outgoing connections
- set<AsyncConnectionRef> anon_conns;
+ s
td::set<AsyncConnectionRef> anon_conns;
/**
* list of connection are closed which need to be clean up
* AsyncConnection in this set.
*/
ceph::mutex deleted_lock = ceph::make_mutex("AsyncMessenger::deleted_lock");
- set<AsyncConnectionRef> deleted_conns;
+ s
td::set<AsyncConnectionRef> deleted_conns;
EventCallbackRef reap_handler;
* about the poller. The name of the superclass is probably sufficient
* for most cases.
*/
-EventCenter::Poller::Poller(EventCenter* center, const string& name)
+EventCenter::Poller::Poller(EventCenter* center, const std::string& name)
: owner(center), poller_name(name), slot(owner->pollers.size())
{
owner->pollers.push_back(this);
slot = -1;
}
-ostream& EventCenter::_event_prefix(std::ostream *_dout)
+std::ostream& EventCenter::_event_prefix(std::ostream *_dout)
{
return *_dout << "Event(" << this << " nevent=" << nevent
<< " time_id=" << time_event_next_id << ").";
{
int processed = 0;
clock_type::time_point now = clock_type::now();
+ using ceph::operator <<;
ldout(cct, 30) << __func__ << " cur time is " << now << dendl;
while (!time_events.empty()) {
tv.tv_usec = timeout_microseconds % 1000000;
ldout(cct, 30) << __func__ << " wait second " << tv.tv_sec << " usec " << tv.tv_usec << dendl;
- vector<FiredFileEvent> fired_events;
+ std::vector<FiredFileEvent> fired_events;
numevents = driver->event_wait(fired_events, &tv);
auto working_start = ceph::mono_clock::now();
for (int event_id = 0; event_id < numevents; event_id++) {
if (external_num_events.load()) {
external_lock.lock();
- deque<EventCallbackRef> cur_process;
+ std::deque<EventCallbackRef> cur_process;
cur_process.swap(external_events);
external_num_events.store(0);
external_lock.unlock();
virtual int init(EventCenter *center, int nevent) = 0;
virtual int add_event(int fd, int cur_mask, int mask) = 0;
virtual int del_event(int fd, int cur_mask, int del_mask) = 0;
- virtual int event_wait(vector<FiredFileEvent> &fired_events, struct timeval *tp) = 0;
+ virtual int event_wait(std::vector<FiredFileEvent> &fired_events, struct timeval *tp) = 0;
virtual int resize_events(int newsize) = 0;
virtual bool need_wakeup() { return true; }
};
*/
class Poller {
public:
- explicit Poller(EventCenter* center, const string& pollerName);
+ explicit Poller(EventCenter* center, const std::string& pollerName);
virtual ~Poller();
/**
/// Human-readable string name given to the poller to make it
/// easy to identify for debugging. For most pollers just passing
/// in the subclass name probably makes sense.
- string poller_name;
+ std::string poller_name;
/// Index of this Poller in EventCenter::pollers. Allows deletion
/// without having to scan all the entries in pollers. -1 means
pthread_t owner = 0;
std::mutex external_lock;
std::atomic_ulong external_num_events;
- deque<EventCallbackRef> external_events;
- vector<FileEvent> file_events;
+ std::deque<EventCallbackRef> external_events;
+ std::vector<FileEvent> file_events;
EventDriver *driver;
std::multimap<clock_type::time_point, TimeEvent> time_events;
// Keeps track of all of the pollers currently defined. We don't
uint64_t time_event_next_id;
int notify_receive_fd;
int notify_send_fd;
- NetHandler net;
+ ceph::NetHandler net;
EventCallbackRef notify_handler;
unsigned center_id;
AssociatedCenters *global_centers = nullptr;
notify_receive_fd(-1), notify_send_fd(-1), net(c),
notify_handler(NULL), center_id(0) { }
~EventCenter();
- ostream& _event_prefix(std::ostream *_dout);
+ std::ostream& _event_prefix(std::ostream *_dout);
int init(int nevent, unsigned center_id, const std::string &type);
void set_owner();
return 0;
}
-int EpollDriver::event_wait(vector<FiredFileEvent> &fired_events, struct timeval *tvp)
+int EpollDriver::event_wait(std::vector<FiredFileEvent> &fired_events, struct timeval *tvp)
{
int retval, numevents = 0;
int add_event(int fd, int cur_mask, int add_mask) override;
int del_event(int fd, int cur_mask, int del_mask) override;
int resize_events(int newsize) override;
- int event_wait(vector<FiredFileEvent> &fired_events,
+ int event_wait(std::vector<FiredFileEvent> &fired_events,
struct timeval *tp) override;
};
return 0;
}
-int SelectDriver::event_wait(vector<FiredFileEvent> &fired_events, struct timeval *tvp)
+int SelectDriver::event_wait(std::vector<FiredFileEvent> &fired_events, struct timeval *tvp)
{
int retval, numevents = 0;
int add_event(int fd, int cur_mask, int add_mask) override;
int del_event(int fd, int cur_mask, int del_mask) override;
int resize_events(int newsize) override;
- int event_wait(vector<FiredFileEvent> &fired_events,
+ int event_wait(std::vector<FiredFileEvent> &fired_events,
struct timeval *tp) override;
};
#define dout_prefix *_dout << "PosixStack "
class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
- NetHandler &handler;
+ ceph::NetHandler &handler;
int _fd;
entity_addr_t sa;
bool connected;
public:
- explicit PosixConnectedSocketImpl(NetHandler &h, const entity_addr_t &sa, int f, bool connected)
+ explicit PosixConnectedSocketImpl(ceph::NetHandler &h, const entity_addr_t &sa,
+ int f, bool connected)
: handler(h), _fd(f), sa(sa), connected(connected) {}
int is_connected() override {
return (ssize_t)sent;
}
- ssize_t send(bufferlist &bl, bool more) override {
+ ssize_t send(ceph::buffer::list &bl, bool more) override {
size_t sent_bytes = 0;
auto pb = std::cbegin(bl.buffers());
uint64_t left_pbrs = bl.get_num_buffers();
}
if (sent_bytes) {
- bufferlist swapped;
+ ceph::buffer::list swapped;
if (sent_bytes < bl.length()) {
bl.splice(sent_bytes, bl.length()-sent_bytes, &swapped);
bl.swap(swapped);
};
class PosixServerSocketImpl : public ServerSocketImpl {
- NetHandler &handler;
+ ceph::NetHandler &handler;
int _fd;
public:
- explicit PosixServerSocketImpl(NetHandler &h, int f,
+ explicit PosixServerSocketImpl(ceph::NetHandler &h, int f,
const entity_addr_t& listen_addr, unsigned slot)
: ServerSocketImpl(listen_addr.get_type(), slot),
handler(h), _fd(f) {}
return 0;
}
-PosixNetworkStack::PosixNetworkStack(CephContext *c, const string &t)
+PosixNetworkStack::PosixNetworkStack(CephContext *c, const std::string &t)
: NetworkStack(c, t)
{
}
#include "Stack.h"
class PosixWorker : public Worker {
- NetHandler net;
+ ceph::NetHandler net;
void initialize() override;
public:
PosixWorker(CephContext *c, unsigned i)
};
class PosixNetworkStack : public NetworkStack {
- vector<std::thread> threads;
+ std::vector<std::thread> threads;
public:
- explicit PosixNetworkStack(CephContext *c, const string &t);
+ explicit PosixNetworkStack(CephContext *c, const std::string &t);
void spawn_worker(unsigned i, std::function<void ()> &&func) override {
threads.resize(i+1);
};
using rx_buffer_t =
- std::unique_ptr<buffer::ptr_node, buffer::ptr_node::disposer>;
+ std::unique_ptr<ceph::buffer::ptr_node, ceph::buffer::ptr_node::disposer>;
template <class C>
class CtRxNode : public Ct<C> {
#define dout_subsys ceph_subsys_ms
#undef dout_prefix
#define dout_prefix _conn_prefix(_dout)
-ostream &ProtocolV1::_conn_prefix(std::ostream *_dout) {
+std::ostream &ProtocolV1::_conn_prefix(std::ostream *_dout) {
return *_dout << "--1- " << messenger->get_myaddrs() << " >> "
<< *connection->peer_addrs
<< " conn("
using namespace std;
-static void alloc_aligned_buffer(bufferlist &data, unsigned len, unsigned off) {
+static void alloc_aligned_buffer(ceph::buffer::list &data, unsigned len, unsigned off) {
// create a buffer to read into that matches the data alignment
unsigned alloc_len = 0;
unsigned left = len;
left -= head;
}
alloc_len += left;
- bufferptr ptr(buffer::create_small_page_aligned(alloc_len));
+ ceph::bufferptr ptr(ceph::buffer::create_small_page_aligned(alloc_len));
if (head) ptr.set_offset(CEPH_PAGE_SIZE - head);
data.push_back(std::move(ptr));
}
}
void ProtocolV1::send_message(Message *m) {
- bufferlist bl;
+ ceph::buffer::list bl;
uint64_t f = connection->get_features();
// TODO: Currently not all messages supports reencode like MOSDMap, so here
}
void ProtocolV1::prepare_send_message(uint64_t features, Message *m,
- bufferlist &bl) {
+ ceph::buffer::list &bl) {
ldout(cct, 20) << __func__ << " m " << *m << dendl;
// associate message with Connection (for benefit of encode_payload)
auto start = ceph::mono_clock::now();
bool more;
do {
- bufferlist data;
+ ceph::buffer::list data;
Message *m = _get_next_outgoing(&data);
if (!m) {
break;
}
CtPtr ProtocolV1::write(CONTINUATION_TX_TYPE<ProtocolV1> &next,
- bufferlist &buffer) {
+ ceph::buffer::list &buffer) {
ssize_t r = connection->write(buffer, [&next, this](int r) {
next.setParams(r);
CONTINUATION_RUN(next);
unsigned front_len = current_header.front_len;
if (front_len) {
if (!front.length()) {
- front.push_back(buffer::create(front_len));
+ front.push_back(ceph::buffer::create(front_len));
}
return READB(front_len, front.c_str(), handle_message_front);
}
if (current_header.middle_len) {
if (!middle.length()) {
- middle.push_back(buffer::create(current_header.middle_len));
+ middle.push_back(ceph::buffer::create(current_header.middle_len));
}
return READB(current_header.middle_len, middle.c_str(),
handle_message_middle);
#if 0
// rx_buffers is broken by design... see
// http://tracker.ceph.com/issues/22480
- map<ceph_tid_t, pair<bufferlist, int> >::iterator p =
+ map<ceph_tid_t, pair<ceph::buffer::list, int> >::iterator p =
connection->rx_buffers.find(current_header.tid);
if (p != connection->rx_buffers.end()) {
ldout(cct, 10) << __func__ << " seleting rx buffer v " << p->second.second
ldout(cct, 20) << __func__ << " msg_left=" << msg_left << dendl;
if (msg_left > 0) {
- bufferptr bp = data_blp.get_current_ptr();
+ auto bp = data_blp.get_current_ptr();
unsigned read_len = std::min(bp.length(), msg_left);
return READB(read_len, bp.c_str(), handle_message_data);
return _fault();
}
- bufferptr bp = data_blp.get_current_ptr();
+ auto bp = data_blp.get_current_ptr();
unsigned read_len = std::min(bp.length(), msg_left);
ceph_assert(read_len <
static_cast<unsigned>(std::numeric_limits<int>::max()));
}
}
-ssize_t ProtocolV1::write_message(Message *m, bufferlist &bl, bool more) {
+ssize_t ProtocolV1::write_message(Message *m, ceph::buffer::list &bl, bool more) {
FUNCTRACE(cct);
ceph_assert(connection->center->in_thread());
m->set_seq(++out_seq);
return;
}
- list<pair<bufferlist, Message *> > &rq = out_q[CEPH_MSG_PRIO_HIGHEST];
+ list<pair<ceph::buffer::list, Message *> > &rq = out_q[CEPH_MSG_PRIO_HIGHEST];
out_seq -= sent.size();
while (!sent.empty()) {
Message *m = sent.back();
ldout(cct, 10) << __func__ << " " << *m << " for resend "
<< " (" << m->get_seq() << ")" << dendl;
m->clear_payload();
- rq.push_front(make_pair(bufferlist(), m));
+ rq.push_front(make_pair(ceph::buffer::list(), m));
}
}
if (out_q.count(CEPH_MSG_PRIO_HIGHEST) == 0) {
return seq;
}
- list<pair<bufferlist, Message *> > &rq = out_q[CEPH_MSG_PRIO_HIGHEST];
+ list<pair<ceph::buffer::list, Message *> > &rq = out_q[CEPH_MSG_PRIO_HIGHEST];
uint64_t count = out_seq;
while (!rq.empty()) {
- pair<bufferlist, Message *> p = rq.front();
+ pair<ceph::buffer::list, Message *> p = rq.front();
if (p.second->get_seq() == 0 || p.second->get_seq() > seq) break;
ldout(cct, 10) << __func__ << " " << *(p.second) << " for resend seq "
<< p.second->get_seq() << " <= " << seq << ", discarding"
(*p)->put();
}
sent.clear();
- for (map<int, list<pair<bufferlist, Message *> > >::iterator p =
+ for (map<int, list<pair<ceph::buffer::list, Message *> > >::iterator p =
out_q.begin();
p != out_q.end(); ++p) {
- for (list<pair<bufferlist, Message *> >::iterator r = p->second.begin();
+ for (list<pair<ceph::buffer::list, Message *> >::iterator r = p->second.begin();
r != p->second.end(); ++r) {
ldout(cct, 20) << __func__ << " discard " << r->second << dendl;
r->second->put();
}
}
-Message *ProtocolV1::_get_next_outgoing(bufferlist *bl) {
+Message *ProtocolV1::_get_next_outgoing(ceph::buffer::list *bl) {
Message *m = 0;
if (!out_q.empty()) {
- map<int, list<pair<bufferlist, Message *> > >::reverse_iterator it =
+ map<int, list<pair<ceph::buffer::list, Message *> > >::reverse_iterator it =
out_q.rbegin();
ceph_assert(!it->second.empty());
- list<pair<bufferlist, Message *> >::iterator p = it->second.begin();
+ list<pair<ceph::buffer::list, Message *> >::iterator p = it->second.begin();
m = p->second;
if (p->first.length() && bl) {
assert(bl->length() == 0);
ldout(cct, 20) << __func__ << dendl;
state = CONNECTING;
- bufferlist bl;
+ ceph::buffer::list bl;
bl.append(CEPH_BANNER, strlen(CEPH_BANNER));
return WRITE(bl, handle_client_banner_write);
}
ldout(cct, 20) << __func__ << dendl;
- bufferlist myaddrbl;
+ ceph::buffer::list myaddrbl;
unsigned banner_len = strlen(CEPH_BANNER);
unsigned need_len = banner_len + sizeof(ceph_entity_addr) * 2;
return READ(need_len, handle_server_banner_and_identify);
return _fault();
}
- bufferlist bl;
+ ceph::buffer::list bl;
entity_addr_t paddr, peer_addr_for_me;
bl.append(buffer + banner_len, sizeof(ceph_entity_addr) * 2);
try {
decode(paddr, p);
decode(peer_addr_for_me, p);
- } catch (const buffer::error &e) {
+ } catch (const ceph::buffer::error &e) {
lderr(cct) << __func__ << " decode peer addr failed " << dendl;
return _fault();
}
}
}
- bufferlist myaddrbl;
+ ceph::buffer::list myaddrbl;
encode(messenger->get_myaddr_legacy(), myaddrbl, 0); // legacy
return WRITE(myaddrbl, handle_my_addr_write);
}
ldout(cct, 20) << __func__ << dendl;
ceph_assert(messenger->auth_client);
- bufferlist auth_bl;
+ ceph::buffer::list auth_bl;
vector<uint32_t> preferred_modes;
if (connection->peer_type != CEPH_ENTITY_TYPE_MON ||
CEPH_MSG_CONNECT_LOSSY; // this is fyi, actually, server decides!
}
- bufferlist bl;
+ ceph::buffer::list bl;
bl.append((char *)&connect, sizeof(connect));
if (auth_bl.length()) {
bl.append(auth_bl.c_str(), auth_bl.length());
return _fault();
}
- bufferlist authorizer_reply;
+ ceph::buffer::list authorizer_reply;
authorizer_reply.append(buffer, connect_reply.authorizer_len);
if (connection->peer_type != CEPH_ENTITY_TYPE_MON ||
messenger->get_myname().type() == CEPH_ENTITY_TYPE_MON) {
auto am = auth_meta;
bool more = (connect_reply.tag == CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER);
- bufferlist auth_retry_bl;
+ ceph::buffer::list auth_retry_bl;
int r;
connection->lock.unlock();
if (more) {
<< " vs out_seq " << out_seq << dendl;
out_seq = discard_requeued_up_to(out_seq, newly_acked_seq);
- bufferlist bl;
+ ceph::buffer::list bl;
uint64_t s = in_seq;
bl.append((char *)&s, sizeof(s));
ldout(cct, 20) << __func__ << dendl;
state = ACCEPTING;
- bufferlist bl;
+ ceph::buffer::list bl;
bl.append(CEPH_BANNER, strlen(CEPH_BANNER));
return _fault();
}
- bufferlist addr_bl;
+ ceph::buffer::list addr_bl;
entity_addr_t peer_addr;
addr_bl.append(buffer + strlen(CEPH_BANNER), sizeof(ceph_entity_addr));
try {
auto ti = addr_bl.cbegin();
decode(peer_addr, ti);
- } catch (const buffer::error &e) {
+ } catch (const ceph::buffer::error &e) {
lderr(cct) << __func__ << " decode peer_addr failed " << dendl;
return _fault();
}
CtPtr ProtocolV1::wait_connect_message_auth() {
ldout(cct, 20) << __func__ << dendl;
authorizer_buf.clear();
- authorizer_buf.push_back(buffer::create(connect_msg.authorizer_len));
+ authorizer_buf.push_back(ceph::buffer::create(connect_msg.authorizer_len));
return READB(connect_msg.authorizer_len, authorizer_buf.c_str(),
handle_connect_message_auth);
}
<< dendl;
ceph_msg_connect_reply reply;
- bufferlist authorizer_reply;
+ ceph::buffer::list authorizer_reply;
// FIPS zeroization audit 20191115: this memset is not security related.
memset(&reply, 0, sizeof(reply));
authorizer_reply);
}
- bufferlist auth_bl_copy = authorizer_buf;
+ ceph::buffer::list auth_bl_copy = authorizer_buf;
auto am = auth_meta;
am->auth_method = connect_msg.authorizer_protocol;
connection->lock.unlock();
CtPtr ProtocolV1::send_connect_message_reply(char tag,
ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply) {
+ ceph::buffer::list &authorizer_reply) {
ldout(cct, 20) << __func__ << dendl;
- bufferlist reply_bl;
+ ceph::buffer::list reply_bl;
reply.tag = tag;
reply.features =
((uint64_t)connect_msg.features & connection->policy.features_supported) |
CtPtr ProtocolV1::replace(const AsyncConnectionRef& existing,
ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply) {
+ ceph::buffer::list &authorizer_reply) {
ldout(cct, 10) << __func__ << " accept replacing " << existing << dendl;
connection->inject_delay();
}
CtPtr ProtocolV1::open(ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply) {
+ ceph::buffer::list &authorizer_reply) {
ldout(cct, 20) << __func__ << dendl;
connect_seq = connect_msg.connect_seq + 1;
auth_meta->session_key,
connection->get_features()));
- bufferlist reply_bl;
+ ceph::buffer::list reply_bl;
reply_bl.append((char *)&reply, sizeof(reply));
if (reply.authorizer_len) {
enum class WriteStatus { NOWRITE, REPLACING, CANWRITE, CLOSED };
std::atomic<WriteStatus> can_write;
- std::list<Message *> sent; // the first bufferlist need to inject seq
+ std::list<Message *> sent; // the first ceph::buffer::list need to inject seq
// priority queue for outbound msgs
- std::map<int, std::list<std::pair<bufferlist, Message *>>> out_q;
+ std::map<int, std::list<std::pair<ceph::buffer::list, Message *>>> out_q;
bool keepalive;
bool write_in_progress = false;
// Open state
ceph_msg_connect connect_msg;
ceph_msg_connect_reply connect_reply;
- bufferlist authorizer_buf; // auth(orizer) payload read off the wire
- bufferlist authorizer_more; // connect-side auth retry (we added challenge)
+ ceph::buffer::list authorizer_buf; // auth(orizer) payload read off the wire
+ ceph::buffer::list authorizer_more; // connect-side auth retry (we added challenge)
utime_t backoff; // backoff time
utime_t recv_stamp;
unsigned msg_left;
uint64_t cur_msg_size;
ceph_msg_header current_header;
- bufferlist data_buf;
- bufferlist::iterator data_blp;
- bufferlist front, middle, data;
+ ceph::buffer::list data_buf;
+ ceph::buffer::list::iterator data_blp;
+ ceph::buffer::list front, middle, data;
bool replacing; // when replacing process happened, we will reply connect
// side with RETRY tag and accept side will clear replaced
void run_continuation(CtPtr pcontinuation);
CtPtr read(CONTINUATION_RX_TYPE<ProtocolV1> &next, int len,
char *buffer = nullptr);
- CtPtr write(CONTINUATION_TX_TYPE<ProtocolV1> &next,bufferlist &bl);
+ CtPtr write(CONTINUATION_TX_TYPE<ProtocolV1> &next,ceph::buffer::list &bl);
inline CtPtr _fault() { // helper fault method that stops continuation
fault();
return nullptr;
void session_reset();
void randomize_out_seq();
- Message *_get_next_outgoing(bufferlist *bl);
+ Message *_get_next_outgoing(ceph::buffer::list *bl);
- void prepare_send_message(uint64_t features, Message *m, bufferlist &bl);
- ssize_t write_message(Message *m, bufferlist &bl, bool more);
+ void prepare_send_message(uint64_t features, Message *m, ceph::buffer::list &bl);
+ ssize_t write_message(Message *m, ceph::buffer::list &bl, bool more);
void requeue_sent();
uint64_t discard_requeued_up_to(uint64_t out_seq, uint64_t seq);
void reset_recv_state();
- ostream &_conn_prefix(std::ostream *_dout);
+ std::ostream& _conn_prefix(std::ostream *_dout);
public:
ProtocolV1(AsyncConnection *connection);
CtPtr handle_connect_message_auth(char *buffer, int r);
CtPtr handle_connect_message_2();
CtPtr send_connect_message_reply(char tag, ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply);
+ ceph::buffer::list &authorizer_reply);
CtPtr handle_connect_message_reply_write(int r);
CtPtr replace(const AsyncConnectionRef& existing, ceph_msg_connect_reply &reply,
- bufferlist &authorizer_reply);
- CtPtr open(ceph_msg_connect_reply &reply, bufferlist &authorizer_reply);
+ ceph::buffer::list &authorizer_reply);
+ CtPtr open(ceph_msg_connect_reply &reply, ceph::buffer::list &authorizer_reply);
CtPtr handle_ready_connect_message_reply_write(int r);
CtPtr wait_seq();
CtPtr handle_seq(char *buffer, int r);
#define dout_subsys ceph_subsys_ms
#undef dout_prefix
#define dout_prefix _conn_prefix(_dout)
-ostream &ProtocolV2::_conn_prefix(std::ostream *_dout) {
+std::ostream &ProtocolV2::_conn_prefix(std::ostream *_dout) {
return *_dout << "--2- " << messenger->get_myaddrs() << " >> "
<< *connection->peer_addrs << " conn(" << connection << " "
<< this
void ProtocolV2::run_continuation(CtRef continuation) {
try {
CONTINUATION_RUN(continuation)
- } catch (const buffer::error &e) {
+ } catch (const ceph::buffer::error &e) {
lderr(cct) << __func__ << " failed decoding of frame header: " << e
<< dendl;
_fault();
#define WRITE(B, D, C) write(D, CONTINUATION(C), B)
-#define READ(L, C) read(CONTINUATION(C), buffer::ptr_node::create(buffer::create(L)))
+#define READ(L, C) read(CONTINUATION(C), ceph::buffer::ptr_node::create(ceph::buffer::create(L)))
#define READ_RXBUF(B, C) read(CONTINUATION(C), B)
void ProtocolV2::discard_out_queue() {
ldout(cct, 10) << __func__ << " started" << dendl;
- for (list<Message *>::iterator p = sent.begin(); p != sent.end(); ++p) {
+ for (auto p = sent.begin(); p != sent.end(); ++p) {
ldout(cct, 20) << __func__ << " discard " << *p << dendl;
(*p)->put();
}
CtPtr ProtocolV2::write(const std::string &desc,
CONTINUATION_TYPE<ProtocolV2> &next,
- bufferlist &buffer) {
+ ceph::bufferlist &buffer) {
if (unlikely(pre_auth.enabled)) {
pre_auth.txbuf.append(buffer);
ceph_assert(!cct->_conf->ms_die_on_bug ||
ldout(cct, 20) << __func__ << dendl;
bannerExchangeCallback = &callback;
- bufferlist banner_payload;
+ ceph::bufferlist banner_payload;
+ using ceph::encode;
encode((uint64_t)CEPH_MSGR2_SUPPORTED_FEATURES, banner_payload, 0);
encode((uint64_t)CEPH_MSGR2_REQUIRED_FEATURES, banner_payload, 0);
- bufferlist bl;
+ ceph::bufferlist bl;
bl.append(CEPH_BANNER_V2_PREFIX, strlen(CEPH_BANNER_V2_PREFIX));
encode((uint16_t)banner_payload.length(), bl, 0);
bl.claim_append(banner_payload);
}
uint16_t payload_len;
- bufferlist bl;
+ ceph::bufferlist bl;
buffer->set_offset(banner_prefix_len);
buffer->set_length(sizeof(ceph_le16));
bl.push_back(std::move(buffer));
auto ti = bl.cbegin();
+ using ceph::decode;
try {
decode(payload_len, ti);
- } catch (const buffer::error &e) {
+ } catch (const ceph::buffer::error &e) {
lderr(cct) << __func__ << " decode banner payload len failed " << dendl;
return _fault();
}
uint64_t peer_supported_features;
uint64_t peer_required_features;
- bufferlist bl;
+ ceph::bufferlist bl;
+ using ceph::decode;
bl.push_back(std::move(buffer));
auto ti = bl.cbegin();
try {
decode(peer_supported_features, ti);
decode(peer_required_features, ti);
- } catch (const buffer::error &e) {
+ } catch (const ceph::buffer::error &e) {
lderr(cct) << __func__ << " decode banner payload failed " << dendl;
return _fault();
}
const auto& cur_rx_desc = rx_segments_desc.at(rx_segments_data.size());
rx_buffer_t rx_buffer;
try {
- rx_buffer = buffer::ptr_node::create(buffer::create_aligned(
+ rx_buffer = ceph::buffer::ptr_node::create(ceph::buffer::create_aligned(
get_onwire_size(cur_rx_desc.length), cur_rx_desc.alignment));
} catch (std::bad_alloc&) {
// Catching because of potential issues with satisfying alignment.
ldout(cct, 20) << __func__ << " peer_type " << (int)connection->peer_type
<< " auth_client " << messenger->auth_client << dendl;
- bufferlist bl;
- vector<uint32_t> preferred_modes;
+ ceph::bufferlist bl;
+ std::vector<uint32_t> preferred_modes;
auto am = auth_meta;
connection->lock.unlock();
int r = messenger->auth_client->get_auth_request(
return WRITE(bad_method, "bad auth method", read_frame);
}
-CtPtr ProtocolV2::_handle_auth_request(bufferlist& auth_payload, bool more)
+CtPtr ProtocolV2::_handle_auth_request(ceph::bufferlist& auth_payload, bool more)
{
if (!messenger->auth_server) {
return _fault();
}
- bufferlist reply;
+ ceph::bufferlist reply;
auto am = auth_meta;
connection->lock.unlock();
int r = messenger->auth_server->handle_auth_request(
bool keepalive;
bool write_in_progress = false;
- ostream &_conn_prefix(std::ostream *_dout);
+ std::ostream& _conn_prefix(std::ostream *_dout);
void run_continuation(Ct<ProtocolV2> *pcontinuation);
void run_continuation(Ct<ProtocolV2> &continuation);
F &frame);
Ct<ProtocolV2> *write(const std::string &desc,
CONTINUATION_TYPE<ProtocolV2> &next,
- bufferlist &buffer);
+ ceph::bufferlist &buffer);
void requeue_sent();
uint64_t discard_requeued_up_to(uint64_t out_seq, uint64_t seq);
Ct<ProtocolV2> *post_server_banner_exchange();
Ct<ProtocolV2> *handle_auth_request(ceph::bufferlist &payload);
Ct<ProtocolV2> *handle_auth_request_more(ceph::bufferlist &payload);
- Ct<ProtocolV2> *_handle_auth_request(bufferlist& auth_payload, bool more);
+ Ct<ProtocolV2> *_handle_auth_request(ceph::bufferlist& auth_payload, bool more);
Ct<ProtocolV2> *_auth_bad_method(int r);
Ct<ProtocolV2> *handle_client_ident(ceph::bufferlist &payload);
Ct<ProtocolV2> *handle_ident_missing_features_write(int r);
};
}
-std::shared_ptr<NetworkStack> NetworkStack::create(CephContext *c, const string &t)
+std::shared_ptr<NetworkStack> NetworkStack::create(CephContext *c,
+ const std::string &t)
{
if (t == "posix")
return std::make_shared<PosixNetworkStack>(c, t);
return nullptr;
}
-Worker* NetworkStack::create_worker(CephContext *c, const string &type, unsigned worker_id)
+Worker* NetworkStack::create_worker(CephContext *c, const std::string &type, unsigned worker_id)
{
if (type == "posix")
return new PosixWorker(c, worker_id);
return nullptr;
}
-NetworkStack::NetworkStack(CephContext *c, const string &t): type(t), started(false), cct(c)
+NetworkStack::NetworkStack(CephContext *c, const std:: string &t): type(t), started(false), cct(c)
{
ceph_assert(cct->_conf->ms_async_op_threads > 0);
virtual ~ConnectedSocketImpl() {}
virtual int is_connected() = 0;
virtual ssize_t read(char*, size_t) = 0;
- virtual ssize_t send(bufferlist &bl, bool more) = 0;
+ virtual ssize_t send(ceph::buffer::list &bl, bool more) = 0;
virtual void shutdown() = 0;
virtual void close() = 0;
virtual int fd() const = 0;
/// Gets the output stream.
///
/// Gets an object that sends data to the remote endpoint.
- ssize_t send(bufferlist &bl, bool more) {
+ ssize_t send(ceph::buffer::list &bl, bool more) {
return _csi->send(bl, more);
}
/// Disables output to the socket.
protected:
CephContext *cct;
- vector<Worker*> workers;
+ std::vector<Worker*> workers;
- explicit NetworkStack(CephContext *c, const string &t);
+ explicit NetworkStack(CephContext *c, const std::string &t);
public:
NetworkStack(const NetworkStack &) = delete;
NetworkStack& operator=(const NetworkStack &) = delete;
}
static std::shared_ptr<NetworkStack> create(
- CephContext *c, const string &type);
+ CephContext *c, const std::string &type);
static Worker* create_worker(
- CephContext *c, const string &t, unsigned i);
+ CephContext *c, const std::string &t, unsigned i);
// backend need to override this method if backend doesn't support shared
// listen table.
// For example, posix backend has in kernel global listen table. If one
struct AuthRequestFrame : public ControlFrame<AuthRequestFrame,
uint32_t, // auth method
- vector<uint32_t>, // preferred modes
+ std::vector<uint32_t>, // preferred modes
bufferlist> { // auth payload
static const Tag tag = Tag::AUTH_REQUEST;
using ControlFrame::Encode;
using ControlFrame::Decode;
inline uint32_t &method() { return get_val<0>(); }
- inline vector<uint32_t> &preferred_modes() { return get_val<1>(); }
+ inline std::vector<uint32_t> &preferred_modes() { return get_val<1>(); }
inline bufferlist &auth_payload() { return get_val<2>(); }
protected:
<< cpp_strerror(errno) << dendl;
return -1;
}
- int rc = NetHandler(cct).set_nonblock(channel->fd);
+ int rc = ceph::NetHandler(cct).set_nonblock(channel->fd);
if (rc < 0) {
ibv_destroy_comp_channel(channel);
return -1;
device->binding_port(cct, port_num);
ib_physical_port = device->active_port->get_port_num();
pd = new ProtectionDomain(cct, device);
- ceph_assert(NetHandler(cct).set_nonblock(device->ctxt->async_fd) == 0);
+ ceph_assert(ceph::NetHandler(cct).set_nonblock(device->ctxt->async_fd) == 0);
support_srq = cct->_conf->ms_async_rdma_support_srq;
if (support_srq) {
uint32_t max_recv_wr;
uint32_t q_key;
bool dead;
- vector<Chunk*> recv_queue;
+ std::vector<Chunk*> recv_queue;
ceph::mutex lock = ceph::make_mutex("queue_pair_lock");
};
#undef dout_prefix
#define dout_prefix *_dout << " RDMAConnectedSocketImpl "
-RDMAConnectedSocketImpl::RDMAConnectedSocketImpl(CephContext *cct, shared_ptr<Infiniband> &ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher,
+RDMAConnectedSocketImpl::RDMAConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband> &ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
RDMAWorker *w)
: cct(cct), connected(0), error(0), ib(ib),
dispatcher(rdma_dispatcher), worker(w),
int RDMAConnectedSocketImpl::try_connect(const entity_addr_t& peer_addr, const SocketOptions &opts) {
ldout(cct, 20) << __func__ << " nonblock:" << opts.nonblock << ", nodelay:"
<< opts.nodelay << ", rbuf_size: " << opts.rcbuf_size << dendl;
- NetHandler net(cct);
+ ceph::NetHandler net(cct);
// we construct a socket to transport ib sync message
// but we shouldn't block in tcp connecting
return read_size;
}
-ssize_t RDMAConnectedSocketImpl::send(bufferlist &bl, bool more)
+ssize_t RDMAConnectedSocketImpl::send(ceph::buffer::list &bl, bool more)
{
if (error) {
if (!active)
if (total_copied == 0)
return -EAGAIN;
ceph_assert(total_copied <= pending_bl.length());
- bufferlist swapped;
+ ceph::buffer::list swapped;
if (total_copied < pending_bl.length()) {
worker->perf_logger->inc(l_msgr_rdma_tx_parital_mem);
pending_bl.splice(total_copied, pending_bl.length() - total_copied, &swapped);
int RDMAConnectedSocketImpl::post_work_request(std::vector<Chunk*> &tx_buffers)
{
ldout(cct, 20) << __func__ << " QP: " << local_qpn << " " << tx_buffers[0] << dendl;
- vector<Chunk*>::iterator current_buffer = tx_buffers.begin();
+ auto current_buffer = tx_buffers.begin();
ibv_sge isge[tx_buffers.size()];
uint32_t current_sge = 0;
ibv_send_wr iswr[tx_buffers.size()];
#define TIMEOUT_MS 3000
#define RETRY_COUNT 7
-RDMAIWARPConnectedSocketImpl::RDMAIWARPConnectedSocketImpl(CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher,
+RDMAIWARPConnectedSocketImpl::RDMAIWARPConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
RDMAWorker *w, RDMACMInfo *info)
: RDMAConnectedSocketImpl(cct, ib, rdma_dispatcher, w), cm_con_handler(new C_handle_cm_connection(this))
{
#define dout_prefix *_dout << " RDMAIWARPServerSocketImpl "
RDMAIWARPServerSocketImpl::RDMAIWARPServerSocketImpl(
- CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w,
+ CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w,
entity_addr_t& a, unsigned addr_slot)
: RDMAServerSocketImpl(cct, ib, rdma_dispatcher, w, a, addr_slot)
{
#define dout_prefix *_dout << " RDMAServerSocketImpl "
RDMAServerSocketImpl::RDMAServerSocketImpl(
- CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher,
+ CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
RDMAWorker *w, entity_addr_t& a, unsigned slot)
: ServerSocketImpl(a.get_type(), slot),
cct(cct), net(cct), server_setup_socket(-1), ib(ib),
ceph_assert(dead_queue_pairs.empty());
}
-RDMADispatcher::RDMADispatcher(CephContext* c, shared_ptr<Infiniband>& ib)
+RDMADispatcher::RDMADispatcher(CephContext* c, std::shared_ptr<Infiniband>& ib)
: cct(c), ib(ib)
{
PerfCountersBuilder plb(cct, "AsyncMessenger::RDMADispatcher", l_msgr_rdma_dispatcher_first, l_msgr_rdma_dispatcher_last);
dispatcher->notify_pending_workers();
}
-RDMAStack::RDMAStack(CephContext *cct, const string &t)
- : NetworkStack(cct, t), ib(make_shared<Infiniband>(cct)),
- rdma_dispatcher(make_shared<RDMADispatcher>(cct, ib))
+RDMAStack::RDMAStack(CephContext *cct, const std::string &t)
+ : NetworkStack(cct, t), ib(std::make_shared<Infiniband>(cct)),
+ rdma_dispatcher(std::make_shared<RDMADispatcher>(cct, ib))
{
ldout(cct, 20) << __func__ << " constructing RDMAStack..." << dendl;
std::thread t;
CephContext *cct;
- shared_ptr<Infiniband> ib;
+ std::shared_ptr<Infiniband> ib;
Infiniband::CompletionQueue* tx_cq = nullptr;
Infiniband::CompletionQueue* rx_cq = nullptr;
Infiniband::CompletionChannel *tx_cc = nullptr, *rx_cc = nullptr;
public:
PerfCounters *perf_logger;
- explicit RDMADispatcher(CephContext* c, shared_ptr<Infiniband>& ib);
+ explicit RDMADispatcher(CephContext* c, std::shared_ptr<Infiniband>& ib);
virtual ~RDMADispatcher();
void handle_async_event();
typedef Infiniband::MemoryManager::Chunk Chunk;
typedef Infiniband::MemoryManager MemoryManager;
typedef std::vector<Chunk*>::iterator ChunkIter;
- shared_ptr<Infiniband> ib;
+ std::shared_ptr<Infiniband> ib;
EventCallbackRef tx_handler;
std::list<RDMAConnectedSocketImpl*> pending_sent_conns;
- shared_ptr<RDMADispatcher> dispatcher;
+ std::shared_ptr<RDMADispatcher> dispatcher;
ceph::mutex lock = ceph::make_mutex("RDMAWorker::lock");
class C_handle_cq_tx : public EventCallback {
pending_sent_conns.remove(o);
}
void handle_pending_message();
- void set_dispatcher(shared_ptr<RDMADispatcher>& dispatcher) { this->dispatcher = dispatcher; }
- void set_ib(shared_ptr<Infiniband> &ib) {this->ib = ib;}
+ void set_dispatcher(std::shared_ptr<RDMADispatcher>& dispatcher) { this->dispatcher = dispatcher; }
+ void set_ib(std::shared_ptr<Infiniband> &ib) {this->ib = ib;}
void notify_worker() {
center.dispatch_event_external(tx_handler);
}
uint32_t local_qpn = 0;
int connected;
int error;
- shared_ptr<Infiniband> ib;
- shared_ptr<RDMADispatcher> dispatcher;
+ std::shared_ptr<Infiniband> ib;
+ std::shared_ptr<RDMADispatcher> dispatcher;
RDMAWorker* worker;
std::vector<Chunk*> buffers;
int notify_fd = -1;
- bufferlist pending_bl;
+ ceph::buffer::list pending_bl;
ceph::mutex lock = ceph::make_mutex("RDMAConnectedSocketImpl::lock");
std::vector<ibv_wc> wc;
const decltype(std::cbegin(pending_bl.buffers()))& end);
public:
- RDMAConnectedSocketImpl(CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w);
+ RDMAConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w);
virtual ~RDMAConnectedSocketImpl();
void pass_wc(std::vector<ibv_wc> &&v);
virtual int is_connected() override { return connected; }
virtual ssize_t read(char* buf, size_t len) override;
- virtual ssize_t send(bufferlist &bl, bool more) override;
+ virtual ssize_t send(ceph::buffer::list &bl, bool more) override;
virtual void shutdown() override;
virtual void close() override;
virtual int fd() const override { return notify_fd; }
class RDMAIWARPConnectedSocketImpl : public RDMAConnectedSocketImpl {
public:
- RDMAIWARPConnectedSocketImpl(CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher, RDMAWorker *w, RDMACMInfo *info = nullptr);
+ RDMAIWARPConnectedSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
+ RDMAWorker *w, RDMACMInfo *info = nullptr);
~RDMAIWARPConnectedSocketImpl();
virtual int try_connect(const entity_addr_t&, const SocketOptions &opt) override;
virtual void close() override;
class RDMAServerSocketImpl : public ServerSocketImpl {
protected:
CephContext *cct;
- NetHandler net;
+ ceph::NetHandler net;
int server_setup_socket;
- shared_ptr<Infiniband> ib;
- shared_ptr<RDMADispatcher> dispatcher;
+ std::shared_ptr<Infiniband> ib;
+ std::shared_ptr<RDMADispatcher> dispatcher;
RDMAWorker *worker;
entity_addr_t sa;
public:
- RDMAServerSocketImpl(CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher,
+ RDMAServerSocketImpl(CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
RDMAWorker *w, entity_addr_t& a, unsigned slot);
virtual int listen(entity_addr_t &sa, const SocketOptions &opt);
class RDMAIWARPServerSocketImpl : public RDMAServerSocketImpl {
public:
RDMAIWARPServerSocketImpl(
- CephContext *cct, shared_ptr<Infiniband>& ib,
- shared_ptr<RDMADispatcher>& rdma_dispatcher,
+ CephContext *cct, std::shared_ptr<Infiniband>& ib,
+ std::shared_ptr<RDMADispatcher>& rdma_dispatcher,
RDMAWorker* w, entity_addr_t& addr, unsigned addr_slot);
virtual int listen(entity_addr_t &sa, const SocketOptions &opt) override;
virtual int accept(ConnectedSocket *s, const SocketOptions &opts, entity_addr_t *out, Worker *w) override;
};
class RDMAStack : public NetworkStack {
- vector<std::thread> threads;
+ std::vector<std::thread> threads;
PerfCounters *perf_counter;
- shared_ptr<Infiniband> ib;
- shared_ptr<RDMADispatcher> rdma_dispatcher;
+ std::shared_ptr<Infiniband> ib;
+ std::shared_ptr<RDMADispatcher> rdma_dispatcher;
std::atomic<bool> fork_finished = {false};
public:
- explicit RDMAStack(CephContext *cct, const string &t);
+ explicit RDMAStack(CephContext *cct, const std::string &t);
virtual ~RDMAStack();
virtual bool nonblock_connect_need_writable_event() const override { return false; }
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
#include "msg_types.h"
#include "common/Formatter.h"
-void entity_name_t::dump(Formatter *f) const
+void entity_name_t::dump(ceph::Formatter *f) const
{
f->dump_string("type", type_str());
f->dump_unsigned("num", num());
}
-void entity_addr_t::dump(Formatter *f) const
+void entity_addr_t::dump(ceph::Formatter *f) const
{
f->dump_string("type", get_type_name(type));
f->dump_stream("addr") << get_sockaddr();
f->dump_unsigned("nonce", nonce);
}
-void entity_inst_t::dump(Formatter *f) const
+void entity_inst_t::dump(ceph::Formatter *f) const
{
f->dump_object("name", name);
f->dump_object("addr", addr);
}
-void entity_name_t::generate_test_instances(list<entity_name_t*>& o)
+void entity_name_t::generate_test_instances(std::list<entity_name_t*>& o)
{
o.push_back(new entity_name_t(entity_name_t::MON()));
o.push_back(new entity_name_t(entity_name_t::MON(1)));
o.push_back(new entity_name_t(entity_name_t::CLIENT(1)));
}
-void entity_addr_t::generate_test_instances(list<entity_addr_t*>& o)
+void entity_addr_t::generate_test_instances(std::list<entity_addr_t*>& o)
{
o.push_back(new entity_addr_t());
entity_addr_t *a = new entity_addr_t();
o.push_back(b);
}
-void entity_inst_t::generate_test_instances(list<entity_inst_t*>& o)
+void entity_inst_t::generate_test_instances(std::list<entity_inst_t*>& o)
{
o.push_back(new entity_inst_t());
entity_name_t name;
return true;
}
-ostream& operator<<(ostream& out, const entity_addr_t &addr)
+std::ostream& operator<<(std::ostream& out, const entity_addr_t &addr)
{
if (addr.type == entity_addr_t::TYPE_NONE) {
return out << "-";
return out;
}
-ostream& operator<<(ostream& out, const sockaddr *psa)
+std::ostream& operator<<(std::ostream& out, const sockaddr *psa)
{
char buf[NI_MAXHOST] = { 0 };
}
}
-ostream& operator<<(ostream& out, const sockaddr_storage &ss)
+std::ostream& operator<<(std::ostream& out, const sockaddr_storage &ss)
{
return out << (const sockaddr*)&ss;
}
return !v.empty();
}
-void entity_addrvec_t::encode(bufferlist& bl, uint64_t features) const
+void entity_addrvec_t::encode(ceph::buffer::list& bl, uint64_t features) const
{
using ceph::encode;
if ((features & CEPH_FEATURE_MSG_ADDR2) == 0) {
encode(v, bl, features);
}
-void entity_addrvec_t::decode(bufferlist::const_iterator& bl)
+void entity_addrvec_t::decode(ceph::buffer::list::const_iterator& bl)
{
using ceph::decode;
__u8 marker;
return;
}
if (marker > 2)
- throw buffer::malformed_input("entity_addrvec_marker > 2");
+ throw ceph::buffer::malformed_input("entity_addrvec_marker > 2");
decode(v, bl);
}
-void entity_addrvec_t::dump(Formatter *f) const
+void entity_addrvec_t::dump(ceph::Formatter *f) const
{
f->open_array_section("addrvec");
- for (vector<entity_addr_t>::const_iterator p = v.begin();
- p != v.end(); ++p) {
+ for (auto p = v.begin(); p != v.end(); ++p) {
f->dump_object("addr", *p);
}
f->close_section();
}
-void entity_addrvec_t::generate_test_instances(list<entity_addrvec_t*>& ls)
+void entity_addrvec_t::generate_test_instances(std::list<entity_addrvec_t*>& ls)
{
ls.push_back(new entity_addrvec_t());
ls.push_back(new entity_addrvec_t());
#endif
uint16_t ss_family;
if (elen < sizeof(ss_family)) {
- throw buffer::malformed_input("elen smaller than family len");
+ throw ceph::buffer::malformed_input("elen smaller than family len");
}
decode(ss_family, bl);
u.sa.sa_family = ss_family;
elen -= sizeof(ss_family);
if (elen > get_sockaddr_len() - sizeof(u.sa.sa_family)) {
- throw buffer::malformed_input("elen exceeds sockaddr len");
+ throw ceph::buffer::malformed_input("elen exceeds sockaddr len");
}
bl.copy(elen, u.sa.sa_data);
}
projects / ceph.git / commitdiff