#include <unistd.h>
#include "marshall.h"
-connection::connection(connection_delegate *m1, socket_t && f1, int l1)
-: mgr_(m1), fd_(move(f1)), lossy_(l1)
+connection::connection(connection_delegate * delegate, socket_t && f1, int l1)
+: fd(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);
+ poll_mgr::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
+ poll_mgr::shared_mgr.block_remove_fd(fd);
VERIFY(dead_);
VERIFY(!wpdu_.buf.size());
}
-shared_ptr<connection> connection::to_dst(const sockaddr_in &dst, connection_delegate *mgr, int lossy) {
+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) {
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>(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_);
+ return make_shared<connection>(delegate, move(s), lossy);
}
bool connection::send(const string & b) {
if (lossy_) {
if ((random()%100) < lossy_) {
- IF_LEVEL(1) LOG("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();
- poll_mgr::shared_mgr.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
- poll_mgr::shared_mgr.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);
}
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) {
- poll_mgr::shared_mgr.del_callback(fd_,CB_WRONLY);
+ poll_mgr::shared_mgr.del_callback(fd, CB_WRONLY);
return;
}
if (!writepdu()) {
- poll_mgr::shared_mgr.del_callback(fd_, CB_RDWR);
+ poll_mgr::shared_mgr.del_callback(fd, CB_RDWR);
dead_ = true;
} else {
VERIFY(wpdu_.solong != size_t_max);
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);
+ 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");
- poll_mgr::shared_mgr.del_callback(fd_,CB_RDWR);
- dead_ = true;
- send_complete_.notify_one();
+ 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 (mgr_->got_pdu(shared_from_this(), rpdu_.buf)) {
+ if (delegate_->got_pdu(shared_from_this(), rpdu_.buf)) {
// connection_delegate has successfully consumed the pdu
rpdu_.buf.clear();
rpdu_.solong = 0;
if (wpdu_.solong == wpdu_.buf.size())
return true;
- 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("writepdu fd_ " << fd_ << " failure errno=" << errno);
+ IF_LEVEL(1) LOG("writepdu fd " << fd << " failure errno=" << errno);
wpdu_.solong = size_t_max;
wpdu_.buf.clear();
}
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);
+ ssize_t n = fd.read(sz1);
if (n == 0)
return false;
IF_LEVEL(5) LOG("read size of datagram = " << sz);
- VERIFY(rpdu_.buf.size() == 0);
- rpdu_.buf = string(sz+sizeof(sz1), 0);
+ rpdu_.buf.assign(sz+sizeof(sz1), 0);
rpdu_.solong = sizeof(sz1);
}
- ssize_t n = fd_.read(&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");
return true;
rpdu_.buf.clear();
rpdu_.solong = 0;
- return (errno == EAGAIN);
+ return false;
}
rpdu_.solong += (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});
- // 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) {
+ 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);
}
poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);
}
-tcpsconn::~tcpsconn()
-{
+connection_listener::~connection_listener() {
poll_mgr::shared_mgr.block_remove_fd(tcp_);
-
- for (auto & i : conns_)
- i.second->closeconn();
}
-void tcpsconn::read_cb(int) {
+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) << ":" << ntoh(sin.sin_port));
- auto ch = make_shared<connection>(mgr_, s1, lossy_);
+ auto ch = make_shared<connection>(delegate_, s1, lossy_);
// garbage collect dead connections
for (auto i = conns_.begin(); i != conns_.end();) {
++i;
}
- conns_[ch->channo()] = ch;
+ conns_[s1] = ch;
}
projects / invirt/third/libt4.git / blobdiff