tcp_client.cc

Go to the documentation of this file.

// Copyright (C) 2018-2026 Internet Systems Consortium, Inc. ("ISC")
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
#include <config.h>
 
#include <asiolink/asio_wrapper.h>
#include <asiolink/interval_timer.h>
#include <asiolink/io_service_thread_pool.h>
#include <asiolink/tls_socket.h>
#include <tcp/tcp_client.h>
#include <tcp/tcp_log.h>
#include <tcp/tcp_messages.h>
#include <util/boost_time_utils.h>
#include <util/str.h>
#include <util/multi_threading_mgr.h>
#include <util/unlock_guard.h>
 
#include <boost/enable_shared_from_this.hpp>
#include <boost/weak_ptr.hpp>
 
#include <atomic>
#include <array>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <queue>
#include <thread>
 
using namespace isc;
using namespace isc::asiolink;
using namespace isc::tcp;
using namespace isc::util;
using namespace boost::posix_time;
 
namespace ph = std::placeholders;
 
namespace {

typedef std::function<void(boost::system::error_code ec, size_t length)>
SocketCallbackFunction;

class SocketCallback {
public:

    SocketCallback(SocketCallbackFunction socket_callback)
        : callback_(socket_callback) {
    }

    void operator()(boost::system::error_code ec, size_t length = 0) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;
        }
        callback_(ec, length);
    }
 
private:

    SocketCallbackFunction callback_;
 
};
 
class ConnectionPool;

typedef boost::shared_ptr<ConnectionPool> ConnectionPoolPtr;

class Connection : public boost::enable_shared_from_this<Connection> {
public:

    explicit Connection(const IOServicePtr& io_service,
                        const TlsContextPtr& tls_context,
                        const ConnectionPoolPtr& conn_pool,
                        const IOAddress& address,
                        const uint16_t port);

    ~Connection();

    void doTransaction(const WireDataPtr& request,
                       const WireDataPtr& response,
                       const bool persistent,
                       const long request_timeout,
                       const TcpClient::CompleteCheck& complete_check,
                       const TcpClient::RequestHandler& callback,
                       const TcpClient::ConnectHandler& connect_callback,
                       const TcpClient::HandshakeHandler& handshake_callback,
                       const TcpClient::CloseHandler& close_callback);

    void close();

    bool isTransactionOngoing() const {
        return (started_);
    }

    bool isClosed() const {
        return (closed_);
    }

    void isClosedByPeer();

    bool isMySocket(int socket_fd) const;

    bool checkPrematureTimeout(const uint64_t transid);
 
private:

    void doTransactionInternal(const WireDataPtr& request,
                               const WireDataPtr& response,
                               const bool persistent,
                               const long request_timeout,
                               const TcpClient::CompleteCheck& complete_check,
                               const TcpClient::RequestHandler& callback,
                               const TcpClient::ConnectHandler& connect_callback,
                               const TcpClient::HandshakeHandler& handshake_callback,
                               const TcpClient::CloseHandler& close_callback);

    void closeInternal();

    void isClosedByPeerInternal();

    bool checkPrematureTimeoutInternal(const uint64_t transid);

    void resetState();

    void terminate(const boost::system::error_code& ec,
                   const std::string& error_msg = "");

    void terminateInternal(const boost::system::error_code& ec,
                           std::string error_msg = "");

    bool runCompleteCheck(const boost::system::error_code& ec, size_t length);

    bool runCompleteCheckInternal(const boost::system::error_code& ec, size_t length);

    void scheduleTimer(const long request_timeout);

    void doHandshake(const uint64_t transid);

    void doSend(const uint64_t transid);

    void doReceive(const uint64_t transid);

    void connectCallback(TcpClient::ConnectHandler connect_callback,
                         const uint64_t transid,
                         const boost::system::error_code& ec);

    void handshakeCallback(TcpClient::HandshakeHandler handshake_callback,
                           const uint64_t transid,
                           const boost::system::error_code& ec);

    void sendCallback(const uint64_t transid, const boost::system::error_code& ec,
                      size_t length);

    void receiveCallback(const uint64_t transid,
                         const boost::system::error_code& ec,
                         size_t length);

    void timerCallback();

    void closeCallback(const bool clear = false);

    IOServicePtr io_service_;

    boost::weak_ptr<ConnectionPool> conn_pool_;

    IOAddress address_;

    uint16_t port_;

    TlsContextPtr tls_context_;

    std::shared_ptr<TCPSocket<SocketCallback>> tcp_socket_;

    std::shared_ptr<TLSSocket<SocketCallback>> tls_socket_;

    IntervalTimerPtr timer_;

    WireDataPtr current_request_;

    WireDataPtr current_response_;

    bool current_persistent_;

    bool current_response_complete_;

    TcpClient::CompleteCheck current_complete_check_;

    TcpClient::RequestHandler current_callback_;

    std::vector<uint8_t>  buf_;

    std::array<uint8_t, 32768> input_buf_;

    uint64_t current_transid_;

    TcpClient::HandshakeHandler handshake_callback_;

    TcpClient::CloseHandler close_callback_;

    std::atomic<bool> started_;

    std::atomic<bool> need_handshake_;

    std::atomic<bool> closed_;

    std::mutex mutex_;
};

typedef boost::shared_ptr<Connection> ConnectionPtr;

class ConnectionPool : public boost::enable_shared_from_this<ConnectionPool> {
public:

    explicit ConnectionPool(const IOServicePtr& io_service, size_t max_addr_connections)
        : io_service_(io_service), destinations_(), pool_mutex_(),
          max_addr_connections_(max_addr_connections) {
    }

    ~ConnectionPool() {
        closeAll();
    }

    void processNextRequest(const IOAddress& address,
                            const uint16_t port,
                            const TlsContextPtr& tls_context) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            return (processNextRequestInternal(address, port, tls_context));
        } else {
            return (processNextRequestInternal(address, port, tls_context));
        }
    }

    void postProcessNextRequest(const IOAddress& address,
                                const uint16_t port,
                                const TlsContextPtr& tls_context) {
        io_service_->post(std::bind(&ConnectionPool::processNextRequest,
                                    shared_from_this(),
                                    address, port,
                                    tls_context));
    }

    void queueRequest(const IOAddress& address,
                      const uint16_t port,
                      const TlsContextPtr& tls_context,
                      const WireDataPtr& request,
                      const WireDataPtr& response,
                      const bool persistent,
                      const long request_timeout,
                      const TcpClient::CompleteCheck& complete_check,
                      const TcpClient::RequestHandler& request_callback,
                      const TcpClient::ConnectHandler& connect_callback,
                      const TcpClient::HandshakeHandler& handshake_callback,
                      const TcpClient::CloseHandler& close_callback) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            return (queueRequestInternal(address, port, tls_context,
                                         request, response, persistent,
                                         request_timeout, complete_check,
                                         request_callback, connect_callback,
                                         handshake_callback, close_callback));
        } else {
            return (queueRequestInternal(address, port, tls_context,
                                         request, response, persistent,
                                         request_timeout, complete_check,
                                         request_callback, connect_callback,
                                         handshake_callback, close_callback));
        }
    }

    void closeAll() {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            closeAllInternal();
        } else {
            closeAllInternal();
        }
    }

    void closeIfOutOfBand(int socket_fd) {
        if (MultiThreadingMgr::instance().getMode()) {
            std::lock_guard<std::mutex> lk(pool_mutex_);
            closeIfOutOfBandInternal(socket_fd);
        } else {
            closeIfOutOfBandInternal(socket_fd);
        }
    }
 
private:

    void processNextRequestInternal(const IOAddress& address,
                                    const uint16_t port,
                                    const TlsContextPtr& tls_context) {
        // Check if there is a queue for this address. If there is no queue,
        // there is no request queued either.
        DestinationPtr destination = findDestination(address, port, tls_context);
        if (destination) {
            // Remove closed connections.
            destination->garbageCollectConnections();
            if (!destination->queueEmpty()) {
                // We have at least one queued request. Do we have an
                // idle connection?
                ConnectionPtr connection = destination->getIdleConnection();
                if (!connection) {
                    // No idle connections.
                    if (destination->connectionsFull()) {
                        return;
                    }
                    // Room to make another connection with this destination,
                    // so make one.
                    connection.reset(new Connection(io_service_,
                                                    tls_context,
                                                    shared_from_this(),
                                                    address, port));
                    destination->addConnection(connection);
                }
 
                // Dequeue the oldest request and start a transaction for it using
                // the idle connection.
                RequestDescriptor desc = destination->popNextRequest();
                connection->doTransaction(desc.request_,
                                          desc.response_,
                                          desc.persistent_,
                                          desc.request_timeout_,
                                          desc.complete_check_,
                                          desc.callback_,
                                          desc.connect_callback_,
                                          desc.handshake_callback_,
                                          desc.close_callback_);
            }
        }
    }

    void queueRequestInternal(const IOAddress& address,
                              const uint16_t port,
                              const TlsContextPtr& tls_context,
                              const WireDataPtr& request,
                              const WireDataPtr& response,
                              const bool persistent,
                              const long request_timeout,
                              const TcpClient::CompleteCheck& complete_check,
                              const TcpClient::RequestHandler& request_callback,
                              const TcpClient::ConnectHandler& connect_callback,
                              const TcpClient::HandshakeHandler& handshake_callback,
                              const TcpClient::CloseHandler& close_callback) {
        ConnectionPtr connection;
        // Find the destination for the requested address.
        DestinationPtr destination = findDestination(address, port, tls_context);
        if (destination) {
            // Remove closed connections.
            destination->garbageCollectConnections();
            // Found it, look for an idle connection.
            connection = destination->getIdleConnection();
        } else {
            // Doesn't exist yet so it's a new destination.
            destination = addDestination(address, port, tls_context);
        }
 
        if (!connection) {
            if (destination->connectionsFull()) {
                // All connections busy, queue it.
                destination->pushRequest(RequestDescriptor(request,
                                                           response,
                                                           persistent,
                                                           request_timeout,
                                                           complete_check,
                                                           request_callback,
                                                           connect_callback,
                                                           handshake_callback,
                                                           close_callback));
                return;
            }
 
            // Room to make another connection with this destination, so make one.
            connection.reset(new Connection(io_service_, tls_context,
                                            shared_from_this(),
                                            address, port));
            destination->addConnection(connection);
        }
 
        // Use the connection to start the transaction.
        connection->doTransaction(request, response, persistent,
                                  request_timeout, complete_check,
                                  request_callback, connect_callback,
                                  handshake_callback, close_callback);
    }

    void closeAllInternal() {
        for (auto const& destination : destinations_) {
            destination.second->closeAllConnections();
        }
 
        destinations_.clear();
    }

    void closeIfOutOfBandInternal(int socket_fd) {
        for (auto const& destination : destinations_) {
            // First we look for a connection with the socket.
            ConnectionPtr connection = destination.second->findBySocketFd(socket_fd);
            if (connection) {
                if (!connection->isTransactionOngoing()) {
                    // Socket has no transaction, so any ready event is
                    // out-of-band (other end probably closed), so
                    // let's close it.  Note we do not remove any queued
                    // requests, as this might somehow be occurring in
                    // between them.
                    destination.second->closeConnection(connection);
                }
 
                return;
            }
        }
    }

    struct RequestDescriptor {
        RequestDescriptor(const WireDataPtr& request,
                          const WireDataPtr& response,
                          const bool persistent,
                          const long& request_timeout,
                          const TcpClient::CompleteCheck& complete_check,
                          const TcpClient::RequestHandler& callback,
                          const TcpClient::ConnectHandler& connect_callback,
                          const TcpClient::HandshakeHandler& handshake_callback,
                          const TcpClient::CloseHandler& close_callback)
            : request_(request),
              response_(response),
              persistent_(persistent),
              request_timeout_(request_timeout),
              complete_check_(complete_check),
              callback_(callback),
              connect_callback_(connect_callback),
              handshake_callback_(handshake_callback),
              close_callback_(close_callback) {
        }

        WireDataPtr request_;

        WireDataPtr response_;

        bool persistent_;

        long request_timeout_;

        TcpClient::CompleteCheck complete_check_;

        TcpClient::RequestHandler callback_;

        TcpClient::ConnectHandler connect_callback_;

        TcpClient::HandshakeHandler handshake_callback_;

        TcpClient::CloseHandler close_callback_;
    };

    struct DestinationDescriptor {
        // Constructor.
        DestinationDescriptor(const IOAddress& address,
                              const uint16_t port,
                              const TlsContextPtr& tls_context)
            : address_(address), port_(port), tls_context_(tls_context) {
        }
 
        // Members.
        IOAddress address_;
        uint16_t port_;
        TlsContextPtr tls_context_;
 
        // Compare method.
        bool operator<(const DestinationDescriptor& other) const {
            return ((address_ < other.address_) ||
                    ((address_ == other.address_) && (port_ < other.port_)) ||
                    ((address_ == other.address_) && (port_ == other.port_) &&
                     (tls_context_ < other.tls_context_)));
        }
    };

    class Destination {
    public:
        const size_t QUEUE_SIZE_THRESHOLD = 2048;
        const int QUEUE_WARN_SECS = 5;

        Destination(IOAddress const& address,
                    uint16_t const port,
                    TlsContextPtr tls_context,
                    size_t max_connections)
            : address_(address), port_(port), tls_context_(tls_context),
              max_connections_(max_connections), connections_(), queue_(),
              last_queue_warn_time_(min_date_time), last_queue_size_(0) {
        }

        ~Destination() {
            closeAllConnections();
        }

        void addConnection(ConnectionPtr connection) {
            if (connectionsFull()) {
                isc_throw(BadValue, "address: " << address_.toText()
                      << ", already at maximum connections: "
                      << max_connections_);
            }
 
            connections_.push_back(connection);
        }

        void closeConnection(ConnectionPtr connection) {
            for (auto it = connections_.begin(); it != connections_.end(); ++it) {
                if (*it == connection) {
                    (*it)->close();
                    connections_.erase(it);
                    break;
                }
            }
        }

        void closeAllConnections() {
            // Flush the queue.
            while (!queue_.empty()) {
                queue_.pop();
            }
 
            for (auto const& connection : connections_) {
                connection->close();
            }
 
            connections_.clear();
        }

        void garbageCollectConnections() {
            for (auto it = connections_.begin(); it != connections_.end();) {
                (*it)->isClosedByPeer();
                if (!(*it)->isClosed()) {
                    ++it;
                } else {
                    it = connections_.erase(it);
                }
            }
        }

        ConnectionPtr getIdleConnection() {
            for (auto const& connection : connections_) {
                if (!connection->isTransactionOngoing() &&
                    !connection->isClosed()) {
                    return (connection);
                }
            }
 
            return (ConnectionPtr());
        }

        ConnectionPtr findBySocketFd(int socket_fd) {
            for (auto const& connection : connections_) {
                if (connection->isMySocket(socket_fd)) {
                    return (connection);
                }
            }
 
            return (ConnectionPtr());
        }

        bool connectionsEmpty() {
            return (connections_.empty());
        }

        bool connectionsFull() {
            return (connections_.size() >= max_connections_);
        }

        size_t connectionCount() {
            return (connections_.size());
        }

        size_t getMaxConnections() const {
            return (max_connections_);
        }

        bool queueEmpty() const {
            return (queue_.empty());
        }

        void pushRequest(RequestDescriptor const& desc) {
            queue_.push(desc);
            size_t size = queue_.size();
            // If the queue size is larger than the threshold and growing,
            // issue a periodic warning.
            if ((size > QUEUE_SIZE_THRESHOLD) && (size > last_queue_size_)) {
                ptime now = microsec_clock::universal_time();
                if ((now - last_queue_warn_time_) > seconds(QUEUE_WARN_SECS)) {
                    LOG_WARN(tcp_logger, TCP_CLIENT_QUEUE_SIZE_GROWING)
                        .arg(address_.toText())
                        .arg(port_)
                        .arg(size);
                    // Remember the last time we warned.
                    last_queue_warn_time_ = now;
                }
            }
 
            // Remember the previous size.
            last_queue_size_ = size;
        }

        RequestDescriptor popNextRequest() {
            if (queue_.empty()) {
                isc_throw(InvalidOperation, "cannot pop, queue is empty");
            }
 
            RequestDescriptor desc = queue_.front();
            queue_.pop();
            return (desc);
        }
 
    private:
        IOAddress address_;

        uint16_t port_;

        TlsContextPtr tls_context_;

        size_t max_connections_;

        std::list<ConnectionPtr> connections_;

        std::queue<RequestDescriptor> queue_;

        ptime last_queue_warn_time_;

        size_t last_queue_size_;
    };

    typedef boost::shared_ptr<Destination> DestinationPtr;

    DestinationPtr addDestination(const IOAddress& address,
                                  const uint16_t port,
                                  const TlsContextPtr& tls_context) {
        DestinationDescriptor desc(address, port, tls_context);
        DestinationPtr destination(new Destination(address, port, tls_context,
                                                   max_addr_connections_));
        destinations_[desc] = destination;
        return (destination);
    }

    DestinationPtr findDestination(const IOAddress& address,
                                   const uint16_t port,
                                   const TlsContextPtr& tls_context) const {
        DestinationDescriptor desc(address, port, tls_context);
        auto it = destinations_.find(desc);
        if (it != destinations_.end()) {
            return (it->second);
        }
 
        return (DestinationPtr());
    }

    void removeDestination(const IOAddress& address,
                           const uint16_t port,
                           const TlsContextPtr& tls_context) {
        DestinationDescriptor desc(address, port, tls_context);
        auto it = destinations_.find(desc);
        if (it != destinations_.end()) {
            it->second->closeAllConnections();
            destinations_.erase(it);
        }
    }

    IOServicePtr io_service_;

    std::map<DestinationDescriptor, DestinationPtr> destinations_;

    std::mutex pool_mutex_;

    size_t max_addr_connections_;
};
 
Connection::Connection(const IOServicePtr& io_service,
                       const TlsContextPtr& tls_context,
                       const ConnectionPoolPtr& conn_pool,
                       const IOAddress& address,
                       const uint16_t port)
    : io_service_(io_service), conn_pool_(conn_pool), address_(address),
      port_(port), tls_context_(tls_context), tcp_socket_(), tls_socket_(),
      timer_(new IntervalTimer(io_service)), current_request_(),
      current_response_(), current_persistent_(false),
      current_response_complete_(false), current_complete_check_(),
      current_callback_(), buf_(), input_buf_(), current_transid_(0),
      close_callback_(), started_(false), need_handshake_(false),
      closed_(false) {
    if (!tls_context) {
        tcp_socket_.reset(new asiolink::TCPSocket<SocketCallback>(io_service));
    } else {
        tls_socket_.reset(new asiolink::TLSSocket<SocketCallback>(io_service,
                                                                  tls_context));
        need_handshake_ = true;
    }
}
 
Connection::~Connection() {
    close();
}
 
void
Connection::resetState() {
    started_ = false;
    current_request_.reset();
    current_response_.reset();
    current_persistent_ = false;
    current_response_complete_ = false;
    current_callback_ = TcpClient::RequestHandler();
}
 
void
Connection::closeCallback(const bool clear) {
    if (close_callback_) {
        try {
            if (tcp_socket_) {
                close_callback_(tcp_socket_->getNative());
            } else if (tls_socket_) {
                close_callback_(tls_socket_->getNative());
            } else {
                isc_throw(Unexpected,
                          "internal error: can't find a socket to close");
            }
        } catch (...) {
            LOG_ERROR(tcp_logger, TCP_CLIENT_CONNECTION_CLOSE_CALLBACK_FAILED);
        }
    }
 
    if (clear) {
        close_callback_ = TcpClient::CloseHandler();
    }
}
 
void
Connection::isClosedByPeer() {
    // This method applies only to idle connections.
    if (started_ || closed_) {
        return;
    }
    // This code was guarded by a lock so keep this.
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        isClosedByPeerInternal();
    } else {
        isClosedByPeerInternal();
    }
}
 
void
Connection::isClosedByPeerInternal() {
    // If the socket is open we check if it is possible to transmit
    // the data over this socket by reading from it with message
    // peeking. If the socket is not usable, we close it and then
    // re-open it. There is a narrow window of time between checking
    // the socket usability and actually transmitting the data over
    // this socket, when the peer may close the connection. In this
    // case we'll need to re-transmit but we don't handle it here.
    if (tcp_socket_) {
        if (tcp_socket_->getASIOSocket().is_open() &&
            !tcp_socket_->isUsable()) {
            closeCallback();
            closed_ = true;
            tcp_socket_->close();
        }
    } else if (tls_socket_) {
        if (tls_socket_->getASIOSocket().is_open() &&
            !tls_socket_->isUsable()) {
            closeCallback();
            closed_ = true;
            tls_socket_->close();
        }
    } else {
        isc_throw(Unexpected, "internal error: can't find the sending socket");
    }
}
 
void
Connection::doTransaction(const WireDataPtr& request,
                          const WireDataPtr& response,
                          const bool persistent,
                          const long request_timeout,
                          const TcpClient::CompleteCheck& complete_check,
                          const TcpClient::RequestHandler& callback,
                          const TcpClient::ConnectHandler& connect_callback,
                          const TcpClient::HandshakeHandler& handshake_callback,
                          const TcpClient::CloseHandler& close_callback) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        doTransactionInternal(request, response, persistent, request_timeout,
                              complete_check, callback, connect_callback,
                              handshake_callback, close_callback);
    } else {
        doTransactionInternal(request, response, persistent, request_timeout,
                              complete_check, callback, connect_callback,
                              handshake_callback, close_callback);
    }
}
 
void
Connection::doTransactionInternal(const WireDataPtr& request,
                                  const WireDataPtr& response,
                                  const bool persistent,
                                  const long request_timeout,
                                  const TcpClient::CompleteCheck& complete_check,
                                  const TcpClient::RequestHandler& callback,
                                  const TcpClient::ConnectHandler& connect_callback,
                                  const TcpClient::HandshakeHandler& handshake_callback,
                                  const TcpClient::CloseHandler& close_callback) {
    try {
        started_ = true;
        current_request_ = request;
        current_response_ = response;
        current_persistent_ = persistent;
        current_complete_check_ = complete_check;
        current_callback_ = callback;
        handshake_callback_ = handshake_callback;
        close_callback_ = close_callback;
 
        // Starting new transaction. Generate new transaction id.
        ++current_transid_;
 
        buf_ = *request;
 
        size_t to_dump = request->size();
        bool truncated = false;
        if (to_dump > 100) {
            to_dump = 100;
            truncated = true;
        }
        LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_DETAIL,
                  TCP_CLIENT_REQUEST_SEND)
            .arg(str::dumpAsHex(request->data(), to_dump) +
                 (truncated ? "..." : ""))
            .arg(address_.toText())
            .arg(port_);
 
        // Setup request timer.
        scheduleTimer(request_timeout);

        TCPEndpoint endpoint(address_, port_);
        SocketCallback socket_cb(std::bind(&Connection::connectCallback,
                                           shared_from_this(),
                                           connect_callback,
                                           current_transid_,
                                           ph::_1));
 
        // Establish new connection or use existing connection.
        if (tcp_socket_) {
            tcp_socket_->open(&endpoint, socket_cb);
            return;
        }
        if (tls_socket_) {
            tls_socket_->open(&endpoint, socket_cb);
            return;
        }
 
        // Should never reach this point.
        isc_throw(Unexpected, "internal error: can't find a socket to open");
 
    } catch (const std::exception& ex) {
        // Re-throw with the expected exception type.
        isc_throw(TcpClientError, ex.what());
    }
}
 
void
Connection::close() {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (closeInternal());
    } else {
        return (closeInternal());
    }
}
 
void
Connection::closeInternal() {
    // Pass in true to discard the callback.
    closeCallback(true);
 
    closed_ = true;
    timer_->cancel();
    if (tcp_socket_) {
        tcp_socket_->close();
    }
    if (tls_socket_) {
        tls_socket_->close();
    }
 
    resetState();
}
 
bool
Connection::isMySocket(int socket_fd) const {
    if (tcp_socket_) {
        return (tcp_socket_->getNative() == socket_fd);
    } else if (tls_socket_) {
        return (tls_socket_->getNative() == socket_fd);
    }
    // Should never reach this point.
    std::cerr << "internal error: can't find my socket\n";
    return (false);
}
 
bool
Connection::checkPrematureTimeout(const uint64_t transid) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (checkPrematureTimeoutInternal(transid));
    } else {
        return (checkPrematureTimeoutInternal(transid));
    }
}
 
bool
Connection::checkPrematureTimeoutInternal(const uint64_t transid) {
    // If there is no transaction but the handlers are invoked it means
    // that the last transaction in the queue timed out prematurely.
    // Also, if there is a transaction in progress but the ID of that
    // transaction doesn't match the one associated with the handler it,
    // also means that the transaction timed out prematurely.
    if (!isTransactionOngoing() || (transid != current_transid_)) {
        LOG_WARN(tcp_logger, TCP_CLIENT_PREMATURE_CONNECTION_TIMEOUT_OCCURRED)
                .arg(isTransactionOngoing())
                .arg(transid)
                .arg(current_transid_);
        return (true);
    }
 
    return (false);
}
 
void
Connection::terminate(const boost::system::error_code& ec,
                      const std::string& error_msg) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        terminateInternal(ec, error_msg);
    } else {
        terminateInternal(ec, error_msg);
    }
}
 
void
Connection::terminateInternal(const boost::system::error_code& ec,
                              std::string error_msg) {
    WireDataPtr response;
    if (isTransactionOngoing()) {
 
        timer_->cancel();
        if (tcp_socket_) {
            tcp_socket_->cancel();
        }
        if (tls_socket_) {
            tls_socket_->cancel();
        }
 
        if (!ec && current_response_complete_) {
            response = current_response_;
 
            LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_BASIC,
                      TCP_CLIENT_SERVER_RESPONSE_RECEIVED)
                .arg(address_.toText())
                .arg(port_);
        } else {
            if (error_msg.empty()) {
                error_msg = ec.message();
            }
            LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_BASIC,
                      TCP_CLIENT_BAD_SERVER_RESPONSE_RECEIVED)
                .arg(address_.toText())
                .arg(port_)
                .arg(error_msg);
 
            // Only log the details if we have received anything.
            if (!current_response_->empty()) {
                size_t to_dump = current_response_->size();
                bool truncated = false;
                if (to_dump > 100) {
                    to_dump = 100;
                    truncated = true;
                }
                LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_BASIC_DATA,
                          TCP_CLIENT_BAD_SERVER_RESPONSE_RECEIVED_DETAILS)
                    .arg(address_.toText())
                    .arg(port_)
                    .arg(str::dumpAsHex(current_response_->data(), to_dump) +
                         (truncated ? "..." : ""));
            }
        }
 
        try {
            // The callback should take care of its own exceptions but one
            // never knows.
            if (MultiThreadingMgr::instance().getMode()) {
                UnlockGuard<std::mutex> lock(mutex_);
                current_callback_(ec, response, error_msg);
            } else {
                current_callback_(ec, response, error_msg);
            }
        } catch (...) {
        }
 
        // If we're not requesting connection persistence or the
        // connection has timed out, we should close the socket.
        if (!closed_ &&
            (!current_persistent_ || (ec == boost::asio::error::timed_out))) {
            closeInternal();
        }
 
        resetState();
    }
 
    // Check if there are any requests queued for this destination and start
    // another transaction if there is at least one.
    ConnectionPoolPtr conn_pool = conn_pool_.lock();
    if (conn_pool) {
        conn_pool->postProcessNextRequest(address_, port_, tls_context_);
    }
}
 
void
Connection::scheduleTimer(const long request_timeout) {
    if (request_timeout > 0) {
        timer_->setup(std::bind(&Connection::timerCallback, this), request_timeout,
                      IntervalTimer::ONE_SHOT);
    }
}
 
void
Connection::doHandshake(const uint64_t transid) {
    // Skip the handshake if it is not needed.
    if (!need_handshake_) {
        doSend(transid);
        return;
    }
 
    SocketCallback socket_cb(std::bind(&Connection::handshakeCallback,
                                       shared_from_this(),
                                       handshake_callback_,
                                       transid,
                                       ph::_1));
    try {
       tls_socket_->handshake(socket_cb);
 
    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}
 
void
Connection::doSend(const uint64_t transid) {
    SocketCallback socket_cb(std::bind(&Connection::sendCallback,
                                       shared_from_this(),
                                       transid,
                                       ph::_1,
                                       ph::_2));
    try {
        if (tcp_socket_) {
            tcp_socket_->asyncSend(&buf_[0], buf_.size(), socket_cb);
            return;
        }
 
        if (tls_socket_) {
            tls_socket_->asyncSend(&buf_[0], buf_.size(), socket_cb);
            return;
        }
 
        // Should never reach this point.
        std::cerr << "internal error: can't find a socket to send to\n";
        isc_throw(Unexpected,
                  "internal error: can't find a socket to send to");
    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}
 
void
Connection::doReceive(const uint64_t transid) {
    TCPEndpoint endpoint;
    SocketCallback socket_cb(std::bind(&Connection::receiveCallback,
                                       shared_from_this(),
                                       transid,
                                       ph::_1,
                                       ph::_2));
    try {
        if (tcp_socket_) {
            tcp_socket_->asyncReceive(static_cast<void*>(input_buf_.data()),
                                      input_buf_.size(), 0,
                                      &endpoint, socket_cb);
            return;
        }
        if (tls_socket_) {
            tls_socket_->asyncReceive(static_cast<void*>(input_buf_.data()),
                                      input_buf_.size(), 0,
                                      &endpoint, socket_cb);
            return;
        }
        // Should never reach this point.
        std::cerr << "internal error: can't find a socket to receive from\n";
        isc_throw(Unexpected,
                  "internal error: can't find a socket to receive from");
 
    } catch (...) {
        terminate(boost::asio::error::not_connected);
    }
}
 
void
Connection::connectCallback(TcpClient::ConnectHandler connect_callback,
                            const uint64_t transid,
                            const boost::system::error_code& ec) {
    if (checkPrematureTimeout(transid)) {
        return;
    }
 
    // Run user defined connect callback if specified.
    if (connect_callback) {
        // If the user defined callback indicates that the connection
        // should not be continued.
        if (tcp_socket_) {
            if (!connect_callback(ec, tcp_socket_->getNative())) {
                return;
            }
        } else if (tls_socket_) {
            if (!connect_callback(ec, tls_socket_->getNative())) {
                return;
            }
        } else {
            // Should never reach this point.
            std::cerr << "internal error: can't find a socket to connect\n";
        }
    }
 
    if (ec && (ec.value() == boost::asio::error::operation_aborted)) {
        return;
 
    // In some cases the "in progress" status code may be returned. It doesn't
    // indicate an error. Sending the request over the socket is expected to
    // be successful. Getting such status appears to be highly dependent on
    // the operating system.
    } else if (ec &&
        (ec.value() != boost::asio::error::in_progress) &&
        (ec.value() != boost::asio::error::already_connected)) {
        terminate(ec);
 
    } else {
        // Start the TLS handshake asynchronously.
        doHandshake(transid);
    }
}
 
void
Connection::handshakeCallback(TcpClient::ConnectHandler handshake_callback,
                              const uint64_t transid,
                              const boost::system::error_code& ec) {
    need_handshake_ = false;
    if (checkPrematureTimeout(transid)) {
        return;
    }
 
    // Run user defined handshake callback if specified.
    if (handshake_callback) {
        // If the user defined callback indicates that the connection
        // should not be continued.
        if (tls_socket_) {
            if (!handshake_callback(ec, tls_socket_->getNative())) {
                return;
            }
        } else {
            // Should never reach this point.
            std::cerr << "internal error: can't find TLS socket\n";
        }
    }
 
    if (ec && (ec.value() == boost::asio::error::operation_aborted)) {
        return;
    } else if (ec) {
        terminate(ec);
 
    } else {
        // Start sending the request asynchronously.
        doSend(transid);
    }
}
 
void
Connection::sendCallback(const uint64_t transid,
                         const boost::system::error_code& ec,
                         size_t length) {
    if (checkPrematureTimeout(transid)) {
        return;
    }
 
    if (ec) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;
 
        // EAGAIN and EWOULDBLOCK don't really indicate an error. The length
        // should be 0 in this case but let's be sure.
        } else if ((ec.value() == boost::asio::error::would_block) ||
            (ec.value() == boost::asio::error::try_again)) {
            length = 0;
 
        } else {
            // Any other error should cause the transaction to terminate.
            terminate(ec);
            return;
        }
    }
 
    // Sending is in progress, so push back the timeout.
    scheduleTimer(timer_->getInterval());
 
    // If any data have been sent, remove it from the buffer and only leave the
    // portion that still has to be sent.
    if (length > 0) {
        if (length >= buf_.size()) {
            buf_.clear();
        } else {
            buf_.erase(buf_.begin(), buf_.begin() + length);
        }
    }
 
    // If there is no more data to be sent, start receiving a response. Otherwise,
    // continue sending.
    if (buf_.empty()) {
        doReceive(transid);
 
    } else {
        doSend(transid);
    }
}
 
void
Connection::receiveCallback(const uint64_t transid,
                            const boost::system::error_code& ec,
                            size_t length) {
    if (checkPrematureTimeout(transid)) {
        return;
    }
 
    if (ec) {
        if (ec.value() == boost::asio::error::operation_aborted) {
            return;
 
        // EAGAIN and EWOULDBLOCK don't indicate an error in this case. All
        // other errors should terminate the transaction.
        }
        if ((ec.value() != boost::asio::error::try_again) &&
            (ec.value() != boost::asio::error::would_block)) {
            terminate(ec);
            return;
 
        } else {
            // For EAGAIN and EWOULDBLOCK the length should be 0 anyway, but let's
            // make sure.
            length = 0;
        }
    }
 
    // Receiving is in progress, so push back the timeout.
    scheduleTimer(timer_->getInterval());
 
    if (runCompleteCheck(ec, length)) {
        doReceive(transid);
    }
}
 
bool
Connection::runCompleteCheck(const boost::system::error_code& ec, size_t length) {
    if (MultiThreadingMgr::instance().getMode()) {
        std::lock_guard<std::mutex> lk(mutex_);
        return (runCompleteCheckInternal(ec, length));
    } else {
        return (runCompleteCheckInternal(ec, length));
    }
}
 
bool
Connection::runCompleteCheckInternal(const boost::system::error_code& ec,
                                     size_t length) {
    // If we have received any data, let's store it.
    if (length != 0) {
        current_response_->insert(current_response_->end(),
                                  input_buf_.begin(),
                                  input_buf_.begin() + length);
    }
 
    // If data is still needed, let's schedule another receive.
    int status = -1;
    std::string err = "";
    if (current_complete_check_) {
        status = current_complete_check_(current_response_, err);
    } else {
        err = "Internal error: no completion checker?";
    }
    if (status == 0) {
        return (true);
    } else if (status > 0) {
        // No more data needed.
        current_response_complete_ = true;
        terminateInternal(ec);
    } else {
        // Error case.
        terminateInternal(ec, err);
    }
 
    return (false);
}
 
void
Connection::timerCallback() {
    // Request timeout occurred.
    terminate(boost::asio::error::timed_out);
}
 
}
 
namespace isc {
namespace tcp {

class TcpClientImpl {
public:
    TcpClientImpl(const IOServicePtr& io_service, size_t thread_pool_size = 0,
                   bool defer_thread_start = false)
        : thread_pool_size_(thread_pool_size), thread_pool_() {
        if (thread_pool_size_ > 0) {
            // Create our own private IOService.
            thread_io_service_.reset(new IOService());
 
            // Create the connection pool. Note that we use the thread_pool_size
            // as the maximum connections per address.
            conn_pool_.reset(new ConnectionPool(thread_io_service_, thread_pool_size_));
 
            // Create the thread pool.
            thread_pool_.reset(new IoServiceThreadPool(thread_io_service_, thread_pool_size_,
                                                       defer_thread_start));
 
            LOG_DEBUG(tcp_logger, isc::log::DBGLVL_TRACE_BASIC,
                      TCP_CLIENT_MT_STARTED)
                .arg(thread_pool_size_);
        } else {
            // Single-threaded mode: use the caller's IOService,
            // one connection per address.
            conn_pool_.reset(new ConnectionPool(io_service, 1));
        }
    }

    ~TcpClientImpl() {
        stop();
    }

    void checkPermissions() {
        if (thread_pool_) {
            thread_pool_->checkPausePermissions();
        }
    }

    void start() {
        if (thread_pool_) {
            thread_pool_->run();
        }
    }

    void stop() {
        // Close all the connections.
        conn_pool_->closeAll();
 
        // Stop the thread pool.
        if (thread_pool_) {
            thread_pool_->stop();
        }
 
        if (thread_io_service_) {
            thread_io_service_->stopAndPoll();
            thread_io_service_->stop();
        }
    }

    void pause() {
        if (!thread_pool_) {
            isc_throw(InvalidOperation, "TcpClient::pause - no thread pool");
        }
 
        // Pause the thread pool.
        thread_pool_->pause();
    }

    void resume() {
        if (!thread_pool_) {
            isc_throw(InvalidOperation, "TcpClient::resume - no thread pool");
        }
 
        // Resume running the thread pool.
        thread_pool_->run();
    }

    bool isRunning() {
        if (thread_pool_) {
            return (thread_pool_->isRunning());
        }
 
        return (false);
    }

    bool isStopped() {
        if (thread_pool_) {
            return (thread_pool_->isStopped());
        }
 
        return (false);
    }

    bool isPaused() {
        if (thread_pool_) {
            return (thread_pool_->isPaused());
        }
 
        return (false);
    }

    asiolink::IOServicePtr getThreadIOService() {
        return (thread_io_service_);
    };

    uint16_t getThreadPoolSize() {
        return (thread_pool_size_);
    }

    uint16_t getThreadCount() {
        if (!thread_pool_) {
            return (0);
        }
        return (thread_pool_->getThreadCount());
    }

    ConnectionPoolPtr conn_pool_;
 
private:

    size_t thread_pool_size_;

    asiolink::IOServicePtr thread_io_service_;

    IoServiceThreadPoolPtr thread_pool_;
};
 
TcpClient::TcpClient(const IOServicePtr& io_service, bool multi_threading_enabled,
                       size_t thread_pool_size, bool defer_thread_start) {
    if (!multi_threading_enabled && thread_pool_size) {
        isc_throw(InvalidOperation,
                  "TcpClient thread_pool_size must be zero "
                  "when Kea core multi-threading is disabled");
    }
 
    impl_.reset(new TcpClientImpl(io_service, thread_pool_size,
                                   defer_thread_start));
}
 
TcpClient::~TcpClient() {
    impl_->stop();
}
 
void
TcpClient::asyncSendRequest(const IOAddress& address,
                            const uint16_t port,
                            const TlsContextPtr& tls_context,
                            const WireDataPtr& request,
                            const WireDataPtr& response,
                            const bool persistent,
                            const TcpClient::CompleteCheck& complete_check,
                            const TcpClient::RequestHandler& request_callback,
                            const TcpClient::RequestTimeout& request_timeout,
                            const TcpClient::ConnectHandler& connect_callback,
                            const TcpClient::HandshakeHandler& handshake_callback,
                            const TcpClient::CloseHandler& close_callback) {
    if (!request) {
        isc_throw(TcpClientError, "TCP request must not be null");
    }
 
    if (!response) {
        isc_throw(TcpClientError, "TCP response must not be null");
    }
 
    if (!complete_check) {
        isc_throw(TcpClientError, "TCP response completion checker must not be null");
    }
 
    if (!request_callback) {
        isc_throw(TcpClientError, "callback for TCP transaction must not be null");
    }
 
    impl_->conn_pool_->queueRequest(address, port, tls_context,
                                    request, response, persistent,
                                    request_timeout.value_,
                                    complete_check,
                                    request_callback, connect_callback,
                                    handshake_callback, close_callback);
}
 
void
TcpClient::closeIfOutOfBand(int socket_fd) {
    return (impl_->conn_pool_->closeIfOutOfBand(socket_fd));
}
 
void
TcpClient::start() {
    impl_->start();
}
 
void
TcpClient::checkPermissions() {
    impl_->checkPermissions();
}
 
void
TcpClient::pause() {
    impl_->pause();
}
 
void
TcpClient::resume() {
    impl_->resume();
}
 
void
TcpClient::stop() {
    impl_->stop();
}
 
const IOServicePtr
TcpClient::getThreadIOService() const {
    return (impl_->getThreadIOService());
}
 
uint16_t
TcpClient::getThreadPoolSize() const {
    return (impl_->getThreadPoolSize());
}
 
uint16_t
TcpClient::getThreadCount() const {
    return (impl_->getThreadCount());
}
 
bool
TcpClient::isRunning() {
    return (impl_->isRunning());
}
 
bool
TcpClient::isStopped() {
    return (impl_->isStopped());
}
 
bool
TcpClient::isPaused() {
    return (impl_->isPaused());
}
 
} // end of namespace isc::tcp
} // end of namespace isc