rpc/connection.cc

   1 #include "connection.h"
   2 #include "rpc_protocol.h"
   3 #include <cerrno>
   4 #include <csignal>
   5 #include <sys/types.h>
   6 #include <netinet/tcp.h>
   7 #include <unistd.h>
   8 #include "marshall.h"
   9
  10 connection::connection(chanmgr *m1, int f1, int l1)
  11 : mgr_(m1), fd_(f1), lossy_(l1)
  12 {
  13     fd_.flags() |= O_NONBLOCK;
  14
  15     signal(SIGPIPE, SIG_IGN);
  16
  17     create_time_ = steady_clock::now();
  18
  19     PollMgr::Instance().add_callback(fd_, CB_RDONLY, this);
  20 }
  21
  22 connection::~connection() {
  23     VERIFY(dead_);
  24     VERIFY(!wpdu_.buf.size());
  25 }
  26
  27 void connection::incref() {
  28     lock rl(ref_m_);
  29     refno_++;
  30 }
  31
  32 bool connection::isdead() {
  33     lock ml(m_);
  34     return dead_;
  35 }
  36
  37 void connection::closeconn() {
  38     {
  39         lock ml(m_);
  40         if (dead_)
  41             return;
  42         dead_ = true;
  43         shutdown(fd_,SHUT_RDWR);
  44     }
  45     //after block_remove_fd, select will never wait on fd_
  46     //and no callbacks will be active
  47     PollMgr::Instance().block_remove_fd(fd_);
  48 }
  49
  50 void connection::decref() {
  51     bool dead = false;
  52     {
  53         lock rl(ref_m_);
  54         refno_--;
  55         VERIFY(refno_>=0);
  56         if (refno_==0) {
  57             lock ml(m_);
  58             dead = dead_;
  59         }
  60     }
  61     if (dead)
  62         delete this;
  63 }
  64
  65 int connection::compare(connection *another) {
  66     if (create_time_ > another->create_time_)
  67         return 1;
  68     if (create_time_ < another->create_time_)
  69         return -1;
  70     return 0;
  71 }
  72
  73 bool connection::send(const string & b) {
  74     lock ml(m_);
  75
  76     waiters_++;
  77     while (!dead_ && wpdu_.buf.size())
  78         send_wait_.wait(ml);
  79     waiters_--;
  80
  81     if (dead_)
  82         return false;
  83
  84     wpdu_.buf = b;
  85     wpdu_.solong = 0;
  86
  87     if (lossy_) {
  88         if ((random()%100) < lossy_) {
  89             IF_LEVEL(1) LOG("send LOSSY TEST shutdown fd_ " << fd_);
  90             shutdown(fd_,SHUT_RDWR);
  91         }
  92     }
  93
  94     if (!writepdu()) {
  95         dead_ = true;
  96         ml.unlock();
  97         PollMgr::Instance().block_remove_fd(fd_);
  98         ml.lock();
  99     } else if (wpdu_.solong != wpdu_.buf.size()) {
 100         // should be rare to need to explicitly add write callback
 101         PollMgr::Instance().add_callback(fd_, CB_WRONLY, this);
 102         while (!dead_ && wpdu_.solong != size_t_max && wpdu_.solong < wpdu_.buf.size())
 103             send_complete_.wait(ml);
 104     }
 105     bool ret = (!dead_ && wpdu_.solong == wpdu_.buf.size());
 106     wpdu_.solong = 0;
 107     wpdu_.buf.clear();
 108     if (waiters_ > 0)
 109         send_wait_.notify_all();
 110     return ret;
 111 }
 112
 113 //fd_ is ready to be written
 114 void connection::write_cb(int s) {
 115     lock ml(m_);
 116     VERIFY(!dead_);
 117     VERIFY(fd_ == s);
 118     if (wpdu_.buf.size() == 0) {
 119         PollMgr::Instance().del_callback(fd_,CB_WRONLY);
 120         return;
 121     }
 122     if (!writepdu()) {
 123         PollMgr::Instance().del_callback(fd_, CB_RDWR);
 124         dead_ = true;
 125     } else {
 126         VERIFY(wpdu_.solong != size_t_max);
 127         if (wpdu_.solong < wpdu_.buf.size()) {
 128             return;
 129         }
 130     }
 131     send_complete_.notify_one();
 132 }
 133
 134 //fd_ is ready to be read
 135 void connection::read_cb(int s) {
 136     lock ml(m_);
 137     VERIFY(fd_ == s);
 138     if (dead_)  {
 139         return;
 140     }
 141
 142     IF_LEVEL(5) LOG("got data on fd " << s);
 143
 144     bool succ = true;
 145     if (!rpdu_.buf.size() || rpdu_.solong < rpdu_.buf.size()) {
 146         succ = readpdu();
 147     }
 148
 149     if (!succ) {
 150         IF_LEVEL(5) LOG("readpdu on fd " << s << " failed; dying");
 151         PollMgr::Instance().del_callback(fd_,CB_RDWR);
 152         dead_ = true;
 153         send_complete_.notify_one();
 154     }
 155
 156     if (rpdu_.buf.size() && rpdu_.buf.size() == rpdu_.solong) {
 157         if (mgr_->got_pdu(this, rpdu_.buf)) {
 158             //chanmgr has successfully consumed the pdu
 159             rpdu_.buf.clear();
 160             rpdu_.solong = 0;
 161         }
 162     }
 163 }
 164
 165 bool connection::writepdu() {
 166     VERIFY(wpdu_.solong != size_t_max);
 167     if (wpdu_.solong == wpdu_.buf.size())
 168         return true;
 169
 170     ssize_t n = write(fd_, &wpdu_.buf[wpdu_.solong], (wpdu_.buf.size()-wpdu_.solong));
 171     if (n < 0) {
 172         if (errno != EAGAIN) {
 173             IF_LEVEL(1) LOG("writepdu fd_ " << fd_ << " failure errno=" << errno);
 174             wpdu_.solong = size_t_max;
 175             wpdu_.buf.clear();
 176         }
 177         return (errno == EAGAIN);
 178     }
 179     wpdu_.solong += (size_t)n;
 180     return true;
 181 }
 182
 183 bool connection::readpdu() {
 184     IF_LEVEL(5) LOG("the receive buffer has length " << rpdu_.buf.size());
 185     if (!rpdu_.buf.size()) {
 186         rpc_sz_t sz1;
 187         ssize_t n = fd_.read(sz1);
 188
 189         if (n == 0)
 190             return false;
 191
 192         if (n < 0) {
 193             VERIFY(errno!=EAGAIN);
 194             return false;
 195         }
 196
 197         if (n > 0 && n != sizeof(sz1)) {
 198             IF_LEVEL(0) LOG("short read of sz");
 199             return false;
 200         }
 201
 202         size_t sz = ntoh(sz1);
 203
 204         if (sz > MAX_PDU) {
 205             IF_LEVEL(2) LOG("read pdu TOO BIG " << sz << " network order=" << hex << sz1);
 206             return false;
 207         }
 208
 209         IF_LEVEL(5) LOG("read size of datagram = " << sz);
 210
 211         VERIFY(rpdu_.buf.size() == 0);
 212         rpdu_.buf = string(sz+sizeof(sz1), 0);
 213         rpdu_.solong = sizeof(sz1);
 214     }
 215
 216     ssize_t n = fd_.read(&rpdu_.buf[rpdu_.solong], rpdu_.buf.size() - rpdu_.solong);
 217
 218     IF_LEVEL(5) LOG("read " << n << " bytes");
 219
 220     if (n <= 0) {
 221         if (errno == EAGAIN)
 222             return true;
 223         rpdu_.buf.clear();
 224         rpdu_.solong = 0;
 225         return (errno == EAGAIN);
 226     }
 227     rpdu_.solong += (size_t)n;
 228     return true;
 229 }
 230
 231 tcpsconn::tcpsconn(chanmgr *m1, in_port_t port, int lossytest)
 232 : tcp_(socket(AF_INET, SOCK_STREAM, 0)), mgr_(m1), lossy_(lossytest)
 233 {
 234     struct sockaddr_in sin;
 235     memset(&sin, 0, sizeof(sin));
 236     sin.sin_family = AF_INET;
 237     sin.sin_port = hton(port);
 238
 239     tcp_.setsockopt(SOL_SOCKET, SO_REUSEADDR, (int)1);
 240     tcp_.setsockopt(IPPROTO_TCP, TCP_NODELAY, (int)1);
 241
 242     struct timeval timeout = {0, 50000};
 243
 244     if (tcp_.setsockopt(SOL_SOCKET, SO_RCVTIMEO, timeout) < 0)
 245         perror("accept_loop setsockopt");
 246
 247     if (tcp_.setsockopt(SOL_SOCKET, SO_SNDTIMEO, timeout) < 0)
 248         perror("accept_loop setsockopt");
 249
 250     // careful to exactly match type signature of bind arguments so we don't
 251     // get std::bind instead
 252     if (bind((int)tcp_, (const struct sockaddr *)&sin, (socklen_t)sizeof(sin)) < 0) {
 253         perror("accept_loop bind");
 254         VERIFY(0);
 255     }
 256
 257     if (listen(tcp_, 1000) < 0) {
 258         perror("accept_loop listen");
 259         VERIFY(0);
 260     }
 261
 262     socklen_t addrlen = sizeof(sin);
 263     VERIFY(getsockname(tcp_, (sockaddr *)&sin, &addrlen) == 0);
 264     port_ = ntoh(sin.sin_port);
 265
 266     IF_LEVEL(2) LOG("listen on " << port_ << " " << sin.sin_port);
 267
 268     file_t::pipe(pipe_);
 269
 270     pipe_[0].flags() |= O_NONBLOCK;
 271
 272     th_ = thread(&tcpsconn::accept_conn, this);
 273 }
 274
 275 tcpsconn::~tcpsconn()
 276 {
 277     pipe_[1].close();
 278     th_.join();
 279
 280     // close all the active connections
 281     map<int, connection *>::iterator i;
 282     for (i = conns_.begin(); i != conns_.end(); i++) {
 283         i->second->closeconn();
 284         i->second->decref();
 285     }
 286 }
 287
 288 void tcpsconn::process_accept() {
 289     sockaddr_in sin;
 290     socklen_t slen = sizeof(sin);
 291     int s1 = accept(tcp_, (sockaddr *)&sin, &slen);
 292     if (s1 < 0) {
 293         perror("tcpsconn::accept_conn error");
 294         throw thread_exit_exception();
 295     }
 296
 297     IF_LEVEL(2) LOG("accept_loop got connection fd=" << s1 << " " << inet_ntoa(sin.sin_addr) << ":" << ntoh(sin.sin_port));
 298     connection *ch = new connection(mgr_, s1, lossy_);
 299
 300     // garbage collect all dead connections with refcount of 1
 301     for (auto i = conns_.begin(); i != conns_.end();) {
 302         if (i->second->isdead() && i->second->ref() == 1) {
 303             IF_LEVEL(2) LOG("accept_loop garbage collected fd=" << i->second->channo());
 304             i->second->decref();
 305             // Careful not to reuse i right after erase. (i++) will
 306             // be evaluated before the erase call because in C++,
 307             // there is a sequence point before a function call.
 308             // See http://en.wikipedia.org/wiki/Sequence_point.
 309             conns_.erase(i++);
 310         } else
 311             ++i;
 312     }
 313
 314     conns_[ch->channo()] = ch;
 315 }
 316
 317 void tcpsconn::accept_conn() {
 318     fd_set rfds;
 319     int max_fd = max((int)pipe_[0], (int)tcp_);
 320
 321     while (1) {
 322         FD_ZERO(&rfds);
 323         FD_SET(pipe_[0], &rfds);
 324         FD_SET(tcp_, &rfds);
 325
 326         int ret = select(max_fd+1, &rfds, NULL, NULL, NULL);
 327
 328         if (ret < 0 && errno == EINTR)
 329             continue;
 330         else if (ret < 0) {
 331             perror("accept_conn select:");
 332             IF_LEVEL(0) LOG("accept_conn failure errno " << errno);
 333             VERIFY(0);
 334         }
 335
 336         if (FD_ISSET(pipe_[0], &rfds))
 337             return;
 338
 339         if (!FD_ISSET(tcp_, &rfds))
 340             VERIFY(0);
 341
 342         try {
 343             process_accept();
 344         } catch (thread_exit_exception e) {
 345             break;
 346         }
 347     }
 348 }
 349
 350 connection * connect_to_dst(const sockaddr_in &dst, chanmgr *mgr, int lossy) {
 351     int s = socket(AF_INET, SOCK_STREAM, 0);
 352     int yes = 1;
 353     setsockopt(s, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes));
 354     if (connect(s, (sockaddr*)&dst, sizeof(dst)) < 0) {
 355         IF_LEVEL(1) LOG_NONMEMBER("failed to " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
 356         close(s);
 357         return NULL;
 358     }
 359     IF_LEVEL(2) LOG_NONMEMBER("connect_to_dst fd=" << s << " to dst " << inet_ntoa(dst.sin_addr) << ":" << ntoh(dst.sin_port));
 360     return new connection(mgr, s, lossy);
 361 }
 362