-#include <fcntl.h>
-#include <sys/types.h>
-#include <sys/time.h>
+#include "connection.h"
+#include "rpc_protocol.h"
+#include <cerrno>
+#include <csignal>
#include <netinet/tcp.h>
-#include <errno.h>
-#include <signal.h>
#include <unistd.h>
+#include "marshall.h"
-#include "connection.h"
-#include "pollmgr.h"
-#include "jsl_log.h"
-#include "lang/verify.h"
-#include "lock.h"
-
-#define MAX_PDU (10<<20) //maximum PDF is 10M
-
-
-connection::connection(chanmgr *m1, int f1, int l1)
-: mgr_(m1), fd_(f1), dead_(false),waiters_(0), refno_(1),lossy_(l1)
+connection::connection(connection_delegate *m1, socket_t && f1, int l1)
+: mgr_(m1), fd_(move(f1)), lossy_(l1)
{
-
- int flags = fcntl(fd_, F_GETFL, NULL);
- flags |= O_NONBLOCK;
- fcntl(fd_, F_SETFL, flags);
+ fd_.flags() |= O_NONBLOCK;
signal(SIGPIPE, SIG_IGN);
- VERIFY(gettimeofday(&create_time_, NULL) == 0);
+ 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()
-{
+connection::~connection() {
+ closeconn();
VERIFY(dead_);
- if (rpdu_.buf)
- free(rpdu_.buf);
- VERIFY(!wpdu_.buf);
- close(fd_);
+ VERIFY(!wpdu_.buf.size());
}
-void
-connection::incref()
-{
- lock rl(ref_m_);
- refno_++;
-}
-
-bool
-connection::isdead()
-{
- lock ml(m_);
- return dead_;
+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()
-{
+void connection::closeconn() {
{
lock ml(m_);
- if (!dead_) {
- dead_ = true;
- shutdown(fd_,SHUT_RDWR);
- } else {
+ 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
- PollMgr::Instance()->block_remove_fd(fd_);
+ poll_mgr::shared_mgr.block_remove_fd(fd_);
}
-void
-connection::decref()
-{
- bool dead = false;
- {
- lock rl(ref_m_);
- refno_--;
- VERIFY(refno_>=0);
- if (refno_==0) {
- lock ml(m_);
- dead = dead_;
- }
- }
- if (dead) {
- delete this;
- }
-}
+bool connection::send(const string & b) {
+ lock ml(m_);
-int
-connection::ref()
-{
- lock rl(ref_m_);
- return refno_;
-}
+ waiters_++;
+ while (!dead_ && wpdu_.buf.size())
+ send_wait_.wait(ml);
+ waiters_--;
-int
-connection::compare(connection *another)
-{
- if (create_time_.tv_sec > another->create_time_.tv_sec)
- return 1;
- if (create_time_.tv_sec < another->create_time_.tv_sec)
- return -1;
- if (create_time_.tv_usec > another->create_time_.tv_usec)
- return 1;
- if (create_time_.tv_usec < another->create_time_.tv_usec)
- return -1;
- return 0;
-}
+ if (dead_)
+ return false;
-bool
-connection::send(char *b, int sz)
-{
- lock ml(m_);
- waiters_++;
- while (!dead_ && wpdu_.buf) {
- send_wait_.wait(ml);
- }
- waiters_--;
- if (dead_) {
- return false;
- }
- wpdu_.buf = b;
- wpdu_.sz = sz;
- wpdu_.solong = 0;
-
- if (lossy_) {
- if ((random()%100) < lossy_) {
- jsl_log(JSL_DBG_1, "connection::send LOSSY TEST shutdown fd_ %d\n", fd_);
- shutdown(fd_,SHUT_RDWR);
- }
- }
-
- if (!writepdu()) {
- dead_ = true;
+ wpdu_.buf = b;
+ wpdu_.solong = 0;
+
+ if (lossy_) {
+ if ((random()%100) < lossy_) {
+ IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd_ " << fd_);
+ shutdown(fd_,SHUT_RDWR);
+ }
+ }
+
+ 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_.sz) {
- }else{
- //should be rare to need to explicitly add write callback
- PollMgr::Instance()->add_callback(fd_, CB_WRONLY, this);
- while (!dead_ && wpdu_.solong >= 0 && wpdu_.solong < wpdu_.sz) {
- send_complete_.wait(ml);
- }
- }
- }
- bool ret = (!dead_ && wpdu_.solong == wpdu_.sz);
- wpdu_.solong = wpdu_.sz = 0;
- wpdu_.buf = NULL;
- if (waiters_ > 0)
+ } 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);
+ while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.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();
- return ret;
+ return ret;
}
//fd_ is ready to be written
-void
-connection::write_cb(int s)
-{
+void connection::write_cb(int s) {
lock ml(m_);
- VERIFY(!dead_);
- VERIFY(fd_ == s);
- if (wpdu_.sz == 0) {
- PollMgr::Instance()->del_callback(fd_,CB_WRONLY);
- return;
- }
- if (!writepdu()) {
- PollMgr::Instance()->del_callback(fd_, CB_RDWR);
- dead_ = true;
- } else {
- VERIFY(wpdu_.solong >= 0);
- if (wpdu_.solong < wpdu_.sz) {
- return;
- }
+ VERIFY(!dead_);
+ VERIFY(fd_ == s);
+ if (wpdu_.buf.size() == 0) {
+ poll_mgr::shared_mgr.del_callback(fd_,CB_WRONLY);
+ return;
+ }
+ if (!writepdu()) {
+ poll_mgr::shared_mgr.del_callback(fd_, CB_RDWR);
+ dead_ = true;
+ } else {
+ VERIFY(wpdu_.solong != size_t_max);
+ if (wpdu_.solong < wpdu_.buf.size()) {
+ return;
+ }
}
- send_complete_.notify_one();
+ send_complete_.notify_one();
}
-//fd_ is ready to be read
-void
-connection::read_cb(int s)
-{
+// fd_ is ready to be read
+void connection::read_cb(int s) {
lock ml(m_);
- VERIFY(fd_ == s);
- if (dead_) {
- return;
- }
-
- bool succ = true;
- if (!rpdu_.buf || rpdu_.solong < rpdu_.sz) {
- succ = readpdu();
- }
-
- if (!succ) {
- PollMgr::Instance()->del_callback(fd_,CB_RDWR);
- dead_ = true;
- send_complete_.notify_one();
- }
-
- if (rpdu_.buf && rpdu_.sz == rpdu_.solong) {
- if (mgr_->got_pdu(this, rpdu_.buf, rpdu_.sz)) {
- //chanmgr has successfully consumed the pdu
- rpdu_.buf = NULL;
- rpdu_.sz = rpdu_.solong = 0;
- }
- }
+ 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_.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;
+ }
+ }
}
-bool
-connection::writepdu()
-{
- VERIFY(wpdu_.solong >= 0);
- if (wpdu_.solong == wpdu_.sz)
- return true;
-
- if (wpdu_.solong == 0) {
- int sz = htonl(wpdu_.sz);
- bcopy(&sz,wpdu_.buf,sizeof(sz));
- }
- int n = write(fd_, wpdu_.buf + wpdu_.solong, (wpdu_.sz-wpdu_.solong));
- if (n < 0) {
- if (errno != EAGAIN) {
- jsl_log(JSL_DBG_1, "connection::writepdu fd_ %d failure errno=%d\n", fd_, errno);
- wpdu_.solong = -1;
- wpdu_.sz = 0;
- }
- return (errno == EAGAIN);
- }
- wpdu_.solong += n;
- return true;
+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 (!rpdu_.sz) {
- int sz, sz1;
- int n = read(fd_, &sz1, sizeof(sz1));
-
- if (n == 0) {
- return false;
- }
-
- if (n < 0) {
- VERIFY(errno!=EAGAIN);
- return false;
- }
-
- if (n >0 && n!= sizeof(sz)) {
- jsl_log(JSL_DBG_OFF, "connection::readpdu short read of sz\n");
- return false;
- }
-
- sz = ntohl(sz1);
-
- if (sz > MAX_PDU) {
- char *tmpb = (char *)&sz1;
- jsl_log(JSL_DBG_2, "connection::readpdu read pdu TOO BIG %d network order=%x %x %x %x %x\n", sz,
- sz1, tmpb[0],tmpb[1],tmpb[2],tmpb[3]);
- return false;
- }
-
- rpdu_.sz = sz;
- VERIFY(rpdu_.buf == NULL);
- rpdu_.buf = (char *)malloc(sz+sizeof(sz));
- VERIFY(rpdu_.buf);
- bcopy(&sz1,rpdu_.buf,sizeof(sz));
- rpdu_.solong = sizeof(sz);
- }
-
- int n = read(fd_, rpdu_.buf + rpdu_.solong, rpdu_.sz - rpdu_.solong);
- if (n <= 0) {
- if (errno == EAGAIN)
- return true;
- if (rpdu_.buf)
- free(rpdu_.buf);
- rpdu_.buf = NULL;
- rpdu_.sz = rpdu_.solong = 0;
- return (errno == EAGAIN);
- }
- rpdu_.solong += 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 (n == 0)
+ return false;
+
+ if (n < 0) {
+ VERIFY(errno!=EAGAIN);
+ return false;
+ }
+
+ if (n > 0 && n != sizeof(sz1)) {
+ 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);
+ return false;
+ }
+
+ 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);
+ }
+
+ ssize_t n = fd_.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);
+
+ 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_.solong += (size_t)n;
+ return true;
}
-tcpsconn::tcpsconn(chanmgr *m1, int port, int lossytest)
-: mgr_(m1), lossy_(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));
- sin.sin_family = AF_INET;
- sin.sin_port = htons(port);
-
- tcp_ = socket(AF_INET, SOCK_STREAM, 0);
- if(tcp_ < 0){
- perror("tcpsconn::tcpsconn accept_loop socket:");
- VERIFY(0);
- }
-
- int yes = 1;
- setsockopt(tcp_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
- setsockopt(tcp_, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
-
- if(bind(tcp_, (sockaddr *)&sin, sizeof(sin)) < 0){
- perror("accept_loop tcp bind:");
- VERIFY(0);
- }
-
- if(listen(tcp_, 1000) < 0) {
- perror("tcpsconn::tcpsconn listen:");
- VERIFY(0);
- }
+ sockaddr_in sin;
+ memset(&sin, 0, sizeof(sin));
+ 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);
+
+ 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");
+
+ // 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) {
+ perror("accept_loop bind");
+ VERIFY(0);
+ }
+
+ if (listen(tcp_, 1000) < 0) {
+ perror("accept_loop listen");
+ VERIFY(0);
+ }
socklen_t addrlen = sizeof(sin);
VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
- port_ = ntohs(sin.sin_port);
+ port_ = ntoh(sin.sin_port);
- jsl_log(JSL_DBG_2, "tcpsconn::tcpsconn listen on %d %d\n", port_,
- sin.sin_port);
+ IF_LEVEL(2) LOG("listen on " << port_ << " " << sin.sin_port);
- if (pipe(pipe_) < 0) {
- perror("accept_loop pipe:");
- VERIFY(0);
- }
+ file_t::pipe(pipe_);
- int flags = fcntl(pipe_[0], F_GETFL, NULL);
- flags |= O_NONBLOCK;
- fcntl(pipe_[0], F_SETFL, flags);
+ pipe_[0].flags() |= O_NONBLOCK;
- th_ = std::thread(&tcpsconn::accept_conn, this);
+ th_ = thread(&tcpsconn::accept_conn, this);
}
tcpsconn::~tcpsconn()
{
- VERIFY(close(pipe_[1]) == 0);
+ pipe_[1].close();
th_.join();
- //close all the active connections
- std::map<int, connection *>::iterator i;
- for (i = conns_.begin(); i != conns_.end(); i++) {
- i->second->closeconn();
- i->second->decref();
- }
+ for (auto & i : conns_)
+ i.second->closeconn();
}
-void
-tcpsconn::process_accept()
-{
- 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();
- }
-
- jsl_log(JSL_DBG_2, "accept_loop got connection fd=%d %s:%d\n",
- s1, inet_ntoa(sin.sin_addr), ntohs(sin.sin_port));
- connection *ch = new connection(mgr_, s1, lossy_);
-
- // garbage collect all dead connections with refcount of 1
- std::map<int, connection *>::iterator i;
- for (i = conns_.begin(); i != conns_.end();) {
- if (i->second->isdead() && i->second->ref() == 1) {
- jsl_log(JSL_DBG_2, "accept_loop garbage collected fd=%d\n",
- i->second->channo());
- i->second->decref();
- // Careful not to reuse i right after erase. (i++) will
- // be evaluated before the erase call because in C++,
- // there is a sequence point before a function call.
- // See http://en.wikipedia.org/wiki/Sequence_point.
- conns_.erase(i++);
- } else
- ++i;
- }
+void tcpsconn::process_accept() {
+ 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();
+ }
- conns_[ch->channo()] = ch;
-}
+ 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_);
-void
-tcpsconn::accept_conn()
-{
- fd_set rfds;
- int max_fd = pipe_[0] > tcp_ ? pipe_[0] : tcp_;
-
- try {
-
- 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) {
- if (errno == EINTR) {
- continue;
- } else {
- perror("accept_conn select:");
- jsl_log(JSL_DBG_OFF, "tcpsconn::accept_conn failure errno %d\n",errno);
- VERIFY(0);
- }
- }
-
- if (FD_ISSET(pipe_[0], &rfds)) {
- close(pipe_[0]);
- close(tcp_);
- return;
- }
- else if (FD_ISSET(tcp_, &rfds)) {
- process_accept();
- } else {
- VERIFY(0);
- }
- }
- }
- catch (thread_exit_exception e)
- {
- return;
+ // garbage collect dead connections
+ for (auto i = conns_.begin(); i != conns_.end();) {
+ if (i->second->isdead())
+ conns_.erase(i++);
+ else
+ ++i;
}
+
+ conns_[ch->channo()] = ch;
}
-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) {
- jsl_log(JSL_DBG_1, "rpcc::connect_to_dst failed to %s:%d\n",
- inet_ntoa(dst.sin_addr), (int)ntohs(dst.sin_port));
- close(s);
- return NULL;
- }
- jsl_log(JSL_DBG_2, "connect_to_dst fd=%d to dst %s:%d\n",
- s, inet_ntoa(dst.sin_addr), (int)ntohs(dst.sin_port));
- return new connection(mgr, s, 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;
+ }
+ }
}