unix_domain_socket.cc

Go to the documentation of this file.

// Copyright (C) 2017-2020 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/unix_domain_socket.h>
#include <boost/enable_shared_from_this.hpp>
#include <functional>
#include <iostream>

using namespace boost::asio::local;
namespace ph = std::placeholders;

namespace isc {
namespace asiolink {

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

    UnixDomainSocketImpl(IOService& io_service)
        : socket_(io_service.get_io_service()) {
    }

    ~UnixDomainSocketImpl() {
        close();
    }

    void asyncConnect(const stream_protocol::endpoint& endpoint,
                      const UnixDomainSocket::ConnectHandler& handler);

    void connectHandler(const UnixDomainSocket::ConnectHandler& remote_handler,
                        const boost::system::error_code& ec);

    void asyncSend(const void* data, const size_t length,
                   const UnixDomainSocket::Handler& handler);

    void doSend(const boost::asio::const_buffers_1& buffer,
                const UnixDomainSocket::Handler& handler);


    void sendHandler(const UnixDomainSocket::Handler& remote_handler,
                     const boost::asio::const_buffers_1& buffer,
                     const boost::system::error_code& ec,
                     size_t length);

    void asyncReceive(void* data, const size_t length,
                      const UnixDomainSocket::Handler& handler);

    void doReceive(const boost::asio::mutable_buffers_1& buffer,
                   const UnixDomainSocket::Handler& handler);

    void receiveHandler(const UnixDomainSocket::Handler& remote_handler,
                        const boost::asio::mutable_buffers_1& buffer,
                        const boost::system::error_code& ec,
                        size_t length);

    void shutdown();

    void cancel();

    void close();

    stream_protocol::socket socket_;
};

void
UnixDomainSocketImpl::asyncConnect(const stream_protocol::endpoint& endpoint,
                                   const UnixDomainSocket::ConnectHandler& handler) {
    auto local_handler = std::bind(&UnixDomainSocketImpl::connectHandler,
                                   shared_from_this(),
                                   handler, ph::_1);
    socket_.async_connect(endpoint, local_handler);
}

void
UnixDomainSocketImpl::connectHandler(const UnixDomainSocket::ConnectHandler& remote_handler,
                                     const boost::system::error_code& ec) {
    // It was observed on Debian and Fedora that asynchronous connect may result
    // in EINPROGRESS error. This doesn't really indicate a problem with a
    // connection. If we continue transmitting data over the socket it will
    // succeed. So we suppress this error and return 'success' to the user's
    // handler.
    if (ec.value() == boost::asio::error::in_progress) {
        remote_handler(boost::system::error_code());
    } else {
        remote_handler(ec);
    }
}

void
UnixDomainSocketImpl::asyncSend(const void* data, const size_t length,
                                const UnixDomainSocket::Handler& handler) {
    doSend(boost::asio::buffer(data, length), handler);
}

void
UnixDomainSocketImpl::doSend(const boost::asio::const_buffers_1& buffer,
                             const UnixDomainSocket::Handler& handler) {
    auto local_handler = std::bind(&UnixDomainSocketImpl::sendHandler,
                                   shared_from_this(),
                                   handler, buffer, ph::_1, ph::_2);
    socket_.async_send(buffer, local_handler);
}

void
UnixDomainSocketImpl::sendHandler(const UnixDomainSocket::Handler& remote_handler,
                                  const boost::asio::const_buffers_1& buffer,
                                  const boost::system::error_code& ec,
                                  size_t length) {
    // The asynchronous send may return EWOULDBLOCK or EAGAIN on some
    // operating systems. In this case, we simply retry hoping that it
    // will succeed next time. The user's callback never sees these
    // errors.
    if ((ec.value() == boost::asio::error::would_block) ||
        (ec.value() == boost::asio::error::try_again)) {
        doSend(buffer, remote_handler);

    } else {
        remote_handler(ec, length);
    }
}

void
UnixDomainSocketImpl::asyncReceive(void* data, const size_t length,
                                   const UnixDomainSocket::Handler& handler) {
    doReceive(boost::asio::buffer(data, length), handler);
}

void
UnixDomainSocketImpl::doReceive(const boost::asio::mutable_buffers_1& buffer,
                                const UnixDomainSocket::Handler& handler) {
    auto local_handler = std::bind(&UnixDomainSocketImpl::receiveHandler,
                                   shared_from_this(),
                                   handler, buffer, ph::_1, ph::_2);
    socket_.async_receive(buffer, 0, local_handler);
}

void
UnixDomainSocketImpl::receiveHandler(const UnixDomainSocket::Handler& remote_handler,
                                     const boost::asio::mutable_buffers_1& buffer,
                                     const boost::system::error_code& ec,
                                     size_t length) {
    // The asynchronous receive may return EWOULDBLOCK or EAGAIN on some
    // operating systems. In this case, we simply retry hoping that it
    // will succeed next time. The user's callback never sees these
    // errors.
    if ((ec.value() == boost::asio::error::would_block) ||
        (ec.value() == boost::asio::error::try_again)) {
        doReceive(buffer, remote_handler);

    } else {
        remote_handler(ec, length);
    }
}

void
UnixDomainSocketImpl::shutdown() {
    boost::system::error_code ec;
    static_cast<void>(socket_.shutdown(stream_protocol::socket::shutdown_both, ec));
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
}

void
UnixDomainSocketImpl::cancel() {
    boost::system::error_code ec;
    static_cast<void>(socket_.cancel(ec));
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
}

void
UnixDomainSocketImpl::close() {
    boost::system::error_code ec;
    static_cast<void>(socket_.close(ec));
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
}

UnixDomainSocket::UnixDomainSocket(IOService& io_service)
    : impl_(new UnixDomainSocketImpl(io_service)) {
}

int
UnixDomainSocket::getNative() const {
#if BOOST_VERSION < 106600
    return (impl_->socket_.native());
#else
    return (impl_->socket_.native_handle());
#endif
}

int
UnixDomainSocket::getProtocol() const {
    return (0);
}

void
UnixDomainSocket::connect(const std::string& path) {
    boost::system::error_code ec;
    impl_->socket_.connect(stream_protocol::endpoint(path.c_str()), ec);
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
}

void
UnixDomainSocket::asyncConnect(const std::string& path, const ConnectHandler& handler) {
    impl_->asyncConnect(stream_protocol::endpoint(path.c_str()), handler);
}

size_t
UnixDomainSocket::write(const void* data, size_t length) {
    boost::system::error_code ec;
    size_t res = boost::asio::write(impl_->socket_,
                                    boost::asio::buffer(data, length),
                                    boost::asio::transfer_all(),
                                    ec);
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
    return (res);
}

void
UnixDomainSocket::asyncSend(const void* data, const size_t length,
                            const Handler& handler) {
    impl_->asyncSend(data, length, handler);
}

size_t
UnixDomainSocket::receive(void* data, size_t length) {
    boost::system::error_code ec;
    size_t res = impl_->socket_.receive(boost::asio::buffer(data, length), 0, ec);
    if (ec) {
        isc_throw(UnixDomainSocketError, ec.message());
    }
    return (res);
}

void
UnixDomainSocket::asyncReceive(void* data, const size_t length,
                               const Handler& handler) {
    impl_->asyncReceive(data, length, handler);
}

void
UnixDomainSocket::shutdown() {
    impl_->shutdown();
}

void
UnixDomainSocket::cancel() {
    impl_->cancel();
}

void
UnixDomainSocket::close() {
    impl_->close();
}

boost::asio::local::stream_protocol::socket&
UnixDomainSocket::getASIOSocket() const {
    return (impl_->socket_);
}

} // end of namespace asiolink
} // end of namespace isc