X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/5d99dbf06a14904944f5593c63705934bdfdcfb7..f0dcb6b97d6d40f67698d1f71ac26970f1776f82:/rpc/connection.cc diff --git a/rpc/connection.cc b/rpc/connection.cc index 55e374a..4e49305 100644 --- a/rpc/connection.cc +++ b/rpc/connection.cc @@ -1,79 +1,46 @@ -// 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 "rpc_protocol.h" #include #include -#include -#include #include #include -#include +#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); - fcntl(fd_, F_SETFL, flags | O_NONBLOCK); + fd.flags() |= O_NONBLOCK; 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() { - VERIFY(dead_); - VERIFY(!wpdu_.buf.size()); - close(fd_); -} - -void connection::incref() { - lock rl(ref_m_); - refno_++; -} - -bool connection::isdead() { - lock ml(m_); - return dead_; -} - -void connection::closeconn() { { lock ml(m_); if (dead_) return; dead_ = true; - shutdown(fd_,SHUT_RDWR); + 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::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(delegate, move(s), lossy); } bool connection::send(const string & b) { @@ -92,19 +59,19 @@ bool connection::send(const string & b) { 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(); - PollMgr::Instance()->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 - PollMgr::Instance()->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); } @@ -116,17 +83,17 @@ bool connection::send(const string & b) { 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_); - VERIFY(fd_ == s); + VERIFY(fd == s); if (wpdu_.buf.size() == 0) { - PollMgr::Instance()->del_callback(fd_,CB_WRONLY); + poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY); 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); @@ -137,31 +104,27 @@ void connection::write_cb(int s) { 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_); - VERIFY(fd_ == s); - if (dead_) { + VERIFY(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"); - PollMgr::Instance()->del_callback(fd_,CB_RDWR); - dead_ = true; - send_complete_.notify_one(); + 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 (mgr_->got_pdu(this, rpdu_.buf)) { - //chanmgr has successfully consumed the pdu + if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) { + // connection_delegate has successfully consumed the pdu rpdu_.buf.clear(); rpdu_.solong = 0; } @@ -173,14 +136,10 @@ bool connection::writepdu() { if (wpdu_.solong == wpdu_.buf.size()) return true; - if (wpdu_.solong == 0) { - uint32_t sz = htonl((uint32_t)wpdu_.buf.size() - sizeof(uint32_t)); - copy((const char *)&sz, (const char *)(&sz+1), &wpdu_.buf[0]); - } - 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_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_.buf.clear(); } @@ -193,12 +152,11 @@ bool connection::writepdu() { bool connection::readpdu() { IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size()); if (!rpdu_.buf.size()) { - uint32_t sz1; - ssize_t n = read(fd_, &sz1, sizeof(sz1)); + rpc_protocol::rpc_sz_t sz1; + ssize_t n = fd.read(sz1); - if (n == 0) { + if (n == 0) return false; - } if (n < 0) { VERIFY(errno!=EAGAIN); @@ -210,22 +168,20 @@ bool connection::readpdu() { return false; } - size_t sz = ntohl(sz1); + size_t sz = ntoh(sz1); - if (sz > MAX_PDU) { + if (sz > rpc_protocol::MAX_PDU) { IF_LEVEL(2) LOG("read pdu TOO BIG " << sz << " network order=" << hex << sz1); return false; } IF_LEVEL(5) LOG("read size of datagram = " << sz); - VERIFY(rpdu_.buf.size() == 0); - rpdu_.buf = string(sz+sizeof(sz1), 0); - copy((const char *)&sz1, (const char *)(&sz1 + 1), &rpdu_.buf[0]); + rpdu_.buf.assign(sz+sizeof(sz1), 0); rpdu_.solong = sizeof(sz1); } - ssize_t n = read(fd_, &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"); @@ -234,149 +190,66 @@ bool connection::readpdu() { return true; rpdu_.buf.clear(); rpdu_.solong = 0; - return (errno == EAGAIN); + return false; } 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("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) { - perror("accept_loop tcp bind:"); + perror("accept_loop bind"); VERIFY(0); } if (listen(tcp_, 1000) < 0) { - perror("listen:"); + perror("accept_loop listen"); VERIFY(0); } socklen_t addrlen = sizeof(sin); VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0); - port_ = ntohs(sin.sin_port); + port_ = ntoh(sin.sin_port); IF_LEVEL(2) LOG("listen on " << 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); - - 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::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) { - 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) << ":" << 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(delegate_, s1, lossy_); - // garbage collect all dead connections with refcount of 1 + // garbage collect dead connections 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++); - } else + else ++i; } - conns_[ch->channo()] = 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) - { - } + conns_[s1] = ch; } - -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); -} -