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