folly/io/async/AsyncSocket.cpp

   1 /*
   2  * Copyright 2017 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/ExceptionWrapper.h>
  20 #include <folly/Portability.h>
  21 #include <folly/SocketAddress.h>
  22 #include <folly/io/Cursor.h>
  23 #include <folly/io/IOBuf.h>
  24 #include <folly/io/IOBufQueue.h>
  25 #include <folly/portability/Fcntl.h>
  26 #include <folly/portability/Sockets.h>
  27 #include <folly/portability/SysUio.h>
  28 #include <folly/portability/Unistd.h>
  29
  30 #include <boost/preprocessor/control/if.hpp>
  31 #include <errno.h>
  32 #include <limits.h>
  33 #include <sys/types.h>
  34 #include <thread>
  35
  36 using std::string;
  37 using std::unique_ptr;
  38
  39 namespace fsp = folly::portability::sockets;
  40
  41 namespace folly {
  42
  43 static constexpr bool msgErrQueueSupported =
  44 #ifdef MSG_ERRQUEUE
  45     true;
  46 #else
  47     false;
  48 #endif // MSG_ERRQUEUE
  49
  50 // static members initializers
  51 const AsyncSocket::OptionMap AsyncSocket::emptyOptionMap;
  52
  53 const AsyncSocketException socketClosedLocallyEx(
  54     AsyncSocketException::END_OF_FILE, "socket closed locally");
  55 const AsyncSocketException socketShutdownForWritesEx(
  56     AsyncSocketException::END_OF_FILE, "socket shutdown for writes");
  57
  58 // TODO: It might help performance to provide a version of BytesWriteRequest that
  59 // users could derive from, so we can avoid the extra allocation for each call
  60 // to write()/writev().  We could templatize TFramedAsyncChannel just like the
  61 // protocols are currently templatized for transports.
  62 //
  63 // We would need the version for external users where they provide the iovec
  64 // storage space, and only our internal version would allocate it at the end of
  65 // the WriteRequest.
  66
  67 /* The default WriteRequest implementation, used for write(), writev() and
  68  * writeChain()
  69  *
  70  * A new BytesWriteRequest operation is allocated on the heap for all write
  71  * operations that cannot be completed immediately.
  72  */
  73 class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest {
  74  public:
  75   static BytesWriteRequest* newRequest(AsyncSocket* socket,
  76                                        WriteCallback* callback,
  77                                        const iovec* ops,
  78                                        uint32_t opCount,
  79                                        uint32_t partialWritten,
  80                                        uint32_t bytesWritten,
  81                                        unique_ptr<IOBuf>&& ioBuf,
  82                                        WriteFlags flags) {
  83     assert(opCount > 0);
  84     // Since we put a variable size iovec array at the end
  85     // of each BytesWriteRequest, we have to manually allocate the memory.
  86     void* buf = malloc(sizeof(BytesWriteRequest) +
  87                        (opCount * sizeof(struct iovec)));
  88     if (buf == nullptr) {
  89       throw std::bad_alloc();
  90     }
  91
  92     return new(buf) BytesWriteRequest(socket, callback, ops, opCount,
  93                                       partialWritten, bytesWritten,
  94                                       std::move(ioBuf), flags);
  95   }
  96
  97   void destroy() override {
  98     this->~BytesWriteRequest();
  99     free(this);
 100   }
 101
 102   WriteResult performWrite() override {
 103     WriteFlags writeFlags = flags_;
 104     if (getNext() != nullptr) {
 105       writeFlags |= WriteFlags::CORK;
 106     }
 107     auto writeResult = socket_->performWrite(
 108         getOps(), getOpCount(), writeFlags, &opsWritten_, &partialBytes_);
 109     bytesWritten_ = writeResult.writeReturn > 0 ? writeResult.writeReturn : 0;
 110     return writeResult;
 111   }
 112
 113   bool isComplete() override {
 114     return opsWritten_ == getOpCount();
 115   }
 116
 117   void consume() override {
 118     // Advance opIndex_ forward by opsWritten_
 119     opIndex_ += opsWritten_;
 120     assert(opIndex_ < opCount_);
 121
 122     // If we've finished writing any IOBufs, release them
 123     if (ioBuf_) {
 124       for (uint32_t i = opsWritten_; i != 0; --i) {
 125         assert(ioBuf_);
 126         ioBuf_ = ioBuf_->pop();
 127       }
 128     }
 129
 130     // Move partialBytes_ forward into the current iovec buffer
 131     struct iovec* currentOp = writeOps_ + opIndex_;
 132     assert((partialBytes_ < currentOp->iov_len) || (currentOp->iov_len == 0));
 133     currentOp->iov_base =
 134       reinterpret_cast<uint8_t*>(currentOp->iov_base) + partialBytes_;
 135     currentOp->iov_len -= partialBytes_;
 136
 137     // Increment the totalBytesWritten_ count by bytesWritten_;
 138     assert(bytesWritten_ >= 0);
 139     totalBytesWritten_ += uint32_t(bytesWritten_);
 140   }
 141
 142  private:
 143   BytesWriteRequest(AsyncSocket* socket,
 144                     WriteCallback* callback,
 145                     const struct iovec* ops,
 146                     uint32_t opCount,
 147                     uint32_t partialBytes,
 148                     uint32_t bytesWritten,
 149                     unique_ptr<IOBuf>&& ioBuf,
 150                     WriteFlags flags)
 151     : AsyncSocket::WriteRequest(socket, callback)
 152     , opCount_(opCount)
 153     , opIndex_(0)
 154     , flags_(flags)
 155     , ioBuf_(std::move(ioBuf))
 156     , opsWritten_(0)
 157     , partialBytes_(partialBytes)
 158     , bytesWritten_(bytesWritten) {
 159     memcpy(writeOps_, ops, sizeof(*ops) * opCount_);
 160   }
 161
 162   // private destructor, to ensure callers use destroy()
 163   ~BytesWriteRequest() override = default;
 164
 165   const struct iovec* getOps() const {
 166     assert(opCount_ > opIndex_);
 167     return writeOps_ + opIndex_;
 168   }
 169
 170   uint32_t getOpCount() const {
 171     assert(opCount_ > opIndex_);
 172     return opCount_ - opIndex_;
 173   }
 174
 175   uint32_t opCount_;            ///< number of entries in writeOps_
 176   uint32_t opIndex_;            ///< current index into writeOps_
 177   WriteFlags flags_;            ///< set for WriteFlags
 178   unique_ptr<IOBuf> ioBuf_;     ///< underlying IOBuf, or nullptr if N/A
 179
 180   // for consume(), how much we wrote on the last write
 181   uint32_t opsWritten_;         ///< complete ops written
 182   uint32_t partialBytes_;       ///< partial bytes of incomplete op written
 183   ssize_t bytesWritten_;        ///< bytes written altogether
 184
 185   struct iovec writeOps_[];     ///< write operation(s) list
 186 };
 187
 188 int AsyncSocket::SendMsgParamsCallback::getDefaultFlags(folly::WriteFlags flags)
 189                                                                       noexcept {
 190   int msg_flags = MSG_DONTWAIT;
 191
 192 #ifdef MSG_NOSIGNAL // Linux-only
 193   msg_flags |= MSG_NOSIGNAL;
 194 #ifdef MSG_MORE
 195   if (isSet(flags, WriteFlags::CORK)) {
 196     // MSG_MORE tells the kernel we have more data to send, so wait for us to
 197     // give it the rest of the data rather than immediately sending a partial
 198     // frame, even when TCP_NODELAY is enabled.
 199     msg_flags |= MSG_MORE;
 200   }
 201 #endif // MSG_MORE
 202 #endif // MSG_NOSIGNAL
 203   if (isSet(flags, WriteFlags::EOR)) {
 204     // marks that this is the last byte of a record (response)
 205     msg_flags |= MSG_EOR;
 206   }
 207
 208   return msg_flags;
 209 }
 210
 211 namespace {
 212 static AsyncSocket::SendMsgParamsCallback defaultSendMsgParamsCallback;
 213 }
 214
 215 AsyncSocket::AsyncSocket()
 216     : eventBase_(nullptr),
 217       writeTimeout_(this, nullptr),
 218       ioHandler_(this, nullptr),
 219       immediateReadHandler_(this) {
 220   VLOG(5) << "new AsyncSocket()";
 221   init();
 222 }
 223
 224 AsyncSocket::AsyncSocket(EventBase* evb)
 225     : eventBase_(evb),
 226       writeTimeout_(this, evb),
 227       ioHandler_(this, evb),
 228       immediateReadHandler_(this) {
 229   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
 230   init();
 231 }
 232
 233 AsyncSocket::AsyncSocket(EventBase* evb,
 234                            const folly::SocketAddress& address,
 235                            uint32_t connectTimeout)
 236   : AsyncSocket(evb) {
 237   connect(nullptr, address, connectTimeout);
 238 }
 239
 240 AsyncSocket::AsyncSocket(EventBase* evb,
 241                            const std::string& ip,
 242                            uint16_t port,
 243                            uint32_t connectTimeout)
 244   : AsyncSocket(evb) {
 245   connect(nullptr, ip, port, connectTimeout);
 246 }
 247
 248 AsyncSocket::AsyncSocket(EventBase* evb, int fd)
 249     : eventBase_(evb),
 250       writeTimeout_(this, evb),
 251       ioHandler_(this, evb, fd),
 252       immediateReadHandler_(this) {
 253   VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ", fd="
 254           << fd << ")";
 255   init();
 256   fd_ = fd;
 257   setCloseOnExec();
 258   state_ = StateEnum::ESTABLISHED;
 259 }
 260
 261 AsyncSocket::AsyncSocket(AsyncSocket::UniquePtr oldAsyncSocket)
 262     : AsyncSocket(oldAsyncSocket->getEventBase(), oldAsyncSocket->detachFd()) {
 263   preReceivedData_ = std::move(oldAsyncSocket->preReceivedData_);
 264 }
 265
 266 // init() method, since constructor forwarding isn't supported in most
 267 // compilers yet.
 268 void AsyncSocket::init() {
 269   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 270   shutdownFlags_ = 0;
 271   state_ = StateEnum::UNINIT;
 272   eventFlags_ = EventHandler::NONE;
 273   fd_ = -1;
 274   sendTimeout_ = 0;
 275   maxReadsPerEvent_ = 16;
 276   connectCallback_ = nullptr;
 277   errMessageCallback_ = nullptr;
 278   readCallback_ = nullptr;
 279   writeReqHead_ = nullptr;
 280   writeReqTail_ = nullptr;
 281   shutdownSocketSet_ = nullptr;
 282   appBytesWritten_ = 0;
 283   appBytesReceived_ = 0;
 284   sendMsgParamCallback_ = &defaultSendMsgParamsCallback;
 285 }
 286
 287 AsyncSocket::~AsyncSocket() {
 288   VLOG(7) << "actual destruction of AsyncSocket(this=" << this
 289           << ", evb=" << eventBase_ << ", fd=" << fd_
 290           << ", state=" << state_ << ")";
 291 }
 292
 293 void AsyncSocket::destroy() {
 294   VLOG(5) << "AsyncSocket::destroy(this=" << this << ", evb=" << eventBase_
 295           << ", fd=" << fd_ << ", state=" << state_;
 296   // When destroy is called, close the socket immediately
 297   closeNow();
 298
 299   // Then call DelayedDestruction::destroy() to take care of
 300   // whether or not we need immediate or delayed destruction
 301   DelayedDestruction::destroy();
 302 }
 303
 304 int AsyncSocket::detachFd() {
 305   VLOG(6) << "AsyncSocket::detachFd(this=" << this << ", fd=" << fd_
 306           << ", evb=" << eventBase_ << ", state=" << state_
 307           << ", events=" << std::hex << eventFlags_ << ")";
 308   // Extract the fd, and set fd_ to -1 first, so closeNow() won't
 309   // actually close the descriptor.
 310   if (shutdownSocketSet_) {
 311     shutdownSocketSet_->remove(fd_);
 312   }
 313   int fd = fd_;
 314   fd_ = -1;
 315   // Call closeNow() to invoke all pending callbacks with an error.
 316   closeNow();
 317   // Update the EventHandler to stop using this fd.
 318   // This can only be done after closeNow() unregisters the handler.
 319   ioHandler_.changeHandlerFD(-1);
 320   return fd;
 321 }
 322
 323 const folly::SocketAddress& AsyncSocket::anyAddress() {
 324   static const folly::SocketAddress anyAddress =
 325     folly::SocketAddress("0.0.0.0", 0);
 326   return anyAddress;
 327 }
 328
 329 void AsyncSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
 330   if (shutdownSocketSet_ == newSS) {
 331     return;
 332   }
 333   if (shutdownSocketSet_ && fd_ != -1) {
 334     shutdownSocketSet_->remove(fd_);
 335   }
 336   shutdownSocketSet_ = newSS;
 337   if (shutdownSocketSet_ && fd_ != -1) {
 338     shutdownSocketSet_->add(fd_);
 339   }
 340 }
 341
 342 void AsyncSocket::setCloseOnExec() {
 343   int rv = fcntl(fd_, F_SETFD, FD_CLOEXEC);
 344   if (rv != 0) {
 345     auto errnoCopy = errno;
 346     throw AsyncSocketException(
 347         AsyncSocketException::INTERNAL_ERROR,
 348         withAddr("failed to set close-on-exec flag"),
 349         errnoCopy);
 350   }
 351 }
 352
 353 void AsyncSocket::connect(ConnectCallback* callback,
 354                            const folly::SocketAddress& address,
 355                            int timeout,
 356                            const OptionMap &options,
 357                            const folly::SocketAddress& bindAddr) noexcept {
 358   DestructorGuard dg(this);
 359   assert(eventBase_->isInEventBaseThread());
 360
 361   addr_ = address;
 362
 363   // Make sure we're in the uninitialized state
 364   if (state_ != StateEnum::UNINIT) {
 365     return invalidState(callback);
 366   }
 367
 368   connectTimeout_ = std::chrono::milliseconds(timeout);
 369   connectStartTime_ = std::chrono::steady_clock::now();
 370   // Make connect end time at least >= connectStartTime.
 371   connectEndTime_ = connectStartTime_;
 372
 373   assert(fd_ == -1);
 374   state_ = StateEnum::CONNECTING;
 375   connectCallback_ = callback;
 376
 377   sockaddr_storage addrStorage;
 378   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 379
 380   try {
 381     // Create the socket
 382     // Technically the first parameter should actually be a protocol family
 383     // constant (PF_xxx) rather than an address family (AF_xxx), but the
 384     // distinction is mainly just historical.  In pretty much all
 385     // implementations the PF_foo and AF_foo constants are identical.
 386     fd_ = fsp::socket(address.getFamily(), SOCK_STREAM, 0);
 387     if (fd_ < 0) {
 388       auto errnoCopy = errno;
 389       throw AsyncSocketException(
 390           AsyncSocketException::INTERNAL_ERROR,
 391           withAddr("failed to create socket"),
 392           errnoCopy);
 393     }
 394     if (shutdownSocketSet_) {
 395       shutdownSocketSet_->add(fd_);
 396     }
 397     ioHandler_.changeHandlerFD(fd_);
 398
 399     setCloseOnExec();
 400
 401     // Put the socket in non-blocking mode
 402     int flags = fcntl(fd_, F_GETFL, 0);
 403     if (flags == -1) {
 404       auto errnoCopy = errno;
 405       throw AsyncSocketException(
 406           AsyncSocketException::INTERNAL_ERROR,
 407           withAddr("failed to get socket flags"),
 408           errnoCopy);
 409     }
 410     int rv = fcntl(fd_, F_SETFL, flags | O_NONBLOCK);
 411     if (rv == -1) {
 412       auto errnoCopy = errno;
 413       throw AsyncSocketException(
 414           AsyncSocketException::INTERNAL_ERROR,
 415           withAddr("failed to put socket in non-blocking mode"),
 416           errnoCopy);
 417     }
 418
 419 #if !defined(MSG_NOSIGNAL) && defined(F_SETNOSIGPIPE)
 420     // iOS and OS X don't support MSG_NOSIGNAL; set F_SETNOSIGPIPE instead
 421     rv = fcntl(fd_, F_SETNOSIGPIPE, 1);
 422     if (rv == -1) {
 423       auto errnoCopy = errno;
 424       throw AsyncSocketException(
 425           AsyncSocketException::INTERNAL_ERROR,
 426           "failed to enable F_SETNOSIGPIPE on socket",
 427           errnoCopy);
 428     }
 429 #endif
 430
 431     // By default, turn on TCP_NODELAY
 432     // If setNoDelay() fails, we continue anyway; this isn't a fatal error.
 433     // setNoDelay() will log an error message if it fails.
 434     if (address.getFamily() != AF_UNIX) {
 435       (void)setNoDelay(true);
 436     }
 437
 438     VLOG(5) << "AsyncSocket::connect(this=" << this << ", evb=" << eventBase_
 439             << ", fd=" << fd_ << ", host=" << address.describe().c_str();
 440
 441     // bind the socket
 442     if (bindAddr != anyAddress()) {
 443       int one = 1;
 444       if (setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one))) {
 445         auto errnoCopy = errno;
 446         doClose();
 447         throw AsyncSocketException(
 448             AsyncSocketException::NOT_OPEN,
 449             "failed to setsockopt prior to bind on " + bindAddr.describe(),
 450             errnoCopy);
 451       }
 452
 453       bindAddr.getAddress(&addrStorage);
 454
 455       if (bind(fd_, saddr, bindAddr.getActualSize()) != 0) {
 456         auto errnoCopy = errno;
 457         doClose();
 458         throw AsyncSocketException(
 459             AsyncSocketException::NOT_OPEN,
 460             "failed to bind to async socket: " + bindAddr.describe(),
 461             errnoCopy);
 462       }
 463     }
 464
 465     // Apply the additional options if any.
 466     for (const auto& opt: options) {
 467       rv = opt.first.apply(fd_, opt.second);
 468       if (rv != 0) {
 469         auto errnoCopy = errno;
 470         throw AsyncSocketException(
 471             AsyncSocketException::INTERNAL_ERROR,
 472             withAddr("failed to set socket option"),
 473             errnoCopy);
 474       }
 475     }
 476
 477     // Perform the connect()
 478     address.getAddress(&addrStorage);
 479
 480     if (tfoEnabled_) {
 481       state_ = StateEnum::FAST_OPEN;
 482       tfoAttempted_ = true;
 483     } else {
 484       if (socketConnect(saddr, addr_.getActualSize()) < 0) {
 485         return;
 486       }
 487     }
 488
 489     // If we're still here the connect() succeeded immediately.
 490     // Fall through to call the callback outside of this try...catch block
 491   } catch (const AsyncSocketException& ex) {
 492     return failConnect(__func__, ex);
 493   } catch (const std::exception& ex) {
 494     // shouldn't happen, but handle it just in case
 495     VLOG(4) << "AsyncSocket::connect(this=" << this << ", fd=" << fd_
 496                << "): unexpected " << typeid(ex).name() << " exception: "
 497                << ex.what();
 498     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 499                             withAddr(string("unexpected exception: ") +
 500                                      ex.what()));
 501     return failConnect(__func__, tex);
 502   }
 503
 504   // The connection succeeded immediately
 505   // The read callback may not have been set yet, and no writes may be pending
 506   // yet, so we don't have to register for any events at the moment.
 507   VLOG(8) << "AsyncSocket::connect succeeded immediately; this=" << this;
 508   assert(errMessageCallback_ == nullptr);
 509   assert(readCallback_ == nullptr);
 510   assert(writeReqHead_ == nullptr);
 511   if (state_ != StateEnum::FAST_OPEN) {
 512     state_ = StateEnum::ESTABLISHED;
 513   }
 514   invokeConnectSuccess();
 515 }
 516
 517 int AsyncSocket::socketConnect(const struct sockaddr* saddr, socklen_t len) {
 518 #if __linux__
 519   if (noTransparentTls_) {
 520     // Ignore return value, errors are ok
 521     setsockopt(fd_, SOL_SOCKET, SO_NO_TRANSPARENT_TLS, nullptr, 0);
 522   }
 523 #endif
 524   int rv = fsp::connect(fd_, saddr, len);
 525   if (rv < 0) {
 526     auto errnoCopy = errno;
 527     if (errnoCopy == EINPROGRESS) {
 528       scheduleConnectTimeout();
 529       registerForConnectEvents();
 530     } else {
 531       throw AsyncSocketException(
 532           AsyncSocketException::NOT_OPEN,
 533           "connect failed (immediately)",
 534           errnoCopy);
 535     }
 536   }
 537   return rv;
 538 }
 539
 540 void AsyncSocket::scheduleConnectTimeout() {
 541   // Connection in progress.
 542   auto timeout = connectTimeout_.count();
 543   if (timeout > 0) {
 544     // Start a timer in case the connection takes too long.
 545     if (!writeTimeout_.scheduleTimeout(uint32_t(timeout))) {
 546       throw AsyncSocketException(
 547           AsyncSocketException::INTERNAL_ERROR,
 548           withAddr("failed to schedule AsyncSocket connect timeout"));
 549     }
 550   }
 551 }
 552
 553 void AsyncSocket::registerForConnectEvents() {
 554   // Register for write events, so we'll
 555   // be notified when the connection finishes/fails.
 556   // Note that we don't register for a persistent event here.
 557   assert(eventFlags_ == EventHandler::NONE);
 558   eventFlags_ = EventHandler::WRITE;
 559   if (!ioHandler_.registerHandler(eventFlags_)) {
 560     throw AsyncSocketException(
 561         AsyncSocketException::INTERNAL_ERROR,
 562         withAddr("failed to register AsyncSocket connect handler"));
 563   }
 564 }
 565
 566 void AsyncSocket::connect(ConnectCallback* callback,
 567                            const string& ip, uint16_t port,
 568                            int timeout,
 569                            const OptionMap &options) noexcept {
 570   DestructorGuard dg(this);
 571   try {
 572     connectCallback_ = callback;
 573     connect(callback, folly::SocketAddress(ip, port), timeout, options);
 574   } catch (const std::exception& ex) {
 575     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 576                             ex.what());
 577     return failConnect(__func__, tex);
 578   }
 579 }
 580
 581 void AsyncSocket::cancelConnect() {
 582   connectCallback_ = nullptr;
 583   if (state_ == StateEnum::CONNECTING || state_ == StateEnum::FAST_OPEN) {
 584     closeNow();
 585   }
 586 }
 587
 588 void AsyncSocket::setSendTimeout(uint32_t milliseconds) {
 589   sendTimeout_ = milliseconds;
 590   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 591
 592   // If we are currently pending on write requests, immediately update
 593   // writeTimeout_ with the new value.
 594   if ((eventFlags_ & EventHandler::WRITE) &&
 595       (state_ != StateEnum::CONNECTING && state_ != StateEnum::FAST_OPEN)) {
 596     assert(state_ == StateEnum::ESTABLISHED);
 597     assert((shutdownFlags_ & SHUT_WRITE) == 0);
 598     if (sendTimeout_ > 0) {
 599       if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
 600         AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
 601             withAddr("failed to reschedule send timeout in setSendTimeout"));
 602         return failWrite(__func__, ex);
 603       }
 604     } else {
 605       writeTimeout_.cancelTimeout();
 606     }
 607   }
 608 }
 609
 610 void AsyncSocket::setErrMessageCB(ErrMessageCallback* callback) {
 611   VLOG(6) << "AsyncSocket::setErrMessageCB() this=" << this
 612           << ", fd=" << fd_ << ", callback=" << callback
 613           << ", state=" << state_;
 614
 615   // Short circuit if callback is the same as the existing timestampCallback_.
 616   if (callback == errMessageCallback_) {
 617     return;
 618   }
 619
 620   if (!msgErrQueueSupported) {
 621       // Per-socket error message queue is not supported on this platform.
 622       return invalidState(callback);
 623   }
 624
 625   DestructorGuard dg(this);
 626   assert(eventBase_->isInEventBaseThread());
 627
 628   switch ((StateEnum)state_) {
 629     case StateEnum::CONNECTING:
 630     case StateEnum::FAST_OPEN:
 631     case StateEnum::ESTABLISHED: {
 632       errMessageCallback_ = callback;
 633       return;
 634     }
 635     case StateEnum::CLOSED:
 636     case StateEnum::ERROR:
 637       // We should never reach here.  SHUT_READ should always be set
 638       // if we are in STATE_CLOSED or STATE_ERROR.
 639       assert(false);
 640       return invalidState(callback);
 641     case StateEnum::UNINIT:
 642       // We do not allow setReadCallback() to be called before we start
 643       // connecting.
 644       return invalidState(callback);
 645   }
 646
 647   // We don't put a default case in the switch statement, so that the compiler
 648   // will warn us to update the switch statement if a new state is added.
 649   return invalidState(callback);
 650 }
 651
 652 AsyncSocket::ErrMessageCallback* AsyncSocket::getErrMessageCallback() const {
 653   return errMessageCallback_;
 654 }
 655
 656 void AsyncSocket::setSendMsgParamCB(SendMsgParamsCallback* callback) {
 657   sendMsgParamCallback_ = callback;
 658 }
 659
 660 AsyncSocket::SendMsgParamsCallback* AsyncSocket::getSendMsgParamsCB() const {
 661   return sendMsgParamCallback_;
 662 }
 663
 664 void AsyncSocket::setReadCB(ReadCallback *callback) {
 665   VLOG(6) << "AsyncSocket::setReadCallback() this=" << this << ", fd=" << fd_
 666           << ", callback=" << callback << ", state=" << state_;
 667
 668   // Short circuit if callback is the same as the existing readCallback_.
 669   //
 670   // Note that this is needed for proper functioning during some cleanup cases.
 671   // During cleanup we allow setReadCallback(nullptr) to be called even if the
 672   // read callback is already unset and we have been detached from an event
 673   // base.  This check prevents us from asserting
 674   // eventBase_->isInEventBaseThread() when eventBase_ is nullptr.
 675   if (callback == readCallback_) {
 676     return;
 677   }
 678
 679   /* We are removing a read callback */
 680   if (callback == nullptr &&
 681       immediateReadHandler_.isLoopCallbackScheduled()) {
 682     immediateReadHandler_.cancelLoopCallback();
 683   }
 684
 685   if (shutdownFlags_ & SHUT_READ) {
 686     // Reads have already been shut down on this socket.
 687     //
 688     // Allow setReadCallback(nullptr) to be called in this case, but don't
 689     // allow a new callback to be set.
 690     //
 691     // For example, setReadCallback(nullptr) can happen after an error if we
 692     // invoke some other error callback before invoking readError().  The other
 693     // error callback that is invoked first may go ahead and clear the read
 694     // callback before we get a chance to invoke readError().
 695     if (callback != nullptr) {
 696       return invalidState(callback);
 697     }
 698     assert((eventFlags_ & EventHandler::READ) == 0);
 699     readCallback_ = nullptr;
 700     return;
 701   }
 702
 703   DestructorGuard dg(this);
 704   assert(eventBase_->isInEventBaseThread());
 705
 706   switch ((StateEnum)state_) {
 707     case StateEnum::CONNECTING:
 708     case StateEnum::FAST_OPEN:
 709       // For convenience, we allow the read callback to be set while we are
 710       // still connecting.  We just store the callback for now.  Once the
 711       // connection completes we'll register for read events.
 712       readCallback_ = callback;
 713       return;
 714     case StateEnum::ESTABLISHED:
 715     {
 716       readCallback_ = callback;
 717       uint16_t oldFlags = eventFlags_;
 718       if (readCallback_) {
 719         eventFlags_ |= EventHandler::READ;
 720       } else {
 721         eventFlags_ &= ~EventHandler::READ;
 722       }
 723
 724       // Update our registration if our flags have changed
 725       if (eventFlags_ != oldFlags) {
 726         // We intentionally ignore the return value here.
 727         // updateEventRegistration() will move us into the error state if it
 728         // fails, and we don't need to do anything else here afterwards.
 729         (void)updateEventRegistration();
 730       }
 731
 732       if (readCallback_) {
 733         checkForImmediateRead();
 734       }
 735       return;
 736     }
 737     case StateEnum::CLOSED:
 738     case StateEnum::ERROR:
 739       // We should never reach here.  SHUT_READ should always be set
 740       // if we are in STATE_CLOSED or STATE_ERROR.
 741       assert(false);
 742       return invalidState(callback);
 743     case StateEnum::UNINIT:
 744       // We do not allow setReadCallback() to be called before we start
 745       // connecting.
 746       return invalidState(callback);
 747   }
 748
 749   // We don't put a default case in the switch statement, so that the compiler
 750   // will warn us to update the switch statement if a new state is added.
 751   return invalidState(callback);
 752 }
 753
 754 AsyncSocket::ReadCallback* AsyncSocket::getReadCallback() const {
 755   return readCallback_;
 756 }
 757
 758 void AsyncSocket::write(WriteCallback* callback,
 759                          const void* buf, size_t bytes, WriteFlags flags) {
 760   iovec op;
 761   op.iov_base = const_cast<void*>(buf);
 762   op.iov_len = bytes;
 763   writeImpl(callback, &op, 1, unique_ptr<IOBuf>(), flags);
 764 }
 765
 766 void AsyncSocket::writev(WriteCallback* callback,
 767                           const iovec* vec,
 768                           size_t count,
 769                           WriteFlags flags) {
 770   writeImpl(callback, vec, count, unique_ptr<IOBuf>(), flags);
 771 }
 772
 773 void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf,
 774                               WriteFlags flags) {
 775   constexpr size_t kSmallSizeMax = 64;
 776   size_t count = buf->countChainElements();
 777   if (count <= kSmallSizeMax) {
 778     // suppress "warning: variable length array 'vec' is used [-Wvla]"
 779     FOLLY_PUSH_WARNING
 780     FOLLY_GCC_DISABLE_WARNING("-Wvla")
 781     iovec vec[BOOST_PP_IF(FOLLY_HAVE_VLA, count, kSmallSizeMax)];
 782     FOLLY_POP_WARNING
 783
 784     writeChainImpl(callback, vec, count, std::move(buf), flags);
 785   } else {
 786     iovec* vec = new iovec[count];
 787     writeChainImpl(callback, vec, count, std::move(buf), flags);
 788     delete[] vec;
 789   }
 790 }
 791
 792 void AsyncSocket::writeChainImpl(WriteCallback* callback, iovec* vec,
 793     size_t count, unique_ptr<IOBuf>&& buf, WriteFlags flags) {
 794   size_t veclen = buf->fillIov(vec, count);
 795   writeImpl(callback, vec, veclen, std::move(buf), flags);
 796 }
 797
 798 void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
 799                              size_t count, unique_ptr<IOBuf>&& buf,
 800                              WriteFlags flags) {
 801   VLOG(6) << "AsyncSocket::writev() this=" << this << ", fd=" << fd_
 802           << ", callback=" << callback << ", count=" << count
 803           << ", state=" << state_;
 804   DestructorGuard dg(this);
 805   unique_ptr<IOBuf>ioBuf(std::move(buf));
 806   assert(eventBase_->isInEventBaseThread());
 807
 808   if (shutdownFlags_ & (SHUT_WRITE | SHUT_WRITE_PENDING)) {
 809     // No new writes may be performed after the write side of the socket has
 810     // been shutdown.
 811     //
 812     // We could just call callback->writeError() here to fail just this write.
 813     // However, fail hard and use invalidState() to fail all outstanding
 814     // callbacks and move the socket into the error state.  There's most likely
 815     // a bug in the caller's code, so we abort everything rather than trying to
 816     // proceed as best we can.
 817     return invalidState(callback);
 818   }
 819
 820   uint32_t countWritten = 0;
 821   uint32_t partialWritten = 0;
 822   ssize_t bytesWritten = 0;
 823   bool mustRegister = false;
 824   if ((state_ == StateEnum::ESTABLISHED || state_ == StateEnum::FAST_OPEN) &&
 825       !connecting()) {
 826     if (writeReqHead_ == nullptr) {
 827       // If we are established and there are no other writes pending,
 828       // we can attempt to perform the write immediately.
 829       assert(writeReqTail_ == nullptr);
 830       assert((eventFlags_ & EventHandler::WRITE) == 0);
 831
 832       auto writeResult = performWrite(
 833           vec, uint32_t(count), flags, &countWritten, &partialWritten);
 834       bytesWritten = writeResult.writeReturn;
 835       if (bytesWritten < 0) {
 836         auto errnoCopy = errno;
 837         if (writeResult.exception) {
 838           return failWrite(__func__, callback, 0, *writeResult.exception);
 839         }
 840         AsyncSocketException ex(
 841             AsyncSocketException::INTERNAL_ERROR,
 842             withAddr("writev failed"),
 843             errnoCopy);
 844         return failWrite(__func__, callback, 0, ex);
 845       } else if (countWritten == count) {
 846         // We successfully wrote everything.
 847         // Invoke the callback and return.
 848         if (callback) {
 849           callback->writeSuccess();
 850         }
 851         return;
 852       } else { // continue writing the next writeReq
 853         if (bufferCallback_) {
 854           bufferCallback_->onEgressBuffered();
 855         }
 856       }
 857       if (!connecting()) {
 858         // Writes might put the socket back into connecting state
 859         // if TFO is enabled, and using TFO fails.
 860         // This means that write timeouts would not be active, however
 861         // connect timeouts would affect this stage.
 862         mustRegister = true;
 863       }
 864     }
 865   } else if (!connecting()) {
 866     // Invalid state for writing
 867     return invalidState(callback);
 868   }
 869
 870   // Create a new WriteRequest to add to the queue
 871   WriteRequest* req;
 872   try {
 873     req = BytesWriteRequest::newRequest(
 874         this,
 875         callback,
 876         vec + countWritten,
 877         uint32_t(count - countWritten),
 878         partialWritten,
 879         uint32_t(bytesWritten),
 880         std::move(ioBuf),
 881         flags);
 882   } catch (const std::exception& ex) {
 883     // we mainly expect to catch std::bad_alloc here
 884     AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
 885         withAddr(string("failed to append new WriteRequest: ") + ex.what()));
 886     return failWrite(__func__, callback, size_t(bytesWritten), tex);
 887   }
 888   req->consume();
 889   if (writeReqTail_ == nullptr) {
 890     assert(writeReqHead_ == nullptr);
 891     writeReqHead_ = writeReqTail_ = req;
 892   } else {
 893     writeReqTail_->append(req);
 894     writeReqTail_ = req;
 895   }
 896
 897   // Register for write events if are established and not currently
 898   // waiting on write events
 899   if (mustRegister) {
 900     assert(state_ == StateEnum::ESTABLISHED);
 901     assert((eventFlags_ & EventHandler::WRITE) == 0);
 902     if (!updateEventRegistration(EventHandler::WRITE, 0)) {
 903       assert(state_ == StateEnum::ERROR);
 904       return;
 905     }
 906     if (sendTimeout_ > 0) {
 907       // Schedule a timeout to fire if the write takes too long.
 908       if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
 909         AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
 910                                withAddr("failed to schedule send timeout"));
 911         return failWrite(__func__, ex);
 912       }
 913     }
 914   }
 915 }
 916
 917 void AsyncSocket::writeRequest(WriteRequest* req) {
 918   if (writeReqTail_ == nullptr) {
 919     assert(writeReqHead_ == nullptr);
 920     writeReqHead_ = writeReqTail_ = req;
 921     req->start();
 922   } else {
 923     writeReqTail_->append(req);
 924     writeReqTail_ = req;
 925   }
 926 }
 927
 928 void AsyncSocket::close() {
 929   VLOG(5) << "AsyncSocket::close(): this=" << this << ", fd_=" << fd_
 930           << ", state=" << state_ << ", shutdownFlags="
 931           << std::hex << (int) shutdownFlags_;
 932
 933   // close() is only different from closeNow() when there are pending writes
 934   // that need to drain before we can close.  In all other cases, just call
 935   // closeNow().
 936   //
 937   // Note that writeReqHead_ can be non-nullptr even in STATE_CLOSED or
 938   // STATE_ERROR if close() is invoked while a previous closeNow() or failure
 939   // is still running.  (e.g., If there are multiple pending writes, and we
 940   // call writeError() on the first one, it may call close().  In this case we
 941   // will already be in STATE_CLOSED or STATE_ERROR, but the remaining pending
 942   // writes will still be in the queue.)
 943   //
 944   // We only need to drain pending writes if we are still in STATE_CONNECTING
 945   // or STATE_ESTABLISHED
 946   if ((writeReqHead_ == nullptr) ||
 947       !(state_ == StateEnum::CONNECTING ||
 948       state_ == StateEnum::ESTABLISHED)) {
 949     closeNow();
 950     return;
 951   }
 952
 953   // Declare a DestructorGuard to ensure that the AsyncSocket cannot be
 954   // destroyed until close() returns.
 955   DestructorGuard dg(this);
 956   assert(eventBase_->isInEventBaseThread());
 957
 958   // Since there are write requests pending, we have to set the
 959   // SHUT_WRITE_PENDING flag, and wait to perform the real close until the
 960   // connect finishes and we finish writing these requests.
 961   //
 962   // Set SHUT_READ to indicate that reads are shut down, and set the
 963   // SHUT_WRITE_PENDING flag to mark that we want to shutdown once the
 964   // pending writes complete.
 965   shutdownFlags_ |= (SHUT_READ | SHUT_WRITE_PENDING);
 966
 967   // If a read callback is set, invoke readEOF() immediately to inform it that
 968   // the socket has been closed and no more data can be read.
 969   if (readCallback_) {
 970     // Disable reads if they are enabled
 971     if (!updateEventRegistration(0, EventHandler::READ)) {
 972       // We're now in the error state; callbacks have been cleaned up
 973       assert(state_ == StateEnum::ERROR);
 974       assert(readCallback_ == nullptr);
 975     } else {
 976       ReadCallback* callback = readCallback_;
 977       readCallback_ = nullptr;
 978       callback->readEOF();
 979     }
 980   }
 981 }
 982
 983 void AsyncSocket::closeNow() {
 984   VLOG(5) << "AsyncSocket::closeNow(): this=" << this << ", fd_=" << fd_
 985           << ", state=" << state_ << ", shutdownFlags="
 986           << std::hex << (int) shutdownFlags_;
 987   DestructorGuard dg(this);
 988   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 989
 990   switch (state_) {
 991     case StateEnum::ESTABLISHED:
 992     case StateEnum::CONNECTING:
 993     case StateEnum::FAST_OPEN: {
 994       shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
 995       state_ = StateEnum::CLOSED;
 996
 997       // If the write timeout was set, cancel it.
 998       writeTimeout_.cancelTimeout();
 999
1000       // If we are registered for I/O events, unregister.
1001       if (eventFlags_ != EventHandler::NONE) {
1002         eventFlags_ = EventHandler::NONE;
1003         if (!updateEventRegistration()) {
1004           // We will have been moved into the error state.
1005           assert(state_ == StateEnum::ERROR);
1006           return;
1007         }
1008       }
1009
1010       if (immediateReadHandler_.isLoopCallbackScheduled()) {
1011         immediateReadHandler_.cancelLoopCallback();
1012       }
1013
1014       if (fd_ >= 0) {
1015         ioHandler_.changeHandlerFD(-1);
1016         doClose();
1017       }
1018
1019       invokeConnectErr(socketClosedLocallyEx);
1020
1021       failAllWrites(socketClosedLocallyEx);
1022
1023       if (readCallback_) {
1024         ReadCallback* callback = readCallback_;
1025         readCallback_ = nullptr;
1026         callback->readEOF();
1027       }
1028       return;
1029     }
1030     case StateEnum::CLOSED:
1031       // Do nothing.  It's possible that we are being called recursively
1032       // from inside a callback that we invoked inside another call to close()
1033       // that is still running.
1034       return;
1035     case StateEnum::ERROR:
1036       // Do nothing.  The error handling code has performed (or is performing)
1037       // cleanup.
1038       return;
1039     case StateEnum::UNINIT:
1040       assert(eventFlags_ == EventHandler::NONE);
1041       assert(connectCallback_ == nullptr);
1042       assert(readCallback_ == nullptr);
1043       assert(writeReqHead_ == nullptr);
1044       shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
1045       state_ = StateEnum::CLOSED;
1046       return;
1047   }
1048
1049   LOG(DFATAL) << "AsyncSocket::closeNow() (this=" << this << ", fd=" << fd_
1050               << ") called in unknown state " << state_;
1051 }
1052
1053 void AsyncSocket::closeWithReset() {
1054   // Enable SO_LINGER, with the linger timeout set to 0.
1055   // This will trigger a TCP reset when we close the socket.
1056   if (fd_ >= 0) {
1057     struct linger optLinger = {1, 0};
1058     if (setSockOpt(SOL_SOCKET, SO_LINGER, &optLinger) != 0) {
1059       VLOG(2) << "AsyncSocket::closeWithReset(): error setting SO_LINGER "
1060               << "on " << fd_ << ": errno=" << errno;
1061     }
1062   }
1063
1064   // Then let closeNow() take care of the rest
1065   closeNow();
1066 }
1067
1068 void AsyncSocket::shutdownWrite() {
1069   VLOG(5) << "AsyncSocket::shutdownWrite(): this=" << this << ", fd=" << fd_
1070           << ", state=" << state_ << ", shutdownFlags="
1071           << std::hex << (int) shutdownFlags_;
1072
1073   // If there are no pending writes, shutdownWrite() is identical to
1074   // shutdownWriteNow().
1075   if (writeReqHead_ == nullptr) {
1076     shutdownWriteNow();
1077     return;
1078   }
1079
1080   assert(eventBase_->isInEventBaseThread());
1081
1082   // There are pending writes.  Set SHUT_WRITE_PENDING so that the actual
1083   // shutdown will be performed once all writes complete.
1084   shutdownFlags_ |= SHUT_WRITE_PENDING;
1085 }
1086
1087 void AsyncSocket::shutdownWriteNow() {
1088   VLOG(5) << "AsyncSocket::shutdownWriteNow(): this=" << this
1089           << ", fd=" << fd_ << ", state=" << state_
1090           << ", shutdownFlags=" << std::hex << (int) shutdownFlags_;
1091
1092   if (shutdownFlags_ & SHUT_WRITE) {
1093     // Writes are already shutdown; nothing else to do.
1094     return;
1095   }
1096
1097   // If SHUT_READ is already set, just call closeNow() to completely
1098   // close the socket.  This can happen if close() was called with writes
1099   // pending, and then shutdownWriteNow() is called before all pending writes
1100   // complete.
1101   if (shutdownFlags_ & SHUT_READ) {
1102     closeNow();
1103     return;
1104   }
1105
1106   DestructorGuard dg(this);
1107   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
1108
1109   switch (static_cast<StateEnum>(state_)) {
1110     case StateEnum::ESTABLISHED:
1111     {
1112       shutdownFlags_ |= SHUT_WRITE;
1113
1114       // If the write timeout was set, cancel it.
1115       writeTimeout_.cancelTimeout();
1116
1117       // If we are registered for write events, unregister.
1118       if (!updateEventRegistration(0, EventHandler::WRITE)) {
1119         // We will have been moved into the error state.
1120         assert(state_ == StateEnum::ERROR);
1121         return;
1122       }
1123
1124       // Shutdown writes on the file descriptor
1125       shutdown(fd_, SHUT_WR);
1126
1127       // Immediately fail all write requests
1128       failAllWrites(socketShutdownForWritesEx);
1129       return;
1130     }
1131     case StateEnum::CONNECTING:
1132     {
1133       // Set the SHUT_WRITE_PENDING flag.
1134       // When the connection completes, it will check this flag,
1135       // shutdown the write half of the socket, and then set SHUT_WRITE.
1136       shutdownFlags_ |= SHUT_WRITE_PENDING;
1137
1138       // Immediately fail all write requests
1139       failAllWrites(socketShutdownForWritesEx);
1140       return;
1141     }
1142     case StateEnum::UNINIT:
1143       // Callers normally shouldn't call shutdownWriteNow() before the socket
1144       // even starts connecting.  Nonetheless, go ahead and set
1145       // SHUT_WRITE_PENDING.  Once the socket eventually connects it will
1146       // immediately shut down the write side of the socket.
1147       shutdownFlags_ |= SHUT_WRITE_PENDING;
1148       return;
1149     case StateEnum::FAST_OPEN:
1150       // In fast open state we haven't call connected yet, and if we shutdown
1151       // the writes, we will never try to call connect, so shut everything down
1152       shutdownFlags_ |= SHUT_WRITE;
1153       // Immediately fail all write requests
1154       failAllWrites(socketShutdownForWritesEx);
1155       return;
1156     case StateEnum::CLOSED:
1157     case StateEnum::ERROR:
1158       // We should never get here.  SHUT_WRITE should always be set
1159       // in STATE_CLOSED and STATE_ERROR.
1160       VLOG(4) << "AsyncSocket::shutdownWriteNow() (this=" << this
1161                  << ", fd=" << fd_ << ") in unexpected state " << state_
1162                  << " with SHUT_WRITE not set ("
1163                  << std::hex << (int) shutdownFlags_ << ")";
1164       assert(false);
1165       return;
1166   }
1167
1168   LOG(DFATAL) << "AsyncSocket::shutdownWriteNow() (this=" << this << ", fd="
1169               << fd_ << ") called in unknown state " << state_;
1170 }
1171
1172 bool AsyncSocket::readable() const {
1173   if (fd_ == -1) {
1174     return false;
1175   }
1176   struct pollfd fds[1];
1177   fds[0].fd = fd_;
1178   fds[0].events = POLLIN;
1179   fds[0].revents = 0;
1180   int rc = poll(fds, 1, 0);
1181   return rc == 1;
1182 }
1183
1184 bool AsyncSocket::isPending() const {
1185   return ioHandler_.isPending();
1186 }
1187
1188 bool AsyncSocket::hangup() const {
1189   if (fd_ == -1) {
1190     // sanity check, no one should ask for hangup if we are not connected.
1191     assert(false);
1192     return false;
1193   }
1194 #ifdef POLLRDHUP // Linux-only
1195   struct pollfd fds[1];
1196   fds[0].fd = fd_;
1197   fds[0].events = POLLRDHUP|POLLHUP;
1198   fds[0].revents = 0;
1199   poll(fds, 1, 0);
1200   return (fds[0].revents & (POLLRDHUP|POLLHUP)) != 0;
1201 #else
1202   return false;
1203 #endif
1204 }
1205
1206 bool AsyncSocket::good() const {
1207   return (
1208       (state_ == StateEnum::CONNECTING || state_ == StateEnum::FAST_OPEN ||
1209        state_ == StateEnum::ESTABLISHED) &&
1210       (shutdownFlags_ == 0) && (eventBase_ != nullptr));
1211 }
1212
1213 bool AsyncSocket::error() const {
1214   return (state_ == StateEnum::ERROR);
1215 }
1216
1217 void AsyncSocket::attachEventBase(EventBase* eventBase) {
1218   VLOG(5) << "AsyncSocket::attachEventBase(this=" << this << ", fd=" << fd_
1219           << ", old evb=" << eventBase_ << ", new evb=" << eventBase
1220           << ", state=" << state_ << ", events="
1221           << std::hex << eventFlags_ << ")";
1222   assert(eventBase_ == nullptr);
1223   assert(eventBase->isInEventBaseThread());
1224
1225   eventBase_ = eventBase;
1226   ioHandler_.attachEventBase(eventBase);
1227   writeTimeout_.attachEventBase(eventBase);
1228   if (evbChangeCb_) {
1229     evbChangeCb_->evbAttached(this);
1230   }
1231 }
1232
1233 void AsyncSocket::detachEventBase() {
1234   VLOG(5) << "AsyncSocket::detachEventBase(this=" << this << ", fd=" << fd_
1235           << ", old evb=" << eventBase_ << ", state=" << state_
1236           << ", events=" << std::hex << eventFlags_ << ")";
1237   assert(eventBase_ != nullptr);
1238   assert(eventBase_->isInEventBaseThread());
1239
1240   eventBase_ = nullptr;
1241   ioHandler_.detachEventBase();
1242   writeTimeout_.detachEventBase();
1243   if (evbChangeCb_) {
1244     evbChangeCb_->evbDetached(this);
1245   }
1246 }
1247
1248 bool AsyncSocket::isDetachable() const {
1249   DCHECK(eventBase_ != nullptr);
1250   DCHECK(eventBase_->isInEventBaseThread());
1251
1252   return !ioHandler_.isHandlerRegistered() && !writeTimeout_.isScheduled();
1253 }
1254
1255 void AsyncSocket::getLocalAddress(folly::SocketAddress* address) const {
1256   if (!localAddr_.isInitialized()) {
1257     localAddr_.setFromLocalAddress(fd_);
1258   }
1259   *address = localAddr_;
1260 }
1261
1262 void AsyncSocket::getPeerAddress(folly::SocketAddress* address) const {
1263   if (!addr_.isInitialized()) {
1264     addr_.setFromPeerAddress(fd_);
1265   }
1266   *address = addr_;
1267 }
1268
1269 bool AsyncSocket::getTFOSucceded() const {
1270   return detail::tfo_succeeded(fd_);
1271 }
1272
1273 int AsyncSocket::setNoDelay(bool noDelay) {
1274   if (fd_ < 0) {
1275     VLOG(4) << "AsyncSocket::setNoDelay() called on non-open socket "
1276                << this << "(state=" << state_ << ")";
1277     return EINVAL;
1278
1279   }
1280
1281   int value = noDelay ? 1 : 0;
1282   if (setsockopt(fd_, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)) != 0) {
1283     int errnoCopy = errno;
1284     VLOG(2) << "failed to update TCP_NODELAY option on AsyncSocket "
1285             << this << " (fd=" << fd_ << ", state=" << state_ << "): "
1286             << strerror(errnoCopy);
1287     return errnoCopy;
1288   }
1289
1290   return 0;
1291 }
1292
1293 int AsyncSocket::setCongestionFlavor(const std::string &cname) {
1294
1295   #ifndef TCP_CONGESTION
1296   #define TCP_CONGESTION  13
1297   #endif
1298
1299   if (fd_ < 0) {
1300     VLOG(4) << "AsyncSocket::setCongestionFlavor() called on non-open "
1301                << "socket " << this << "(state=" << state_ << ")";
1302     return EINVAL;
1303
1304   }
1305
1306   if (setsockopt(
1307           fd_,
1308           IPPROTO_TCP,
1309           TCP_CONGESTION,
1310           cname.c_str(),
1311           socklen_t(cname.length() + 1)) != 0) {
1312     int errnoCopy = errno;
1313     VLOG(2) << "failed to update TCP_CONGESTION option on AsyncSocket "
1314             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1315             << strerror(errnoCopy);
1316     return errnoCopy;
1317   }
1318
1319   return 0;
1320 }
1321
1322 int AsyncSocket::setQuickAck(bool quickack) {
1323   (void)quickack;
1324   if (fd_ < 0) {
1325     VLOG(4) << "AsyncSocket::setQuickAck() called on non-open socket "
1326                << this << "(state=" << state_ << ")";
1327     return EINVAL;
1328
1329   }
1330
1331 #ifdef TCP_QUICKACK // Linux-only
1332   int value = quickack ? 1 : 0;
1333   if (setsockopt(fd_, IPPROTO_TCP, TCP_QUICKACK, &value, sizeof(value)) != 0) {
1334     int errnoCopy = errno;
1335     VLOG(2) << "failed to update TCP_QUICKACK option on AsyncSocket"
1336             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1337             << strerror(errnoCopy);
1338     return errnoCopy;
1339   }
1340
1341   return 0;
1342 #else
1343   return ENOSYS;
1344 #endif
1345 }
1346
1347 int AsyncSocket::setSendBufSize(size_t bufsize) {
1348   if (fd_ < 0) {
1349     VLOG(4) << "AsyncSocket::setSendBufSize() called on non-open socket "
1350                << this << "(state=" << state_ << ")";
1351     return EINVAL;
1352   }
1353
1354   if (setsockopt(fd_, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)) !=0) {
1355     int errnoCopy = errno;
1356     VLOG(2) << "failed to update SO_SNDBUF option on AsyncSocket"
1357             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1358             << strerror(errnoCopy);
1359     return errnoCopy;
1360   }
1361
1362   return 0;
1363 }
1364
1365 int AsyncSocket::setRecvBufSize(size_t bufsize) {
1366   if (fd_ < 0) {
1367     VLOG(4) << "AsyncSocket::setRecvBufSize() called on non-open socket "
1368                << this << "(state=" << state_ << ")";
1369     return EINVAL;
1370   }
1371
1372   if (setsockopt(fd_, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)) !=0) {
1373     int errnoCopy = errno;
1374     VLOG(2) << "failed to update SO_RCVBUF option on AsyncSocket"
1375             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1376             << strerror(errnoCopy);
1377     return errnoCopy;
1378   }
1379
1380   return 0;
1381 }
1382
1383 int AsyncSocket::setTCPProfile(int profd) {
1384   if (fd_ < 0) {
1385     VLOG(4) << "AsyncSocket::setTCPProfile() called on non-open socket "
1386                << this << "(state=" << state_ << ")";
1387     return EINVAL;
1388   }
1389
1390   if (setsockopt(fd_, SOL_SOCKET, SO_SET_NAMESPACE, &profd, sizeof(int)) !=0) {
1391     int errnoCopy = errno;
1392     VLOG(2) << "failed to set socket namespace option on AsyncSocket"
1393             << this << "(fd=" << fd_ << ", state=" << state_ << "): "
1394             << strerror(errnoCopy);
1395     return errnoCopy;
1396   }
1397
1398   return 0;
1399 }
1400
1401 void AsyncSocket::ioReady(uint16_t events) noexcept {
1402   VLOG(7) << "AsyncSocket::ioRead() this=" << this << ", fd=" << fd_
1403           << ", events=" << std::hex << events << ", state=" << state_;
1404   DestructorGuard dg(this);
1405   assert(events & EventHandler::READ_WRITE);
1406   assert(eventBase_->isInEventBaseThread());
1407
1408   uint16_t relevantEvents = uint16_t(events & EventHandler::READ_WRITE);
1409   EventBase* originalEventBase = eventBase_;
1410   // If we got there it means that either EventHandler::READ or
1411   // EventHandler::WRITE is set. Any of these flags can
1412   // indicate that there are messages available in the socket
1413   // error message queue.
1414   handleErrMessages();
1415
1416   // Return now if handleErrMessages() detached us from our EventBase
1417   if (eventBase_ != originalEventBase) {
1418     return;
1419   }
1420
1421   if (relevantEvents == EventHandler::READ) {
1422     handleRead();
1423   } else if (relevantEvents == EventHandler::WRITE) {
1424     handleWrite();
1425   } else if (relevantEvents == EventHandler::READ_WRITE) {
1426     // If both read and write events are ready, process writes first.
1427     handleWrite();
1428
1429     // Return now if handleWrite() detached us from our EventBase
1430     if (eventBase_ != originalEventBase) {
1431       return;
1432     }
1433
1434     // Only call handleRead() if a read callback is still installed.
1435     // (It's possible that the read callback was uninstalled during
1436     // handleWrite().)
1437     if (readCallback_) {
1438       handleRead();
1439     }
1440   } else {
1441     VLOG(4) << "AsyncSocket::ioRead() called with unexpected events "
1442                << std::hex << events << "(this=" << this << ")";
1443     abort();
1444   }
1445 }
1446
1447 AsyncSocket::ReadResult
1448 AsyncSocket::performRead(void** buf, size_t* buflen, size_t* /* offset */) {
1449   VLOG(5) << "AsyncSocket::performRead() this=" << this << ", buf=" << *buf
1450           << ", buflen=" << *buflen;
1451
1452   if (preReceivedData_ && !preReceivedData_->empty()) {
1453     VLOG(5) << "AsyncSocket::performRead() this=" << this
1454             << ", reading pre-received data";
1455
1456     io::Cursor cursor(preReceivedData_.get());
1457     auto len = cursor.pullAtMost(*buf, *buflen);
1458
1459     IOBufQueue queue;
1460     queue.append(std::move(preReceivedData_));
1461     queue.trimStart(len);
1462     preReceivedData_ = queue.move();
1463
1464     appBytesReceived_ += len;
1465     return ReadResult(len);
1466   }
1467
1468   ssize_t bytes = recv(fd_, *buf, *buflen, MSG_DONTWAIT);
1469   if (bytes < 0) {
1470     if (errno == EAGAIN || errno == EWOULDBLOCK) {
1471       // No more data to read right now.
1472       return ReadResult(READ_BLOCKING);
1473     } else {
1474       return ReadResult(READ_ERROR);
1475     }
1476   } else {
1477     appBytesReceived_ += bytes;
1478     return ReadResult(bytes);
1479   }
1480 }
1481
1482 void AsyncSocket::prepareReadBuffer(void** buf, size_t* buflen) {
1483   // no matter what, buffer should be preapared for non-ssl socket
1484   CHECK(readCallback_);
1485   readCallback_->getReadBuffer(buf, buflen);
1486 }
1487
1488 void AsyncSocket::handleErrMessages() noexcept {
1489   // This method has non-empty implementation only for platforms
1490   // supporting per-socket error queues.
1491   VLOG(5) << "AsyncSocket::handleErrMessages() this=" << this << ", fd=" << fd_
1492           << ", state=" << state_;
1493   if (errMessageCallback_ == nullptr) {
1494     VLOG(7) << "AsyncSocket::handleErrMessages(): "
1495             << "no callback installed - exiting.";
1496     return;
1497   }
1498
1499 #ifdef MSG_ERRQUEUE
1500   uint8_t ctrl[1024];
1501   unsigned char data;
1502   struct msghdr msg;
1503   iovec entry;
1504
1505   entry.iov_base = &data;
1506   entry.iov_len = sizeof(data);
1507   msg.msg_iov = &entry;
1508   msg.msg_iovlen = 1;
1509   msg.msg_name = nullptr;
1510   msg.msg_namelen = 0;
1511   msg.msg_control = ctrl;
1512   msg.msg_controllen = sizeof(ctrl);
1513   msg.msg_flags = 0;
1514
1515   int ret;
1516   while (true) {
1517     ret = recvmsg(fd_, &msg, MSG_ERRQUEUE);
1518     VLOG(5) << "AsyncSocket::handleErrMessages(): recvmsg returned " << ret;
1519
1520     if (ret < 0) {
1521       if (errno != EAGAIN) {
1522         auto errnoCopy = errno;
1523         LOG(ERROR) << "::recvmsg exited with code " << ret
1524                    << ", errno: " << errnoCopy;
1525         AsyncSocketException ex(
1526           AsyncSocketException::INTERNAL_ERROR,
1527           withAddr("recvmsg() failed"),
1528           errnoCopy);
1529         failErrMessageRead(__func__, ex);
1530       }
1531       return;
1532     }
1533
1534     for (struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
1535          cmsg != nullptr && cmsg->cmsg_len != 0;
1536          cmsg = CMSG_NXTHDR(&msg, cmsg)) {
1537       errMessageCallback_->errMessage(*cmsg);
1538     }
1539   }
1540 #endif //MSG_ERRQUEUE
1541 }
1542
1543 void AsyncSocket::handleRead() noexcept {
1544   VLOG(5) << "AsyncSocket::handleRead() this=" << this << ", fd=" << fd_
1545           << ", state=" << state_;
1546   assert(state_ == StateEnum::ESTABLISHED);
1547   assert((shutdownFlags_ & SHUT_READ) == 0);
1548   assert(readCallback_ != nullptr);
1549   assert(eventFlags_ & EventHandler::READ);
1550
1551   // Loop until:
1552   // - a read attempt would block
1553   // - readCallback_ is uninstalled
1554   // - the number of loop iterations exceeds the optional maximum
1555   // - this AsyncSocket is moved to another EventBase
1556   //
1557   // When we invoke readDataAvailable() it may uninstall the readCallback_,
1558   // which is why need to check for it here.
1559   //
1560   // The last bullet point is slightly subtle.  readDataAvailable() may also
1561   // detach this socket from this EventBase.  However, before
1562   // readDataAvailable() returns another thread may pick it up, attach it to
1563   // a different EventBase, and install another readCallback_.  We need to
1564   // exit immediately after readDataAvailable() returns if the eventBase_ has
1565   // changed.  (The caller must perform some sort of locking to transfer the
1566   // AsyncSocket between threads properly.  This will be sufficient to ensure
1567   // that this thread sees the updated eventBase_ variable after
1568   // readDataAvailable() returns.)
1569   uint16_t numReads = 0;
1570   EventBase* originalEventBase = eventBase_;
1571   while (readCallback_ && eventBase_ == originalEventBase) {
1572     // Get the buffer to read into.
1573     void* buf = nullptr;
1574     size_t buflen = 0, offset = 0;
1575     try {
1576       prepareReadBuffer(&buf, &buflen);
1577       VLOG(5) << "prepareReadBuffer() buf=" << buf << ", buflen=" << buflen;
1578     } catch (const AsyncSocketException& ex) {
1579       return failRead(__func__, ex);
1580     } catch (const std::exception& ex) {
1581       AsyncSocketException tex(AsyncSocketException::BAD_ARGS,
1582                               string("ReadCallback::getReadBuffer() "
1583                                      "threw exception: ") +
1584                               ex.what());
1585       return failRead(__func__, tex);
1586     } catch (...) {
1587       AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
1588                              "ReadCallback::getReadBuffer() threw "
1589                              "non-exception type");
1590       return failRead(__func__, ex);
1591     }
1592     if (!isBufferMovable_ && (buf == nullptr || buflen == 0)) {
1593       AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
1594                              "ReadCallback::getReadBuffer() returned "
1595                              "empty buffer");
1596       return failRead(__func__, ex);
1597     }
1598
1599     // Perform the read
1600     auto readResult = performRead(&buf, &buflen, &offset);
1601     auto bytesRead = readResult.readReturn;
1602     VLOG(4) << "this=" << this << ", AsyncSocket::handleRead() got "
1603             << bytesRead << " bytes";
1604     if (bytesRead > 0) {
1605       if (!isBufferMovable_) {
1606         readCallback_->readDataAvailable(size_t(bytesRead));
1607       } else {
1608         CHECK(kOpenSslModeMoveBufferOwnership);
1609         VLOG(5) << "this=" << this << ", AsyncSocket::handleRead() got "
1610                 << "buf=" << buf << ", " << bytesRead << "/" << buflen
1611                 << ", offset=" << offset;
1612         auto readBuf = folly::IOBuf::takeOwnership(buf, buflen);
1613         readBuf->trimStart(offset);
1614         readBuf->trimEnd(buflen - offset - bytesRead);
1615         readCallback_->readBufferAvailable(std::move(readBuf));
1616       }
1617
1618       // Fall through and continue around the loop if the read
1619       // completely filled the available buffer.
1620       // Note that readCallback_ may have been uninstalled or changed inside
1621       // readDataAvailable().
1622       if (size_t(bytesRead) < buflen) {
1623         return;
1624       }
1625     } else if (bytesRead == READ_BLOCKING) {
1626         // No more data to read right now.
1627         return;
1628     } else if (bytesRead == READ_ERROR) {
1629       readErr_ = READ_ERROR;
1630       if (readResult.exception) {
1631         return failRead(__func__, *readResult.exception);
1632       }
1633       auto errnoCopy = errno;
1634       AsyncSocketException ex(
1635           AsyncSocketException::INTERNAL_ERROR,
1636           withAddr("recv() failed"),
1637           errnoCopy);
1638       return failRead(__func__, ex);
1639     } else {
1640       assert(bytesRead == READ_EOF);
1641       readErr_ = READ_EOF;
1642       // EOF
1643       shutdownFlags_ |= SHUT_READ;
1644       if (!updateEventRegistration(0, EventHandler::READ)) {
1645         // we've already been moved into STATE_ERROR
1646         assert(state_ == StateEnum::ERROR);
1647         assert(readCallback_ == nullptr);
1648         return;
1649       }
1650
1651       ReadCallback* callback = readCallback_;
1652       readCallback_ = nullptr;
1653       callback->readEOF();
1654       return;
1655     }
1656     if (maxReadsPerEvent_ && (++numReads >= maxReadsPerEvent_)) {
1657       if (readCallback_ != nullptr) {
1658         // We might still have data in the socket.
1659         // (e.g. see comment in AsyncSSLSocket::checkForImmediateRead)
1660         scheduleImmediateRead();
1661       }
1662       return;
1663     }
1664   }
1665 }
1666
1667 /**
1668  * This function attempts to write as much data as possible, until no more data
1669  * can be written.
1670  *
1671  * - If it sends all available data, it unregisters for write events, and stops
1672  *   the writeTimeout_.
1673  *
1674  * - If not all of the data can be sent immediately, it reschedules
1675  *   writeTimeout_ (if a non-zero timeout is set), and ensures the handler is
1676  *   registered for write events.
1677  */
1678 void AsyncSocket::handleWrite() noexcept {
1679   VLOG(5) << "AsyncSocket::handleWrite() this=" << this << ", fd=" << fd_
1680           << ", state=" << state_;
1681   DestructorGuard dg(this);
1682
1683   if (state_ == StateEnum::CONNECTING) {
1684     handleConnect();
1685     return;
1686   }
1687
1688   // Normal write
1689   assert(state_ == StateEnum::ESTABLISHED);
1690   assert((shutdownFlags_ & SHUT_WRITE) == 0);
1691   assert(writeReqHead_ != nullptr);
1692
1693   // Loop until we run out of write requests,
1694   // or until this socket is moved to another EventBase.
1695   // (See the comment in handleRead() explaining how this can happen.)
1696   EventBase* originalEventBase = eventBase_;
1697   while (writeReqHead_ != nullptr && eventBase_ == originalEventBase) {
1698     auto writeResult = writeReqHead_->performWrite();
1699     if (writeResult.writeReturn < 0) {
1700       if (writeResult.exception) {
1701         return failWrite(__func__, *writeResult.exception);
1702       }
1703       auto errnoCopy = errno;
1704       AsyncSocketException ex(
1705           AsyncSocketException::INTERNAL_ERROR,
1706           withAddr("writev() failed"),
1707           errnoCopy);
1708       return failWrite(__func__, ex);
1709     } else if (writeReqHead_->isComplete()) {
1710       // We finished this request
1711       WriteRequest* req = writeReqHead_;
1712       writeReqHead_ = req->getNext();
1713
1714       if (writeReqHead_ == nullptr) {
1715         writeReqTail_ = nullptr;
1716         // This is the last write request.
1717         // Unregister for write events and cancel the send timer
1718         // before we invoke the callback.  We have to update the state properly
1719         // before calling the callback, since it may want to detach us from
1720         // the EventBase.
1721         if (eventFlags_ & EventHandler::WRITE) {
1722           if (!updateEventRegistration(0, EventHandler::WRITE)) {
1723             assert(state_ == StateEnum::ERROR);
1724             return;
1725           }
1726           // Stop the send timeout
1727           writeTimeout_.cancelTimeout();
1728         }
1729         assert(!writeTimeout_.isScheduled());
1730
1731         // If SHUT_WRITE_PENDING is set, we should shutdown the socket after
1732         // we finish sending the last write request.
1733         //
1734         // We have to do this before invoking writeSuccess(), since
1735         // writeSuccess() may detach us from our EventBase.
1736         if (shutdownFlags_ & SHUT_WRITE_PENDING) {
1737           assert(connectCallback_ == nullptr);
1738           shutdownFlags_ |= SHUT_WRITE;
1739
1740           if (shutdownFlags_ & SHUT_READ) {
1741             // Reads have already been shutdown.  Fully close the socket and
1742             // move to STATE_CLOSED.
1743             //
1744             // Note: This code currently moves us to STATE_CLOSED even if
1745             // close() hasn't ever been called.  This can occur if we have
1746             // received EOF from the peer and shutdownWrite() has been called
1747             // locally.  Should we bother staying in STATE_ESTABLISHED in this
1748             // case, until close() is actually called?  I can't think of a
1749             // reason why we would need to do so.  No other operations besides
1750             // calling close() or destroying the socket can be performed at
1751             // this point.
1752             assert(readCallback_ == nullptr);
1753             state_ = StateEnum::CLOSED;
1754             if (fd_ >= 0) {
1755               ioHandler_.changeHandlerFD(-1);
1756               doClose();
1757             }
1758           } else {
1759             // Reads are still enabled, so we are only doing a half-shutdown
1760             shutdown(fd_, SHUT_WR);
1761           }
1762         }
1763       }
1764
1765       // Invoke the callback
1766       WriteCallback* callback = req->getCallback();
1767       req->destroy();
1768       if (callback) {
1769         callback->writeSuccess();
1770       }
1771       // We'll continue around the loop, trying to write another request
1772     } else {
1773       // Partial write.
1774       if (bufferCallback_) {
1775         bufferCallback_->onEgressBuffered();
1776       }
1777       writeReqHead_->consume();
1778       // Stop after a partial write; it's highly likely that a subsequent write
1779       // attempt will just return EAGAIN.
1780       //
1781       // Ensure that we are registered for write events.
1782       if ((eventFlags_ & EventHandler::WRITE) == 0) {
1783         if (!updateEventRegistration(EventHandler::WRITE, 0)) {
1784           assert(state_ == StateEnum::ERROR);
1785           return;
1786         }
1787       }
1788
1789       // Reschedule the send timeout, since we have made some write progress.
1790       if (sendTimeout_ > 0) {
1791         if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
1792           AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
1793               withAddr("failed to reschedule write timeout"));
1794           return failWrite(__func__, ex);
1795         }
1796       }
1797       return;
1798     }
1799   }
1800   if (!writeReqHead_ && bufferCallback_) {
1801     bufferCallback_->onEgressBufferCleared();
1802   }
1803 }
1804
1805 void AsyncSocket::checkForImmediateRead() noexcept {
1806   // We currently don't attempt to perform optimistic reads in AsyncSocket.
1807   // (However, note that some subclasses do override this method.)
1808   //
1809   // Simply calling handleRead() here would be bad, as this would call
1810   // readCallback_->getReadBuffer(), forcing the callback to allocate a read
1811   // buffer even though no data may be available.  This would waste lots of
1812   // memory, since the buffer will sit around unused until the socket actually
1813   // becomes readable.
1814   //
1815   // Checking if the socket is readable now also seems like it would probably
1816   // be a pessimism.  In most cases it probably wouldn't be readable, and we
1817   // would just waste an extra system call.  Even if it is readable, waiting to
1818   // find out from libevent on the next event loop doesn't seem that bad.
1819   //
1820   // The exception to this is if we have pre-received data. In that case there
1821   // is definitely data available immediately.
1822   if (preReceivedData_ && !preReceivedData_->empty()) {
1823     handleRead();
1824   }
1825 }
1826
1827 void AsyncSocket::handleInitialReadWrite() noexcept {
1828   // Our callers should already be holding a DestructorGuard, but grab
1829   // one here just to make sure, in case one of our calling code paths ever
1830   // changes.
1831   DestructorGuard dg(this);
1832   // If we have a readCallback_, make sure we enable read events.  We
1833   // may already be registered for reads if connectSuccess() set
1834   // the read calback.
1835   if (readCallback_ && !(eventFlags_ & EventHandler::READ)) {
1836     assert(state_ == StateEnum::ESTABLISHED);
1837     assert((shutdownFlags_ & SHUT_READ) == 0);
1838     if (!updateEventRegistration(EventHandler::READ, 0)) {
1839       assert(state_ == StateEnum::ERROR);
1840       return;
1841     }
1842     checkForImmediateRead();
1843   } else if (readCallback_ == nullptr) {
1844     // Unregister for read events.
1845     updateEventRegistration(0, EventHandler::READ);
1846   }
1847
1848   // If we have write requests pending, try to send them immediately.
1849   // Since we just finished accepting, there is a very good chance that we can
1850   // write without blocking.
1851   //
1852   // However, we only process them if EventHandler::WRITE is not already set,
1853   // which means that we're already blocked on a write attempt.  (This can
1854   // happen if connectSuccess() called write() before returning.)
1855   if (writeReqHead_ && !(eventFlags_ & EventHandler::WRITE)) {
1856     // Call handleWrite() to perform write processing.
1857     handleWrite();
1858   } else if (writeReqHead_ == nullptr) {
1859     // Unregister for write event.
1860     updateEventRegistration(0, EventHandler::WRITE);
1861   }
1862 }
1863
1864 void AsyncSocket::handleConnect() noexcept {
1865   VLOG(5) << "AsyncSocket::handleConnect() this=" << this << ", fd=" << fd_
1866           << ", state=" << state_;
1867   assert(state_ == StateEnum::CONNECTING);
1868   // SHUT_WRITE can never be set while we are still connecting;
1869   // SHUT_WRITE_PENDING may be set, be we only set SHUT_WRITE once the connect
1870   // finishes
1871   assert((shutdownFlags_ & SHUT_WRITE) == 0);
1872
1873   // In case we had a connect timeout, cancel the timeout
1874   writeTimeout_.cancelTimeout();
1875   // We don't use a persistent registration when waiting on a connect event,
1876   // so we have been automatically unregistered now.  Update eventFlags_ to
1877   // reflect reality.
1878   assert(eventFlags_ == EventHandler::WRITE);
1879   eventFlags_ = EventHandler::NONE;
1880
1881   // Call getsockopt() to check if the connect succeeded
1882   int error;
1883   socklen_t len = sizeof(error);
1884   int rv = getsockopt(fd_, SOL_SOCKET, SO_ERROR, &error, &len);
1885   if (rv != 0) {
1886     auto errnoCopy = errno;
1887     AsyncSocketException ex(
1888         AsyncSocketException::INTERNAL_ERROR,
1889         withAddr("error calling getsockopt() after connect"),
1890         errnoCopy);
1891     VLOG(4) << "AsyncSocket::handleConnect(this=" << this << ", fd="
1892                << fd_ << " host=" << addr_.describe()
1893                << ") exception:" << ex.what();
1894     return failConnect(__func__, ex);
1895   }
1896
1897   if (error != 0) {
1898     AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
1899                            "connect failed", error);
1900     VLOG(1) << "AsyncSocket::handleConnect(this=" << this << ", fd="
1901             << fd_ << " host=" << addr_.describe()
1902             << ") exception: " << ex.what();
1903     return failConnect(__func__, ex);
1904   }
1905
1906   // Move into STATE_ESTABLISHED
1907   state_ = StateEnum::ESTABLISHED;
1908
1909   // If SHUT_WRITE_PENDING is set and we don't have any write requests to
1910   // perform, immediately shutdown the write half of the socket.
1911   if ((shutdownFlags_ & SHUT_WRITE_PENDING) && writeReqHead_ == nullptr) {
1912     // SHUT_READ shouldn't be set.  If close() is called on the socket while we
1913     // are still connecting we just abort the connect rather than waiting for
1914     // it to complete.
1915     assert((shutdownFlags_ & SHUT_READ) == 0);
1916     shutdown(fd_, SHUT_WR);
1917     shutdownFlags_ |= SHUT_WRITE;
1918   }
1919
1920   VLOG(7) << "AsyncSocket " << this << ": fd " << fd_
1921           << "successfully connected; state=" << state_;
1922
1923   // Remember the EventBase we are attached to, before we start invoking any
1924   // callbacks (since the callbacks may call detachEventBase()).
1925   EventBase* originalEventBase = eventBase_;
1926
1927   invokeConnectSuccess();
1928   // Note that the connect callback may have changed our state.
1929   // (set or unset the read callback, called write(), closed the socket, etc.)
1930   // The following code needs to handle these situations correctly.
1931   //
1932   // If the socket has been closed, readCallback_ and writeReqHead_ will
1933   // always be nullptr, so that will prevent us from trying to read or write.
1934   //
1935   // The main thing to check for is if eventBase_ is still originalEventBase.
1936   // If not, we have been detached from this event base, so we shouldn't
1937   // perform any more operations.
1938   if (eventBase_ != originalEventBase) {
1939     return;
1940   }
1941
1942   handleInitialReadWrite();
1943 }
1944
1945 void AsyncSocket::timeoutExpired() noexcept {
1946   VLOG(7) << "AsyncSocket " << this << ", fd " << fd_ << ": timeout expired: "
1947           << "state=" << state_ << ", events=" << std::hex << eventFlags_;
1948   DestructorGuard dg(this);
1949   assert(eventBase_->isInEventBaseThread());
1950
1951   if (state_ == StateEnum::CONNECTING) {
1952     // connect() timed out
1953     // Unregister for I/O events.
1954     if (connectCallback_) {
1955       AsyncSocketException ex(
1956           AsyncSocketException::TIMED_OUT,
1957           folly::sformat(
1958               "connect timed out after {}ms", connectTimeout_.count()));
1959       failConnect(__func__, ex);
1960     } else {
1961       // we faced a connect error without a connect callback, which could
1962       // happen due to TFO.
1963       AsyncSocketException ex(
1964           AsyncSocketException::TIMED_OUT, "write timed out during connection");
1965       failWrite(__func__, ex);
1966     }
1967   } else {
1968     // a normal write operation timed out
1969     AsyncSocketException ex(
1970         AsyncSocketException::TIMED_OUT,
1971         folly::sformat("write timed out after {}ms", sendTimeout_));
1972     failWrite(__func__, ex);
1973   }
1974 }
1975
1976 ssize_t AsyncSocket::tfoSendMsg(int fd, struct msghdr* msg, int msg_flags) {
1977   return detail::tfo_sendmsg(fd, msg, msg_flags);
1978 }
1979
1980 AsyncSocket::WriteResult
1981 AsyncSocket::sendSocketMessage(int fd, struct msghdr* msg, int msg_flags) {
1982   ssize_t totalWritten = 0;
1983   if (state_ == StateEnum::FAST_OPEN) {
1984     sockaddr_storage addr;
1985     auto len = addr_.getAddress(&addr);
1986     msg->msg_name = &addr;
1987     msg->msg_namelen = len;
1988     totalWritten = tfoSendMsg(fd_, msg, msg_flags);
1989     if (totalWritten >= 0) {
1990       tfoFinished_ = true;
1991       state_ = StateEnum::ESTABLISHED;
1992       // We schedule this asynchrously so that we don't end up
1993       // invoking initial read or write while a write is in progress.
1994       scheduleInitialReadWrite();
1995     } else if (errno == EINPROGRESS) {
1996       VLOG(4) << "TFO falling back to connecting";
1997       // A normal sendmsg doesn't return EINPROGRESS, however
1998       // TFO might fallback to connecting if there is no
1999       // cookie.
2000       state_ = StateEnum::CONNECTING;
2001       try {
2002         scheduleConnectTimeout();
2003         registerForConnectEvents();
2004       } catch (const AsyncSocketException& ex) {
2005         return WriteResult(
2006             WRITE_ERROR, folly::make_unique<AsyncSocketException>(ex));
2007       }
2008       // Let's fake it that no bytes were written and return an errno.
2009       errno = EAGAIN;
2010       totalWritten = -1;
2011     } else if (errno == EOPNOTSUPP) {
2012       // Try falling back to connecting.
2013       VLOG(4) << "TFO not supported";
2014       state_ = StateEnum::CONNECTING;
2015       try {
2016         int ret = socketConnect((const sockaddr*)&addr, len);
2017         if (ret == 0) {
2018           // connect succeeded immediately
2019           // Treat this like no data was written.
2020           state_ = StateEnum::ESTABLISHED;
2021           scheduleInitialReadWrite();
2022         }
2023         // If there was no exception during connections,
2024         // we would return that no bytes were written.
2025         errno = EAGAIN;
2026         totalWritten = -1;
2027       } catch (const AsyncSocketException& ex) {
2028         return WriteResult(
2029             WRITE_ERROR, folly::make_unique<AsyncSocketException>(ex));
2030       }
2031     } else if (errno == EAGAIN) {
2032       // Normally sendmsg would indicate that the write would block.
2033       // However in the fast open case, it would indicate that sendmsg
2034       // fell back to a connect. This is a return code from connect()
2035       // instead, and is an error condition indicating no fds available.
2036       return WriteResult(
2037           WRITE_ERROR,
2038           folly::make_unique<AsyncSocketException>(
2039               AsyncSocketException::UNKNOWN, "No more free local ports"));
2040     }
2041   } else {
2042     totalWritten = ::sendmsg(fd, msg, msg_flags);
2043   }
2044   return WriteResult(totalWritten);
2045 }
2046
2047 AsyncSocket::WriteResult AsyncSocket::performWrite(
2048     const iovec* vec,
2049     uint32_t count,
2050     WriteFlags flags,
2051     uint32_t* countWritten,
2052     uint32_t* partialWritten) {
2053   // We use sendmsg() instead of writev() so that we can pass in MSG_NOSIGNAL
2054   // We correctly handle EPIPE errors, so we never want to receive SIGPIPE
2055   // (since it may terminate the program if the main program doesn't explicitly
2056   // ignore it).
2057   struct msghdr msg;
2058   msg.msg_name = nullptr;
2059   msg.msg_namelen = 0;
2060   msg.msg_iov = const_cast<iovec *>(vec);
2061   msg.msg_iovlen = std::min<size_t>(count, kIovMax);
2062   msg.msg_flags = 0;
2063   msg.msg_controllen = sendMsgParamCallback_->getAncillaryDataSize(flags);
2064   CHECK_GE(AsyncSocket::SendMsgParamsCallback::maxAncillaryDataSize,
2065            msg.msg_controllen);
2066
2067   if (msg.msg_controllen != 0) {
2068     msg.msg_control = reinterpret_cast<char*>(alloca(msg.msg_controllen));
2069     sendMsgParamCallback_->getAncillaryData(flags, msg.msg_control);
2070   } else {
2071     msg.msg_control = nullptr;
2072   }
2073   int msg_flags = sendMsgParamCallback_->getFlags(flags);
2074
2075   auto writeResult = sendSocketMessage(fd_, &msg, msg_flags);
2076   auto totalWritten = writeResult.writeReturn;
2077   if (totalWritten < 0) {
2078     bool tryAgain = (errno == EAGAIN);
2079 #ifdef __APPLE__
2080     // Apple has a bug where doing a second write on a socket which we
2081     // have opened with TFO causes an ENOTCONN to be thrown. However the
2082     // socket is really connected, so treat ENOTCONN as a EAGAIN until
2083     // this bug is fixed.
2084     tryAgain |= (errno == ENOTCONN);
2085 #endif
2086     if (!writeResult.exception && tryAgain) {
2087       // TCP buffer is full; we can't write any more data right now.
2088       *countWritten = 0;
2089       *partialWritten = 0;
2090       return WriteResult(0);
2091     }
2092     // error
2093     *countWritten = 0;
2094     *partialWritten = 0;
2095     return writeResult;
2096   }
2097
2098   appBytesWritten_ += totalWritten;
2099
2100   uint32_t bytesWritten;
2101   uint32_t n;
2102   for (bytesWritten = uint32_t(totalWritten), n = 0; n < count; ++n) {
2103     const iovec* v = vec + n;
2104     if (v->iov_len > bytesWritten) {
2105       // Partial write finished in the middle of this iovec
2106       *countWritten = n;
2107       *partialWritten = bytesWritten;
2108       return WriteResult(totalWritten);
2109     }
2110
2111     bytesWritten -= uint32_t(v->iov_len);
2112   }
2113
2114   assert(bytesWritten == 0);
2115   *countWritten = n;
2116   *partialWritten = 0;
2117   return WriteResult(totalWritten);
2118 }
2119
2120 /**
2121  * Re-register the EventHandler after eventFlags_ has changed.
2122  *
2123  * If an error occurs, fail() is called to move the socket into the error state
2124  * and call all currently installed callbacks.  After an error, the
2125  * AsyncSocket is completely unregistered.
2126  *
2127  * @return Returns true on succcess, or false on error.
2128  */
2129 bool AsyncSocket::updateEventRegistration() {
2130   VLOG(5) << "AsyncSocket::updateEventRegistration(this=" << this
2131           << ", fd=" << fd_ << ", evb=" << eventBase_ << ", state=" << state_
2132           << ", events=" << std::hex << eventFlags_;
2133   assert(eventBase_->isInEventBaseThread());
2134   if (eventFlags_ == EventHandler::NONE) {
2135     ioHandler_.unregisterHandler();
2136     return true;
2137   }
2138
2139   // Always register for persistent events, so we don't have to re-register
2140   // after being called back.
2141   if (!ioHandler_.registerHandler(
2142           uint16_t(eventFlags_ | EventHandler::PERSIST))) {
2143     eventFlags_ = EventHandler::NONE; // we're not registered after error
2144     AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
2145         withAddr("failed to update AsyncSocket event registration"));
2146     fail("updateEventRegistration", ex);
2147     return false;
2148   }
2149
2150   return true;
2151 }
2152
2153 bool AsyncSocket::updateEventRegistration(uint16_t enable,
2154                                            uint16_t disable) {
2155   uint16_t oldFlags = eventFlags_;
2156   eventFlags_ |= enable;
2157   eventFlags_ &= ~disable;
2158   if (eventFlags_ == oldFlags) {
2159     return true;
2160   } else {
2161     return updateEventRegistration();
2162   }
2163 }
2164
2165 void AsyncSocket::startFail() {
2166   // startFail() should only be called once
2167   assert(state_ != StateEnum::ERROR);
2168   assert(getDestructorGuardCount() > 0);
2169   state_ = StateEnum::ERROR;
2170   // Ensure that SHUT_READ and SHUT_WRITE are set,
2171   // so all future attempts to read or write will be rejected
2172   shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
2173
2174   if (eventFlags_ != EventHandler::NONE) {
2175     eventFlags_ = EventHandler::NONE;
2176     ioHandler_.unregisterHandler();
2177   }
2178   writeTimeout_.cancelTimeout();
2179
2180   if (fd_ >= 0) {
2181     ioHandler_.changeHandlerFD(-1);
2182     doClose();
2183   }
2184 }
2185
2186 void AsyncSocket::invokeAllErrors(const AsyncSocketException& ex) {
2187   invokeConnectErr(ex);
2188   failAllWrites(ex);
2189
2190   if (readCallback_) {
2191     ReadCallback* callback = readCallback_;
2192     readCallback_ = nullptr;
2193     callback->readErr(ex);
2194   }
2195 }
2196
2197 void AsyncSocket::finishFail() {
2198   assert(state_ == StateEnum::ERROR);
2199   assert(getDestructorGuardCount() > 0);
2200
2201   AsyncSocketException ex(
2202       AsyncSocketException::INTERNAL_ERROR,
2203       withAddr("socket closing after error"));
2204   invokeAllErrors(ex);
2205 }
2206
2207 void AsyncSocket::finishFail(const AsyncSocketException& ex) {
2208   assert(state_ == StateEnum::ERROR);
2209   assert(getDestructorGuardCount() > 0);
2210   invokeAllErrors(ex);
2211 }
2212
2213 void AsyncSocket::fail(const char* fn, const AsyncSocketException& ex) {
2214   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2215              << state_ << " host=" << addr_.describe()
2216              << "): failed in " << fn << "(): "
2217              << ex.what();
2218   startFail();
2219   finishFail();
2220 }
2221
2222 void AsyncSocket::failConnect(const char* fn, const AsyncSocketException& ex) {
2223   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2224                << state_ << " host=" << addr_.describe()
2225                << "): failed while connecting in " << fn << "(): "
2226                << ex.what();
2227   startFail();
2228
2229   invokeConnectErr(ex);
2230   finishFail(ex);
2231 }
2232
2233 void AsyncSocket::failRead(const char* fn, const AsyncSocketException& ex) {
2234   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2235                << state_ << " host=" << addr_.describe()
2236                << "): failed while reading in " << fn << "(): "
2237                << ex.what();
2238   startFail();
2239
2240   if (readCallback_ != nullptr) {
2241     ReadCallback* callback = readCallback_;
2242     readCallback_ = nullptr;
2243     callback->readErr(ex);
2244   }
2245
2246   finishFail();
2247 }
2248
2249 void AsyncSocket::failErrMessageRead(const char* fn,
2250                                      const AsyncSocketException& ex) {
2251   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2252                << state_ << " host=" << addr_.describe()
2253                << "): failed while reading message in " << fn << "(): "
2254                << ex.what();
2255   startFail();
2256
2257   if (errMessageCallback_ != nullptr) {
2258     ErrMessageCallback* callback = errMessageCallback_;
2259     errMessageCallback_ = nullptr;
2260     callback->errMessageError(ex);
2261   }
2262
2263   finishFail();
2264 }
2265
2266 void AsyncSocket::failWrite(const char* fn, const AsyncSocketException& ex) {
2267   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2268                << state_ << " host=" << addr_.describe()
2269                << "): failed while writing in " << fn << "(): "
2270                << ex.what();
2271   startFail();
2272
2273   // Only invoke the first write callback, since the error occurred while
2274   // writing this request.  Let any other pending write callbacks be invoked in
2275   // finishFail().
2276   if (writeReqHead_ != nullptr) {
2277     WriteRequest* req = writeReqHead_;
2278     writeReqHead_ = req->getNext();
2279     WriteCallback* callback = req->getCallback();
2280     uint32_t bytesWritten = req->getTotalBytesWritten();
2281     req->destroy();
2282     if (callback) {
2283       callback->writeErr(bytesWritten, ex);
2284     }
2285   }
2286
2287   finishFail();
2288 }
2289
2290 void AsyncSocket::failWrite(const char* fn, WriteCallback* callback,
2291                              size_t bytesWritten,
2292                              const AsyncSocketException& ex) {
2293   // This version of failWrite() is used when the failure occurs before
2294   // we've added the callback to writeReqHead_.
2295   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
2296              << state_ << " host=" << addr_.describe()
2297              <<"): failed while writing in " << fn << "(): "
2298              << ex.what();
2299   startFail();
2300
2301   if (callback != nullptr) {
2302     callback->writeErr(bytesWritten, ex);
2303   }
2304
2305   finishFail();
2306 }
2307
2308 void AsyncSocket::failAllWrites(const AsyncSocketException& ex) {
2309   // Invoke writeError() on all write callbacks.
2310   // This is used when writes are forcibly shutdown with write requests
2311   // pending, or when an error occurs with writes pending.
2312   while (writeReqHead_ != nullptr) {
2313     WriteRequest* req = writeReqHead_;
2314     writeReqHead_ = req->getNext();
2315     WriteCallback* callback = req->getCallback();
2316     if (callback) {
2317       callback->writeErr(req->getTotalBytesWritten(), ex);
2318     }
2319     req->destroy();
2320   }
2321 }
2322
2323 void AsyncSocket::invalidState(ConnectCallback* callback) {
2324   VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_
2325           << "): connect() called in invalid state " << state_;
2326
2327   /*
2328    * The invalidState() methods don't use the normal failure mechanisms,
2329    * since we don't know what state we are in.  We don't want to call
2330    * startFail()/finishFail() recursively if we are already in the middle of
2331    * cleaning up.
2332    */
2333
2334   AsyncSocketException ex(AsyncSocketException::ALREADY_OPEN,
2335                          "connect() called with socket in invalid state");
2336   connectEndTime_ = std::chrono::steady_clock::now();
2337   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
2338     if (callback) {
2339       callback->connectErr(ex);
2340     }
2341   } else {
2342     // We can't use failConnect() here since connectCallback_
2343     // may already be set to another callback.  Invoke this ConnectCallback
2344     // here; any other connectCallback_ will be invoked in finishFail()
2345     startFail();
2346     if (callback) {
2347       callback->connectErr(ex);
2348     }
2349     finishFail();
2350   }
2351 }
2352
2353 void AsyncSocket::invalidState(ErrMessageCallback* callback) {
2354   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
2355           << "): setErrMessageCB(" << callback
2356           << ") called in invalid state " << state_;
2357
2358   AsyncSocketException ex(
2359       AsyncSocketException::NOT_OPEN,
2360       msgErrQueueSupported
2361       ? "setErrMessageCB() called with socket in invalid state"
2362       : "This platform does not support socket error message notifications");
2363   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
2364     if (callback) {
2365       callback->errMessageError(ex);
2366     }
2367   } else {
2368     startFail();
2369     if (callback) {
2370       callback->errMessageError(ex);
2371     }
2372     finishFail();
2373   }
2374 }
2375
2376 void AsyncSocket::invokeConnectErr(const AsyncSocketException& ex) {
2377   connectEndTime_ = std::chrono::steady_clock::now();
2378   if (connectCallback_) {
2379     ConnectCallback* callback = connectCallback_;
2380     connectCallback_ = nullptr;
2381     callback->connectErr(ex);
2382   }
2383 }
2384
2385 void AsyncSocket::invokeConnectSuccess() {
2386   connectEndTime_ = std::chrono::steady_clock::now();
2387   if (connectCallback_) {
2388     ConnectCallback* callback = connectCallback_;
2389     connectCallback_ = nullptr;
2390     callback->connectSuccess();
2391   }
2392 }
2393
2394 void AsyncSocket::invalidState(ReadCallback* callback) {
2395   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
2396              << "): setReadCallback(" << callback
2397              << ") called in invalid state " << state_;
2398
2399   AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
2400                          "setReadCallback() called with socket in "
2401                          "invalid state");
2402   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
2403     if (callback) {
2404       callback->readErr(ex);
2405     }
2406   } else {
2407     startFail();
2408     if (callback) {
2409       callback->readErr(ex);
2410     }
2411     finishFail();
2412   }
2413 }
2414
2415 void AsyncSocket::invalidState(WriteCallback* callback) {
2416   VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
2417              << "): write() called in invalid state " << state_;
2418
2419   AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
2420                          withAddr("write() called with socket in invalid state"));
2421   if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
2422     if (callback) {
2423       callback->writeErr(0, ex);
2424     }
2425   } else {
2426     startFail();
2427     if (callback) {
2428       callback->writeErr(0, ex);
2429     }
2430     finishFail();
2431   }
2432 }
2433
2434 void AsyncSocket::doClose() {
2435   if (fd_ == -1) return;
2436   if (shutdownSocketSet_) {
2437     shutdownSocketSet_->close(fd_);
2438   } else {
2439     ::close(fd_);
2440   }
2441   fd_ = -1;
2442 }
2443
2444 std::ostream& operator << (std::ostream& os,
2445                            const AsyncSocket::StateEnum& state) {
2446   os << static_cast<int>(state);
2447   return os;
2448 }
2449
2450 std::string AsyncSocket::withAddr(const std::string& s) {
2451   // Don't use addr_ directly because it may not be initialized
2452   // e.g. if constructed from fd
2453   folly::SocketAddress peer, local;
2454   try {
2455     getPeerAddress(&peer);
2456     getLocalAddress(&local);
2457   } catch (const std::exception&) {
2458     // ignore
2459   } catch (...) {
2460     // ignore
2461   }
2462   return s + " (peer=" + peer.describe() + ", local=" + local.describe() + ")";
2463 }
2464
2465 void AsyncSocket::setBufferCallback(BufferCallback* cb) {
2466   bufferCallback_ = cb;
2467 }
2468
2469 } // folly