From e7686a4776ec8c93a265009ce99874dffd7eac11 Mon Sep 17 00:00:00 2001
From: Vu Pham <vu@mellanox.com>
Date: Mon, 8 Dec 2014 06:46:55 -0800
Subject: [PATCH] xio_messenger: Enable flow-control on total msgs and bytes
 per connection

* Enable flow-control in general
* Set flow-control on total msgs and bytes per connection
* Temporarily set total bytes = 256 MB, queue depth msgs = 512
* Need to get flow-control policy from messenger

Signed-off-by: Vu Pham <vu@mellanox.com>
---
 src/msg/xio/XioConnection.cc |  4 ++--
 src/msg/xio/XioMessenger.cc  | 13 +++++++++++++
 src/msg/xio/XioPortal.h      |  3 +++
 3 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/src/msg/xio/XioConnection.cc b/src/msg/xio/XioConnection.cc
index 75109476d8..4ca72ea4e7 100644
--- a/src/msg/xio/XioConnection.cc
+++ b/src/msg/xio/XioConnection.cc
@@ -416,7 +416,7 @@ int XioConnection::on_ow_msg_send_complete(struct xio_session *session,
 
 void XioConnection::msg_send_fail(XioMsg *xmsg, int code)
 {
-  ldout(msgr->cct,4) << "xio_send_msg FAILED " << &xmsg->req_0.msg << " code=" << code <<
+  ldout(msgr->cct,0) << "xio_send_msg FAILED " << &xmsg->req_0.msg << " code=" << code <<
     " (" << xio_strerror(code) << ")" << dendl;
   /* return refs taken for each xio_msg */
   xmsg->put_msg_refs();
@@ -424,7 +424,7 @@ void XioConnection::msg_send_fail(XioMsg *xmsg, int code)
 
 void XioConnection::msg_release_fail(struct xio_msg *msg, int code)
 {
-  ldout(msgr->cct,4) << "xio_release_msg FAILED " << msg <<  "code=" << code <<
+  ldout(msgr->cct,0) << "xio_release_msg FAILED " << msg <<  "code=" << code <<
     " (" << xio_strerror(code) << ")" << dendl;
 } /* msg_release_fail */
 
diff --git a/src/msg/xio/XioMessenger.cc b/src/msg/xio/XioMessenger.cc
index 49f01427d2..009743ed1a 100644
--- a/src/msg/xio/XioMessenger.cc
+++ b/src/msg/xio/XioMessenger.cc
@@ -289,12 +289,25 @@ XioMessenger::XioMessenger(CephContext *cct, entity_name_t name,
       xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_MAX_OUT_IOVLEN,
 		  &xopt, sizeof(xopt));
 
+      /* enable flow-control */
+      xopt = 1;
+      xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_ENABLE_FLOW_CONTROL,
+		  &xopt, sizeof(xopt));
+
       xopt = cct->_conf->xio_queue_depth; // defaults to 512
       xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_SND_QUEUE_DEPTH_MSGS,
 		  &xopt, sizeof(xopt));
       xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_RCV_QUEUE_DEPTH_MSGS,
 		  &xopt, sizeof(xopt));
 
+      /* flow-control total bytes per connection; should get from messenger's policy */
+      uint64_t bytes_opt;
+      bytes_opt = (2 << 27); // 256MB
+      xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_SND_QUEUE_DEPTH_BYTES,
+		  &bytes_opt, sizeof(bytes_opt));
+      xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_RCV_QUEUE_DEPTH_BYTES,
+		  &bytes_opt, sizeof(bytes_opt));
+
       /* and set threshold for buffer callouts */
       xopt = 16384;
       xio_set_opt(NULL, XIO_OPTLEVEL_ACCELIO, XIO_OPTNAME_MAX_INLINE_DATA,
diff --git a/src/msg/xio/XioPortal.h b/src/msg/xio/XioPortal.h
index ca50a8d65e..3c30d4a934 100644
--- a/src/msg/xio/XioPortal.h
+++ b/src/msg/xio/XioPortal.h
@@ -230,6 +230,9 @@ class XioPortal : public Thread
                 }
 	      }
 	      if (unlikely(code)) {
+                std::cout << "xio_send_msg FAILED " << &xmsg->req_0.msg <<
+                             " code=" << code <<
+                             " (" << xio_strerror(code) << ")" << std::endl;
                 xs->xcon->msg_send_fail(xmsg, code);
 	      } else {
                 xs->xcon->send.set(msg->timestamp); // need atomic?