X-Git-Url: http://xvm.mit.edu/gitweb/invirt/third/libt4.git/blobdiff_plain/26ade07ab0e62b98b452fbbd18edba0450035e35..03b35a9a1bd1f583e32b27d260b223a0989d6c75:/rpc/connection.cc diff --git a/rpc/connection.cc b/rpc/connection.cc index d118539..c7e8f95 100644 --- a/rpc/connection.cc +++ b/rpc/connection.cc @@ -30,7 +30,7 @@ connection::~connection() { // will be active poll_mgr::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 * delegate, int lossy) { @@ -42,28 +42,23 @@ shared_ptr connection::to_dst(const sockaddr_in & dst, connection_de 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, std::move(s), lossy); + 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 (lossy_ && (random()%100) < lossy_) { + IF_LEVEL(1) LOG << "send LOSSY TEST shutdown fd " << fd; + shutdown(fd,SHUT_RDWR); } if (!writepdu()) { @@ -71,17 +66,15 @@ bool connection::send(const string & b) { ml.unlock(); poll_mgr::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()) + 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; } @@ -90,7 +83,7 @@ void connection::write_cb(int s) { lock ml(m_); VERIFY(!dead_); VERIFY(fd == s); - if (wpdu_.buf.size() == 0) { + if (wpdu_.status != inflight) { poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY); return; } @@ -98,14 +91,30 @@ void connection::write_cb(int s) { poll_mgr::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(); } +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_); @@ -115,7 +124,7 @@ void connection::read_cb(int s) { IF_LEVEL(5) LOG << "got data on fd " << s; - if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size()) { + if (rpdu_.status == unused || rpdu_.cursor < rpdu_.buf.size()) { if (!readpdu()) { IF_LEVEL(5) LOG << "readpdu on fd " << s << " failed; dying"; poll_mgr::shared_mgr.del_callback(fd, CB_RDWR); @@ -124,36 +133,17 @@ void connection::read_cb(int s) { } } - if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) { + 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_.buf.clear(); - rpdu_.solong = 0; + rpdu_ = {unused, "", 0}; } } } -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; -} - bool connection::readpdu() { IF_LEVEL(5) LOG << "the receive buffer has length " << rpdu_.buf.size(); - if (!rpdu_.buf.size()) { + if (rpdu_.status == unused) { rpc_protocol::rpc_sz_t sz1; ssize_t n = fd.read(sz1); @@ -179,22 +169,20 @@ bool connection::readpdu() { IF_LEVEL(5) LOG << "read size of datagram = " << sz; - rpdu_.buf.assign(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 (n <= 0) { if (errno == EAGAIN) return true; - rpdu_.buf.clear(); - rpdu_.solong = 0; + rpdu_ = {unused, "", 0}; return false; } - rpdu_.solong += (size_t)n; + rpdu_.cursor += (size_t)n; return true; } @@ -239,11 +227,10 @@ void connection_listener::read_cb(int) { int s1 = accept(tcp_, (sockaddr *)&sin, &slen); if (s1 < 0) { perror("connection_listener::accept_conn error"); - throw runtime_error("connection listener failure"); + 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(delegate_, s1, lossy_); // garbage collect dead connections for (auto i = conns_.begin(); i != conns_.end();) { @@ -253,5 +240,5 @@ void connection_listener::read_cb(int) { ++i; } - conns_[s1] = ch; + conns_[s1] = std::make_shared(delegate_, s1, lossy_); }