Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

msg/async/rdma: refactor tx handle flow to get rid of locks #13680

Merged
merged 5 commits into from Mar 2, 2017
Merged

msg/async/rdma: refactor tx handle flow to get rid of locks #13680

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ab7de95
msg/async/rdma: change notify to notify_worker
yuyuyu101 Feb 27, 2017
06eadd2
msg/async/rdma: remove unused variable
yuyuyu101 Feb 27, 2017
7891484
msg/async/rdma: remove lock protection for pending_sent_conns
yuyuyu101 Feb 27, 2017
bc580b0
msg/async/rdma: refactor RDMAStack to accelerate tx handle
yuyuyu101 Feb 27, 2017
f104300
msg/async: must destruct ConnectedSocket in the his own thread
yuyuyu101 Mar 1, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
30 changes: 13 additions & 17 deletions src/msg/async/rdma/RDMAStack.cc
View file
Open in desktop
Expand Up @@ -247,11 +247,12 @@ void RDMADispatcher::polling()
}

void RDMADispatcher::notify_pending_workers() {
Mutex::Locker l(w_lock);
if (pending_workers.empty())
return ;
pending_workers.front()->pass_wc(std::move(vector<ibv_wc>()));
pending_workers.pop_front();
{
Mutex::Locker l(w_lock);
if (pending_workers.empty())
return ;
}
pending_workers.front()->notify_worker();
}

int RDMADispatcher::register_qp(QueuePair *qp, RDMAConnectedSocketImpl* csi)
Expand Down Expand Up @@ -337,7 +338,7 @@ void RDMADispatcher::handle_post_fork()

RDMAWorker::RDMAWorker(CephContext *c, unsigned i)
: Worker(c, i), stack(nullptr), infiniband(NULL),
tx_handler(new C_handle_cq_tx(this)), memory_manager(NULL), lock("RDMAWorker::lock"), pended(false)
tx_handler(new C_handle_cq_tx(this)), memory_manager(NULL), lock("RDMAWorker::lock")
{
// initialize perf_logger
char name[128];
Expand Down Expand Up @@ -380,7 +381,7 @@ void RDMAWorker::initialize()
}
}

void RDMAWorker::notify()
void RDMAWorker::notify_worker()
{
uint64_t i = 1;
assert(write(notify_fd, &i, sizeof(i)) == sizeof(i));
Expand All @@ -393,7 +394,7 @@ void RDMAWorker::pass_wc(std::vector<ibv_wc> &&v)
wc = std::move(v);
else
wc.insert(wc.end(), v.begin(), v.end());
notify();
notify_worker();
}

void RDMAWorker::get_wc(std::vector<ibv_wc> &w)
Expand Down Expand Up @@ -434,6 +435,7 @@ int RDMAWorker::connect(const entity_addr_t &addr, const SocketOptions &opts, Co

int RDMAWorker::reserve_message_buffer(RDMAConnectedSocketImpl *o, std::vector<Chunk*> &c, size_t bytes)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we call it: "get_reged_mem()"

{
assert(center.in_thread());
int r = infiniband->get_tx_buffers(c, bytes);
if (r > 0) {
stack->get_dispatcher()->inflight += r;
Expand All @@ -443,11 +445,8 @@ int RDMAWorker::reserve_message_buffer(RDMAConnectedSocketImpl *o, std::vector<C
assert(r == 0);

if (o) {
{
Mutex::Locker l(lock);
if (pending_sent_conns.back() != o)
pending_sent_conns.push_back(o);
}
if (pending_sent_conns.back() != o)
pending_sent_conns.push_back(o);
dispatcher->pending_buffers(this);
}
return r;
Expand All @@ -463,25 +462,22 @@ int RDMAWorker::reserve_message_buffer(RDMAConnectedSocketImpl *o, std::vector<C
*/
void RDMAWorker::post_tx_buffer(std::vector<Chunk*> &chunks)
{
assert(center.in_thread());
if (chunks.empty())
return ;

dispatcher->inflight -= chunks.size();
memory_manager->return_tx(chunks);
ldout(cct, 30) << __func__ << " release " << chunks.size() << " chunks, inflight " << stack->get_dispatcher()->inflight << dendl;

pended = false;
std::set<RDMAConnectedSocketImpl*> done;
Mutex::Locker l(lock);
while (!pending_sent_conns.empty()) {
RDMAConnectedSocketImpl *o = pending_sent_conns.front();
pending_sent_conns.pop_front();
if (!done.count(o)) {
lock.Unlock();
done.insert(o);
ssize_t r = o->submit(false);
ldout(cct, 20) << __func__ << " sent pending bl socket=" << o << " r=" << r << dendl;
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks like false param to submit has no effect
?!

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes

lock.Lock();
if (r < 0) {
if (r == -EAGAIN) {
pending_sent_conns.push_front(o);
Expand Down
5 changes: 2 additions & 3 deletions src/msg/async/rdma/RDMAStack.h
View file
Open in desktop
Expand Up @@ -165,7 +165,6 @@ class RDMAWorker : public Worker {
int notify_fd = -1;
Mutex lock;
std::vector<ibv_wc> wc;
bool pended;

class C_handle_cq_tx : public EventCallback {
RDMAWorker *worker;
Expand All @@ -180,7 +179,7 @@ class RDMAWorker : public Worker {
PerfCounters *perf_logger;
explicit RDMAWorker(CephContext *c, unsigned i);
virtual ~RDMAWorker();
void notify();
void notify_worker();
void pass_wc(std::vector<ibv_wc> &&v);
void get_wc(std::vector<ibv_wc> &w);
virtual int listen(entity_addr_t &addr, const SocketOptions &opts, ServerSocket *) override;
Expand All @@ -190,7 +189,7 @@ class RDMAWorker : public Worker {
int reserve_message_buffer(RDMAConnectedSocketImpl *o, std::vector<Chunk*> &c, size_t bytes);
void post_tx_buffer(std::vector<Chunk*> &chunks);
void remove_pending_conn(RDMAConnectedSocketImpl *o) {
Mutex::Locker l(lock);
assert(center.in_thread());
pending_sent_conns.remove(o);
}
void handle_tx_event();
Expand Down