#include <unistd.h>
#include "marshall.h"
+connection_delegate::~connection_delegate() {}
+
connection::connection(connection_delegate * delegate, socket_t && f1, int l1)
-: fd(move(f1)), delegate_(delegate), lossy_(l1)
+: fd(std::move(f1)), delegate_(delegate), lossy_(l1)
{
fd.flags() |= O_NONBLOCK;
// will be active
poll_mgr::shared_mgr.block_remove_fd(fd);
VERIFY(dead_);
- VERIFY(!wpdu_.buf.size());
+ VERIFY(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));
+ 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<connection>(delegate, move(s), lossy);
+ 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_);
- 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()) {
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;
}
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;
}
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_);
if (dead_)
return;
- IF_LEVEL(5) LOG("got data on fd " << 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");
+ 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 (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_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 && 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 > rpc_protocol::MAX_PDU) {
- IF_LEVEL(2) LOG("read pdu TOO BIG " << sz << " network order=" << hex << sz1);
+ 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;
- 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_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;
}
tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeval{0, 50000});
tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeval{0, 50000});
- sockaddr_in sin{}; // zero initialize
+ sockaddr_in sin = sockaddr_in(); // zero initialize
sin.sin_family = AF_INET;
sin.sin_port = hton(port);
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;
poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);
}
int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
if (s1 < 0) {
perror("connection_listener::accept_conn error");
- throw thread_exit_exception();
+ 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<connection>(delegate_, 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();) {
++i;
}
- conns_[s1] = ch;
+ conns_[s1] = std::make_shared<connection>(delegate_, s1, lossy_);
}
projects / invirt/third/libt4.git / blobdiff