X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/dfe8486473094c0769fd1922329c3f0dfd8f43c0..f0dcb6b97d6d40f67698d1f71ac26970f1776f82:/rpc/connection.cc diff --git a/rpc/connection.cc b/rpc/connection.cc index 6e865e8..4e49305 100644 --- a/rpc/connection.cc +++ b/rpc/connection.cc @@ -1,434 +1,255 @@ -#include -#include -#include +#include "connection.h" +#include "rpc_protocol.h" +#include +#include #include -#include -#include #include +#include "marshall.h" -#include "connection.h" -#include "pollmgr.h" -#include "jsl_log.h" -#include "lang/verify.h" -#include "lock.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) +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); - create_time_ = std::chrono::steady_clock::now(); - - PollMgr::Instance()->add_callback(fd_, CB_RDONLY, this); -} - -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_; + poll_mgr::shared_mgr.add_callback(fd, CB_RDONLY, this); } -void -connection::closeconn() -{ +connection::~connection() { { lock ml(m_); - if (!dead_) { - dead_ = true; - shutdown(fd_,SHUT_RDWR); - } else { + 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 - 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_; - } - } - if (dead) { - delete this; +shared_ptr 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 make_shared(delegate, move(s), lossy); } -int -connection::ref() -{ - lock rl(ref_m_); - return refno_; -} - -int -connection::compare(connection *another) -{ - if (create_time_ > another->create_time_) - return 1; - if (create_time_ < another->create_time_) - return -1; - return 0; -} - -bool -connection::send(char *b, int sz) -{ +bool connection::send(const string & b) { lock ml(m_); - waiters_++; - while (!dead_ && wpdu_.buf) { + + waiters_++; + while (!dead_ && wpdu_.buf.size()) 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_.solong = 0; + + if (lossy_) { + if ((random()%100) < lossy_) { + IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd " << fd); + shutdown(fd,SHUT_RDWR); + } + } + + if (!writepdu()) { + dead_ = true; ml.unlock(); - PollMgr::Instance()->block_remove_fd(fd_); + poll_mgr::shared_mgr.block_remove_fd(fd); 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 >= 0 && wpdu_.solong < wpdu_.sz) { - send_complete_.wait(ml); - } - } - } - bool ret = (!dead_ && wpdu_.solong == wpdu_.sz); - wpdu_.solong = wpdu_.sz = 0; - wpdu_.buf = NULL; - if (waiters_ > 0) + } 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_.buf.size()); + wpdu_.solong = 0; + wpdu_.buf.clear(); + if (waiters_ > 0) send_wait_.notify_all(); - return ret; + return ret; } -//fd_ is ready to be written -void -connection::write_cb(int s) -{ +// fd is ready to be written +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); - return; - } - if (!writepdu()) { - PollMgr::Instance()->del_callback(fd_, CB_RDWR); - dead_ = true; - } else { - VERIFY(wpdu_.solong >= 0); - if (wpdu_.solong < wpdu_.sz) { - return; - } + VERIFY(!dead_); + VERIFY(fd == s); + if (wpdu_.buf.size() == 0) { + poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY); + return; } - send_complete_.notify_one(); + if (!writepdu()) { + poll_mgr::shared_mgr.del_callback(fd, CB_RDWR); + dead_ = true; + } else { + VERIFY(wpdu_.solong != size_t_max); + if (wpdu_.solong < wpdu_.buf.size()) { + return; + } + } + send_complete_.notify_one(); } -//fd_ is ready to be read -void -connection::read_cb(int s) -{ +// fd is ready to be read +void connection::read_cb(int s) { 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 -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("got data on fd " << s); + + 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 (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) { + // connection_delegate has successfully consumed the pdu + rpdu_.buf.clear(); + rpdu_.solong = 0; + } + } } -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; +bool connection::writepdu() { + VERIFY(wpdu_.solong != size_t_max); + if (wpdu_.solong == wpdu_.buf.size()) + return true; + + ssize_t n = write(fd, &wpdu_.buf[wpdu_.solong], (wpdu_.buf.size()-wpdu_.solong)); + if (n < 0) { + if (errno != EAGAIN) { + IF_LEVEL(1) LOG("writepdu fd " << fd << " failure errno=" << errno); + wpdu_.solong = size_t_max; + wpdu_.buf.clear(); + } + return (errno == EAGAIN); + } + wpdu_.solong += (size_t)n; + return true; } -tcpsconn::tcpsconn(chanmgr *m1, 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); - } +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; + ssize_t n = fd.read(sz1); - socklen_t addrlen = sizeof(sin); - VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0); - port_ = ntohs(sin.sin_port); + if (n == 0) + return false; - jsl_log(JSL_DBG_2, "tcpsconn::tcpsconn listen on %d %d\n", port_, - sin.sin_port); + if (n < 0) { + VERIFY(errno!=EAGAIN); + return false; + } - if (pipe(pipe_) < 0) { - perror("accept_loop pipe:"); - VERIFY(0); - } + if (n > 0 && n != sizeof(sz1)) { + IF_LEVEL(0) LOG("short read of sz"); + return false; + } - int flags = fcntl(pipe_[0], F_GETFL, NULL); - flags |= O_NONBLOCK; - fcntl(pipe_[0], F_SETFL, flags); + size_t sz = ntoh(sz1); - th_ = std::thread(&tcpsconn::accept_conn, this); -} + if (sz > rpc_protocol::MAX_PDU) { + IF_LEVEL(2) LOG("read pdu TOO BIG " << sz << " network order=" << hex << sz1); + return false; + } -tcpsconn::~tcpsconn() -{ - VERIFY(close(pipe_[1]) == 0); - th_.join(); - - //close all the active connections - std::map::iterator i; - for (i = conns_.begin(); i != conns_.end(); i++) { - i->second->closeconn(); - i->second->decref(); - } -} + IF_LEVEL(5) LOG("read size of datagram = " << sz); -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_); - - // garbage collect all dead connections with refcount of 1 - std::map::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; + rpdu_.buf.assign(sz+sizeof(sz1), 0); + rpdu_.solong = sizeof(sz1); } - conns_[ch->channo()] = ch; + 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; + rpdu_.buf.clear(); + rpdu_.solong = 0; + return false; + } + rpdu_.solong += (size_t)n; + return true; } -void -tcpsconn::accept_conn() +connection_listener::connection_listener(connection_delegate * delegate, in_port_t port, int lossytest) +: tcp_(socket(AF_INET, SOCK_STREAM, 0)), delegate_(delegate), lossy_(lossytest) { - 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:"); - 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); - } - } + 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}); + + 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); } - catch (thread_exit_exception e) - { + + if (listen(tcp_, 1000) < 0) { + perror("accept_loop listen"); + VERIFY(0); } + + socklen_t addrlen = sizeof(sin); + VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0); + port_ = ntoh(sin.sin_port); + + IF_LEVEL(2) LOG("listen on " << port_ << " " << sin.sin_port); + + poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this); } -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_listener::~connection_listener() { + poll_mgr::shared_mgr.block_remove_fd(tcp_); } +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 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(delegate_, s1, lossy_); + + // garbage collect dead connections + for (auto i = conns_.begin(); i != conns_.end();) { + if (i->second->isdead()) + conns_.erase(i++); + else + ++i; + } + + conns_[s1] = ch; +}