MOAR TEMPLATE MAGIC

[invirt/third/libt4.git] / rpc / connection.cc

diff --git a/rpc/connection.cc b/rpc/connection.cc
index 3f60c69..86d4ec5 100644 (file)
--- a/rpc/connection.cc
+++ b/rpc/connection.cc
@@ -7,15 +7,13 @@
 #include <errno.h>
 #include <signal.h>
 #include <unistd.h>
 #include <errno.h>
 #include <signal.h>
 #include <unistd.h>

-#include "jsl_log.h"

 #include <sys/socket.h>
 
 #define MAX_PDU (10<<20) //maximum PDF is 10M
 
 connection::connection(chanmgr *m1, int f1, int l1)
 #include <sys/socket.h>
 
 #define MAX_PDU (10<<20) //maximum PDF is 10M
 
 connection::connection(chanmgr *m1, int f1, int l1)

-: mgr_(m1), fd_(f1), dead_(false),waiters_(0), refno_(1),lossy_(l1)
+: mgr_(m1), fd_(f1), lossy_(l1)

 {
 {

-

     int flags = fcntl(fd_, F_GETFL, NULL);
     flags |= O_NONBLOCK;
     fcntl(fd_, F_SETFL, flags);
     int flags = fcntl(fd_, F_GETFL, NULL);
     flags |= O_NONBLOCK;
     fcntl(fd_, F_SETFL, flags);

@@ -27,8 +25,7 @@ connection::connection(chanmgr *m1, int f1, int l1)
     PollMgr::Instance()->add_callback(fd_, CB_RDONLY, this);
 }
 
     PollMgr::Instance()->add_callback(fd_, CB_RDONLY, this);
 }
 

-connection::~connection()
-{
+connection::~connection() {

     VERIFY(dead_);
     if (rpdu_.buf)
         free(rpdu_.buf);
     VERIFY(dead_);
     if (rpdu_.buf)
         free(rpdu_.buf);

@@ -36,23 +33,17 @@ connection::~connection()
     close(fd_);
 }
 
     close(fd_);
 }
 

-void
-connection::incref()
-{
+void connection::incref() {

     lock rl(ref_m_);
     refno_++;
 }
 
     lock rl(ref_m_);
     refno_++;
 }
 

-bool
-connection::isdead()
-{
+bool connection::isdead() {

     lock ml(m_);
     return dead_;
 }
 
     lock ml(m_);
     return dead_;
 }
 

-void
-connection::closeconn()
-{
+void connection::closeconn() {

     {
         lock ml(m_);
         if (!dead_) {
     {
         lock ml(m_);
         if (!dead_) {

@@ -67,9 +58,7 @@ connection::closeconn()
     PollMgr::Instance()->block_remove_fd(fd_);
 }
 
     PollMgr::Instance()->block_remove_fd(fd_);
 }
 

-void
-connection::decref()
-{
+void connection::decref() {

     bool dead = false;
     {
         lock rl(ref_m_);
     bool dead = false;
     {
         lock rl(ref_m_);

@@ -80,21 +69,11 @@ connection::decref()
             dead = dead_;
         }
     }
             dead = dead_;
         }
     }

-    if (dead) {
+    if (dead)

         delete this;
         delete this;

-    }
-}
-
-int
-connection::ref()
-{
-    lock rl(ref_m_);
-       return refno_;

 }
 
 }
 

-int
-connection::compare(connection *another)
-{
+int connection::compare(connection *another) {

     if (create_time_ > another->create_time_)
         return 1;
     if (create_time_ < another->create_time_)
     if (create_time_ > another->create_time_)
         return 1;
     if (create_time_ < another->create_time_)

@@ -102,261 +81,244 @@ connection::compare(connection *another)
     return 0;
 }
 
     return 0;
 }
 

-bool
-connection::send(char *b, size_t sz)
-{
+bool connection::send(char *b, size_t sz) {

     lock ml(m_);
     lock ml(m_);

-       waiters_++;
-       while (!dead_ && wpdu_.buf) {
+    waiters_++;
+    while (!dead_ && wpdu_.buf) {

         send_wait_.wait(ml);
         send_wait_.wait(ml);

-       }
-       waiters_--;
-       if (dead_) {
-               return false;
-       }
-       wpdu_.buf = b;
-       wpdu_.sz = sz;
-       wpdu_.solong = 0;
-
-       if (lossy_) {
-               if ((random()%100) < lossy_) {
-                       jsl_log(JSL_DBG_1, "connection::send LOSSY TEST shutdown fd_ %d\n", fd_);
-                       shutdown(fd_,SHUT_RDWR);
-               }
-       }
-
-       if (!writepdu()) {
-               dead_ = true;
+    }
+    waiters_--;
+    if (dead_) {
+        return false;
+    }
+    wpdu_.buf = b;
+    wpdu_.sz = sz;
+    wpdu_.solong = 0;
+
+    if (lossy_) {
+        if ((random()%100) < lossy_) {
+            IF_LEVEL(1) LOG("connection::send LOSSY TEST shutdown fd_ " << fd_);
+            shutdown(fd_,SHUT_RDWR);
+        }
+    }
+
+    if (!writepdu()) {
+        dead_ = true;

         ml.unlock();
         ml.unlock();

-               PollMgr::Instance()->block_remove_fd(fd_);
+        PollMgr::Instance()->block_remove_fd(fd_);

         ml.lock();
         ml.lock();

-       } else {
-               if (wpdu_.solong == wpdu_.sz) {
-               } else {
-                       //should be rare to need to explicitly add write callback
-                       PollMgr::Instance()->add_callback(fd_, CB_WRONLY, this);
-                       while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.sz) {
+    } else {
+        if (wpdu_.solong == wpdu_.sz) {
+        } else {
+            //should be rare to need to explicitly add write callback
+            PollMgr::Instance()->add_callback(fd_, CB_WRONLY, this);
+            while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.sz) {

                 send_complete_.wait(ml);
                 send_complete_.wait(ml);

-                       }
-               }
-       }
-       bool ret = (!dead_ && wpdu_.solong == wpdu_.sz);
-       wpdu_.solong = wpdu_.sz = 0;
-       wpdu_.buf = NULL;
-       if (waiters_ > 0)
+            }
+        }
+    }
+    bool ret = (!dead_ && wpdu_.solong == wpdu_.sz);
+    wpdu_.solong = wpdu_.sz = 0;
+    wpdu_.buf = NULL;
+    if (waiters_ > 0)

         send_wait_.notify_all();
         send_wait_.notify_all();

-       return ret;
+    return ret;

 }
 
 //fd_ is ready to be written
 }
 
 //fd_ is ready to be written

-void
-connection::write_cb(int s)
-{
+void connection::write_cb(int s) {

     lock ml(m_);
     lock ml(m_);

-       VERIFY(!dead_);
-       VERIFY(fd_ == s);
-       if (wpdu_.sz == 0) {
-               PollMgr::Instance()->del_callback(fd_,CB_WRONLY);
-               return;
-       }
-       if (!writepdu()) {
-               PollMgr::Instance()->del_callback(fd_, CB_RDWR);
-               dead_ = true;
-       } else {
-               VERIFY(wpdu_.solong != size_t_max);
-               if (wpdu_.solong < wpdu_.sz) {
-                       return;
-               }
+    VERIFY(!dead_);
+    VERIFY(fd_ == s);
+    if (wpdu_.sz == 0) {
+        PollMgr::Instance()->del_callback(fd_,CB_WRONLY);
+        return;
+    }
+    if (!writepdu()) {
+        PollMgr::Instance()->del_callback(fd_, CB_RDWR);
+        dead_ = true;
+    } else {
+        VERIFY(wpdu_.solong != size_t_max);
+        if (wpdu_.solong < wpdu_.sz) {
+            return;
+        }

     }
     }

-       send_complete_.notify_one();
+    send_complete_.notify_one();

 }
 
 //fd_ is ready to be read
 }
 
 //fd_ is ready to be read

-void
-connection::read_cb(int s)
-{
+void connection::read_cb(int s) {

     lock ml(m_);
     lock ml(m_);

-       VERIFY(fd_ == s);
-       if (dead_)  {
-               return;
-       }
-
-       bool succ = true;
-       if (!rpdu_.buf || rpdu_.solong < rpdu_.sz) {
-               succ = readpdu();
-       }
-
-       if (!succ) {
-               PollMgr::Instance()->del_callback(fd_,CB_RDWR);
-               dead_ = true;
-               send_complete_.notify_one();
-       }
-
-       if (rpdu_.buf && rpdu_.sz == rpdu_.solong) {
-               if (mgr_->got_pdu(this, rpdu_.buf, rpdu_.sz)) {
-                       //chanmgr has successfully consumed the pdu
-                       rpdu_.buf = NULL;
-                       rpdu_.sz = rpdu_.solong = 0;
-               }
-       }
+    VERIFY(fd_ == s);
+    if (dead_)  {
+        return;
+    }
+
+    bool succ = true;
+    if (!rpdu_.buf || rpdu_.solong < rpdu_.sz) {
+        succ = readpdu();
+    }
+
+    if (!succ) {
+        PollMgr::Instance()->del_callback(fd_,CB_RDWR);
+        dead_ = true;
+        send_complete_.notify_one();
+    }
+
+    if (rpdu_.buf && rpdu_.sz == rpdu_.solong) {
+        if (mgr_->got_pdu(this, rpdu_.buf, rpdu_.sz)) {
+            //chanmgr has successfully consumed the pdu
+            rpdu_.buf = NULL;
+            rpdu_.sz = rpdu_.solong = 0;
+        }
+    }

 }
 
 }
 

-bool
-connection::writepdu()
-{
-       VERIFY(wpdu_.solong != size_t_max);
-       if (wpdu_.solong == wpdu_.sz)
-               return true;
-
-       if (wpdu_.solong == 0) {
-               uint32_t sz = htonl((uint32_t)wpdu_.sz);
-               bcopy(&sz,wpdu_.buf,sizeof(sz));
-       }
-       ssize_t n = write(fd_, wpdu_.buf + wpdu_.solong, (wpdu_.sz-wpdu_.solong));
-       if (n < 0) {
-               if (errno != EAGAIN) {
-                       jsl_log(JSL_DBG_1, "connection::writepdu fd_ %d failure errno=%d\n", fd_, errno);
-                       wpdu_.solong = size_t_max;
-                       wpdu_.sz = 0;
-               }
-               return (errno == EAGAIN);
-       }
-       wpdu_.solong += (size_t)n;
-       return true;
+bool connection::writepdu() {
+    VERIFY(wpdu_.solong != size_t_max);
+    if (wpdu_.solong == wpdu_.sz)
+        return true;
+
+    if (wpdu_.solong == 0) {
+        uint32_t sz = htonl((uint32_t)wpdu_.sz);
+        bcopy(&sz,wpdu_.buf,sizeof(sz));
+    }
+    ssize_t n = write(fd_, wpdu_.buf + wpdu_.solong, (wpdu_.sz-wpdu_.solong));
+    if (n < 0) {
+        if (errno != EAGAIN) {
+            IF_LEVEL(1) LOG("connection::writepdu fd_ " << fd_ << " failure errno=" << errno);
+            wpdu_.solong = size_t_max;
+            wpdu_.sz = 0;
+        }
+        return (errno == EAGAIN);
+    }
+    wpdu_.solong += (size_t)n;
+    return true;

 }
 
 }
 

-bool
-connection::readpdu()
-{
-       if (!rpdu_.sz) {
-               uint32_t sz1;
-               ssize_t n = read(fd_, &sz1, sizeof(sz1));
-
-               if (n == 0) {
-                       return false;
-               }
-
-               if (n < 0) {
-                       VERIFY(errno!=EAGAIN);
-                       return false;
-               }
-
-               if (n > 0 && n != sizeof(sz1)) {
-                       jsl_log(JSL_DBG_OFF, "connection::readpdu short read of sz\n");
-                       return false;
-               }
-
-               size_t sz = ntohl(sz1);
-
-               if (sz > MAX_PDU) {
-                       char *tmpb = (char *)&sz1;
-                       jsl_log(JSL_DBG_2, "connection::readpdu read pdu TOO BIG %lu network order=%x %x %x %x %x\n", sz,
-                                       sz1, tmpb[0],tmpb[1],tmpb[2],tmpb[3]);
-                       return false;
-               }
-
-               rpdu_.sz = sz;
-               VERIFY(rpdu_.buf == NULL);
-               rpdu_.buf = (char *)malloc(sz+sizeof(sz1));
-               VERIFY(rpdu_.buf);
-               bcopy(&sz1,rpdu_.buf,sizeof(sz1));
-               rpdu_.solong = sizeof(sz1);
-       }
-
-       ssize_t n = read(fd_, rpdu_.buf + rpdu_.solong, rpdu_.sz - rpdu_.solong);
-       if (n <= 0) {
-               if (errno == EAGAIN)
-                       return true;
-               if (rpdu_.buf)
-                       free(rpdu_.buf);
-               rpdu_.buf = NULL;
-               rpdu_.sz = rpdu_.solong = 0;
-               return (errno == EAGAIN);
-       }
-       rpdu_.solong += (size_t)n;
-       return true;
+bool connection::readpdu() {
+    if (!rpdu_.sz) {
+        uint32_t sz1;
+        ssize_t n = read(fd_, &sz1, sizeof(sz1));
+
+        if (n == 0) {
+            return false;
+        }
+
+        if (n < 0) {
+            VERIFY(errno!=EAGAIN);
+            return false;
+        }
+
+        if (n > 0 && n != sizeof(sz1)) {
+            IF_LEVEL(0) LOG("connection::readpdu short read of sz");
+            return false;
+        }
+
+        size_t sz = ntohl(sz1);
+
+        if (sz > MAX_PDU) {
+            IF_LEVEL(2) LOG("connection::readpdu read pdu TOO BIG " << sz << " network order=" << hex << sz1);
+            return false;
+        }
+
+        rpdu_.sz = sz;
+        VERIFY(rpdu_.buf == NULL);
+        rpdu_.buf = (char *)malloc(sz+sizeof(sz1));
+        VERIFY(rpdu_.buf);
+        bcopy(&sz1,rpdu_.buf,sizeof(sz1));
+        rpdu_.solong = sizeof(sz1);
+    }
+
+    ssize_t n = read(fd_, rpdu_.buf + rpdu_.solong, rpdu_.sz - rpdu_.solong);
+    if (n <= 0) {
+        if (errno == EAGAIN)
+            return true;
+        if (rpdu_.buf)
+            free(rpdu_.buf);
+        rpdu_.buf = NULL;
+        rpdu_.sz = rpdu_.solong = 0;
+        return (errno == EAGAIN);
+    }
+    rpdu_.solong += (size_t)n;
+    return true;

 }
 
 tcpsconn::tcpsconn(chanmgr *m1, unsigned int port, int lossytest)
 : mgr_(m1), lossy_(lossytest)
 {
 }
 
 tcpsconn::tcpsconn(chanmgr *m1, unsigned int port, int lossytest)
 : mgr_(m1), lossy_(lossytest)
 {

-       struct sockaddr_in sin;
-       memset(&sin, 0, sizeof(sin));
-       sin.sin_family = AF_INET;
-       sin.sin_port = htons(port);
-
-       tcp_ = socket(AF_INET, SOCK_STREAM, 0);
-       if (tcp_ < 0) {
-               perror("tcpsconn::tcpsconn accept_loop socket:");
-               VERIFY(0);
-       }
-
-       int yes = 1;
-       setsockopt(tcp_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
-       setsockopt(tcp_, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
-
-       if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) {
-               perror("accept_loop tcp bind:");
-               VERIFY(0);
-       }
-
-       if (listen(tcp_, 1000) < 0) {
-               perror("tcpsconn::tcpsconn listen:");
-               VERIFY(0);
-       }
+    struct sockaddr_in sin;
+    memset(&sin, 0, sizeof(sin));
+    sin.sin_family = AF_INET;
+    sin.sin_port = htons(port);
+
+    tcp_ = socket(AF_INET, SOCK_STREAM, 0);
+    if (tcp_ < 0) {
+        perror("tcpsconn::tcpsconn accept_loop socket:");
+        VERIFY(0);
+    }
+
+    int yes = 1;
+    setsockopt(tcp_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
+    setsockopt(tcp_, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
+
+    if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) {
+        perror("accept_loop tcp bind:");
+        VERIFY(0);
+    }
+
+    if (listen(tcp_, 1000) < 0) {
+        perror("tcpsconn::tcpsconn listen:");
+        VERIFY(0);
+    }

 
     socklen_t addrlen = sizeof(sin);
     VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
     port_ = ntohs(sin.sin_port);
 
 
     socklen_t addrlen = sizeof(sin);
     VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
     port_ = ntohs(sin.sin_port);
 

-       jsl_log(JSL_DBG_2, "tcpsconn::tcpsconn listen on %d %d\n", port_,
-               sin.sin_port);
+    IF_LEVEL(2) LOG("tcpsconn::tcpsconn listen on " << port_ << " " << sin.sin_port);

 
 

-       if (pipe(pipe_) < 0) {
-               perror("accept_loop pipe:");
-               VERIFY(0);
-       }
+    if (pipe(pipe_) < 0) {
+        perror("accept_loop pipe:");
+        VERIFY(0);
+    }

 
 

-       int flags = fcntl(pipe_[0], F_GETFL, NULL);
-       flags |= O_NONBLOCK;
-       fcntl(pipe_[0], F_SETFL, flags);
+    int flags = fcntl(pipe_[0], F_GETFL, NULL);
+    flags |= O_NONBLOCK;
+    fcntl(pipe_[0], F_SETFL, flags);

 
     th_ = thread(&tcpsconn::accept_conn, this);
 }
 
 tcpsconn::~tcpsconn()
 {
 
     th_ = thread(&tcpsconn::accept_conn, this);
 }
 
 tcpsconn::~tcpsconn()
 {

-       VERIFY(close(pipe_[1]) == 0);
+    VERIFY(close(pipe_[1]) == 0);

     th_.join();
 
     th_.join();
 

-       //close all the active connections
-       map<int, connection *>::iterator i;
-       for (i = conns_.begin(); i != conns_.end(); i++) {
-               i->second->closeconn();
-               i->second->decref();
-       }
+    //close all the active connections
+    map<int, connection *>::iterator i;
+    for (i = conns_.begin(); i != conns_.end(); i++) {
+        i->second->closeconn();
+        i->second->decref();
+    }

 }
 
 }
 

-void
-tcpsconn::process_accept()
-{
-       sockaddr_in sin;
-       socklen_t slen = sizeof(sin);
-       int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
-       if (s1 < 0) {
-               perror("tcpsconn::accept_conn error");
-               throw thread_exit_exception();
-       }
-
-       jsl_log(JSL_DBG_2, "accept_loop got connection fd=%d %s:%d\n",
-                       s1, inet_ntoa(sin.sin_addr), ntohs(sin.sin_port));
-       connection *ch = new connection(mgr_, s1, lossy_);
+void tcpsconn::process_accept() {
+    sockaddr_in sin;
+    socklen_t slen = sizeof(sin);
+    int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
+    if (s1 < 0) {
+        perror("tcpsconn::accept_conn error");
+        throw thread_exit_exception();
+    }
+
+    IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntohs(sin.sin_port));
+    connection *ch = new connection(mgr_, s1, lossy_);

 
     // garbage collect all dead connections with refcount of 1
     for (auto i = conns_.begin(); i != conns_.end();) {
         if (i->second->isdead() && i->second->ref() == 1) {
 
     // garbage collect all dead connections with refcount of 1
     for (auto i = conns_.begin(); i != conns_.end();) {
         if (i->second->isdead() && i->second->ref() == 1) {

-            jsl_log(JSL_DBG_2, "accept_loop garbage collected fd=%d\n",
-                    i->second->channo());
+            IF_LEVEL(2) LOG("accept_loop garbage collected fd=" << i->second->channo());

             i->second->decref();
             // Careful not to reuse i right after erase. (i++) will
             // be evaluated before the erase call because in C++,
             i->second->decref();
             // Careful not to reuse i right after erase. (i++) will
             // be evaluated before the erase call because in C++,

@@ -367,14 +329,12 @@ tcpsconn::process_accept()
             ++i;
     }
 
             ++i;
     }
 

-       conns_[ch->channo()] = ch;
+    conns_[ch->channo()] = ch;

 }
 
 }
 

-void
-tcpsconn::accept_conn()
-{
-       fd_set rfds;
-       int max_fd = pipe_[0] > tcp_ ? pipe_[0] : tcp_;
+void tcpsconn::accept_conn() {
+    fd_set rfds;
+    int max_fd = pipe_[0] > tcp_ ? pipe_[0] : tcp_;

 
     try {
         while (1) {
 
     try {
         while (1) {

@@ -389,7 +349,7 @@ tcpsconn::accept_conn()
                     continue;
                 } else {
                     perror("accept_conn select:");
                     continue;
                 } else {
                     perror("accept_conn select:");

-                    jsl_log(JSL_DBG_OFF, "tcpsconn::accept_conn failure errno %d\n",errno);
+                    IF_LEVEL(0) LOG("tcpsconn::accept_conn failure errno " << errno);

                     VERIFY(0);
                 }
             }
                     VERIFY(0);
                 }
             }

@@ -411,20 +371,16 @@ tcpsconn::accept_conn()
     }
 }
 
     }
 }
 

-connection *
-connect_to_dst(const sockaddr_in &dst, chanmgr *mgr, int lossy)
-{
-       int s = socket(AF_INET, SOCK_STREAM, 0);
-       int yes = 1;
-       setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
-       if (connect(s, (sockaddr*)&dst, sizeof(dst)) < 0) {
-               jsl_log(JSL_DBG_1, "rpcc::connect_to_dst failed to %s:%d\n",
-                               inet_ntoa(dst.sin_addr), (int)ntohs(dst.sin_port));
-               close(s);
-               return NULL;
-       }
-       jsl_log(JSL_DBG_2, "connect_to_dst fd=%d to dst %s:%d\n",
-                       s, inet_ntoa(dst.sin_addr), (int)ntohs(dst.sin_port));
-       return new connection(mgr, s, lossy);
+connection * connect_to_dst(const sockaddr_in &dst, chanmgr *mgr, int lossy) {
+    int s = socket(AF_INET, SOCK_STREAM, 0);
+    int yes = 1;
+    setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
+    if (connect(s, (sockaddr*)&dst, sizeof(dst)) < 0) {
+        IF_LEVEL(1) LOG_NONMEMBER("rpcc::connect_to_dst failed to " << inet_ntoa(dst.sin_addr) << ":" << ntohs(dst.sin_port));
+        close(s);
+        return NULL;
+    }
+    IF_LEVEL(2) LOG_NONMEMBER("connect_to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntohs(dst.sin_port));
+    return new connection(mgr, s, lossy);

 }
 
 }