folly/io/async/AsyncSocket.cpp

   1 /*
   2  * Copyright 2014 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <folly/io/async/AsyncSocket.h>
  18
  19 #include <folly/io/async/EventBase.h>
  20 #include <folly/SocketAddress.h>
  21 #include <folly/io/IOBuf.h>
  22
  23 #include <poll.h>
  24 #include <errno.h>
  25 #include <limits.h>
  26 #include <unistd.h>
  27 #include <fcntl.h>
  28 #include <sys/types.h>
  29 #include <sys/socket.h>
  30 #include <netinet/in.h>
  31 #include <netinet/tcp.h>
  32
  33 using std::string;
  34 using std::unique_ptr;
  35
  36 namespace folly {
  37
  38 // static members initializers
  39 const AsyncSocket::OptionMap AsyncSocket::emptyOptionMap;
  40 const folly::SocketAddress AsyncSocket::anyAddress =
  41   folly::SocketAddress("0.0.0.0", 0);
  42
  43 const AsyncSocketException socketClosedLocallyEx(
  44     AsyncSocketException::END_OF_FILE, "socket closed locally");
  45 const AsyncSocketException socketShutdownForWritesEx(
  46     AsyncSocketException::END_OF_FILE, "socket shutdown for writes");
  47
  48 // TODO: It might help performance to provide a version of WriteRequest that
  49 // users could derive from, so we can avoid the extra allocation for each call
  50 // to write()/writev().  We could templatize TFramedAsyncChannel just like the
  51 // protocols are currently templatized for transports.
  52 //
  53 // We would need the version for external users where they provide the iovec
  54 // storage space, and only our internal version would allocate it at the end of
  55 // the WriteRequest.
  56
  57 /**
  58  * A WriteRequest object tracks information about a pending write() or writev()
  59  * operation.
  60  *
  61  * A new WriteRequest operation is allocated on the heap for all write
  62  * operations that cannot be completed immediately.
  63  */
  64 class AsyncSocket::WriteRequest {
  65  public:
  66   static WriteRequest* newRequest(WriteCallback* callback,
  67                                   const iovec* ops,
  68                                   uint32_t opCount,
  69                                   unique_ptr<IOBuf>&& ioBuf,
  70                                   WriteFlags flags) {
  71     assert(opCount > 0);
  72     // Since we put a variable size iovec array at the end
  73     // of each WriteRequest, we have to manually allocate the memory.
  74     void* buf = malloc(sizeof(WriteRequest) +
  75                        (opCount * sizeof(struct iovec)));
  76     if (buf == nullptr) {
  77       throw std::bad_alloc();
  78     }
  79
  80     return new(buf) WriteRequest(callback, ops, opCount, std::move(ioBuf),
  81                                  flags);
  82   }
  83
  84   void destroy() {
  85     this->~WriteRequest();
  86     free(this);
  87   }
  88
  89   bool cork() const {
  90     return isSet(flags_, WriteFlags::CORK);
  91   }
  92
  93   WriteFlags flags() const {
  94     return flags_;
  95   }
  96
  97   WriteRequest* getNext() const {
  98     return next_;
  99   }
 100
 101   WriteCallback* getCallback() const {
 102     return callback_;
 103   }
 104
 105   uint32_t getBytesWritten() const {
 106     return bytesWritten_;
 107   }
 108
 109   const struct iovec* getOps() const {
 110     assert(opCount_ > opIndex_);
 111     return writeOps_ + opIndex_;
 112   }
 113
 114   uint32_t getOpCount() const {
 115     assert(opCount_ > opIndex_);
 116     return opCount_ - opIndex_;
 117   }
 118
 119   void consume(uint32_t wholeOps, uint32_t partialBytes,
 120                uint32_t totalBytesWritten) {
 121     // Advance opIndex_ forward by wholeOps
 122     opIndex_ += wholeOps;
 123     assert(opIndex_ < opCount_);
 124
 125     // If we've finished writing any IOBufs, release them
 126     if (ioBuf_) {
 127       for (uint32_t i = wholeOps; i != 0; --i) {
 128         assert(ioBuf_);
 129         ioBuf_ = ioBuf_->pop();
 130       }
 131     }
 132
 133     // Move partialBytes forward into the current iovec buffer
 134     struct iovec* currentOp = writeOps_ + opIndex_;
 135     assert((partialBytes < currentOp->iov_len) || (currentOp->iov_len == 0));
 136     currentOp->iov_base =
 137       reinterpret_cast<uint8_t*>(currentOp->iov_base) + partialBytes;
 138     currentOp->iov_len -= partialBytes;
 139
 140     // Increment the bytesWritten_ count by totalBytesWritten
 141     bytesWritten_ += totalBytesWritten;
 142   }
 143
 144   void append(WriteRequest* next) {
 145     assert(next_ == nullptr);
 146     next_ = next;
 147   }
 148
 149  private:
 150   WriteRequest(WriteCallback* callback,
 151                const struct iovec* ops,
 152                uint32_t opCount,
 153                unique_ptr<IOBuf>&& ioBuf,
 154                WriteFlags flags)
 155     : next_(nullptr)
 156     , callback_(callback)
 157     , bytesWritten_(0)
 158     , opCount_(opCount)
 159     , opIndex_(0)
 160     , flags_(flags)
 161     , ioBuf_(std::move(ioBuf)) {
 162     memcpy(writeOps_, ops, sizeof(*ops) * opCount_);
 163   }
 164
 165   // Private destructor, to ensure callers use destroy()
 166   ~WriteRequest() {}
 167
 168   WriteRequest* next_;          ///< pointer to next WriteRequest
 169   WriteCallback* callback_;     ///< completion callback
 170   uint32_t bytesWritten_;       ///< bytes written
 171   uint32_t opCount_;            ///< number of entries in writeOps_
 172   uint32_t opIndex_;            ///< current index into writeOps_
 173   WriteFlags flags_;            ///< set for WriteFlags
 174   unique_ptr<IOBuf> ioBuf_;     ///< underlying IOBuf, or nullptr if N/A
 175   struct iovec writeOps_[];     ///< write operation(s) list
 176 };
 177
 178 AsyncSocket::AsyncSocket()
 179   : eventBase_(nullptr)
 180   , writeTimeout_(this, nullptr)
 181   , ioHandler_(this, nullptr) {
 182   VLOG(5) << "new AsyncSocket(" << ")";
 183   init();
 184 }
 185
 186 AsyncSocket::AsyncSocket(EventBase* evb)
 187   : eventBase_(evb)
 188   , writeTimeout_(this, evb)
 189   , ioHandler_(this, evb) {
 190   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
 191   init();
 192 }
 193
 194 AsyncSocket::AsyncSocket(EventBase* evb,
 195                            const folly::SocketAddress& address,
 196                            uint32_t connectTimeout)
 197   : eventBase_(evb)
 198   , writeTimeout_(this, evb)
 199   , ioHandler_(this, evb) {
 200   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
 201   init();
 202   connect(nullptr, address, connectTimeout);
 203 }
 204
 205 AsyncSocket::AsyncSocket(EventBase* evb,
 206                            const std::string& ip,
 207                            uint16_t port,
 208                            uint32_t connectTimeout)
 209   : eventBase_(evb)
 210   , writeTimeout_(this, evb)
 211   , ioHandler_(this, evb) {
 212   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
 213   init();
 214   connect(nullptr, ip, port, connectTimeout);
 215 }
 216
 217 AsyncSocket::AsyncSocket(EventBase* evb, int fd)
 218   : eventBase_(evb)
 219   , writeTimeout_(this, evb)
 220   , ioHandler_(this, evb, fd) {
 221   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ", fd="
 222           << fd << ")";
 223   init();
 224   fd_ = fd;
 225   state_ = StateEnum::ESTABLISHED;
 226 }
 227
 228 // init() method, since constructor forwarding isn't supported in most
 229 // compilers yet.
 230 void AsyncSocket::init() {
 231   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 232   shutdownFlags_ = 0;
 233   state_ = StateEnum::UNINIT;
 234   eventFlags_ = EventHandler::NONE;
 235   fd_ = -1;
 236   sendTimeout_ = 0;
 237   maxReadsPerEvent_ = 16;
 238   connectCallback_ = nullptr;
 239   readCallback_ = nullptr;
 240   writeReqHead_ = nullptr;
 241   writeReqTail_ = nullptr;
 242   shutdownSocketSet_ = nullptr;
 243   appBytesWritten_ = 0;
 244   appBytesReceived_ = 0;
 245 }
 246
 247 AsyncSocket::~AsyncSocket() {
 248   VLOG(7) << "actual destruction of AsyncSocket(this=" << this
 249           << ", evb=" << eventBase_ << ", fd=" << fd_
 250           << ", state=" << state_ << ")";
 251 }
 252
 253 void AsyncSocket::destroy() {
 254   VLOG(5) << "AsyncSocket::destroy(this=" << this << ", evb=" << eventBase_
 255           << ", fd=" << fd_ << ", state=" << state_;
 256   // When destroy is called, close the socket immediately
 257   closeNow();
 258
 259   // Then call DelayedDestruction::destroy() to take care of
 260   // whether or not we need immediate or delayed destruction
 261   DelayedDestruction::destroy();
 262 }
 263
 264 int AsyncSocket::detachFd() {
 265   VLOG(6) << "AsyncSocket::detachFd(this=" << this << ", fd=" << fd_
 266           << ", evb=" << eventBase_ << ", state=" << state_
 267           << ", events=" << std::hex << eventFlags_ << ")";
 268   // Extract the fd, and set fd_ to -1 first, so closeNow() won't
 269   // actually close the descriptor.
 270   if (shutdownSocketSet_) {
 271     shutdownSocketSet_->remove(fd_);
 272   }
 273   int fd = fd_;
 274   fd_ = -1;
 275   // Call closeNow() to invoke all pending callbacks with an error.
 276   closeNow();
 277   // Update the EventHandler to stop using this fd.
 278   // This can only be done after closeNow() unregisters the handler.
 279   ioHandler_.changeHandlerFD(-1);
 280   return fd;
 281 }
 282
 283 void AsyncSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
 284   if (shutdownSocketSet_ == newSS) {
 285     return;
 286   }
 287   if (shutdownSocketSet_ && fd_ != -1) {
 288     shutdownSocketSet_->remove(fd_);
 289   }
 290   shutdownSocketSet_ = newSS;
 291   if (shutdownSocketSet_ && fd_ != -1) {
 292     shutdownSocketSet_->add(fd_);
 293   }
 294 }
 295
 296 void AsyncSocket::connect(ConnectCallback* callback,
 297                            const folly::SocketAddress& address,
 298                            int timeout,
 299                            const OptionMap &options,
 300                            const folly::SocketAddress& bindAddr) noexcept {
 301   DestructorGuard dg(this);
 302   assert(eventBase_->isInEventBaseThread());
 303
 304   addr_ = address;
 305
 306   // Make sure we're in the uninitialized state
 307   if (state_ != StateEnum::UNINIT) {
 308     return invalidState(callback);
 309   }
 310
 311   assert(fd_ == -1);
 312   state_ = StateEnum::CONNECTING;
 313   connectCallback_ = callback;
 314
 315   sockaddr_storage addrStorage;
 316   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 317
 318   try {
 319     // Create the socket
 320     // Technically the first parameter should actually be a protocol family
 321     // constant (PF_xxx) rather than an address family (AF_xxx), but the
 322     // distinction is mainly just historical.  In pretty much all
 323     // implementations the PF_foo and AF_foo constants are identical.
 324     fd_ = socket(address.getFamily(), SOCK_STREAM, 0);
 325     if (fd_ < 0) {
 326       throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 327                                 withAddr("failed to create socket"), errno);
 328     }
 329     if (shutdownSocketSet_) {
 330       shutdownSocketSet_->add(fd_);
 331     }
 332     ioHandler_.changeHandlerFD(fd_);
 333
 334     // Set the FD_CLOEXEC flag so that the socket will be closed if the program
 335     // later forks and execs.
 336     int rv = fcntl(fd_, F_SETFD, FD_CLOEXEC);
 337     if (rv != 0) {
 338       throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 339                                 withAddr("failed to set close-on-exec flag"),
 340                                 errno);
 341     }
 342
 343     // Put the socket in non-blocking mode
 344     int flags = fcntl(fd_, F_GETFL, 0);
 345     if (flags == -1) {
 346       throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 347                                 withAddr("failed to get socket flags"), errno);
 348     }
 349     rv = fcntl(fd_, F_SETFL, flags | O_NONBLOCK);
 350     if (rv == -1) {
 351       throw AsyncSocketException(
 352           AsyncSocketException::INTERNAL_ERROR,
 353           withAddr("failed to put socket in non-blocking mode"),
 354           errno);
 355     }
 356
 357 #if !defined(MSG_NOSIGNAL) && defined(F_SETNOSIGPIPE)
 358     // iOS and OS X don't support MSG_NOSIGNAL; set F_SETNOSIGPIPE instead
 359     rv = fcntl(fd_, F_SETNOSIGPIPE, 1);
 360     if (rv == -1) {
 361       throw AsyncSocketException(
 362           AsyncSocketException::INTERNAL_ERROR,
 363           "failed to enable F_SETNOSIGPIPE on socket",
 364           errno);
 365     }
 366 #endif
 367
 368     // By default, turn on TCP_NODELAY
 369     // If setNoDelay() fails, we continue anyway; this isn't a fatal error.
 370     // setNoDelay() will log an error message if it fails.
 371     if (address.getFamily() != AF_UNIX) {
 372       (void)setNoDelay(true);
 373     }
 374
 375     VLOG(5) << "AsyncSocket::connect(this=" << this << ", evb=" << eventBase_
 376             << ", fd=" << fd_ << ", host=" << address.describe().c_str();
 377
 378     // bind the socket
 379     if (bindAddr != anyAddress) {
 380       int one = 1;
 381       if (::setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one))) {
 382         doClose();
 383         throw AsyncSocketException(
 384           AsyncSocketException::NOT_OPEN,
 385           "failed to setsockopt prior to bind on " + bindAddr.describe(),
 386           errno);
 387       }
 388
 389       bindAddr.getAddress(&addrStorage);
 390
 391       if (::bind(fd_, saddr, bindAddr.getActualSize()) != 0) {
 392         doClose();
 393         throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
 394                                   "failed to bind to async socket: " +
 395                                   bindAddr.describe(),
 396                                   errno);
 397       }
 398     }
 399
 400     // Apply the additional options if any.
 401     for (const auto& opt: options) {
 402       int rv = opt.first.apply(fd_, opt.second);
 403       if (rv != 0) {
 404         throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 405                                   withAddr("failed to set socket option"),
 406                                   errno);
 407       }
 408     }
 409
 410     // Perform the connect()
 411     address.getAddress(&addrStorage);
 412
 413     rv = ::connect(fd_, saddr, address.getActualSize());
 414     if (rv < 0) {
 415       if (errno == EINPROGRESS) {
 416         // Connection in progress.
 417         if (timeout > 0) {
 418           // Start a timer in case the connection takes too long.
 419           if (!writeTimeout_.scheduleTimeout(timeout)) {
 420             throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 421                 withAddr("failed to schedule AsyncSocket connect timeout"));
 422           }
 423         }
 424
 425         // Register for write events, so we'll
 426         // be notified when the connection finishes/fails.
 427         // Note that we don't register for a persistent event here.
 428         assert(eventFlags_ == EventHandler::NONE);
 429         eventFlags_ = EventHandler::WRITE;
 430         if (!ioHandler_.registerHandler(eventFlags_)) {
 431           throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
 432               withAddr("failed to register AsyncSocket connect handler"));
 433         }
 434         return;
 435       } else {
 436         throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
 437                                   "connect failed (immediately)", errno);
 438       }
 439     }
 440
 441     // If we're still here the connect() succeeded immediately.
 442     // Fall through to call the callback outside of this try...catch block
 443   } catch (const AsyncSocketException& ex) {
 444     return failConnect(__func__, ex);
 445   } catch (const std::exception& ex) {
 446     // shouldn't happen, but handle it just in case
 447     VLOG(4) << "AsyncSocket::connect(this=" << this << ", fd=" << fd_
 448                << "): unexpected " << typeid(ex).name() << " exception: "
 449                << ex.what();
 450     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 451                             withAddr(string("unexpected exception: ") +
 452                                      ex.what()));
 453     return failConnect(__func__, tex);
 454   }
 455
 456   // The connection succeeded immediately
 457   // The read callback may not have been set yet, and no writes may be pending
 458   // yet, so we don't have to register for any events at the moment.
 459   VLOG(8) << "AsyncSocket::connect succeeded immediately; this=" << this;
 460   assert(readCallback_ == nullptr);
 461   assert(writeReqHead_ == nullptr);
 462   state_ = StateEnum::ESTABLISHED;
 463   if (callback) {
 464     connectCallback_ = nullptr;
 465     callback->connectSuccess();
 466   }
 467 }
 468
 469 void AsyncSocket::connect(ConnectCallback* callback,
 470                            const string& ip, uint16_t port,
 471                            int timeout,
 472                            const OptionMap &options) noexcept {
 473   DestructorGuard dg(this);
 474   try {
 475     connectCallback_ = callback;
 476     connect(callback, folly::SocketAddress(ip, port), timeout, options);
 477   } catch (const std::exception& ex) {
 478     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 479                             ex.what());
 480     return failConnect(__func__, tex);
 481   }
 482 }
 483
 484 void AsyncSocket::cancelConnect() {
 485   connectCallback_ = nullptr;
 486   if (state_ == StateEnum::CONNECTING) {
 487     closeNow();
 488   }
 489 }
 490
 491 void AsyncSocket::setSendTimeout(uint32_t milliseconds) {
 492   sendTimeout_ = milliseconds;
 493   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 494
 495   // If we are currently pending on write requests, immediately update
 496   // writeTimeout_ with the new value.
 497   if ((eventFlags_ & EventHandler::WRITE) &&
 498       (state_ != StateEnum::CONNECTING)) {
 499     assert(state_ == StateEnum::ESTABLISHED);
 500     assert((shutdownFlags_ & SHUT_WRITE) == 0);
 501     if (sendTimeout_ > 0) {
 502       if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
 503         AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
 504             withAddr("failed to reschedule send timeout in setSendTimeout"));
 505         return failWrite(__func__, ex);
 506       }
 507     } else {
 508       writeTimeout_.cancelTimeout();
 509     }
 510   }
 511 }
 512
 513 void AsyncSocket::setReadCB(ReadCallback *callback) {
 514   VLOG(6) << "AsyncSocket::setReadCallback() this=" << this << ", fd=" << fd_
 515           << ", callback=" << callback << ", state=" << state_;
 516
 517   // Short circuit if callback is the same as the existing readCallback_.
 518   //
 519   // Note that this is needed for proper functioning during some cleanup cases.
 520   // During cleanup we allow setReadCallback(nullptr) to be called even if the
 521   // read callback is already unset and we have been detached from an event
 522   // base.  This check prevents us from asserting
 523   // eventBase_->isInEventBaseThread() when eventBase_ is nullptr.
 524   if (callback == readCallback_) {
 525     return;
 526   }
 527
 528   if (shutdownFlags_ & SHUT_READ) {
 529     // Reads have already been shut down on this socket.
 530     //
 531     // Allow setReadCallback(nullptr) to be called in this case, but don't
 532     // allow a new callback to be set.
 533     //
 534     // For example, setReadCallback(nullptr) can happen after an error if we
 535     // invoke some other error callback before invoking readError().  The other
 536     // error callback that is invoked first may go ahead and clear the read
 537     // callback before we get a chance to invoke readError().
 538     if (callback != nullptr) {
 539       return invalidState(callback);
 540     }
 541     assert((eventFlags_ & EventHandler::READ) == 0);
 542     readCallback_ = nullptr;
 543     return;
 544   }
 545
 546   DestructorGuard dg(this);
 547   assert(eventBase_->isInEventBaseThread());
 548
 549   switch ((StateEnum)state_) {
 550     case StateEnum::CONNECTING:
 551       // For convenience, we allow the read callback to be set while we are
 552       // still connecting.  We just store the callback for now.  Once the
 553       // connection completes we'll register for read events.
 554       readCallback_ = callback;
 555       return;
 556     case StateEnum::ESTABLISHED:
 557     {
 558       readCallback_ = callback;
 559       uint16_t oldFlags = eventFlags_;
 560       if (readCallback_) {
 561         eventFlags_ |= EventHandler::READ;
 562       } else {
 563         eventFlags_ &= ~EventHandler::READ;
 564       }
 565
 566       // Update our registration if our flags have changed
 567       if (eventFlags_ != oldFlags) {
 568         // We intentionally ignore the return value here.
 569         // updateEventRegistration() will move us into the error state if it
 570         // fails, and we don't need to do anything else here afterwards.
 571         (void)updateEventRegistration();
 572       }
 573
 574       if (readCallback_) {
 575         checkForImmediateRead();
 576       }
 577       return;
 578     }
 579     case StateEnum::CLOSED:
 580     case StateEnum::ERROR:
 581       // We should never reach here.  SHUT_READ should always be set
 582       // if we are in STATE_CLOSED or STATE_ERROR.
 583       assert(false);
 584       return invalidState(callback);
 585     case StateEnum::UNINIT:
 586       // We do not allow setReadCallback() to be called before we start
 587       // connecting.
 588       return invalidState(callback);
 589   }
 590
 591   // We don't put a default case in the switch statement, so that the compiler
 592   // will warn us to update the switch statement if a new state is added.
 593   return invalidState(callback);
 594 }
 595
 596 AsyncSocket::ReadCallback* AsyncSocket::getReadCallback() const {
 597   return readCallback_;
 598 }
 599
 600 void AsyncSocket::write(WriteCallback* callback,
 601                          const void* buf, size_t bytes, WriteFlags flags) {
 602   iovec op;
 603   op.iov_base = const_cast<void*>(buf);
 604   op.iov_len = bytes;
 605   writeImpl(callback, &op, 1, std::move(unique_ptr<IOBuf>()), flags);
 606 }
 607
 608 void AsyncSocket::writev(WriteCallback* callback,
 609                           const iovec* vec,
 610                           size_t count,
 611                           WriteFlags flags) {
 612   writeImpl(callback, vec, count, std::move(unique_ptr<IOBuf>()), flags);
 613 }
 614
 615 void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf,
 616                               WriteFlags flags) {
 617   size_t count = buf->countChainElements();
 618   if (count <= 64) {
 619     iovec vec[count];
 620     writeChainImpl(callback, vec, count, std::move(buf), flags);
 621   } else {
 622     iovec* vec = new iovec[count];
 623     writeChainImpl(callback, vec, count, std::move(buf), flags);
 624     delete[] vec;
 625   }
 626 }
 627
 628 void AsyncSocket::writeChainImpl(WriteCallback* callback, iovec* vec,
 629     size_t count, unique_ptr<IOBuf>&& buf, WriteFlags flags) {
 630   const IOBuf* head = buf.get();
 631   const IOBuf* next = head;
 632   unsigned i = 0;
 633   do {
 634     vec[i].iov_base = const_cast<uint8_t *>(next->data());
 635     vec[i].iov_len = next->length();
 636     // IOBuf can get confused by empty iovec buffers, so increment the
 637     // output pointer only if the iovec buffer is non-empty.  We could
 638     // end the loop with i < count, but that's ok.
 639     if (vec[i].iov_len != 0) {
 640       i++;
 641     }
 642     next = next->next();
 643   } while (next != head);
 644   writeImpl(callback, vec, i, std::move(buf), flags);
 645 }
 646
 647 void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
 648                              size_t count, unique_ptr<IOBuf>&& buf,
 649                              WriteFlags flags) {
 650   VLOG(6) << "AsyncSocket::writev() this=" << this << ", fd=" << fd_
 651           << ", callback=" << callback << ", count=" << count
 652           << ", state=" << state_;
 653   DestructorGuard dg(this);
 654   unique_ptr<IOBuf>ioBuf(std::move(buf));
 655   assert(eventBase_->isInEventBaseThread());
 656
 657   if (shutdownFlags_ & (SHUT_WRITE | SHUT_WRITE_PENDING)) {
 658     // No new writes may be performed after the write side of the socket has
 659     // been shutdown.
 660     //
 661     // We could just call callback->writeError() here to fail just this write.
 662     // However, fail hard and use invalidState() to fail all outstanding
 663     // callbacks and move the socket into the error state.  There's most likely
 664     // a bug in the caller's code, so we abort everything rather than trying to
 665     // proceed as best we can.
 666     return invalidState(callback);
 667   }
 668
 669   uint32_t countWritten = 0;
 670   uint32_t partialWritten = 0;
 671   int bytesWritten = 0;
 672   bool mustRegister = false;
 673   if (state_ == StateEnum::ESTABLISHED && !connecting()) {
 674     if (writeReqHead_ == nullptr) {
 675       // If we are established and there are no other writes pending,
 676       // we can attempt to perform the write immediately.
 677       assert(writeReqTail_ == nullptr);
 678       assert((eventFlags_ & EventHandler::WRITE) == 0);
 679
 680       bytesWritten = performWrite(vec, count, flags,
 681                                   &countWritten, &partialWritten);
 682       if (bytesWritten < 0) {
 683         AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
 684                                withAddr("writev failed"), errno);
 685         return failWrite(__func__, callback, 0, ex);
 686       } else if (countWritten == count) {
 687         // We successfully wrote everything.
 688         // Invoke the callback and return.
 689         if (callback) {
 690           callback->writeSuccess();
 691         }
 692         return;
 693       } // else { continue writing the next writeReq }
 694       mustRegister = true;
 695     }
 696   } else if (!connecting()) {
 697     // Invalid state for writing
 698     return invalidState(callback);
 699   }
 700
 701   // Create a new WriteRequest to add to the queue
 702   WriteRequest* req;
 703   try {
 704     req = WriteRequest::newRequest(callback, vec + countWritten,
 705                                    count - countWritten, std::move(ioBuf),
 706                                    flags);
 707   } catch (const std::exception& ex) {
 708     // we mainly expect to catch std::bad_alloc here
 709     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 710         withAddr(string("failed to append new WriteRequest: ") + ex.what()));
 711     return failWrite(__func__, callback, bytesWritten, tex);
 712   }
 713   req->consume(0, partialWritten, bytesWritten);
 714   if (writeReqTail_ == nullptr) {
 715     assert(writeReqHead_ == nullptr);
 716     writeReqHead_ = writeReqTail_ = req;
 717   } else {
 718     writeReqTail_->append(req);
 719     writeReqTail_ = req;
 720   }
 721
 722   // Register for write events if are established and not currently
 723   // waiting on write events
 724   if (mustRegister) {
 725     assert(state_ == StateEnum::ESTABLISHED);
 726     assert((eventFlags_ & EventHandler::WRITE) == 0);
 727     if (!updateEventRegistration(EventHandler::WRITE, 0)) {
 728       assert(state_ == StateEnum::ERROR);
 729       return;
 730     }
 731     if (sendTimeout_ > 0) {
 732       // Schedule a timeout to fire if the write takes too long.
 733       if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
 734         AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
 735                                withAddr("failed to schedule send timeout"));
 736         return failWrite(__func__, ex);
 737       }
 738     }
 739   }
 740 }
 741
 742 void AsyncSocket::close() {
 743   VLOG(5) << "AsyncSocket::close(): this=" << this << ", fd_=" << fd_
 744           << ", state=" << state_ << ", shutdownFlags="
 745           << std::hex << (int) shutdownFlags_;
 746
 747   // close() is only different from closeNow() when there are pending writes
 748   // that need to drain before we can close.  In all other cases, just call
 749   // closeNow().
 750   //
 751   // Note that writeReqHead_ can be non-nullptr even in STATE_CLOSED or
 752   // STATE_ERROR if close() is invoked while a previous closeNow() or failure
 753   // is still running.  (e.g., If there are multiple pending writes, and we
 754   // call writeError() on the first one, it may call close().  In this case we
 755   // will already be in STATE_CLOSED or STATE_ERROR, but the remaining pending
 756   // writes will still be in the queue.)
 757   //
 758   // We only need to drain pending writes if we are still in STATE_CONNECTING
 759   // or STATE_ESTABLISHED
 760   if ((writeReqHead_ == nullptr) ||
 761       !(state_ == StateEnum::CONNECTING ||
 762       state_ == StateEnum::ESTABLISHED)) {
 763     closeNow();
 764     return;
 765   }
 766
 767   // Declare a DestructorGuard to ensure that the AsyncSocket cannot be
 768   // destroyed until close() returns.
 769   DestructorGuard dg(this);
 770   assert(eventBase_->isInEventBaseThread());
 771
 772   // Since there are write requests pending, we have to set the
 773   // SHUT_WRITE_PENDING flag, and wait to perform the real close until the
 774   // connect finishes and we finish writing these requests.
 775   //
 776   // Set SHUT_READ to indicate that reads are shut down, and set the
 777   // SHUT_WRITE_PENDING flag to mark that we want to shutdown once the
 778   // pending writes complete.
 779   shutdownFlags_ |= (SHUT_READ | SHUT_WRITE_PENDING);
 780
 781   // If a read callback is set, invoke readEOF() immediately to inform it that
 782   // the socket has been closed and no more data can be read.
 783   if (readCallback_) {
 784     // Disable reads if they are enabled
 785     if (!updateEventRegistration(0, EventHandler::READ)) {
 786       // We're now in the error state; callbacks have been cleaned up
 787       assert(state_ == StateEnum::ERROR);
 788       assert(readCallback_ == nullptr);
 789     } else {
 790       ReadCallback* callback = readCallback_;
 791       readCallback_ = nullptr;
 792       callback->readEOF();
 793     }
 794   }
 795 }
 796
 797 void AsyncSocket::closeNow() {
 798   VLOG(5) << "AsyncSocket::closeNow(): this=" << this << ", fd_=" << fd_
 799           << ", state=" << state_ << ", shutdownFlags="
 800           << std::hex << (int) shutdownFlags_;
 801   DestructorGuard dg(this);
 802   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 803
 804   switch (state_) {
 805     case StateEnum::ESTABLISHED:
 806     case StateEnum::CONNECTING:
 807     {
 808       shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
 809       state_ = StateEnum::CLOSED;
 810
 811       // If the write timeout was set, cancel it.
 812       writeTimeout_.cancelTimeout();
 813
 814       // If we are registered for I/O events, unregister.
 815       if (eventFlags_ != EventHandler::NONE) {
 816         eventFlags_ = EventHandler::NONE;
 817         if (!updateEventRegistration()) {
 818           // We will have been moved into the error state.
 819           assert(state_ == StateEnum::ERROR);
 820           return;
 821         }
 822       }
 823
 824       if (fd_ >= 0) {
 825         ioHandler_.changeHandlerFD(-1);
 826         doClose();
 827       }
 828
 829       if (connectCallback_) {
 830         ConnectCallback* callback = connectCallback_;
 831         connectCallback_ = nullptr;
 832         callback->connectErr(socketClosedLocallyEx);
 833       }
 834
 835       failAllWrites(socketClosedLocallyEx);
 836
 837       if (readCallback_) {
 838         ReadCallback* callback = readCallback_;
 839         readCallback_ = nullptr;
 840         callback->readEOF();
 841       }
 842       return;
 843     }
 844     case StateEnum::CLOSED:
 845       // Do nothing.  It's possible that we are being called recursively
 846       // from inside a callback that we invoked inside another call to close()
 847       // that is still running.
 848       return;
 849     case StateEnum::ERROR:
 850       // Do nothing.  The error handling code has performed (or is performing)
 851       // cleanup.
 852       return;
 853     case StateEnum::UNINIT:
 854       assert(eventFlags_ == EventHandler::NONE);
 855       assert(connectCallback_ == nullptr);
 856       assert(readCallback_ == nullptr);
 857       assert(writeReqHead_ == nullptr);
 858       shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
 859       state_ = StateEnum::CLOSED;
 860       return;
 861   }
 862
 863   LOG(DFATAL) << "AsyncSocket::closeNow() (this=" << this << ", fd=" << fd_
 864               << ") called in unknown state " << state_;
 865 }
 866
 867 void AsyncSocket::closeWithReset() {
 868   // Enable SO_LINGER, with the linger timeout set to 0.
 869   // This will trigger a TCP reset when we close the socket.
 870   if (fd_ >= 0) {
 871     struct linger optLinger = {1, 0};
 872     if (setSockOpt(SOL_SOCKET, SO_LINGER, &optLinger) != 0) {
 873       VLOG(2) << "AsyncSocket::closeWithReset(): error setting SO_LINGER "
 874               << "on " << fd_ << ": errno=" << errno;
 875     }
 876   }
 877
 878   // Then let closeNow() take care of the rest
 879   closeNow();
 880 }
 881
 882 void AsyncSocket::shutdownWrite() {
 883   VLOG(5) << "AsyncSocket::shutdownWrite(): this=" << this << ", fd=" << fd_
 884           << ", state=" << state_ << ", shutdownFlags="
 885           << std::hex << (int) shutdownFlags_;
 886
 887   // If there are no pending writes, shutdownWrite() is identical to
 888   // shutdownWriteNow().
 889   if (writeReqHead_ == nullptr) {
 890     shutdownWriteNow();
 891     return;
 892   }
 893
 894   assert(eventBase_->isInEventBaseThread());
 895
 896   // There are pending writes.  Set SHUT_WRITE_PENDING so that the actual
 897   // shutdown will be performed once all writes complete.
 898   shutdownFlags_ |= SHUT_WRITE_PENDING;
 899 }
 900
 901 void AsyncSocket::shutdownWriteNow() {
 902   VLOG(5) << "AsyncSocket::shutdownWriteNow(): this=" << this
 903           << ", fd=" << fd_ << ", state=" << state_
 904           << ", shutdownFlags=" << std::hex << (int) shutdownFlags_;
 905
 906   if (shutdownFlags_ & SHUT_WRITE) {
 907     // Writes are already shutdown; nothing else to do.
 908     return;
 909   }
 910
 911   // If SHUT_READ is already set, just call closeNow() to completely
 912   // close the socket.  This can happen if close() was called with writes
 913   // pending, and then shutdownWriteNow() is called before all pending writes
 914   // complete.
 915   if (shutdownFlags_ & SHUT_READ) {
 916     closeNow();
 917     return;
 918   }
 919
 920   DestructorGuard dg(this);
 921   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 922
 923   switch (static_cast<StateEnum>(state_)) {
 924     case StateEnum::ESTABLISHED:
 925     {
 926       shutdownFlags_ |= SHUT_WRITE;
 927
 928       // If the write timeout was set, cancel it.
 929       writeTimeout_.cancelTimeout();
 930
 931       // If we are registered for write events, unregister.
 932       if (!updateEventRegistration(0, EventHandler::WRITE)) {
 933         // We will have been moved into the error state.
 934         assert(state_ == StateEnum::ERROR);
 935         return;
 936       }
 937
 938       // Shutdown writes on the file descriptor
 939       ::shutdown(fd_, SHUT_WR);
 940
 941       // Immediately fail all write requests
 942       failAllWrites(socketShutdownForWritesEx);
 943       return;
 944     }
 945     case StateEnum::CONNECTING:
 946     {
 947       // Set the SHUT_WRITE_PENDING flag.
 948       // When the connection completes, it will check this flag,
 949       // shutdown the write half of the socket, and then set SHUT_WRITE.
 950       shutdownFlags_ |= SHUT_WRITE_PENDING;
 951
 952       // Immediately fail all write requests
 953       failAllWrites(socketShutdownForWritesEx);
 954       return;
 955     }
 956     case StateEnum::UNINIT:
 957       // Callers normally shouldn't call shutdownWriteNow() before the socket
 958       // even starts connecting.  Nonetheless, go ahead and set
 959       // SHUT_WRITE_PENDING.  Once the socket eventually connects it will
 960       // immediately shut down the write side of the socket.
 961       shutdownFlags_ |= SHUT_WRITE_PENDING;
 962       return;
 963     case StateEnum::CLOSED:
 964     case StateEnum::ERROR:
 965       // We should never get here.  SHUT_WRITE should always be set
 966       // in STATE_CLOSED and STATE_ERROR.
 967       VLOG(4) << "AsyncSocket::shutdownWriteNow() (this=" << this
 968                  << ", fd=" << fd_ << ") in unexpected state " << state_
 969                  << " with SHUT_WRITE not set ("
 970                  << std::hex << (int) shutdownFlags_ << ")";
 971       assert(false);
 972       return;
 973   }
 974
 975   LOG(DFATAL) << "AsyncSocket::shutdownWriteNow() (this=" << this << ", fd="
 976               << fd_ << ") called in unknown state " << state_;
 977 }
 978
 979 bool AsyncSocket::readable() const {
 980   if (fd_ == -1) {
 981     return false;
 982   }
 983   struct pollfd fds[1];
 984   fds[0].fd = fd_;
 985   fds[0].events = POLLIN;
 986   fds[0].revents = 0;
 987   int rc = poll(fds, 1, 0);
 988   return rc == 1;
 989 }
 990
 991 bool AsyncSocket::isPending() const {
 992   return ioHandler_.isPending();
 993 }
 994
 995 bool AsyncSocket::hangup() const {
 996   if (fd_ == -1) {
 997     // sanity check, no one should ask for hangup if we are not connected.
 998     assert(false);
 999     return false;
1000   }
1001 #ifdef POLLRDHUP // Linux-only
1002   struct pollfd fds[1];
1003   fds[0].fd = fd_;
1004   fds[0].events = POLLRDHUP|POLLHUP;
1005   fds[0].revents = 0;
1006   poll(fds, 1, 0);
1007   return (fds[0].revents & (POLLRDHUP|POLLHUP)) != 0;
1008 #else
1009   return false;
1010 #endif
1011 }
1012
1013 bool AsyncSocket::good() const {
1014   return ((state_ == StateEnum::CONNECTING ||
1015           state_ == StateEnum::ESTABLISHED) &&
1016           (shutdownFlags_ == 0) && (eventBase_ != nullptr));
1017 }
1018
1019 bool AsyncSocket::error() const {
1020   return (state_ == StateEnum::ERROR);
1021 }
1022
1023 void AsyncSocket::attachEventBase(EventBase* eventBase) {
1024   VLOG(5) << "AsyncSocket::attachEventBase(this=" << this << ", fd=" << fd_
1025           << ", old evb=" << eventBase_ << ", new evb=" << eventBase
1026           << ", state=" << state_ << ", events="
1027           << std::hex << eventFlags_ << ")";
1028   assert(eventBase_ == nullptr);
1029   assert(eventBase->isInEventBaseThread());
1030
1031   eventBase_ = eventBase;
1032   ioHandler_.attachEventBase(eventBase);
1033   writeTimeout_.attachEventBase(eventBase);
1034 }
1035
1036 void AsyncSocket::detachEventBase() {
1037   VLOG(5) << "AsyncSocket::detachEventBase(this=" << this << ", fd=" << fd_
1038           << ", old evb=" << eventBase_ << ", state=" << state_
1039           << ", events=" << std::hex << eventFlags_ << ")";
1040   assert(eventBase_ != nullptr);
1041   assert(eventBase_->isInEventBaseThread());
1042
1043   eventBase_ = nullptr;
1044   ioHandler_.detachEventBase();
1045   writeTimeout_.detachEventBase();
1046 }
1047
1048 bool AsyncSocket::isDetachable() const {
1049   DCHECK(eventBase_ != nullptr);
1050   DCHECK(eventBase_->isInEventBaseThread());
1051
1052   return !ioHandler_.isHandlerRegistered() && !writeTimeout_.isScheduled();
1053 }
1054
1055 void AsyncSocket::getLocalAddress(folly::SocketAddress* address) const {
1056   address->setFromLocalAddress(fd_);
1057 }
1058
1059 void AsyncSocket::getPeerAddress(folly::SocketAddress* address) const {
1060   if (!addr_.isInitialized()) {
1061     addr_.setFromPeerAddress(fd_);
1062   }
1063   *address = addr_;
1064 }
1065
1066 int AsyncSocket::setNoDelay(bool noDelay) {
1067   if (fd_ < 0) {
1068     VLOG(4) << "AsyncSocket::setNoDelay() called on non-open socket "
1069                << this << "(state=" << state_ << ")";
1070     return EINVAL;
1071
1072   }
1073
1074   int value = noDelay ? 1 : 0;
1075   if (setsockopt(fd_, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)) != 0) {
1076     int errnoCopy = errno;
1077     VLOG(2) << "failed to update TCP_NODELAY option on AsyncSocket "
1078             << this << " (fd=" << fd_ << ", state=" << state_ << "): "
1079             << strerror(errnoCopy);
1080     return errnoCopy;
1081   }
1082
1083   return 0;
1084 }
1085
1086 int AsyncSocket::setCongestionFlavor(const std::string &cname) {
1087
1088   #ifndef TCP_CONGESTION
1089   #define TCP_CONGESTION  13
1090   #endif
1091
1092   if (fd_ < 0) {
1093     VLOG(4) << "AsyncSocket::setCongestionFlavor() called on non-open "
1094                << "socket " << this << "(state=" << state_ << ")";
1095     return EINVAL;
1096
1097   }
1098
1099   if (setsockopt(fd_, IPPROTO_TCP, TCP_CONGESTION, cname.c_str(),
1100         cname.length() + 1) != 0) {
1101     int errnoCopy = errno;
1102     VLOG(2) << "failed to update TCP_CONGESTION option on AsyncSocket "
1103             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1104             << strerror(errnoCopy);
1105     return errnoCopy;
1106   }
1107
1108   return 0;
1109 }
1110
1111 int AsyncSocket::setQuickAck(bool quickack) {
1112   if (fd_ < 0) {
1113     VLOG(4) << "AsyncSocket::setQuickAck() called on non-open socket "
1114                << this << "(state=" << state_ << ")";
1115     return EINVAL;
1116
1117   }
1118
1119 #ifdef TCP_QUICKACK // Linux-only
1120   int value = quickack ? 1 : 0;
1121   if (setsockopt(fd_, IPPROTO_TCP, TCP_QUICKACK, &value, sizeof(value)) != 0) {
1122     int errnoCopy = errno;
1123     VLOG(2) << "failed to update TCP_QUICKACK option on AsyncSocket"
1124             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1125             << strerror(errnoCopy);
1126     return errnoCopy;
1127   }
1128
1129   return 0;
1130 #else
1131   return ENOSYS;
1132 #endif
1133 }
1134
1135 int AsyncSocket::setSendBufSize(size_t bufsize) {
1136   if (fd_ < 0) {
1137     VLOG(4) << "AsyncSocket::setSendBufSize() called on non-open socket "
1138                << this << "(state=" << state_ << ")";
1139     return EINVAL;
1140   }
1141
1142   if (setsockopt(fd_, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)) !=0) {
1143     int errnoCopy = errno;
1144     VLOG(2) << "failed to update SO_SNDBUF option on AsyncSocket"
1145             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1146             << strerror(errnoCopy);
1147     return errnoCopy;
1148   }
1149
1150   return 0;
1151 }
1152
1153 int AsyncSocket::setRecvBufSize(size_t bufsize) {
1154   if (fd_ < 0) {
1155     VLOG(4) << "AsyncSocket::setRecvBufSize() called on non-open socket "
1156                << this << "(state=" << state_ << ")";
1157     return EINVAL;
1158   }
1159
1160   if (setsockopt(fd_, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)) !=0) {
1161     int errnoCopy = errno;
1162     VLOG(2) << "failed to update SO_RCVBUF option on AsyncSocket"
1163             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1164             << strerror(errnoCopy);
1165     return errnoCopy;
1166   }
1167
1168   return 0;
1169 }
1170
1171 int AsyncSocket::setTCPProfile(int profd) {
1172   if (fd_ < 0) {
1173     VLOG(4) << "AsyncSocket::setTCPProfile() called on non-open socket "
1174                << this << "(state=" << state_ << ")";
1175     return EINVAL;
1176   }
1177
1178   if (setsockopt(fd_, SOL_SOCKET, SO_SET_NAMESPACE, &profd, sizeof(int)) !=0) {
1179     int errnoCopy = errno;
1180     VLOG(2) << "failed to set socket namespace option on AsyncSocket"
1181             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1182             << strerror(errnoCopy);
1183     return errnoCopy;
1184   }
1185
1186   return 0;
1187 }
1188
1189 void AsyncSocket::ioReady(uint16_t events) noexcept {
1190   VLOG(7) << "AsyncSocket::ioRead() this=" << this << ", fd" << fd_
1191           << ", events=" << std::hex << events << ", state=" << state_;
1192   DestructorGuard dg(this);
1193   assert(events & EventHandler::READ_WRITE);
1194   assert(eventBase_->isInEventBaseThread());
1195
1196   uint16_t relevantEvents = events & EventHandler::READ_WRITE;
1197   if (relevantEvents == EventHandler::READ) {
1198     handleRead();
1199   } else if (relevantEvents == EventHandler::WRITE) {
1200     handleWrite();
1201   } else if (relevantEvents == EventHandler::READ_WRITE) {
1202     EventBase* originalEventBase = eventBase_;
1203     // If both read and write events are ready, process writes first.
1204     handleWrite();
1205
1206     // Return now if handleWrite() detached us from our EventBase
1207     if (eventBase_ != originalEventBase) {
1208       return;
1209     }
1210
1211     // Only call handleRead() if a read callback is still installed.
1212     // (It's possible that the read callback was uninstalled during
1213     // handleWrite().)
1214     if (readCallback_) {
1215       handleRead();
1216     }
1217   } else {
1218     VLOG(4) << "AsyncSocket::ioRead() called with unexpected events "
1219                << std::hex << events << "(this=" << this << ")";
1220     abort();
1221   }
1222 }
1223
1224 ssize_t AsyncSocket::performRead(void* buf, size_t buflen) {
1225   ssize_t bytes = recv(fd_, buf, buflen, MSG_DONTWAIT);
1226   if (bytes < 0) {
1227     if (errno == EAGAIN || errno == EWOULDBLOCK) {
1228       // No more data to read right now.
1229       return READ_BLOCKING;
1230     } else {
1231       return READ_ERROR;
1232     }
1233   } else {
1234     appBytesReceived_ += bytes;
1235     return bytes;
1236   }
1237 }
1238
1239 void AsyncSocket::handleRead() noexcept {
1240   VLOG(5) << "AsyncSocket::handleRead() this=" << this << ", fd=" << fd_
1241           << ", state=" << state_;
1242   assert(state_ == StateEnum::ESTABLISHED);
1243   assert((shutdownFlags_ & SHUT_READ) == 0);
1244   assert(readCallback_ != nullptr);
1245   assert(eventFlags_ & EventHandler::READ);
1246
1247   // Loop until:
1248   // - a read attempt would block
1249   // - readCallback_ is uninstalled
1250   // - the number of loop iterations exceeds the optional maximum
1251   // - this AsyncSocket is moved to another EventBase
1252   //
1253   // When we invoke readDataAvailable() it may uninstall the readCallback_,
1254   // which is why need to check for it here.
1255   //
1256   // The last bullet point is slightly subtle.  readDataAvailable() may also
1257   // detach this socket from this EventBase.  However, before
1258   // readDataAvailable() returns another thread may pick it up, attach it to
1259   // a different EventBase, and install another readCallback_.  We need to
1260   // exit immediately after readDataAvailable() returns if the eventBase_ has
1261   // changed.  (The caller must perform some sort of locking to transfer the
1262   // AsyncSocket between threads properly.  This will be sufficient to ensure
1263   // that this thread sees the updated eventBase_ variable after
1264   // readDataAvailable() returns.)
1265   uint16_t numReads = 0;
1266   EventBase* originalEventBase = eventBase_;
1267   while (readCallback_ && eventBase_ == originalEventBase) {
1268     // Get the buffer to read into.
1269     void* buf = nullptr;
1270     size_t buflen = 0;
1271     try {
1272       readCallback_->getReadBuffer(&buf, &buflen);
1273     } catch (const AsyncSocketException& ex) {
1274       return failRead(__func__, ex);
1275     } catch (const std::exception& ex) {
1276       AsyncSocketException tex(AsyncSocketException::BAD_ARGS,
1277                               string("ReadCallback::getReadBuffer() "
1278                                      "threw exception: ") +
1279                               ex.what());
1280       return failRead(__func__, tex);
1281     } catch (...) {
1282       AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
1283                              "ReadCallback::getReadBuffer() threw "
1284                              "non-exception type");
1285       return failRead(__func__, ex);
1286     }
1287     if (buf == nullptr || buflen == 0) {
1288       AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
1289                              "ReadCallback::getReadBuffer() returned "
1290                              "empty buffer");
1291       return failRead(__func__, ex);
1292     }
1293
1294     // Perform the read
1295     ssize_t bytesRead = performRead(buf, buflen);
1296     if (bytesRead > 0) {
1297       readCallback_->readDataAvailable(bytesRead);
1298       // Fall through and continue around the loop if the read
1299       // completely filled the available buffer.
1300       // Note that readCallback_ may have been uninstalled or changed inside
1301       // readDataAvailable().
1302       if (size_t(bytesRead) < buflen) {
1303         return;
1304       }
1305     } else if (bytesRead == READ_BLOCKING) {
1306         // No more data to read right now.
1307         return;
1308     } else if (bytesRead == READ_ERROR) {
1309       AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1310                              withAddr("recv() failed"), errno);
1311       return failRead(__func__, ex);
1312     } else {
1313       assert(bytesRead == READ_EOF);
1314       // EOF
1315       shutdownFlags_ |= SHUT_READ;
1316       if (!updateEventRegistration(0, EventHandler::READ)) {
1317         // we've already been moved into STATE_ERROR
1318         assert(state_ == StateEnum::ERROR);
1319         assert(readCallback_ == nullptr);
1320         return;
1321       }
1322
1323       ReadCallback* callback = readCallback_;
1324       readCallback_ = nullptr;
1325       callback->readEOF();
1326       return;
1327     }
1328     if (maxReadsPerEvent_ && (++numReads >= maxReadsPerEvent_)) {
1329       return;
1330     }
1331   }
1332 }
1333
1334 /**
1335  * This function attempts to write as much data as possible, until no more data
1336  * can be written.
1337  *
1338  * - If it sends all available data, it unregisters for write events, and stops
1339  *   the writeTimeout_.
1340  *
1341  * - If not all of the data can be sent immediately, it reschedules
1342  *   writeTimeout_ (if a non-zero timeout is set), and ensures the handler is
1343  *   registered for write events.
1344  */
1345 void AsyncSocket::handleWrite() noexcept {
1346   VLOG(5) << "AsyncSocket::handleWrite() this=" << this << ", fd=" << fd_
1347           << ", state=" << state_;
1348   if (state_ == StateEnum::CONNECTING) {
1349     handleConnect();
1350     return;
1351   }
1352
1353   // Normal write
1354   assert(state_ == StateEnum::ESTABLISHED);
1355   assert((shutdownFlags_ & SHUT_WRITE) == 0);
1356   assert(writeReqHead_ != nullptr);
1357
1358   // Loop until we run out of write requests,
1359   // or until this socket is moved to another EventBase.
1360   // (See the comment in handleRead() explaining how this can happen.)
1361   EventBase* originalEventBase = eventBase_;
1362   while (writeReqHead_ != nullptr && eventBase_ == originalEventBase) {
1363     uint32_t countWritten;
1364     uint32_t partialWritten;
1365     WriteFlags writeFlags = writeReqHead_->flags();
1366     if (writeReqHead_->getNext() != nullptr) {
1367       writeFlags = writeFlags | WriteFlags::CORK;
1368     }
1369     int bytesWritten = performWrite(writeReqHead_->getOps(),
1370                                     writeReqHead_->getOpCount(),
1371                                     writeFlags, &countWritten, &partialWritten);
1372     if (bytesWritten < 0) {
1373       AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1374                              withAddr("writev() failed"), errno);
1375       return failWrite(__func__, ex);
1376     } else if (countWritten == writeReqHead_->getOpCount()) {
1377       // We finished this request
1378       WriteRequest* req = writeReqHead_;
1379       writeReqHead_ = req->getNext();
1380
1381       if (writeReqHead_ == nullptr) {
1382         writeReqTail_ = nullptr;
1383         // This is the last write request.
1384         // Unregister for write events and cancel the send timer
1385         // before we invoke the callback.  We have to update the state properly
1386         // before calling the callback, since it may want to detach us from
1387         // the EventBase.
1388         if (eventFlags_ & EventHandler::WRITE) {
1389           if (!updateEventRegistration(0, EventHandler::WRITE)) {
1390             assert(state_ == StateEnum::ERROR);
1391             return;
1392           }
1393           // Stop the send timeout
1394           writeTimeout_.cancelTimeout();
1395         }
1396         assert(!writeTimeout_.isScheduled());
1397
1398         // If SHUT_WRITE_PENDING is set, we should shutdown the socket after
1399         // we finish sending the last write request.
1400         //
1401         // We have to do this before invoking writeSuccess(), since
1402         // writeSuccess() may detach us from our EventBase.
1403         if (shutdownFlags_ & SHUT_WRITE_PENDING) {
1404           assert(connectCallback_ == nullptr);
1405           shutdownFlags_ |= SHUT_WRITE;
1406
1407           if (shutdownFlags_ & SHUT_READ) {
1408             // Reads have already been shutdown.  Fully close the socket and
1409             // move to STATE_CLOSED.
1410             //
1411             // Note: This code currently moves us to STATE_CLOSED even if
1412             // close() hasn't ever been called.  This can occur if we have
1413             // received EOF from the peer and shutdownWrite() has been called
1414             // locally.  Should we bother staying in STATE_ESTABLISHED in this
1415             // case, until close() is actually called?  I can't think of a
1416             // reason why we would need to do so.  No other operations besides
1417             // calling close() or destroying the socket can be performed at
1418             // this point.
1419             assert(readCallback_ == nullptr);
1420             state_ = StateEnum::CLOSED;
1421             if (fd_ >= 0) {
1422               ioHandler_.changeHandlerFD(-1);
1423               doClose();
1424             }
1425           } else {
1426             // Reads are still enabled, so we are only doing a half-shutdown
1427             ::shutdown(fd_, SHUT_WR);
1428           }
1429         }
1430       }
1431
1432       // Invoke the callback
1433       WriteCallback* callback = req->getCallback();
1434       req->destroy();
1435       if (callback) {
1436         callback->writeSuccess();
1437       }
1438       // We'll continue around the loop, trying to write another request
1439     } else {
1440       // Partial write.
1441       writeReqHead_->consume(countWritten, partialWritten, bytesWritten);
1442       // Stop after a partial write; it's highly likely that a subsequent write
1443       // attempt will just return EAGAIN.
1444       //
1445       // Ensure that we are registered for write events.
1446       if ((eventFlags_ & EventHandler::WRITE) == 0) {
1447         if (!updateEventRegistration(EventHandler::WRITE, 0)) {
1448           assert(state_ == StateEnum::ERROR);
1449           return;
1450         }
1451       }
1452
1453       // Reschedule the send timeout, since we have made some write progress.
1454       if (sendTimeout_ > 0) {
1455         if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
1456           AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1457               withAddr("failed to reschedule write timeout"));
1458           return failWrite(__func__, ex);
1459         }
1460       }
1461       return;
1462     }
1463   }
1464 }
1465
1466 void AsyncSocket::checkForImmediateRead() noexcept {
1467   // We currently don't attempt to perform optimistic reads in AsyncSocket.
1468   // (However, note that some subclasses do override this method.)
1469   //
1470   // Simply calling handleRead() here would be bad, as this would call
1471   // readCallback_->getReadBuffer(), forcing the callback to allocate a read
1472   // buffer even though no data may be available.  This would waste lots of
1473   // memory, since the buffer will sit around unused until the socket actually
1474   // becomes readable.
1475   //
1476   // Checking if the socket is readable now also seems like it would probably
1477   // be a pessimism.  In most cases it probably wouldn't be readable, and we
1478   // would just waste an extra system call.  Even if it is readable, waiting to
1479   // find out from libevent on the next event loop doesn't seem that bad.
1480 }
1481
1482 void AsyncSocket::handleInitialReadWrite() noexcept {
1483   // Our callers should already be holding a DestructorGuard, but grab
1484   // one here just to make sure, in case one of our calling code paths ever
1485   // changes.
1486   DestructorGuard dg(this);
1487
1488   // If we have a readCallback_, make sure we enable read events.  We
1489   // may already be registered for reads if connectSuccess() set
1490   // the read calback.
1491   if (readCallback_ && !(eventFlags_ & EventHandler::READ)) {
1492     assert(state_ == StateEnum::ESTABLISHED);
1493     assert((shutdownFlags_ & SHUT_READ) == 0);
1494     if (!updateEventRegistration(EventHandler::READ, 0)) {
1495       assert(state_ == StateEnum::ERROR);
1496       return;
1497     }
1498     checkForImmediateRead();
1499   } else if (readCallback_ == nullptr) {
1500     // Unregister for read events.
1501     updateEventRegistration(0, EventHandler::READ);
1502   }
1503
1504   // If we have write requests pending, try to send them immediately.
1505   // Since we just finished accepting, there is a very good chance that we can
1506   // write without blocking.
1507   //
1508   // However, we only process them if EventHandler::WRITE is not already set,
1509   // which means that we're already blocked on a write attempt.  (This can
1510   // happen if connectSuccess() called write() before returning.)
1511   if (writeReqHead_ && !(eventFlags_ & EventHandler::WRITE)) {
1512     // Call handleWrite() to perform write processing.
1513     handleWrite();
1514   } else if (writeReqHead_ == nullptr) {
1515     // Unregister for write event.
1516     updateEventRegistration(0, EventHandler::WRITE);
1517   }
1518 }
1519
1520 void AsyncSocket::handleConnect() noexcept {
1521   VLOG(5) << "AsyncSocket::handleConnect() this=" << this << ", fd=" << fd_
1522           << ", state=" << state_;
1523   assert(state_ == StateEnum::CONNECTING);
1524   // SHUT_WRITE can never be set while we are still connecting;
1525   // SHUT_WRITE_PENDING may be set, be we only set SHUT_WRITE once the connect
1526   // finishes
1527   assert((shutdownFlags_ & SHUT_WRITE) == 0);
1528
1529   // In case we had a connect timeout, cancel the timeout
1530   writeTimeout_.cancelTimeout();
1531   // We don't use a persistent registration when waiting on a connect event,
1532   // so we have been automatically unregistered now.  Update eventFlags_ to
1533   // reflect reality.
1534   assert(eventFlags_ == EventHandler::WRITE);
1535   eventFlags_ = EventHandler::NONE;
1536
1537   // Call getsockopt() to check if the connect succeeded
1538   int error;
1539   socklen_t len = sizeof(error);
1540   int rv = getsockopt(fd_, SOL_SOCKET, SO_ERROR, &error, &len);
1541   if (rv != 0) {
1542     AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1543                            withAddr("error calling getsockopt() after connect"),
1544                            errno);
1545     VLOG(4) << "AsyncSocket::handleConnect(this=" << this << ", fd="
1546                << fd_ << " host=" << addr_.describe()
1547                << ") exception:" << ex.what();
1548     return failConnect(__func__, ex);
1549   }
1550
1551   if (error != 0) {
1552     AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
1553                            "connect failed", error);
1554     VLOG(1) << "AsyncSocket::handleConnect(this=" << this << ", fd="
1555             << fd_ << " host=" << addr_.describe()
1556             << ") exception: " << ex.what();
1557     return failConnect(__func__, ex);
1558   }
1559
1560   // Move into STATE_ESTABLISHED
1561   state_ = StateEnum::ESTABLISHED;
1562
1563   // If SHUT_WRITE_PENDING is set and we don't have any write requests to
1564   // perform, immediately shutdown the write half of the socket.
1565   if ((shutdownFlags_ & SHUT_WRITE_PENDING) && writeReqHead_ == nullptr) {
1566     // SHUT_READ shouldn't be set.  If close() is called on the socket while we
1567     // are still connecting we just abort the connect rather than waiting for
1568     // it to complete.
1569     assert((shutdownFlags_ & SHUT_READ) == 0);
1570     ::shutdown(fd_, SHUT_WR);
1571     shutdownFlags_ |= SHUT_WRITE;
1572   }
1573
1574   VLOG(7) << "AsyncSocket " << this << ": fd " << fd_
1575           << "successfully connected; state=" << state_;
1576
1577   // Remember the EventBase we are attached to, before we start invoking any
1578   // callbacks (since the callbacks may call detachEventBase()).
1579   EventBase* originalEventBase = eventBase_;
1580
1581   // Call the connect callback.
1582   if (connectCallback_) {
1583     ConnectCallback* callback = connectCallback_;
1584     connectCallback_ = nullptr;
1585     callback->connectSuccess();
1586   }
1587
1588   // Note that the connect callback may have changed our state.
1589   // (set or unset the read callback, called write(), closed the socket, etc.)
1590   // The following code needs to handle these situations correctly.
1591   //
1592   // If the socket has been closed, readCallback_ and writeReqHead_ will
1593   // always be nullptr, so that will prevent us from trying to read or write.
1594   //
1595   // The main thing to check for is if eventBase_ is still originalEventBase.
1596   // If not, we have been detached from this event base, so we shouldn't
1597   // perform any more operations.
1598   if (eventBase_ != originalEventBase) {
1599     return;
1600   }
1601
1602   handleInitialReadWrite();
1603 }
1604
1605 void AsyncSocket::timeoutExpired() noexcept {
1606   VLOG(7) << "AsyncSocket " << this << ", fd " << fd_ << ": timeout expired: "
1607           << "state=" << state_ << ", events=" << std::hex << eventFlags_;
1608   DestructorGuard dg(this);
1609   assert(eventBase_->isInEventBaseThread());
1610
1611   if (state_ == StateEnum::CONNECTING) {
1612     // connect() timed out
1613     // Unregister for I/O events.
1614     AsyncSocketException ex(AsyncSocketException::TIMED_OUT,
1615                            "connect timed out");
1616     failConnect(__func__, ex);
1617   } else {
1618     // a normal write operation timed out
1619     assert(state_ == StateEnum::ESTABLISHED);
1620     AsyncSocketException ex(AsyncSocketException::TIMED_OUT, "write timed out");
1621     failWrite(__func__, ex);
1622   }
1623 }
1624
1625 ssize_t AsyncSocket::performWrite(const iovec* vec,
1626                                    uint32_t count,
1627                                    WriteFlags flags,
1628                                    uint32_t* countWritten,
1629                                    uint32_t* partialWritten) {
1630   // We use sendmsg() instead of writev() so that we can pass in MSG_NOSIGNAL
1631   // We correctly handle EPIPE errors, so we never want to receive SIGPIPE
1632   // (since it may terminate the program if the main program doesn't explicitly
1633   // ignore it).
1634   struct msghdr msg;
1635   msg.msg_name = nullptr;
1636   msg.msg_namelen = 0;
1637   msg.msg_iov = const_cast<iovec *>(vec);
1638 #ifdef IOV_MAX // not defined on Android
1639   msg.msg_iovlen = std::min(count, (uint32_t)IOV_MAX);
1640 #else
1641   msg.msg_iovlen = std::min(count, (uint32_t)UIO_MAXIOV);
1642 #endif
1643   msg.msg_control = nullptr;
1644   msg.msg_controllen = 0;
1645   msg.msg_flags = 0;
1646
1647   int msg_flags = MSG_DONTWAIT;
1648
1649 #ifdef MSG_NOSIGNAL // Linux-only
1650   msg_flags |= MSG_NOSIGNAL;
1651   if (isSet(flags, WriteFlags::CORK)) {
1652     // MSG_MORE tells the kernel we have more data to send, so wait for us to
1653     // give it the rest of the data rather than immediately sending a partial
1654     // frame, even when TCP_NODELAY is enabled.
1655     msg_flags |= MSG_MORE;
1656   }
1657 #endif
1658   if (isSet(flags, WriteFlags::EOR)) {
1659     // marks that this is the last byte of a record (response)
1660     msg_flags |= MSG_EOR;
1661   }
1662   ssize_t totalWritten = ::sendmsg(fd_, &msg, msg_flags);
1663   if (totalWritten < 0) {
1664     if (errno == EAGAIN) {
1665       // TCP buffer is full; we can't write any more data right now.
1666       *countWritten = 0;
1667       *partialWritten = 0;
1668       return 0;
1669     }
1670     // error
1671     *countWritten = 0;
1672     *partialWritten = 0;
1673     return -1;
1674   }
1675
1676   appBytesWritten_ += totalWritten;
1677
1678   uint32_t bytesWritten;
1679   uint32_t n;
1680   for (bytesWritten = totalWritten, n = 0; n < count; ++n) {
1681     const iovec* v = vec + n;
1682     if (v->iov_len > bytesWritten) {
1683       // Partial write finished in the middle of this iovec
1684       *countWritten = n;
1685       *partialWritten = bytesWritten;
1686       return totalWritten;
1687     }
1688
1689     bytesWritten -= v->iov_len;
1690   }
1691
1692   assert(bytesWritten == 0);
1693   *countWritten = n;
1694   *partialWritten = 0;
1695   return totalWritten;
1696 }
1697
1698 /**
1699  * Re-register the EventHandler after eventFlags_ has changed.
1700  *
1701  * If an error occurs, fail() is called to move the socket into the error state
1702  * and call all currently installed callbacks.  After an error, the
1703  * AsyncSocket is completely unregistered.
1704  *
1705  * @return Returns true on succcess, or false on error.
1706  */
1707 bool AsyncSocket::updateEventRegistration() {
1708   VLOG(5) << "AsyncSocket::updateEventRegistration(this=" << this
1709           << ", fd=" << fd_ << ", evb=" << eventBase_ << ", state=" << state_
1710           << ", events=" << std::hex << eventFlags_;
1711   assert(eventBase_->isInEventBaseThread());
1712   if (eventFlags_ == EventHandler::NONE) {
1713     ioHandler_.unregisterHandler();
1714     return true;
1715   }
1716
1717   // Always register for persistent events, so we don't have to re-register
1718   // after being called back.
1719   if (!ioHandler_.registerHandler(eventFlags_ | EventHandler::PERSIST)) {
1720     eventFlags_ = EventHandler::NONE; // we're not registered after error
1721     AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1722         withAddr("failed to update AsyncSocket event registration"));
1723     fail("updateEventRegistration", ex);
1724     return false;
1725   }
1726
1727   return true;
1728 }
1729
1730 bool AsyncSocket::updateEventRegistration(uint16_t enable,
1731                                            uint16_t disable) {
1732   uint16_t oldFlags = eventFlags_;
1733   eventFlags_ |= enable;
1734   eventFlags_ &= ~disable;
1735   if (eventFlags_ == oldFlags) {
1736     return true;
1737   } else {
1738     return updateEventRegistration();
1739   }
1740 }
1741
1742 void AsyncSocket::startFail() {
1743   // startFail() should only be called once
1744   assert(state_ != StateEnum::ERROR);
1745   assert(getDestructorGuardCount() > 0);
1746   state_ = StateEnum::ERROR;
1747   // Ensure that SHUT_READ and SHUT_WRITE are set,
1748   // so all future attempts to read or write will be rejected
1749   shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
1750
1751   if (eventFlags_ != EventHandler::NONE) {
1752     eventFlags_ = EventHandler::NONE;
1753     ioHandler_.unregisterHandler();
1754   }
1755   writeTimeout_.cancelTimeout();
1756
1757   if (fd_ >= 0) {
1758     ioHandler_.changeHandlerFD(-1);
1759     doClose();
1760   }
1761 }
1762
1763 void AsyncSocket::finishFail() {
1764   assert(state_ == StateEnum::ERROR);
1765   assert(getDestructorGuardCount() > 0);
1766
1767   AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1768                          withAddr("socket closing after error"));
1769   if (connectCallback_) {
1770     ConnectCallback* callback = connectCallback_;
1771     connectCallback_ = nullptr;
1772     callback->connectErr(ex);
1773   }
1774
1775   failAllWrites(ex);
1776
1777   if (readCallback_) {
1778     ReadCallback* callback = readCallback_;
1779     readCallback_ = nullptr;
1780     callback->readErr(ex);
1781   }
1782 }
1783
1784 void AsyncSocket::fail(const char* fn, const AsyncSocketException& ex) {
1785   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
1786              << state_ << " host=" << addr_.describe()
1787              << "): failed in " << fn << "(): "
1788              << ex.what();
1789   startFail();
1790   finishFail();
1791 }
1792
1793 void AsyncSocket::failConnect(const char* fn, const AsyncSocketException& ex) {
1794   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
1795                << state_ << " host=" << addr_.describe()
1796                << "): failed while connecting in " << fn << "(): "
1797                << ex.what();
1798   startFail();
1799
1800   if (connectCallback_ != nullptr) {
1801     ConnectCallback* callback = connectCallback_;
1802     connectCallback_ = nullptr;
1803     callback->connectErr(ex);
1804   }
1805
1806   finishFail();
1807 }
1808
1809 void AsyncSocket::failRead(const char* fn, const AsyncSocketException& ex) {
1810   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
1811                << state_ << " host=" << addr_.describe()
1812                << "): failed while reading in " << fn << "(): "
1813                << ex.what();
1814   startFail();
1815
1816   if (readCallback_ != nullptr) {
1817     ReadCallback* callback = readCallback_;
1818     readCallback_ = nullptr;
1819     callback->readErr(ex);
1820   }
1821
1822   finishFail();
1823 }
1824
1825 void AsyncSocket::failWrite(const char* fn, const AsyncSocketException& ex) {
1826   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
1827                << state_ << " host=" << addr_.describe()
1828                << "): failed while writing in " << fn << "(): "
1829                << ex.what();
1830   startFail();
1831
1832   // Only invoke the first write callback, since the error occurred while
1833   // writing this request.  Let any other pending write callbacks be invoked in
1834   // finishFail().
1835   if (writeReqHead_ != nullptr) {
1836     WriteRequest* req = writeReqHead_;
1837     writeReqHead_ = req->getNext();
1838     WriteCallback* callback = req->getCallback();
1839     uint32_t bytesWritten = req->getBytesWritten();
1840     req->destroy();
1841     if (callback) {
1842       callback->writeErr(bytesWritten, ex);
1843     }
1844   }
1845
1846   finishFail();
1847 }
1848
1849 void AsyncSocket::failWrite(const char* fn, WriteCallback* callback,
1850                              size_t bytesWritten,
1851                              const AsyncSocketException& ex) {
1852   // This version of failWrite() is used when the failure occurs before
1853   // we've added the callback to writeReqHead_.
1854   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
1855              << state_ << " host=" << addr_.describe()
1856              <<"): failed while writing in " << fn << "(): "
1857              << ex.what();
1858   startFail();
1859
1860   if (callback != nullptr) {
1861     callback->writeErr(bytesWritten, ex);
1862   }
1863
1864   finishFail();
1865 }
1866
1867 void AsyncSocket::failAllWrites(const AsyncSocketException& ex) {
1868   // Invoke writeError() on all write callbacks.
1869   // This is used when writes are forcibly shutdown with write requests
1870   // pending, or when an error occurs with writes pending.
1871   while (writeReqHead_ != nullptr) {
1872     WriteRequest* req = writeReqHead_;
1873     writeReqHead_ = req->getNext();
1874     WriteCallback* callback = req->getCallback();
1875     if (callback) {
1876       callback->writeErr(req->getBytesWritten(), ex);
1877     }
1878     req->destroy();
1879   }
1880 }
1881
1882 void AsyncSocket::invalidState(ConnectCallback* callback) {
1883   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_
1884              << "): connect() called in invalid state " << state_;
1885
1886   /*
1887    * The invalidState() methods don't use the normal failure mechanisms,
1888    * since we don't know what state we are in.  We don't want to call
1889    * startFail()/finishFail() recursively if we are already in the middle of
1890    * cleaning up.
1891    */
1892
1893   AsyncSocketException ex(AsyncSocketException::ALREADY_OPEN,
1894                          "connect() called with socket in invalid state");
1895   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
1896     if (callback) {
1897       callback->connectErr(ex);
1898     }
1899   } else {
1900     // We can't use failConnect() here since connectCallback_
1901     // may already be set to another callback.  Invoke this ConnectCallback
1902     // here; any other connectCallback_ will be invoked in finishFail()
1903     startFail();
1904     if (callback) {
1905       callback->connectErr(ex);
1906     }
1907     finishFail();
1908   }
1909 }
1910
1911 void AsyncSocket::invalidState(ReadCallback* callback) {
1912   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
1913              << "): setReadCallback(" << callback
1914              << ") called in invalid state " << state_;
1915
1916   AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
1917                          "setReadCallback() called with socket in "
1918                          "invalid state");
1919   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
1920     if (callback) {
1921       callback->readErr(ex);
1922     }
1923   } else {
1924     startFail();
1925     if (callback) {
1926       callback->readErr(ex);
1927     }
1928     finishFail();
1929   }
1930 }
1931
1932 void AsyncSocket::invalidState(WriteCallback* callback) {
1933   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
1934              << "): write() called in invalid state " << state_;
1935
1936   AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
1937                          withAddr("write() called with socket in invalid state"));
1938   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
1939     if (callback) {
1940       callback->writeErr(0, ex);
1941     }
1942   } else {
1943     startFail();
1944     if (callback) {
1945       callback->writeErr(0, ex);
1946     }
1947     finishFail();
1948   }
1949 }
1950
1951 void AsyncSocket::doClose() {
1952   if (fd_ == -1) return;
1953   if (shutdownSocketSet_) {
1954     shutdownSocketSet_->close(fd_);
1955   } else {
1956     ::close(fd_);
1957   }
1958   fd_ = -1;
1959 }
1960
1961 std::ostream& operator << (std::ostream& os,
1962                            const AsyncSocket::StateEnum& state) {
1963   os << static_cast<int>(state);
1964   return os;
1965 }
1966
1967 std::string AsyncSocket::withAddr(const std::string& s) {
1968   // Don't use addr_ directly because it may not be initialized
1969   // e.g. if constructed from fd
1970   folly::SocketAddress peer, local;
1971   try {
1972     getPeerAddress(&peer);
1973     getLocalAddress(&local);
1974   } catch (const std::exception&) {
1975     // ignore
1976   } catch (...) {
1977     // ignore
1978   }
1979   return s + " (peer=" + peer.describe() + ", local=" + local.describe() + ")";
1980 }
1981
1982 } // folly