proactor_epoll_impl.hpp

Go to the documentation of this file.

 
// C.
#include <poll.h>
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : BasicProactor
// =========================================================================
inline join::BasicProactor::BasicProactor ()
: _commands (_queueSize)
, _wakeup (eventfd (0, EFD_NONBLOCK | EFD_CLOEXEC))
, _readOps (256, nullptr)
, _writeOps (256, nullptr)
{
    if (_wakeup == -1)
    {
        throw std::system_error (errno, std::system_category (), "eventfd failed");  // LCOV_EXCL_LINE
    }
 
    _reactor.addHandler (_wakeup, this, true, false, false);
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : ~BasicProactor
// =========================================================================
inline join::BasicProactor::~BasicProactor () noexcept
{
    stop (true);
 
    ::close (_wakeup);
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : run
// =========================================================================
inline void join::BasicProactor::run ()
{
    _wakeupState.store (WakeupState::Sleeping, std::memory_order_seq_cst);
    _reactor.run ();
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : stop
// =========================================================================
inline void join::BasicProactor::stop (bool sync) noexcept
{
    if (isProactorThread ())
    {
        cancelAllOperations ();
        _reactor.stop (false);
        return;
    }
 
    if (JOIN_UNLIKELY (!isRunning ()))
    {
        return;
    }
 
    writeCommand ({CommandType::Stop, nullptr, sync, nullptr, nullptr, nullptr});
 
    if (JOIN_LIKELY (sync))
    {
        waitStopped ();
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : waitStopped
// =========================================================================
inline void join::BasicProactor::waitStopped () const noexcept
{
    _reactor.waitStopped ();
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : registerBufferRing
// =========================================================================
template <size_t Count, size_t Size>
int join::BasicProactor::registerBufferRing (uint16_t group, LocalMem::Allocator<Count, Size>& arena)
{
    int result = 0;
    std::error_code errc;
 
    InvokeHandler fn = [this, group, &arena, &result, &errc] () {
        if (JOIN_UNLIKELY (_bufferRings.count (group)))
        {
            lastError = make_error_code (Errc::InUse);
            errc = lastError;
            result = -1;
            return;
        }
 
        if (JOIN_UNLIKELY (_bufferRings[group].registerBuffer (group, arena) == -1))
        {
            errc = lastError;
            _bufferRings.erase (group);
            result = -1;
            return;
        }
    };
 
    if (JOIN_UNLIKELY (!isRunning ()))
    {
        lastError = make_error_code (Errc::OperationFailed);
        return -1;
    }
 
    if (JOIN_UNLIKELY (invoke (&fn) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_UNLIKELY (result == -1))
    {
        lastError = errc;
        return -1;
    }
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : unregisterBufferRing
// =========================================================================
inline int join::BasicProactor::unregisterBufferRing (uint16_t group)
{
    int result = 0;
    std::error_code errc;
 
    InvokeHandler fn = [this, group, &result, &errc] () {
        auto it = _bufferRings.find (group);
        if (JOIN_UNLIKELY (it == _bufferRings.end ()))
        {
            lastError = make_error_code (Errc::NotFound);
            errc = lastError;
            result = -1;
            return;
        }
 
        if (JOIN_UNLIKELY (it->second.armed ()))
        {
            lastError = make_error_code (Errc::InUse);
            errc = lastError;
            result = -1;
            return;
        }
 
        if (JOIN_UNLIKELY (it->second.unregisterBuffer () == -1))
        {
            // LCOV_EXCL_START
            errc = lastError;
            result = -1;
            return;
            // LCOV_EXCL_STOP
        }
 
        _bufferRings.erase (it);
    };
 
    if (JOIN_UNLIKELY (!isRunning ()))
    {
        lastError = make_error_code (Errc::OperationFailed);
        return -1;
    }
 
    if (JOIN_UNLIKELY (invoke (&fn) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_UNLIKELY (result == -1))
    {
        lastError = errc;
        return -1;
    }
 
    return 0;
}
 
#ifdef JOIN_HAS_NUMA
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : mbind
// =========================================================================
inline int join::BasicProactor::mbind (int numa) const noexcept
{
    if (_commands.mbind (numa) == -1)
    {
        return -1;
    }
 
    return _reactor.mbind (numa);
}
#endif
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : mlock
// =========================================================================
inline int join::BasicProactor::mlock () const noexcept
{
    if (_commands.mlock () == -1)
    {
        return -1;
    }
 
    return _reactor.mlock ();
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : isRunning
// =========================================================================
inline bool join::BasicProactor::isRunning () const noexcept
{
    return _reactor.isRunning ();
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : isProactorThread
// =========================================================================
inline bool join::BasicProactor::isProactorThread () const noexcept
{
    return _reactor.isReactorThread ();
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : writeCommand
// =========================================================================
inline int join::BasicProactor::writeCommand (const Command& cmd) noexcept
{
    if (JOIN_UNLIKELY (_commands.push (cmd) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    // pairs with the fence in the event loop: an acquire load would let the push above be
    // reordered after it, the loop would sleep and the wakeup would be lost.
    std::atomic_thread_fence (std::memory_order_seq_cst);
 
    WakeupState state = _wakeupState.load (std::memory_order_relaxed);
    if (state != WakeupState::Sleeping)
    {
        return 0;
    }
 
    if (_wakeupState.compare_exchange_strong (state, WakeupState::Waking, std::memory_order_seq_cst))
    {
        uint64_t value = 1;
        if (JOIN_UNLIKELY (::write (_wakeup, &value, sizeof (uint64_t)) == -1))
        {
            _wakeupState.store (WakeupState::Sleeping, std::memory_order_seq_cst);  // LCOV_EXCL_LINE
        }
    }
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : readCommands
// =========================================================================
inline void join::BasicProactor::readCommands () noexcept
{
    uint64_t count;
    [[maybe_unused]] ssize_t nread = ::read (_wakeup, &count, sizeof (count));
    _wakeupState.store (WakeupState::Sleeping, std::memory_order_relaxed);
    std::atomic_thread_fence (std::memory_order_seq_cst);
 
    Command cmd;
    while (_commands.tryPop (cmd) == 0)
    {
        processCommand (cmd);
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : processCommand
// =========================================================================
inline void join::BasicProactor::processCommand (const Command& cmd) noexcept
{
    int err = 0;
 
    switch (cmd.type)
    {
        case CommandType::Submit:
            err = submitOperation (*cmd.op, cmd.flush);
            if (JOIN_UNLIKELY ((err != 0) && (cmd.done == nullptr) && (cmd.op->state == IoOperation::State::Idle)))
            {
                dispatchOperation (cmd.op, -lastError.default_error_condition ().value (), false);
            }
            break;
 
        case CommandType::Cancel:
            err = cancelOperation (*cmd.op, cmd.flush);
            break;
 
        case CommandType::Invoke:
            err = invokeFunction (cmd.fn);
            break;
 
        case CommandType::Stop:
            cancelAllOperations ();
            _reactor.stop (false);
            break;
 
        default:    // LCOV_EXCL_LINE
            break;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_UNLIKELY (cmd.done))
    {
        if (cmd.errc && (err != 0))
        {
            *cmd.errc = lastError;
        }
        cmd.done->store (true, std::memory_order_release);
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : submitOperation
// =========================================================================
inline int join::BasicProactor::submitOperation (IoOperation& op, [[maybe_unused]] bool flush) noexcept
{
    if (JOIN_UNLIKELY (op.fd () < 0))
    {
        lastError = std::make_error_code (std::errc::bad_file_descriptor);
        return -1;
    }
 
    Backoff backoff;
    while (JOIN_UNLIKELY (op.state.load (std::memory_order_acquire) == IoOperation::State::Suspended))
    {
        backoff ();
    }
 
    IoOperation::State expected = IoOperation::State::Idle;
    if (!op.state.compare_exchange_strong (expected, IoOperation::State::Submitted, std::memory_order_acquire,
                                           std::memory_order_relaxed) &&
        (expected == IoOperation::State::Busy))
    {
        op.state.store (IoOperation::State::Submitted, std::memory_order_release);
    }
 
    IoRingBuffer* ring = nullptr;
 
    if (JOIN_UNLIKELY (op.multishot && ((op.code == static_cast<uint8_t> (IoOperation::Opcode::RecvMsg)) ||
                                        (op.code == static_cast<uint8_t> (IoOperation::Opcode::Recv)))))
    {
        auto it = _bufferRings.find (op.group);
        if (JOIN_UNLIKELY (it == _bufferRings.end ()))
        {
            resetOperation (op);
            lastError = make_error_code (Errc::NotFound);
            return -1;
        }
 
        ring = &it->second;
    }
 
    if (JOIN_UNLIKELY (static_cast<IoOperation::Opcode> (op.code) == IoOperation::Opcode::Connect))
    {
        if (JOIN_UNLIKELY (::connect (op.data.connect.fd, op.data.connect.addr, op.data.connect.addrlen) == -1 &&
                           errno != EINPROGRESS))
        {
            lastError = std::error_code (errno, std::system_category ());
            resetOperation (op);
            return -1;
        }
    }
 
    if (JOIN_UNLIKELY (static_cast<size_t> (op.fd ()) >= _readOps.size ()))
    {
        size_t newSize = static_cast<size_t> (op.fd ()) + 1;
        _readOps.resize (newSize, nullptr);
        _writeOps.resize (newSize, nullptr);
    }
 
    bool isWrite = isWriteOp (op);
 
    if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] == &op)) || (!isWrite && (_readOps[op.fd ()] == &op))))
    {
        lastError = make_error_code (std::errc::device_or_resource_busy);
        return -1;
    }
 
    if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] != nullptr)) || (!isWrite && (_readOps[op.fd ()] != nullptr))))
    {
        resetOperation (op);
        lastError = make_error_code (Errc::InvalidParam);
        return -1;
    }
 
    if (isWrite)
    {
        _writeOps[op.fd ()] = &op;
    }
    else
    {
        _readOps[op.fd ()] = &op;
    }
 
    if (JOIN_UNLIKELY (ring != nullptr))
    {
        op.ring = ring;
        ring->bind ();
    }
 
    int err = _reactor.addHandler (op.fd (), this, _readOps[op.fd ()] != nullptr, _writeOps[op.fd ()] != nullptr);
    if (JOIN_UNLIKELY (err == -1))
    {
        if (isWrite)
        {
            _writeOps[op.fd ()] = nullptr;
        }
        else
        {
            _readOps[op.fd ()] = nullptr;
        }
 
        if (JOIN_UNLIKELY (ring != nullptr))
        {
            op.ring->unbind ();
            op.ring = nullptr;
        }
 
        resetOperation (op);
    }
 
    return err;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : cancelOperation
// =========================================================================
inline int join::BasicProactor::cancelOperation (IoOperation& op, [[maybe_unused]] bool flush) noexcept
{
    if (JOIN_UNLIKELY (op.fd () < 0))
    {
        lastError = std::make_error_code (std::errc::bad_file_descriptor);
        return -1;
    }
 
    if (JOIN_UNLIKELY (op.state != IoOperation::State::Submitted))
    {
        lastError = make_error_code (Errc::OperationFailed);
        return -1;
    }
 
    if (JOIN_UNLIKELY (static_cast<size_t> (op.fd ()) >= _readOps.size ()))
    {
        lastError = std::make_error_code (std::errc::bad_file_descriptor);
        return -1;
    }
 
    bool isWrite = isWriteOp (op);
 
    if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] != &op)) || (!isWrite && (_readOps[op.fd ()] != &op))))
    {
        lastError = make_error_code (Errc::InvalidParam);
        return -1;
    }
 
    if (isWrite)
    {
        _writeOps[op.fd ()] = nullptr;
    }
    else
    {
        _readOps[op.fd ()] = nullptr;
    }
 
    int ret = 0;
 
    if (_readOps[op.fd ()] == nullptr && _writeOps[op.fd ()] == nullptr)
    {
        ret = _reactor.delHandler (op.fd ());
    }
    else
    {
        ret = _reactor.addHandler (op.fd (), this, _readOps[op.fd ()] != nullptr, _writeOps[op.fd ()] != nullptr);
    }
 
    dispatchOperation (&op, -ECANCELED, true);
 
    return ret;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : cancelAllOperations
// =========================================================================
inline void join::BasicProactor::cancelAllOperations () noexcept
{
    for (size_t fd = 0; fd < _readOps.size (); ++fd)
    {
        IoOperation* rOp = std::exchange (_readOps[fd], nullptr);
        IoOperation* wOp = std::exchange (_writeOps[fd], nullptr);
        if (rOp || wOp)
        {
            _reactor.delHandler (fd);
        }
        dispatchOperation (rOp, -ECANCELED, true);
        dispatchOperation (wOp, -ECANCELED, true);
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : endOperation
// =========================================================================
inline void join::BasicProactor::endOperation (IoOperation& op, int result, bool cancelled) noexcept
{
    int fd = op.fd ();
 
    if (JOIN_UNLIKELY (fd < 0 || static_cast<size_t> (fd) >= _readOps.size ()))
    {
        return;  // LCOV_EXCL_LINE
    }
 
    if (isWriteOp (op))
    {
        _writeOps[fd] = nullptr;
    }
    else
    {
        _readOps[fd] = nullptr;
    }
 
    if (_readOps[fd] == nullptr && _writeOps[fd] == nullptr)
    {
        _reactor.delHandler (fd);
    }
    else
    {
        _reactor.addHandler (fd, this, _readOps[fd] != nullptr, _writeOps[fd] != nullptr);
    }
 
    dispatchOperation (&op, result, cancelled);
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : isWriteOp
// =========================================================================
inline bool join::BasicProactor::isWriteOp (const IoOperation& op) noexcept
{
    switch (static_cast<IoOperation::Opcode> (op.code))
    {
        case IoOperation::Opcode::Poll:
            return (op.data.poll.events & POLLIN) == 0;
        case IoOperation::Opcode::Connect:
        case IoOperation::Opcode::Write:
        case IoOperation::Opcode::WriteFixed:
        case IoOperation::Opcode::SendMsg:
        case IoOperation::Opcode::Send:
            return true;
        default:
            return false;
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : executeOp
// =========================================================================
inline int join::BasicProactor::executeOp (IoOperation& op, uint32_t revents) noexcept
{
    for (;;)
    {
        switch (static_cast<IoOperation::Opcode> (op.code))
        {
            case IoOperation::Opcode::Poll:
                return static_cast<int> (revents & (op.data.poll.events | POLLERR | POLLHUP | POLLRDHUP));
 
            case IoOperation::Opcode::Accept:
                {
                    int fd = ::accept4 (op.data.accept.fd, op.data.accept.addr, op.data.accept.addrlen,
                                        op.data.accept.flags);
                    if (JOIN_UNLIKELY ((fd == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (fd == -1) ? -errno : fd;
                }
 
            case IoOperation::Opcode::Connect:
                {
                    int err = 0;
                    socklen_t len = sizeof (err);
                    if (JOIN_UNLIKELY (::getsockopt (op.data.connect.fd, SOL_SOCKET, SO_ERROR, &err, &len) == -1))
                    {
                        return -errno;
                    }
                    return JOIN_UNLIKELY (err) ? -err : 0;
                }
 
            case IoOperation::Opcode::Read:
            case IoOperation::Opcode::ReadFixed:
                {
                    ssize_t n = ::read (op.data.rw.fd, op.data.rw.buf, op.data.rw.len);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            case IoOperation::Opcode::Write:
            case IoOperation::Opcode::WriteFixed:
                {
                    ssize_t n = ::write (op.data.rw.fd, op.data.rw.buf, op.data.rw.len);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            case IoOperation::Opcode::RecvMsg:
                {
                    ssize_t n = ::recvmsg (op.data.msg.fd, op.data.msg.msg, op.data.msg.flags);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            case IoOperation::Opcode::SendMsg:
                {
                    ssize_t n = ::sendmsg (op.data.msg.fd, op.data.msg.msg, op.data.msg.flags);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            case IoOperation::Opcode::Recv:
                {
                    ssize_t n =
                        ::recv (op.data.stream.fd, op.data.stream.buf, op.data.stream.len, op.data.stream.flags);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            case IoOperation::Opcode::Send:
                {
                    ssize_t n =
                        ::send (op.data.stream.fd, op.data.stream.buf, op.data.stream.len, op.data.stream.flags);
                    if (JOIN_UNLIKELY ((n == -1) && (errno == EINTR)))
                    {
                        continue;  // LCOV_EXCL_LINE
                    }
                    return (n == -1) ? -errno : static_cast<int> (n);
                }
 
            default:
                return -EINVAL;
        }
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : onEvent
// =========================================================================
inline void join::BasicProactor::onEvent (int fd, uint32_t revents) noexcept
{
    if (JOIN_UNLIKELY (fd == _wakeup))
    {
        readCommands ();
        return;
    }
 
    if (JOIN_UNLIKELY (revents & (EPOLLERR | EPOLLRDHUP | EPOLLHUP)))
    {
        IoOperation* rOp = std::exchange (_readOps[fd], nullptr);
        IoOperation* wOp = std::exchange (_writeOps[fd], nullptr);
        if (JOIN_LIKELY (rOp || wOp))
        {
            _reactor.delHandler (fd);
        }
 
        int result = (revents & EPOLLERR) ? -ECONNRESET : 0;
        int rResult = result;
        int wResult = result;
 
        if ((rOp != nullptr) && (rOp->code == static_cast<uint8_t> (IoOperation::Opcode::Poll)))
        {
            rResult = executeOp (*rOp, revents);
        }
 
        if ((wOp != nullptr) && (wOp->code == static_cast<uint8_t> (IoOperation::Opcode::Poll)))
        {
            wResult = executeOp (*wOp, revents);
        }
 
        dispatchOperation (rOp, rResult, false);
        dispatchOperation (wOp, wResult, false);
 
        return;
    }
 
    IoOperation* op = (revents & EPOLLIN) ? _readOps[fd] : _writeOps[fd];
    if (JOIN_UNLIKELY (op == nullptr))
    {
        return;
    }
 
    IoOperation::State current = op->state.load (std::memory_order_acquire);
    Backoff backoff;
 
    while (JOIN_UNLIKELY (current == IoOperation::State::Suspended))
    {
        backoff ();
        current = op->state.load (std::memory_order_acquire);
    }
 
    if (JOIN_UNLIKELY (current == IoOperation::State::Idle))
    {
        return;  // LCOV_EXCL_LINE
    }
 
    IoRingBuffer* br = nullptr;
    uint16_t bid = 0;
 
    if (op->ring != nullptr)
    {
        int selected = op->ring->select ();
        if (JOIN_UNLIKELY (selected == -1))
        {
            endOperation (*op, -ENOBUFS, false);
            return;
        }
 
        br = op->ring;
        bid = static_cast<uint16_t> (selected);
 
        if (op->code == static_cast<uint8_t> (IoOperation::Opcode::RecvMsg))
        {
            uint32_t reserved = op->data.msg.namelen + op->data.msg.controllen;
            if (JOIN_UNLIKELY (reserved >= br->size ()))
            {
                endOperation (*op, -EFAULT, false);
                br->recycle (bid);
                return;
            }
 
            char* base = static_cast<char*> (br->get (bid));
 
            op->data.msg.msg->msg_name = base;
            op->data.msg.msg->msg_namelen = op->data.msg.namelen;
            op->data.msg.msg->msg_control = base + op->data.msg.namelen;
            op->data.msg.msg->msg_controllen = op->data.msg.controllen;
            op->data.msg.msg->msg_iov->iov_base = base + reserved;
            op->data.msg.msg->msg_iov->iov_len = br->size () - reserved;
            op->data.msg.msg->msg_iovlen = 1;
            op->data.msg.msg->msg_flags = 0;
        }
        else
        {
            op->data.stream.buf = br->get (bid);
            op->data.stream.len = br->size ();
        }
    }
 
    int result = executeOp (*op, revents);
 
    if (JOIN_UNLIKELY ((result == -EAGAIN) && (op->code != static_cast<uint8_t> (IoOperation::Opcode::Connect))))
    {
        if (br != nullptr)
        {
            br->recycle (bid);
        }
        return;
    }
 
    if ((br != nullptr) && (result >= 0) && (op->code == static_cast<uint8_t> (IoOperation::Opcode::RecvMsg)))
    {
        op->data.msg.msg->msg_iov->iov_len = static_cast<size_t> (result);
 
        if (op->data.msg.msg->msg_controllen < sizeof (cmsghdr))
        {
            op->data.msg.msg->msg_control = nullptr;
        }
    }
 
    if (op->multishot &&
        ((result > 0) || ((result == 0) && (op->code == static_cast<uint8_t> (IoOperation::Opcode::Accept)))))
    {
        op->more = true;
        notifyOperation (*op, result, false);
    }
    else
    {
        endOperation (*op, result, false);
    }
 
    if (br != nullptr)
    {
        br->recycle (bid);
    }
}