X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/4a160f880ce46153acb23b137f30fd588df5fb9d..c279db4240a3a3c30f069ab9dea8055cf94280da:/rpc/connection.cc diff --git a/rpc/connection.cc b/rpc/connection.cc index 315a82e..8966cc2 100644 --- a/rpc/connection.cc +++ b/rpc/connection.cc @@ -6,162 +6,146 @@ #include #include "marshall.h" -connection::connection(connection_delegate *m1, socket_t && f1, int l1) -: mgr_(m1), fd_(move(f1)), lossy_(l1) +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; + fd.flags() |= O_NONBLOCK; signal(SIGPIPE, SIG_IGN); - create_time_ = steady_clock::now(); - - poll_mgr::shared_mgr.add_callback(fd_, CB_RDONLY, this); + global->shared_mgr.add_callback(fd, CB_RDONLY, this); } 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 + global->shared_mgr.block_remove_fd(fd); VERIFY(dead_); - VERIFY(!wpdu_.buf.size()); + VERIFY(wpdu_.status == unused); } -shared_ptr connection::to_dst(const sockaddr_in &dst, connection_delegate *mgr, int lossy) { +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)); + 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(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_); + 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(delegate, std::move(s), lossy); } bool connection::send(const string & b) { lock ml(m_); - waiters_++; - while (!dead_ && wpdu_.buf.size()) + while (!dead_ && wpdu_.status != unused) send_wait_.wait(ml); - waiters_--; if (dead_) return false; - wpdu_.buf = b; - wpdu_.solong = 0; + wpdu_ = {inflight, b, 0}; - if (lossy_) { - if ((random()%100) < lossy_) { - IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd_ " << fd_); - shutdown(fd_,SHUT_RDWR); - } + if (std::bernoulli_distribution(lossy_*.01)(global->random_generator)) { + IF_LEVEL(1) LOG << "send LOSSY TEST shutdown fd " << fd; + shutdown(fd,SHUT_RDWR); } if (!writepdu()) { dead_ = true; ml.unlock(); - poll_mgr::shared_mgr.block_remove_fd(fd_); + global->shared_mgr.block_remove_fd(fd); ml.lock(); - } else if (wpdu_.solong != wpdu_.buf.size()) { + } else if (wpdu_.status == inflight && wpdu_.cursor < b.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()) + 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_.solong == wpdu_.buf.size()); - wpdu_.solong = 0; - wpdu_.buf.clear(); - if (waiters_ > 0) - send_wait_.notify_all(); + 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 +// fd is ready to be written void connection::write_cb(int s) { lock ml(m_); VERIFY(!dead_); - VERIFY(fd_ == s); - if (wpdu_.buf.size() == 0) { - poll_mgr::shared_mgr.del_callback(fd_,CB_WRONLY); + VERIFY(fd == s); + if (wpdu_.status != inflight) { + global->shared_mgr.del_callback(fd, CB_WRONLY); return; } if (!writepdu()) { - poll_mgr::shared_mgr.del_callback(fd_, CB_RDWR); + global->shared_mgr.del_callback(fd, CB_RDWR); dead_ = true; } else { - VERIFY(wpdu_.solong != size_t_max); - if (wpdu_.solong < wpdu_.buf.size()) { + VERIFY(wpdu_.status != error); + if (wpdu_.cursor < wpdu_.buf.size()) return; - } } send_complete_.notify_one(); } -// fd_ is ready to be read +bool connection::writepdu() { + VERIFY(wpdu_.status == inflight); + if (wpdu_.cursor == wpdu_.buf.size()) + return true; + + ssize_t n = write(fd, &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); + 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_LEVEL(5) LOG << "got data on fd " << s; - 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_.buf.size() == rpdu_.solong) { - if (mgr_->got_pdu(shared_from_this(), rpdu_.buf)) { - // connection_delegate has successfully consumed the pdu - rpdu_.buf.clear(); - rpdu_.solong = 0; + 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(); } } -} -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(); + 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}; } - return (errno == EAGAIN); } - wpdu_.solong += (size_t)n; - return true; } bool connection::readpdu() { - IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size()); - if (!rpdu_.buf.size()) { - rpc_sz_t sz1; - ssize_t n = fd_.read(sz1); + 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); if (n == 0) return false; @@ -172,58 +156,49 @@ bool connection::readpdu() { } if (n > 0 && n != sizeof(sz1)) { - IF_LEVEL(0) LOG("short read of sz"); + IF_LEVEL(0) LOG << "short read of sz"; return false; } size_t sz = ntoh(sz1); - if (sz > MAX_PDU) { - IF_LEVEL(2) LOG("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=" << std::hex << sz1; return false; } - 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_.solong = sizeof(sz1); + rpdu_ = {inflight, string(sz+sizeof(sz1), 0), sizeof(sz1)}; } - ssize_t n = fd_.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong); + ssize_t n = fd.read(&rpdu_.buf[rpdu_.cursor], rpdu_.buf.size() - rpdu_.cursor); - IF_LEVEL(5) LOG("read " << n << " bytes"); + IF_LEVEL(5) LOG << "read " << n << " bytes"; if (n <= 0) { if (errno == EAGAIN) return true; - rpdu_.buf.clear(); - rpdu_.solong = 0; - return (errno == EAGAIN); + rpdu_ = {unused, "", 0}; + return false; } - rpdu_.solong += (size_t)n; + rpdu_.cursor += (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}); - if (tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeval{0, 50000}) < 0) - perror("accept_loop setsockopt"); - - if (tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeval{0, 50000}) < 0) - perror("accept_loop setsockopt"); + sockaddr_in sin = sockaddr_in(); // zero initialize + sin.sin_family = AF_INET; + sin.sin_port = hton(port); - // 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) { + if (bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0) { perror("accept_loop bind"); VERIFY(0); } @@ -237,35 +212,25 @@ tcpsconn::tcpsconn(connection_delegate *m1, in_port_t port, int lossytest) VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0); port_ = ntoh(sin.sin_port); - IF_LEVEL(2) LOG("listen on " << port_ << " " << sin.sin_port); + IF_LEVEL(2) LOG << "listen on " << port_ << " " << sin.sin_port; - file_t::pipe(pipe_); - - pipe_[0].flags() |= O_NONBLOCK; - - th_ = thread(&tcpsconn::accept_conn, this); + global->shared_mgr.add_callback(tcp_, CB_RDONLY, this); } -tcpsconn::~tcpsconn() -{ - pipe_[1].close(); - th_.join(); - - for (auto & i : conns_) - i.second->closeconn(); +connection_listener::~connection_listener() { + global->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"); - throw thread_exit_exception(); + perror("connection_listener::accept_conn error"); + throw std::runtime_error("connection listener failure"); } - IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port)); - auto ch = make_shared(mgr_, s1, lossy_); + IF_LEVEL(2) LOG << "accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port); // garbage collect dead connections for (auto i = conns_.begin(); i != conns_.end();) { @@ -275,38 +240,5 @@ void tcpsconn::process_accept() { ++i; } - conns_[ch->channo()] = ch; + conns_[s1] = std::make_shared(delegate_, s1, lossy_); } - -void tcpsconn::accept_conn() { - fd_set rfds; - int max_fd = max((int)pipe_[0], (int)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 && errno == EINTR) - continue; - else if (ret < 0) { - perror("accept_conn select:"); - IF_LEVEL(0) LOG("accept_conn failure errno " << errno); - VERIFY(0); - } - - if (FD_ISSET(pipe_[0], &rfds)) - return; - - VERIFY(FD_ISSET(tcp_, &rfds)); - - try { - process_accept(); - } catch (thread_exit_exception e) { - break; - } - } -} -