rpc/connection.cc

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