#include "rpc_protocol.h"
#include <cerrno>
#include <csignal>
-#include <sys/types.h>
#include <netinet/tcp.h>
#include <unistd.h>
#include "marshall.h"
-connection::connection(chanmgr *m1, int f1, int l1)
-: mgr_(m1), fd_(f1), lossy_(l1)
+connection::connection(connection_delegate *m1, socket_t && f1, int l1)
+: mgr_(m1), fd_(move(f1)), lossy_(l1)
{
fd_.flags() |= O_NONBLOCK;
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(!wpdu_.buf.size());
}
+shared_ptr<connection> connection::to_dst(const sockaddr_in &dst, connection_delegate *mgr, 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_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_);
}
//after block_remove_fd, select will never wait on fd_
//and no callbacks will be active
- PollMgr::Instance().block_remove_fd(fd_);
+ poll_mgr::shared_mgr.block_remove_fd(fd_);
}
bool connection::send(const string & b) {
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);
}
VERIFY(!dead_);
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);
void connection::read_cb(int s) {
lock ml(m_);
VERIFY(fd_ == s);
- if (dead_) {
+ if (dead_)
return;
- }
IF_LEVEL(5) LOG("got data on fd " << s);
bool succ = true;
- if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size()) {
+ 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);
+ 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)) {
- // chanmgr has successfully consumed the pdu
+ // connection_delegate has successfully consumed the pdu
rpdu_.buf.clear();
rpdu_.solong = 0;
}
bool connection::readpdu() {
IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size());
if (!rpdu_.buf.size()) {
- rpc_sz_t sz1;
+ rpc_protocol::rpc_sz_t sz1;
ssize_t n = fd_.read(sz1);
if (n == 0)
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;
}
return true;
}
-tcpsconn::tcpsconn(chanmgr *m1, in_port_t port, int lossytest)
+tcpsconn::tcpsconn(connection_delegate *m1, in_port_t port, int lossytest)
: tcp_(socket(AF_INET, SOCK_STREAM, 0)), mgr_(m1), lossy_(lossytest)
{
- struct sockaddr_in sin;
- memset(&sin, 0, sizeof(sin));
+ 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);
-
- struct timeval timeout = {0, 50000};
-
- if (tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeout) < 0)
- perror("accept_loop setsockopt");
-
- if (tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeout) < 0)
- perror("accept_loop setsockopt");
+ 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 struct sockaddr *)&sin, (socklen_t)sizeof(sin)) < 0) {
+ if (bind((int)tcp_, (const sockaddr *)&sin, (socklen_t)sizeof(sin)) < 0) {
perror("accept_loop bind");
VERIFY(0);
}
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);
+ poll_mgr::shared_mgr.add_callback(tcp_, CB_RDONLY, this);
}
tcpsconn::~tcpsconn()
{
- pipe_[1].close();
- th_.join();
+ poll_mgr::shared_mgr.block_remove_fd(tcp_);
for (auto & i : conns_)
i.second->closeconn();
}
-void tcpsconn::process_accept() {
+void tcpsconn::read_cb(int) {
sockaddr_in sin;
socklen_t slen = sizeof(sin);
int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
conns_[ch->channo()] = ch;
}
-
-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;
-
- if (!FD_ISSET(tcp_, &rfds))
- VERIFY(0);
-
- try {
- process_accept();
- } catch (thread_exit_exception e) {
- break;
- }
- }
-}
-
-shared_ptr<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("failed to " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
- close(s);
- return nullptr;
- }
- IF_LEVEL(2) LOG_NONMEMBER("connect_to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
- return make_shared<connection>(mgr, s, lossy);
-}
-
projects / invirt/third/libt4.git / blobdiff