text: /**
   *    Copyright (C) 2015 MongoDB Inc.
   *
   *    This program is free software: you can redistribute it and/or  modify
   *    it under the terms of the GNU Affero General Public License, version 3,
   *    as published by the Free Software Foundation.
   *
   *    This program is distributed in the hope that it will be useful,
   *    but WITHOUT ANY WARRANTY; without even the implied warranty of
   *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   *    GNU Affero General Public License for more details.
   *
   *    You should have received a copy of the GNU Affero General Public License
   *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
   *
   *    As a special exception, the copyright holders give permission to link the
   *    code of portions of this program with the OpenSSL library under certain
   *    conditions as described in each individual source file and distribute
   *    linked combinations including the program with the OpenSSL library. You
   *    must comply with the GNU Affero General Public License in all respects for
   *    all of the code used other than as permitted herein. If you modify file(s)
   *    with this exception, you may extend this exception to your version of the
   *    file(s), but you are not obligated to do so. If you do not wish to do so,
   *    delete this exception statement from your version. If you delete this
   *    exception statement from all source files in the program, then also delete
   *    it in the license file.
   */
  #define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kASIO
  
  #include "mongo/platform/basic.h"
  
  #include "mongo/executor/network_interface_asio.h"
  
  #include <type_traits>
  #include <utility>
  
  #include "mongo/base/static_assert.h"
  #include "mongo/db/dbmessage.h"
  #include "mongo/db/jsobj.h"
  #include "mongo/executor/async_stream_interface.h"
  #include "mongo/executor/async_stream_interface.h"
  #include "mongo/executor/connection_pool_asio.h"
  #include "mongo/rpc/factory.h"
  #include "mongo/rpc/metadata/metadata_hook.h"
  #include "mongo/rpc/protocol.h"
  #include "mongo/rpc/reply_interface.h"
  #include "mongo/rpc/request_builder_interface.h"
  #include "mongo/stdx/memory.h"
  #include "mongo/util/assert_util.h"
  #include "mongo/util/fail_point_service.h"
  #include "mongo/util/log.h"
  #include "mongo/util/mongoutils/str.h"
  
  namespace mongo {
  namespace executor {
  
  /**
   * The following send - receive utility functions are "stateless" in that they exist
   * apart from the AsyncOp state machine.
   */
  
  namespace {
  
  MONGO_FP_DECLARE(NetworkInterfaceASIOasyncRunCommandFail);
  
  using asio::ip::tcp;
  using ResponseStatus = TaskExecutor::ResponseStatus;
  
  // A type conforms to the NetworkHandler concept if it is a callable type that takes a
  // std::error_code and std::size_t and returns void. The std::error_code parameter is used
  // to inform the handler if the asynchronous operation it was waiting on succeeded, and the
  // size_t parameter conveys how many bytes were read or written.
  template <typename FunctionLike>
  using IsNetworkHandler =
      std::is_convertible<FunctionLike, stdx::function<void(std::error_code, std::size_t)>>;
  
  template <typename Handler>
  void asyncSendMessage(AsyncStreamInterface& stream, Message* m, Handler&& handler) {
      MONGO_STATIC_ASSERT_MSG(
          IsNetworkHandler<Handler>::value,
          "Handler passed to asyncSendMessage does not conform to NetworkHandler concept");
      m->header().setResponseToMsgId(0);
      m->header().setId(nextMessageId());
      // TODO: Some day we may need to support vector messages.
      fassert(28708, m->buf() != 0);
      stream.write(asio::buffer(m->buf(), m->size()), std::forward<Handler>(handler));
  }
  
  template <typename Handler>
  void asyncRecvMessageHeader(AsyncStreamInterface& stream,
                              MSGHEADER::Value* header,
                              Handler&& handler) {
      MONGO_STATIC_ASSERT_MSG(
          IsNetworkHandler<Handler>::value,
          "Handler passed to asyncRecvMessageHeader does not conform to NetworkHandler concept");
      stream.read(asio::buffer(header->view().view2ptr(), sizeof(decltype(*header))),
                  std::forward<Handler>(handler));
  }
  
  template <typename Handler>
  void asyncRecvMessageBody(AsyncStreamInterface& stream,
                            MSGHEADER::Value* header,
                            Message* m,
                            Handler&& handler) {
      MONGO_STATIC_ASSERT_MSG(
          IsNetworkHandler<Handler>::value,
          "Handler passed to asyncRecvMessageBody does not conform to NetworkHandler concept");
      // validate message length
      int len = header->constView().getMessageLength();
      if (len == 542393671) {
          LOG(3) << "attempt to access MongoDB over HTTP on the native driver port.";
          return handler(make_error_code(ErrorCodes::ProtocolError), 0);
      } else if (static_cast<size_t>(len) < sizeof(MSGHEADER::Value) ||
                 static_cast<size_t>(len) > MaxMessageSizeBytes) {
          warning() << "recv(): message len " << len << " is invalid. "
                    << "Min " << sizeof(MSGHEADER::Value) << " Max: " << MaxMessageSizeBytes;
          return handler(make_error_code(ErrorCodes::InvalidLength), 0);
      }
  
      int z = (len + 1023) & 0xfffffc00;
      invariant(z >= len);
      m->setData(SharedBuffer::allocate(z));
      MsgData::View mdView = m->buf();
  
      // copy header data into master buffer
      int headerLen = sizeof(MSGHEADER::Value);
      memcpy(mdView.view2ptr(), header, headerLen);
      int bodyLength = len - headerLen;
      invariant(bodyLength >= 0);
  
      // receive remaining data into md->data
      stream.read(asio::buffer(mdView.data(), bodyLength), std::forward<Handler>(handler));
  }
  
  ResponseStatus decodeRPC(Message* received,
                           rpc::Protocol protocol,
                           Milliseconds elapsed,
                           const HostAndPort& source,
                           rpc::EgressMetadataHook* metadataHook) {
      try {
          // makeReply will throw if the reply is invalid
          auto reply = rpc::makeReply(received);
          if (reply->getProtocol() != protocol) {
              auto requestProtocol = rpc::toString(static_cast<rpc::ProtocolSet>(protocol));
              if (!requestProtocol.isOK())
                  return {requestProtocol.getStatus(), elapsed};
  
              return {ErrorCodes::RPCProtocolNegotiationFailed,
                      str::stream() << "Mismatched RPC protocols - request was '"
                                    << requestProtocol.getValue().toString()
                                    << "' '"
                                    << " but reply was '"
                                    << networkOpToString(received->operation())
                                    << "'",
                      elapsed};
          }
          auto commandReply = reply->getCommandReply();
          auto replyMetadata = reply->getMetadata();
  
          // Handle incoming reply metadata.
          if (metadataHook) {
              auto listenStatus = callNoexcept(*metadataHook,
                                               &rpc::EgressMetadataHook::readReplyMetadata,
                                               source.toString(),
                                               replyMetadata);
              if (!listenStatus.isOK()) {
                  return {listenStatus, elapsed};
              }
          }
  
          return {RemoteCommandResponse(
              std::move(*received), std::move(commandReply), std::move(replyMetadata), elapsed)};
      } catch (...) {
          return {exceptionToStatus(), elapsed};
      }
  }
  
  }  // namespace
  
  NetworkInterfaceASIO::AsyncCommand::AsyncCommand(AsyncConnection* conn,
                                                   Message&& command,
                                                   Date_t now,
                                                   const HostAndPort& target)
      : _conn(conn), _toSend(std::move(command)), _start(now), _target(target) {
      _toSend.header().setResponseToMsgId(0);
  }
  
  NetworkInterfaceASIO::AsyncConnection& NetworkInterfaceASIO::AsyncCommand::conn() {
      return *_conn;
  }
  
  Message& NetworkInterfaceASIO::AsyncCommand::toSend() {
      return _toSend;
  }
  
  Message& NetworkInterfaceASIO::AsyncCommand::toRecv() {
      return _toRecv;
  }
  
  MSGHEADER::Value& NetworkInterfaceASIO::AsyncCommand::header() {
      return _header;
  }
  
  ResponseStatus NetworkInterfaceASIO::AsyncCommand::response(AsyncOp* op,
                                                              rpc::Protocol protocol,
                                                              Date_t now,
                                                              rpc::EgressMetadataHook* metadataHook) {
      auto& received = _toRecv;
      if (received.operation() == dbCompressed) {
          auto swm = conn().getCompressorManager().decompressMessage(received);
          if (!swm.isOK()) {
              return swm.getStatus();
          }
          received = std::move(swm.getValue());
      }
  
      auto rs = decodeRPC(&received, protocol, now - _start, _target, metadataHook);
      if (rs.isOK())
          op->setResponseMetadata(rs.metadata);
      return rs;
  }
  
  void NetworkInterfaceASIO::_startCommand(AsyncOp* op) {
      LOG(3) << "running command " << redact(op->request().cmdObj) << " against database "
             << op->request().dbname << " across network to " << op->request().target.toString();
      if (inShutdown()) {
          return;
      }
  
      // _connect() will continue the state machine.
      _connect(op);
  }
  
  void NetworkInterfaceASIO::_beginCommunication(AsyncOp* op) {
      // The way that we connect connections for the connection pool is by
      // starting the callback chain with connect(), but getting off at the first
      // _beginCommunication. I.e. all AsyncOp's start off with _inSetup == true
      // and arrive here as they're connected and authed. Once they hit here, we
      // return to the connection pool's get() callback with _inSetup == false,
      // so we can proceed with user operations after they return to this
      // codepath.
      if (op->_inSetup) {
          auto host = op->request().target;
          auto getConnectionDuration = now() - op->start();
          log() << "Successfully connected to " << host << ", took " << getConnectionDuration << " ("
                << _connectionPool.getNumConnectionsPerHost(host) << " connections now open to "
                << host << ")";
          op->_inSetup = false;
          op->finish(RemoteCommandResponse());
          return;
      }
  
      LOG(3) << "Initiating asynchronous command: " << redact(op->request().toString());
  
      auto beginStatus = op->beginCommand(op->request());
      if (!beginStatus.isOK()) {
          return _completeOperation(op, beginStatus);
      }
  
      _asyncRunCommand(op, [this, op](std::error_code ec, size_t bytes) {
          _validateAndRun(op, ec, [this, op]() { _completedOpCallback(op); });
      });
  }
  
  void NetworkInterfaceASIO::_completedOpCallback(AsyncOp* op) {
      auto response =
          op->command()->response(op, op->operationProtocol(), now(), _metadataHook.get());
      _completeOperation(op, response);
  }
  
  void NetworkInterfaceASIO::_networkErrorCallback(AsyncOp* op, const std::error_code& ec) {
      ErrorCodes::Error errorCode = (ec.category() == mongoErrorCategory())
          ? ErrorCodes::fromInt(ec.value())
          : ErrorCodes::HostUnreachable;
      _completeOperation(op, {errorCode, ec.message(), Milliseconds(now() - op->_start)});
  }
  
  // NOTE: This method may only be called by ASIO threads
  // (do not call from methods entered by TaskExecutor threads)
  void NetworkInterfaceASIO::_completeOperation(AsyncOp* op, ResponseStatus resp) {
      auto metadata = op->getResponseMetadata();
      if (!metadata.isEmpty()) {
          resp.metadata = metadata;
      }
  
      // Cancel this operation's timeout. Note that the timeout callback may already be running,
      // may have run, or may have already been scheduled to run in the near future.
      if (op->_timeoutAlarm) {
          op->_timeoutAlarm->cancel();
      }
  
      if (resp.status.code() == ErrorCodes::ExceededTimeLimit ||
          resp.status.code() == ErrorCodes::NetworkInterfaceExceededTimeLimit) {
          _numTimedOutOps.fetchAndAdd(1);
      }
  
      if (op->_inSetup) {
          // If we are in setup we should only be here if we failed to connect.
          MONGO_ASIO_INVARIANT(!resp.isOK(), "Failed to connect in setup", op);
          // If we fail during connection, we won't be able to access any of op's members after
          // calling finish(), so we return here.
          log() << "Failed to connect to " << op->request().target << " - " << resp.status;
          op->finish(std::move(resp));
          return;
      }
  
      if (op->_inRefresh) {
          // If we are in refresh we should only be here if we failed to heartbeat.
          MONGO_ASIO_INVARIANT(!resp.isOK(), "In refresh, but did not fail to heartbeat", op);
          // If we fail during heartbeating, we won't be able to access any of op's members after
          // calling finish(), so we return here.
          log() << "Failed asio heartbeat to " << op->request().target << " - "
                << redact(resp.status);
          _numFailedOps.fetchAndAdd(1);
          op->finish(std::move(resp));
          return;
      }
  
      if (!resp.isOK()) {
          // In the case that resp is not OK, but _inSetup is false, we are using a connection
          // that
          // we got from the pool to execute a command, but it failed for some reason.
          LOG(2) << "Failed to execute command: " << redact(op->request().toString())
                 << " reason: " << redact(resp.status);
  
          if (resp.status.code() != ErrorCodes::CallbackCanceled) {
              _numFailedOps.fetchAndAdd(1);
          }
      } else {
          _numSucceededOps.fetchAndAdd(1);
      }
  
      std::unique_ptr<AsyncOp> ownedOp;
  
      {
          stdx::lock_guard<stdx::mutex> lk(_inProgressMutex);
  
          auto iter = _inProgress.find(op);
  
          MONGO_ASIO_INVARIANT_INLOCK(
              iter != _inProgress.end(), "Could not find AsyncOp in _inProgress", op);
  
          ownedOp = std::move(iter->second);
          _inProgress.erase(iter);
      }
  
      op->finish(std::move(resp));
  
      MONGO_ASIO_INVARIANT(static_cast<bool>(ownedOp), "Invalid AsyncOp", op);
  
      auto conn = std::move(op->_connectionPoolHandle);
      auto asioConn = static_cast<connection_pool_asio::ASIOConnection*>(conn.get());
  
      // Prevent any other threads or callbacks from accessing this op so we may safely complete
      // and destroy it. It is key that we do this after we remove the op from the _inProgress map
      // or someone else in cancelCommand could read the bumped generation and cancel the next
      // command that uses this op. See SERVER-20556.
      {
          stdx::lock_guard<stdx::mutex> lk(op->_access->mutex);
          ++(op->_access->id);
      }
  
      // We need to bump the generation BEFORE we call reset() or we could flip the timeout in the
      // timeout callback before returning the AsyncOp to the pool.
      ownedOp->reset();
  
      asioConn->bindAsyncOp(std::move(ownedOp));
      if (!resp.isOK()) {
          asioConn->indicateFailure(resp.status);
      } else {
          asioConn->indicateUsed();
          asioConn->indicateSuccess();
      }
  
      signalWorkAvailable();
  }
  
  void NetworkInterfaceASIO::_asyncRunCommand(AsyncOp* op, NetworkOpHandler handler) {
      LOG(2) << "Starting asynchronous command " << op->request().id << " on host "
             << op->request().target.toString();
  
      if (MONGO_FAIL_POINT(NetworkInterfaceASIOasyncRunCommandFail)) {
          _validateAndRun(op, asio::error::basic_errors::network_unreachable, [] {});
          return;
      }
  
      // We invert the following steps below to run a command:
      // 1 - send the given command
      // 2 - receive a header for the response
      // 3 - validate and receive response body
      // 4 - advance the state machine by calling handler()
      auto cmd = op->command();
  
      // Step 4
      auto recvMessageCallback = [this, cmd, handler, op](std::error_code ec, size_t bytes) {
          // We don't call _validateAndRun here as we assume the caller will.
          handler(ec, bytes);
      };
  
      // Step 3
      auto recvHeaderCallback = [this, cmd, handler, recvMessageCallback, op](std::error_code ec,
                                                                              size_t bytes) {
          // The operation could have been canceled after starting the command, but before
          // receiving the header
          _validateAndRun(op, ec, [this, op, recvMessageCallback, ec, bytes, cmd, handler] {
              // validate response id
              uint32_t expectedId = cmd->toSend().header().getId();
              uint32_t actualId = cmd->header().constView().getResponseToMsgId();
              if (actualId != expectedId) {
                  LOG(3) << "got wrong response:"
                         << " expected response id: " << expectedId
                         << ", got response id: " << actualId;
                  return handler(make_error_code(ErrorCodes::ProtocolError), bytes);
              }
  
              asyncRecvMessageBody(cmd->conn().stream(),
                                   &cmd->header(),
                                   &cmd->toRecv(),
                                   std::move(recvMessageCallback));
          });
      };
  
      // Step 2
      auto sendMessageCallback = [this, cmd, handler, recvHeaderCallback, op](std::error_code ec,
                                                                              size_t bytes) {
          _validateAndRun(op, ec, [this, cmd, op, recvHeaderCallback] {
              asyncRecvMessageHeader(
                  cmd->conn().stream(), &cmd->header(), std::move(recvHeaderCallback));
          });
  
  
      };
  
      // Step 1
      asyncSendMessage(cmd->conn().stream(), &cmd->toSend(), std::move(sendMessageCallback));
  }
  
  void NetworkInterfaceASIO::_runConnectionHook(AsyncOp* op) {
      if (!_hook) {
          return _beginCommunication(op);
      }
  
      auto swOptionalRequest =
          callNoexcept(*_hook, &NetworkConnectionHook::makeRequest, op->request().target);
  
      if (!swOptionalRequest.isOK()) {
          return _completeOperation(op, swOptionalRequest.getStatus());
      }
  
      auto optionalRequest = std::move(swOptionalRequest.getValue());
  
      if (optionalRequest == boost::none) {
          return _beginCommunication(op);
      }
  
      auto beginStatus = op->beginCommand(*optionalRequest);
      if (!beginStatus.isOK()) {
          return _completeOperation(op, beginStatus);
      }
  
      auto finishHook = [this, op]() {
          auto response =
              op->command()->response(op, op->operationProtocol(), now(), _metadataHook.get());
  
          if (!response.isOK()) {
              return _completeOperation(op, response);
          }
  
          auto handleStatus = callNoexcept(
              *_hook, &NetworkConnectionHook::handleReply, op->request().target, std::move(response));
  
          if (!handleStatus.isOK()) {
              return _completeOperation(op, handleStatus);
          }
  
          return _beginCommunication(op);
      };
  
      return _asyncRunCommand(op, [this, op, finishHook](std::error_code ec, std::size_t bytes) {
          _validateAndRun(op, ec, finishHook);
      });
  }
  
  
  }  // namespace executor
  }  // namespace mongo
  