So many changes. Broken.

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

diff --git a/rpc/connection.cc b/rpc/connection.cc
index c22ad45..7a6371a 100644 (file)
--- a/rpc/connection.cc
+++ b/rpc/connection.cc
@@ -1,448 +1,243 @@
-#include <fcntl.h>
-#include <sys/types.h>
-#include <sys/time.h>
+#include "include/rpc/connection.h"
+#include "include/rpc/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 "include/rpc/marshall.h"
+
+connection_delegate::~connection_delegate() {}
+
+connection::connection(connection_delegate * delegate, socket_t && f1, int l1)
+: fd(std::move(f1)), delegate_(delegate), lossy_(l1)
+{
+    fd.flags() |= O_NONBLOCK;
+
+    signal(SIGPIPE, SIG_IGN);
+
+    global->shared_mgr.add_callback(fd, CB_RDONLY, this);
+}
+
+connection::~connection() {
+    {
+        lock ml(m_);
+        if (dead_)
+            return;
+        dead_ = true;
+        fd.shutdown(SHUT_RDWR);
+    }
+    // after block_remove_fd, select will never wait on fd and no callbacks
+    // will be active
+    global->shared_mgr.block_remove_fd(fd);
+    VERIFY(dead_ && wpdu_.status == unused);
+}
+
+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_LEVEL(2) LOG_NONMEMBER << "connection::to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port);
+    return std::make_shared<connection>(delegate, std::move(s), lossy);
+}
+
+bool connection::send(const string & b) {
+    lock ml(m_);
+
+    while (!dead_ && wpdu_.status != unused)
+        send_wait_.wait(ml);
+
+    if (dead_)
+        return false;
+
+    wpdu_ = {inflight, b, 0};
+
+    if (std::bernoulli_distribution(lossy_*.01)(global->random_generator)) {
+        IF_LEVEL(1) LOG << "send LOSSY TEST shutdown fd " << fd;
+        fd.shutdown(SHUT_RDWR);
+    }
+
+    if (!writepdu()) {
+        dead_ = true;
+        ml.unlock();
+        global->shared_mgr.block_remove_fd(fd);
+        ml.lock();
+    } else if (wpdu_.status == inflight && wpdu_.cursor < b.size()) {
+        // should be rare to need to explicitly add write callback
+        global->shared_mgr.add_callback(fd, CB_WRONLY, this);
+        while (!dead_ && wpdu_.status == inflight && wpdu_.cursor < b.size())
+            send_complete_.wait(ml);
+    }
+    bool ret = (!dead_ && wpdu_.status == inflight && wpdu_.cursor == b.size());
+    wpdu_ = {unused, "", 0};
+    send_wait_.notify_all();
+    return ret;
+}
+
+// fd is ready to be written
+void connection::write_cb(int s) {
+    lock ml(m_);
+    VERIFY(!dead_);
+    VERIFY(fd == s);
+    if (wpdu_.status != inflight) {
+        global->shared_mgr.del_callback(fd, CB_WRONLY);
+        return;
+    }
+    if (!writepdu()) {
+        global->shared_mgr.del_callback(fd, CB_RDWR);
+        dead_ = true;
+    } else {
+        VERIFY(wpdu_.status != error);
+        if (wpdu_.cursor < wpdu_.buf.size())
+            return;
+    }
+    send_complete_.notify_one();
+}
+
+bool connection::writepdu() {
+    VERIFY(wpdu_.status == inflight);
+    if (wpdu_.cursor == wpdu_.buf.size())
+        return true;
+
+    ssize_t n = fd.write(&wpdu_.buf[wpdu_.cursor], (wpdu_.buf.size()-wpdu_.cursor));
+    if (n < 0) {
+        if (errno != EAGAIN) {
+            IF_LEVEL(1) LOG << "writepdu fd " << fd << " failure errno=" << errno;
+            wpdu_ = {error, "", 0};
+        }
+        return (errno == EAGAIN);
+    }
+    wpdu_.cursor += (size_t)n;
+    return true;
+}
+
+// fd is ready to be read
+void connection::read_cb(int s) {
+    lock ml(m_);
+    VERIFY(fd == s);
+    if (dead_)
+        return;
+
+    IF_LEVEL(5) LOG << "got data on fd " << s;
+
+    if (rpdu_.status == unused || rpdu_.cursor < rpdu_.buf.size()) {
+        if (!readpdu()) {
+            IF_LEVEL(5) LOG << "readpdu on fd " << s << " failed; dying";
+            global->shared_mgr.del_callback(fd, CB_RDWR);
+            dead_ = true;
+            send_complete_.notify_one();
+        }
+    }

 
 

-#include "method_thread.h"
-#include "connection.h"
-#include "slock.h"
-#include "pollmgr.h"
-#include "jsl_log.h"
-#include "gettime.h"
-#include "lang/verify.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)
-{
-
-       int flags = fcntl(fd_, F_GETFL, NULL);
-       flags |= O_NONBLOCK;
-       fcntl(fd_, F_SETFL, flags);
-
-       signal(SIGPIPE, SIG_IGN);
-       VERIFY(pthread_mutex_init(&m_,0)==0);
-       VERIFY(pthread_mutex_init(&ref_m_,0)==0);
-       VERIFY(pthread_cond_init(&send_wait_,0)==0);
-       VERIFY(pthread_cond_init(&send_complete_,0)==0);
- 
-        VERIFY(gettimeofday(&create_time_, NULL) == 0); 
-
-       PollMgr::Instance()->add_callback(fd_, CB_RDONLY, this);
-}
-
-connection::~connection()
-{
-       VERIFY(dead_);
-       VERIFY(pthread_mutex_destroy(&m_)== 0);
-       VERIFY(pthread_mutex_destroy(&ref_m_)== 0);
-       VERIFY(pthread_cond_destroy(&send_wait_) == 0);
-       VERIFY(pthread_cond_destroy(&send_complete_) == 0);
-       if (rpdu_.buf)
-               free(rpdu_.buf);
-       VERIFY(!wpdu_.buf);
-       close(fd_);
+    if (rpdu_.status == inflight && rpdu_.buf.size() == rpdu_.cursor) {
+        if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {
+            // connection_delegate has successfully consumed the pdu
+            rpdu_ = {unused, "", 0};
+        }
+    }

 }
 
 }
 

-void
-connection::incref()
-{
-       ScopedLock ml(&ref_m_);
-       refno_++;
-}
+bool connection::readpdu() {
+    IF_LEVEL(5) LOG << "the receive buffer has length " << rpdu_.buf.size();
+    if (rpdu_.status == unused) {
+        rpc_protocol::rpc_sz_t sz1;
+        ssize_t n = fd.read(sz1);

 
 

-bool
-connection::isdead()
-{
-       ScopedLock ml(&m_);
-       return dead_;
-}
+        if (n == 0)
+            return false;

 
 

-void
-connection::closeconn()
-{
-       {
-               ScopedLock ml(&m_);
-               if (!dead_) {
-                       dead_ = true;
-                       shutdown(fd_,SHUT_RDWR);
-               }else{
-                       return;
-               }
-       }
-       //after block_remove_fd, select will never wait on fd_ 
-       //and no callbacks will be active
-       PollMgr::Instance()->block_remove_fd(fd_);
-}
+        if (n < 0) {
+            VERIFY(errno!=EAGAIN);
+            return false;
+        }

 
 

-void
-connection::decref()
-{
-       VERIFY(pthread_mutex_lock(&ref_m_)==0);
-       refno_ --;
-       VERIFY(refno_>=0);
-       if (refno_==0) {
-               VERIFY(pthread_mutex_lock(&m_)==0);
-               if (dead_) {
-                       VERIFY(pthread_mutex_unlock(&ref_m_)==0);
-                       VERIFY(pthread_mutex_unlock(&m_)==0);
-                       delete this;
-                       return;
-               }
-               VERIFY(pthread_mutex_unlock(&m_)==0);
-       }
-       pthread_mutex_unlock(&ref_m_);
-}
+        if (n > 0 && n != sizeof(sz1)) {
+            IF_LEVEL(0) LOG << "short read of sz";
+            return false;
+        }

 
 

-int
-connection::ref()
-{
-       ScopedLock rl(&ref_m_);
-       return refno_;
-}
+        size_t sz = ntoh(sz1);

 
 

-int
-connection::compare(connection *another)
-{
-        if (create_time_.tv_sec > another->create_time_.tv_sec)
-                return 1;
-        if (create_time_.tv_sec < another->create_time_.tv_sec)
-                return -1;
-        if (create_time_.tv_usec > another->create_time_.tv_usec)
-                return 1;
-        if (create_time_.tv_usec < another->create_time_.tv_usec)
-                return -1;
-        return 0;
-}
+        if (sz > rpc_protocol::MAX_PDU) {
+            IF_LEVEL(2) LOG << "read pdu TOO BIG " << sz << " network order=" << std::hex << sz1;
+            return false;
+        }

 
 

-bool
-connection::send(char *b, int sz)
-{
-       ScopedLock ml(&m_);
-       waiters_++;
-       while (!dead_ && wpdu_.buf) {
-               VERIFY(pthread_cond_wait(&send_wait_, &m_)==0);
-       }
-       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;
-               VERIFY(pthread_mutex_unlock(&m_) == 0);
-               PollMgr::Instance()->block_remove_fd(fd_);
-               VERIFY(pthread_mutex_lock(&m_) == 0);
-       }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 >= 0 && wpdu_.solong < wpdu_.sz) {
-                               VERIFY(pthread_cond_wait(&send_complete_,&m_) == 0);
-                       }
-               }
-       }
-       bool ret = (!dead_ && wpdu_.solong == wpdu_.sz);
-       wpdu_.solong = wpdu_.sz = 0;
-       wpdu_.buf = NULL;
-       if (waiters_ > 0)
-               pthread_cond_broadcast(&send_wait_);
-       return ret;
-}
+        IF_LEVEL(5) LOG << "read size of datagram = " << sz;

 
 

-//fd_ is ready to be written
-void
-connection::write_cb(int s)
-{
-       ScopedLock 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 >= 0);
-               if (wpdu_.solong < wpdu_.sz) {
-                       return;
-               }
-       } 
-       pthread_cond_signal(&send_complete_);
-}
+        rpdu_ = {inflight, string(sz+sizeof(sz1), 0), sizeof(sz1)};
+    }

 
 

-//fd_ is ready to be read
-void
-connection::read_cb(int s)
-{
-       ScopedLock 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;
-               pthread_cond_signal(&send_complete_);
-       }
-
-       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;
-               }
-       }
-}
+    ssize_t n = fd.read(&rpdu_.buf[rpdu_.cursor], rpdu_.buf.size() - rpdu_.cursor);

 
 

-bool
-connection::writepdu()
-{
-       VERIFY(wpdu_.solong >= 0);
-       if (wpdu_.solong == wpdu_.sz)
-               return true;
-
-       if (wpdu_.solong == 0) {
-               int sz = htonl(wpdu_.sz);
-               bcopy(&sz,wpdu_.buf,sizeof(sz));
-       }
-       int 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 = -1;
-                       wpdu_.sz = 0;
-               }
-               return (errno == EAGAIN);
-       }
-       wpdu_.solong += n;
-       return true;
-}
+    IF_LEVEL(5) LOG << "read " << n << " bytes";

 
 

-bool
-connection::readpdu()
-{
-       if (!rpdu_.sz) {
-               int sz, sz1;
-               int n = read(fd_, &sz1, sizeof(sz1));
-
-               if (n == 0) {
-                       return false;
-               }
-
-               if (n < 0) {
-                       VERIFY(errno!=EAGAIN);
-                       return false;
-               }
-
-               if (n >0 && n!= sizeof(sz)) {
-                       jsl_log(JSL_DBG_OFF, "connection::readpdu short read of sz\n");
-                       return false;
-               }
-
-               sz = ntohl(sz1);
-
-               if (sz > MAX_PDU) {
-                       char *tmpb = (char *)&sz1;
-                       jsl_log(JSL_DBG_2, "connection::readpdu read pdu TOO BIG %d 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(sz));
-               VERIFY(rpdu_.buf);
-               bcopy(&sz1,rpdu_.buf,sizeof(sz));
-               rpdu_.solong = sizeof(sz);
-       }
-
-       int 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 += n;
-       return true;
+    if (n <= 0) {
+        if (errno == EAGAIN)
+            return true;
+        rpdu_ = {unused, "", 0};
+        return false;
+    }
+    rpdu_.cursor += (size_t)n;
+    return true;

 }
 
 }
 

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

 {
 {

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

 
 

-       VERIFY(pthread_mutex_init(&m_,NULL) == 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);
-       }
+    sockaddr_in sin = sockaddr_in(); // zero initialize
+    sin.sin_family = AF_INET;
+    sin.sin_port = hton(port);

 
 

-       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 bind");
+        VERIFY(0);
+    }

 
 

-       if(bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0){
-               perror("accept_loop tcp bind:");
-               VERIFY(0);
-       }
+    if (listen(tcp_, 1000) < 0) {
+        perror("accept_loop listen");
+        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_ = ntoh(sin.sin_port);

 
 

-        socklen_t addrlen = sizeof(sin);
-        VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
-        port_ = ntohs(sin.sin_port);
+    IF_LEVEL(2) LOG << "listen on " << port_ << " " << sin.sin_port;

 
 

-       jsl_log(JSL_DBG_2, "tcpsconn::tcpsconn listen on %d %d\n", port_, 
-               sin.sin_port);
-
-       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);
-
-       VERIFY((th_ = method_thread(this, false, &tcpsconn::accept_conn)) != 0); 
+    global->shared_mgr.add_callback(tcp_, CB_RDONLY, this);

 }
 
 }
 

-tcpsconn::~tcpsconn()
-{
-       VERIFY(close(pipe_[1]) == 0);
-       VERIFY(pthread_join(th_, NULL) == 0);
-
-       //close all the active connections
-       std::map<int, connection *>::iterator i;
-       for (i = conns_.begin(); i != conns_.end(); i++) {
-               i->second->closeconn();
-               i->second->decref();
-       }       
+connection_listener::~connection_listener() {
+    global->shared_mgr.block_remove_fd(tcp_);

 }
 
 }
 

-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");
-               pthread_exit(NULL);
-       }
-
-       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_);
-
-        // garbage collect all dead connections with refcount of 1
-        std::map<int, connection *>::iterator i;
-        for (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());
-                        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.
-                        conns_.erase(i++);
-                } else
-                        ++i;
-        }
+void connection_listener::read_cb(int) {
+    sockaddr_in sin;
+    socklen_t slen = sizeof(sin);
+    int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
+    if (s1 < 0) {
+        perror("connection_listener::accept_conn error");
+        throw std::runtime_error("connection listener failure");
+    }

 
 

-       conns_[ch->channo()] = ch;
-}
+    IF_LEVEL(2) LOG << "accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port);

 
 

-void
-tcpsconn::accept_conn()
-{
-       fd_set rfds;
-       int max_fd = pipe_[0] > tcp_ ? pipe_[0] : tcp_;
-
-       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:");
-                               jsl_log(JSL_DBG_OFF, "tcpsconn::accept_conn failure errno %d\n",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);
-               }
-       }
-}
+    // garbage collect dead connections
+    for (auto i = conns_.begin(); i != conns_.end();) {
+        if (i->second->isdead())
+            conns_.erase(i++);
+        else
+            ++i;
+    }

 
 

-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);
+    conns_[s1] = std::make_shared<connection>(delegate_, s1, lossy_);

 }
 }

-
-