Cosmetic improvements.

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

diff --git a/rpc/connection.cc b/rpc/connection.cc
index 86d4ec5..4e49305 100644 (file)
--- a/rpc/connection.cc
+++ b/rpc/connection.cc
@@ -1,194 +1,147 @@
-// std::bind and syscall bind have the same name, so don't use std::bind in this file
-#define LIBT4_NO_FUNCTIONAL

 #include "connection.h"
 #include "connection.h"

-#include <fcntl.h>
-#include <sys/types.h>
+#include "rpc_protocol.h"
+#include <cerrno>
+#include <csignal>

 #include <netinet/tcp.h>
 #include <netinet/tcp.h>

-#include <errno.h>
-#include <signal.h>

 #include <unistd.h>
 #include <unistd.h>

-#include <sys/socket.h>
+#include "marshall.h"

 
 

-#define MAX_PDU (10<<20) //maximum PDF is 10M
-
-connection::connection(chanmgr *m1, int f1, int l1)
-: mgr_(m1), fd_(f1), lossy_(l1)
+connection::connection(connection_delegate * delegate, socket_t && f1, int l1)
+: fd(move(f1)), delegate_(delegate), lossy_(l1)

 {
 {

-    int flags = fcntl(fd_, F_GETFL, NULL);
-    flags |= O_NONBLOCK;
-    fcntl(fd_, F_SETFL, flags);
+    fd.flags() |= O_NONBLOCK;

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

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

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

-    VERIFY(dead_);
-    if (rpdu_.buf)
-        free(rpdu_.buf);
-    VERIFY(!wpdu_.buf);
-    close(fd_);
-}
-
-void connection::incref() {
-    lock rl(ref_m_);
-    refno_++;
-}
-
-bool connection::isdead() {
-    lock ml(m_);
-    return dead_;
-}
-
-void connection::closeconn() {

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

-        if (!dead_) {
-            dead_ = true;
-            shutdown(fd_,SHUT_RDWR);
-        } else {
+        if (dead_)

             return;
             return;

-        }
+        dead_ = true;
+        shutdown(fd,SHUT_RDWR);

     }
     }

-    //after block_remove_fd, select will never wait on fd_
-    //and no callbacks will be active
-    PollMgr::Instance()->block_remove_fd(fd_);
+    // 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());

 }
 
 }
 

-void connection::decref() {
-    bool dead = false;
-    {
-        lock rl(ref_m_);
-        refno_--;
-        VERIFY(refno_>=0);
-        if (refno_==0) {
-            lock ml(m_);
-            dead = dead_;
-        }
+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) {
+        IF_LEVEL(1) LOG_NONMEMBER("failed to " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
+        close(s);
+        return nullptr;

     }
     }

-    if (dead)
-        delete this;
-}
-
-int connection::compare(connection *another) {
-    if (create_time_ > another->create_time_)
-        return 1;
-    if (create_time_ < another->create_time_)
-        return -1;
-    return 0;
+    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>(delegate, move(s), lossy);

 }
 
 }
 

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

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

+

     waiters_++;
     waiters_++;

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

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

-    }

     waiters_--;
     waiters_--;

-    if (dead_) {
+
+    if (dead_)

         return false;
         return false;

-    }
+

     wpdu_.buf = b;
     wpdu_.buf = b;

-    wpdu_.sz = sz;

     wpdu_.solong = 0;
 
     if (lossy_) {
         if ((random()%100) < lossy_) {
     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_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();

-        PollMgr::Instance()->block_remove_fd(fd_);
+        poll_mgr::shared_mgr.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) {
-                send_complete_.wait(ml);
-            }
-        }
+    } 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);
+        while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.size())
+            send_complete_.wait(ml);

     }
     }

-    bool ret = (!dead_ && wpdu_.solong == wpdu_.sz);
-    wpdu_.solong = wpdu_.sz = 0;
-    wpdu_.buf = NULL;
+    bool ret = (!dead_ && wpdu_.solong == wpdu_.buf.size());
+    wpdu_.solong = 0;
+    wpdu_.buf.clear();

     if (waiters_ > 0)
         send_wait_.notify_all();
     return ret;
 }
 
     if (waiters_ > 0)
         send_wait_.notify_all();
     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);
-    if (wpdu_.sz == 0) {
-        PollMgr::Instance()->del_callback(fd_,CB_WRONLY);
+    VERIFY(fd == s);
+    if (wpdu_.buf.size() == 0) {
+        poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY);

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

-        PollMgr::Instance()->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);

-        if (wpdu_.solong < wpdu_.sz) {
+        if (wpdu_.solong < wpdu_.buf.size()) {

             return;
         }
     }
     send_complete_.notify_one();
 }
 
             return;
         }
     }
     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);
-    if (dead_)  {
+    VERIFY(fd == s);
+    if (dead_)

         return;
         return;

-    }

 
 

-    bool succ = true;
-    if (!rpdu_.buf || rpdu_.solong < rpdu_.sz) {
-        succ = readpdu();
-    }
+    IF_LEVEL(5) LOG("got data on fd " << s);

 
 

-    if (!succ) {
-        PollMgr::Instance()->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 && 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;
+    if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) {
+        if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {
+            // connection_delegate has successfully consumed the pdu
+            rpdu_.buf.clear();
+            rpdu_.solong = 0;

         }
     }
 }
 
 bool connection::writepdu() {
     VERIFY(wpdu_.solong != size_t_max);
         }
     }
 }
 
 bool connection::writepdu() {
     VERIFY(wpdu_.solong != size_t_max);

-    if (wpdu_.solong == wpdu_.sz)
+    if (wpdu_.solong == wpdu_.buf.size())

         return true;
 
         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));
+    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("connection::writepdu fd_ " << fd_ << " failure errno=" << errno);
+            IF_LEVEL(1) LOG("writepdu fd " << fd << " failure errno=" << errno);

             wpdu_.solong = size_t_max;
             wpdu_.solong = size_t_max;

-            wpdu_.sz = 0;
+            wpdu_.buf.clear();

         }
         return (errno == EAGAIN);
     }
         }
         return (errno == EAGAIN);
     }

@@ -197,13 +150,13 @@ bool connection::writepdu() {
 }
 
 bool connection::readpdu() {
 }
 
 bool connection::readpdu() {

-    if (!rpdu_.sz) {
-        uint32_t sz1;
-        ssize_t n = read(fd_, &sz1, sizeof(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);

 
 

-        if (n == 0) {
+        if (n == 0)

             return false;
             return false;

-        }

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

@@ -211,176 +164,92 @@ bool connection::readpdu() {
         }
 
         if (n > 0 && n != sizeof(sz1)) {
         }
 
         if (n > 0 && n != sizeof(sz1)) {

-            IF_LEVEL(0) LOG("connection::readpdu short read of sz");
+            IF_LEVEL(0) LOG("short read of sz");

             return false;
         }
 
             return false;
         }
 

-        size_t sz = ntohl(sz1);
+        size_t sz = ntoh(sz1);

 
 

-        if (sz > MAX_PDU) {
-            IF_LEVEL(2) LOG("connection::readpdu read pdu TOO BIG " << sz << " network order=" << hex << sz1);
+        if (sz > rpc_protocol::MAX_PDU) {
+            IF_LEVEL(2) LOG("read pdu TOO BIG " << sz << " network order=" << hex << sz1);

             return false;
         }
 
             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));
+        IF_LEVEL(5) LOG("read size of datagram = " << sz);
+
+        rpdu_.buf.assign(sz+sizeof(sz1), 0);

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

-    ssize_t n = read(fd_, rpdu_.buf + rpdu_.solong, rpdu_.sz - rpdu_.solong);
+    ssize_t n = fd.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);
+
+    IF_LEVEL(5) LOG("read " << n << " bytes");
+

     if (n <= 0) {
         if (errno == EAGAIN)
             return true;
     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_.buf.clear();
+        rpdu_.solong = 0;
+        return false;

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

-tcpsconn::tcpsconn(chanmgr *m1, unsigned int port, int lossytest)
-: 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)

 {
 {

-    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);
-    }
+    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});

 
 

-    int yes = 1;
-    setsockopt(tcp_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
-    setsockopt(tcp_, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
+    sockaddr_in sin{}; // zero initialize
+    sin.sin_family = AF_INET;
+    sin.sin_port = hton(port);

 
     if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) {
 
     if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) {

-        perror("accept_loop tcp bind:");
+        perror("accept_loop bind");

         VERIFY(0);
     }
 
     if (listen(tcp_, 1000) < 0) {
         VERIFY(0);
     }
 
     if (listen(tcp_, 1000) < 0) {

-        perror("tcpsconn::tcpsconn listen:");
+        perror("accept_loop listen");

         VERIFY(0);
     }
 
     socklen_t addrlen = sizeof(sin);
     VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
         VERIFY(0);
     }
 
     socklen_t addrlen = sizeof(sin);
     VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);

-    port_ = ntohs(sin.sin_port);
-
-    IF_LEVEL(2) LOG("tcpsconn::tcpsconn listen on " << port_ << " " << sin.sin_port);
+    port_ = ntoh(sin.sin_port);

 
 

-    if (pipe(pipe_) < 0) {
-        perror("accept_loop pipe:");
-        VERIFY(0);
-    }
+    IF_LEVEL(2) LOG("listen on " << port_ << " " << sin.sin_port);

 
 

-    int flags = fcntl(pipe_[0], F_GETFL, NULL);
-    flags |= O_NONBLOCK;
-    fcntl(pipe_[0], F_SETFL, flags);
-
-    th_ = thread(&tcpsconn::accept_conn, this);
+    poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);

 }
 
 }
 

-tcpsconn::~tcpsconn()
-{
-    VERIFY(close(pipe_[1]) == 0);
-    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();
-    }
+connection_listener::~connection_listener() {
+    poll_mgr::shared_mgr.block_remove_fd(tcp_);

 }
 
 }
 

-void tcpsconn::process_accept() {
+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();
     }
 
         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_);
+    IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port));
+    auto ch = make_shared<connection>(delegate_, s1, lossy_);

 
 

-    // garbage collect all dead connections with refcount of 1
+    // garbage collect dead connections

     for (auto i = conns_.begin(); i != conns_.end();) {
     for (auto i = conns_.begin(); i != conns_.end();) {

-        if (i->second->isdead() && i->second->ref() == 1) {
-            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++,
-            // there is a sequence point before a function call.
-            // See http://en.wikipedia.org/wiki/Sequence_point.
+        if (i->second->isdead())

             conns_.erase(i++);
             conns_.erase(i++);

-        } else
+        else

             ++i;
     }
 
             ++i;
     }
 

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

 }
 }

-
-void tcpsconn::accept_conn() {
-    fd_set rfds;
-    int max_fd = pipe_[0] > tcp_ ? pipe_[0] : tcp_;
-
-    try {
-        while (1) {
-            FD_ZERO(&rfds);
-            FD_SET(pipe_[0], &rfds);
-            FD_SET(tcp_, &rfds);
-
-            int ret = select(max_fd+1, &rfds, NULL, NULL, NULL);
-
-            if (ret < 0) {
-                if (errno == EINTR) {
-                    continue;
-                } else {
-                    perror("accept_conn select:");
-                    IF_LEVEL(0) LOG("tcpsconn::accept_conn failure errno " << errno);
-                    VERIFY(0);
-                }
-            }
-
-            if (FD_ISSET(pipe_[0], &rfds)) {
-                close(pipe_[0]);
-                close(tcp_);
-                return;
-            }
-            else if (FD_ISSET(tcp_, &rfds)) {
-                process_accept();
-            } else {
-                VERIFY(0);
-            }
-        }
-    }
-    catch (thread_exit_exception e)
-    {
-    }
-}
-
-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);
-}
-