Cosmetic improvements.

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

diff --git a/rpc/connection.cc b/rpc/connection.cc
index 358a2af..4e49305 100644 (file)
--- a/rpc/connection.cc
+++ b/rpc/connection.cc
@@ -6,25 +6,32 @@
 #include <unistd.h>
 #include "marshall.h"
 
 #include <unistd.h>
 #include "marshall.h"
 

-connection::connection(connection_delegate *m1, socket_t && f1, int l1)
-: mgr_(m1), fd_(move(f1)), lossy_(l1)
+connection::connection(connection_delegate * delegate, socket_t && f1, int l1)
+: fd(move(f1)), delegate_(delegate), lossy_(l1)

 {
 {

-    fd_.flags() |= O_NONBLOCK;
+    fd.flags() |= O_NONBLOCK;

 
     signal(SIGPIPE, SIG_IGN);
 
 
     signal(SIGPIPE, SIG_IGN);
 

-    create_time_ = steady_clock::now();
-
-    poll_mgr::shared_mgr.add_callback(fd_, CB_RDONLY, this);
+    poll_mgr::shared_mgr.add_callback(fd, CB_RDONLY, this);

 }
 
 connection::~connection() {
 }
 
 connection::~connection() {

-    closeconn();
+    {
+        lock ml(m_);
+        if (dead_)
+            return;
+        dead_ = true;
+        shutdown(fd,SHUT_RDWR);
+    }
+    // after block_remove_fd, select will never wait on fd and no callbacks
+    // will be active
+    poll_mgr::shared_mgr.block_remove_fd(fd);

     VERIFY(dead_);
     VERIFY(!wpdu_.buf.size());
 }
 
     VERIFY(dead_);
     VERIFY(!wpdu_.buf.size());
 }
 

-shared_ptr<connection> connection::to_dst(const sockaddr_in &dst, connection_delegate *mgr, int lossy) {
+shared_ptr<connection> connection::to_dst(const sockaddr_in & dst, connection_delegate * delegate, int lossy) {

     socket_t s = socket(AF_INET, SOCK_STREAM, 0);
     s.setsockopt(IPPROTO_TCP, TCP_NODELAY, (int)1);
     if (connect(s, (sockaddr*)&dst, sizeof(dst)) < 0) {
     socket_t s = socket(AF_INET, SOCK_STREAM, 0);
     s.setsockopt(IPPROTO_TCP, TCP_NODELAY, (int)1);
     if (connect(s, (sockaddr*)&dst, sizeof(dst)) < 0) {

@@ -33,20 +40,7 @@ shared_ptr<connection> connection::to_dst(const sockaddr_in &dst, connection_del
         return nullptr;
     }
     IF_LEVEL(2) LOG_NONMEMBER("connection::to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
         return nullptr;
     }
     IF_LEVEL(2) LOG_NONMEMBER("connection::to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));

-    return make_shared<connection>(mgr, move(s), lossy);
-}
-
-void connection::closeconn() {
-    {
-        lock ml(m_);
-        if (dead_)
-            return;
-        dead_ = true;
-        shutdown(fd_,SHUT_RDWR);
-    }
-    //after block_remove_fd, select will never wait on fd_
-    //and no callbacks will be active
-    poll_mgr::shared_mgr.block_remove_fd(fd_);
+    return make_shared<connection>(delegate, move(s), lossy);

 }
 
 bool connection::send(const string & b) {
 }
 
 bool connection::send(const string & b) {

@@ -65,19 +59,19 @@ bool connection::send(const string & b) {
 
     if (lossy_) {
         if ((random()%100) < lossy_) {
 
     if (lossy_) {
         if ((random()%100) < lossy_) {

-            IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd_ " << fd_);
-            shutdown(fd_,SHUT_RDWR);
+            IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd " << fd);
+            shutdown(fd,SHUT_RDWR);

         }
     }
 
     if (!writepdu()) {
         dead_ = true;
         ml.unlock();
         }
     }
 
     if (!writepdu()) {
         dead_ = true;
         ml.unlock();

-        poll_mgr::shared_mgr.block_remove_fd(fd_);
+        poll_mgr::shared_mgr.block_remove_fd(fd);

         ml.lock();
     } else if (wpdu_.solong != wpdu_.buf.size()) {
         // should be rare to need to explicitly add write callback
         ml.lock();
     } else if (wpdu_.solong != wpdu_.buf.size()) {
         // should be rare to need to explicitly add write callback

-        poll_mgr::shared_mgr.add_callback(fd_, CB_WRONLY, this);
+        poll_mgr::shared_mgr.add_callback(fd, CB_WRONLY, this);

         while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.size())
             send_complete_.wait(ml);
     }
         while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.size())
             send_complete_.wait(ml);
     }

@@ -89,17 +83,17 @@ bool connection::send(const string & b) {
     return ret;
 }
 
     return ret;
 }
 

-//fd_ is ready to be written
+// fd is ready to be written

 void connection::write_cb(int s) {
     lock ml(m_);
     VERIFY(!dead_);
 void connection::write_cb(int s) {
     lock ml(m_);
     VERIFY(!dead_);

-    VERIFY(fd_ == s);
+    VERIFY(fd == s);

     if (wpdu_.buf.size() == 0) {
     if (wpdu_.buf.size() == 0) {

-        poll_mgr::shared_mgr.del_callback(fd_,CB_WRONLY);
+        poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY);

         return;
     }
     if (!writepdu()) {
         return;
     }
     if (!writepdu()) {

-        poll_mgr::shared_mgr.del_callback(fd_, CB_RDWR);
+        poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);

         dead_ = true;
     } else {
         VERIFY(wpdu_.solong != size_t_max);
         dead_ = true;
     } else {
         VERIFY(wpdu_.solong != size_t_max);

@@ -110,28 +104,26 @@ void connection::write_cb(int s) {
     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) {
     lock ml(m_);
 void connection::read_cb(int s) {
     lock ml(m_);

-    VERIFY(fd_ == s);
+    VERIFY(fd == s);

     if (dead_)
         return;
 
     IF_LEVEL(5) LOG("got data on fd " << s);
 
     if (dead_)
         return;
 
     IF_LEVEL(5) LOG("got data on fd " << s);
 

-    bool succ = true;
-    if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size())
-        succ = readpdu();
-
-    if (!succ) {
-        IF_LEVEL(5) LOG("readpdu on fd " << s << " failed; dying");
-        poll_mgr::shared_mgr.del_callback(fd_,CB_RDWR);
-        dead_ = true;
-        send_complete_.notify_one();
+    if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size()) {
+        if (!readpdu()) {
+            IF_LEVEL(5) LOG("readpdu on fd " << s << " failed; dying");
+            poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);
+            dead_ = true;
+            send_complete_.notify_one();
+        }

     }
 
     if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) {
     }
 
     if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) {

-        if (mgr_->got_pdu(shared_from_this(), rpdu_.buf)) {
+        if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {

             // connection_delegate has successfully consumed the pdu
             rpdu_.buf.clear();
             rpdu_.solong = 0;
             // connection_delegate has successfully consumed the pdu
             rpdu_.buf.clear();
             rpdu_.solong = 0;

@@ -144,10 +136,10 @@ bool connection::writepdu() {
     if (wpdu_.solong == wpdu_.buf.size())
         return true;
 
     if (wpdu_.solong == wpdu_.buf.size())
         return true;
 

-    ssize_t n = write(fd_, &wpdu_.buf[wpdu_.solong], (wpdu_.buf.size()-wpdu_.solong));
+    ssize_t n = write(fd, &wpdu_.buf[wpdu_.solong], (wpdu_.buf.size()-wpdu_.solong));

     if (n < 0) {
         if (errno != EAGAIN) {
     if (n < 0) {
         if (errno != EAGAIN) {

-            IF_LEVEL(1) LOG("writepdu fd_ " << fd_ << " failure errno=" << errno);
+            IF_LEVEL(1) LOG("writepdu fd " << fd << " failure errno=" << errno);

             wpdu_.solong = size_t_max;
             wpdu_.buf.clear();
         }
             wpdu_.solong = size_t_max;
             wpdu_.buf.clear();
         }

@@ -161,7 +153,7 @@ bool connection::readpdu() {
     IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size());
     if (!rpdu_.buf.size()) {
         rpc_protocol::rpc_sz_t sz1;
     IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size());
     if (!rpdu_.buf.size()) {
         rpc_protocol::rpc_sz_t sz1;

-        ssize_t n = fd_.read(sz1);
+        ssize_t n = fd.read(sz1);

 
         if (n == 0)
             return false;
 
         if (n == 0)
             return false;

@@ -185,12 +177,11 @@ bool connection::readpdu() {
 
         IF_LEVEL(5) LOG("read size of datagram = " << sz);
 
 
         IF_LEVEL(5) LOG("read size of datagram = " << sz);
 

-        VERIFY(rpdu_.buf.size() == 0);
-        rpdu_.buf = string(sz+sizeof(sz1), 0);
+        rpdu_.buf.assign(sz+sizeof(sz1), 0);

         rpdu_.solong = sizeof(sz1);
     }
 
         rpdu_.solong = sizeof(sz1);
     }
 

-    ssize_t n = fd_.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);
+    ssize_t n = fd.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);

 
     IF_LEVEL(5) LOG("read " << n << " bytes");
 
 
     IF_LEVEL(5) LOG("read " << n << " bytes");
 

@@ -199,27 +190,25 @@ bool connection::readpdu() {
             return true;
         rpdu_.buf.clear();
         rpdu_.solong = 0;
             return true;
         rpdu_.buf.clear();
         rpdu_.solong = 0;

-        return (errno == EAGAIN);
+        return false;

     }
     rpdu_.solong += (size_t)n;
     return true;
 }
 
     }
     rpdu_.solong += (size_t)n;
     return true;
 }
 

-tcpsconn::tcpsconn(connection_delegate *m1, in_port_t port, int lossytest)
-: tcp_(socket(AF_INET, SOCK_STREAM, 0)), mgr_(m1), lossy_(lossytest)
+connection_listener::connection_listener(connection_delegate * delegate, in_port_t port, int lossytest)
+: tcp_(socket(AF_INET, SOCK_STREAM, 0)), delegate_(delegate), lossy_(lossytest)

 {
 {

-    sockaddr_in sin{}; // zero initialize
-    sin.sin_family = AF_INET;
-    sin.sin_port = hton(port);
-

     tcp_.setsockopt(SOL_SOCKET, SO_REUSEADDR, (int)1);
     tcp_.setsockopt(IPPROTO_TCP, TCP_NODELAY, (int)1);
     tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeval{0, 50000});
     tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeval{0, 50000});
 
     tcp_.setsockopt(SOL_SOCKET, SO_REUSEADDR, (int)1);
     tcp_.setsockopt(IPPROTO_TCP, TCP_NODELAY, (int)1);
     tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeval{0, 50000});
     tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeval{0, 50000});
 

-    // careful to exactly match type signature of bind arguments so we don't
-    // get std::bind instead
-    if (bind((int)tcp_, (const sockaddr *)&sin, (socklen_t)sizeof(sin)) < 0) {
+    sockaddr_in sin{}; // zero initialize
+    sin.sin_family = AF_INET;
+    sin.sin_port = hton(port);
+
+    if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) {

         perror("accept_loop bind");
         VERIFY(0);
     }
         perror("accept_loop bind");
         VERIFY(0);
     }

@@ -238,25 +227,21 @@ tcpsconn::tcpsconn(connection_delegate *m1, in_port_t port, int lossytest)
     poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);
 }
 
     poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);
 }
 

-tcpsconn::~tcpsconn()
-{
+connection_listener::~connection_listener() {

     poll_mgr::shared_mgr.block_remove_fd(tcp_);
     poll_mgr::shared_mgr.block_remove_fd(tcp_);

-
-    for (auto & i : conns_)
-        i.second->closeconn();

 }
 
 }
 

-void tcpsconn::read_cb(int) {
+void connection_listener::read_cb(int) {

     sockaddr_in sin;
     socklen_t slen = sizeof(sin);
     int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
     if (s1 < 0) {
     sockaddr_in sin;
     socklen_t slen = sizeof(sin);
     int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
     if (s1 < 0) {

-        perror("tcpsconn::accept_conn error");
+        perror("connection_listener::accept_conn error");

         throw thread_exit_exception();
     }
 
     IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port));
         throw thread_exit_exception();
     }
 
     IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port));

-    auto ch = make_shared<connection>(mgr_, s1, lossy_);
+    auto ch = make_shared<connection>(delegate_, s1, lossy_);

 
     // garbage collect dead connections
     for (auto i = conns_.begin(); i != conns_.end();) {
 
     // garbage collect dead connections
     for (auto i = conns_.begin(); i != conns_.end();) {

@@ -266,5 +251,5 @@ void tcpsconn::read_cb(int) {
             ++i;
     }
 
             ++i;
     }
 

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

 }
 }