rte_posix_osadapter_ipc.c

Go to the documentation of this file.

/* _GNU_SOURCE (implies _POSIX_C_SOURCE) rather than _POSIX_C_SOURCE alone:
 * F_SETPIPE_SZ / F_GETPIPE_SZ below are a Linux/glibc extension, not
 * standard POSIX, and are guarded with #if so this file still builds on a
 * non-glibc POSIX target, where the pipe capacity is the compile-time
 * RTE_POSIX_IPC_PIPE_CAPACITY instead. */
#define _GNU_SOURCE

#include "rte/posix_osadapter/rte_posix_osadapter.h"
#include "rte_posix_osadapter_poll.h"
 
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <poll.h>
#include <stdbool.h>
#include <stdint.h>
#include <sys/ioctl.h>
#include <time.h>
#include <unistd.h>

#if defined(PIPE_BUF)
#define POSIX_IPC_ATOMIC_MAX ((size_t)PIPE_BUF)
#else
#define POSIX_IPC_ATOMIC_MAX ((size_t)_POSIX_PIPE_BUF)
#endif
#define POSIX_IPC_POLL_NS   1000000L
#define POSIX_IPC_NS_PER_MS 1000000L    
#define POSIX_IPC_NS_PER_S  1000000000L 
 
#if defined(F_SETPIPE_SZ) && defined(F_GETPIPE_SZ)
#define POSIX_IPC_EXTRA_SLOTS 3U
#else
#ifndef RTE_POSIX_IPC_PIPE_CAPACITY
#define RTE_POSIX_IPC_PIPE_CAPACITY 16384U
#endif
#endif

typedef struct
{
    int read_fd;         
    int write_fd;        
    size_t message_size; 
    size_t queue_depth;  
} posix_ipc_state_t;
 
typedef char posix_ipc_storage_fits_[(sizeof(posix_ipc_state_t) <= sizeof(rte_ipc_storage_t)) ? 1 : -1];

static bool compute_deadline(rte_duration_ms_t timeout_ms, struct timespec *out_deadline);
static rte_status_t read_message(int fd, uint8_t *dst, size_t size, rte_duration_ms_t timeout_ms);
static rte_status_t set_nonblocking(const int fds[2]);
#if defined(F_SETPIPE_SZ) && defined(F_GETPIPE_SZ)
static bool required_pipe_bytes(size_t message_size, size_t queue_depth, size_t *out_bytes);
#endif
static rte_status_t ensure_pipe_capacity(int write_fd, size_t message_size, size_t queue_depth);
static rte_status_t try_send_once(const posix_ipc_state_t *state, const void *message);
static rte_status_t sleep_before_retry(const struct timespec *deadline);
static rte_status_t send_until_deadline(const posix_ipc_state_t *state, const void *message,
                                        rte_duration_ms_t timeout_ms);


static bool compute_deadline(rte_duration_ms_t timeout_ms, struct timespec *out_deadline)
{
    bool ok = false;
 
    if (clock_gettime(CLOCK_MONOTONIC, out_deadline) == 0)
    {
        rte_duration_ms_t whole_s = timeout_ms / 1000U;
        rte_duration_ms_t rest_ms = timeout_ms % 1000U;
 
        out_deadline->tv_sec += (time_t)whole_s;
        out_deadline->tv_nsec += (long)rest_ms * POSIX_IPC_NS_PER_MS;
        if (out_deadline->tv_nsec >= POSIX_IPC_NS_PER_S)
        {
            out_deadline->tv_sec += 1;
            out_deadline->tv_nsec -= POSIX_IPC_NS_PER_S;
        }
        ok = true;
    }
    return ok;
}

static rte_status_t read_message(int fd, uint8_t *dst, size_t size, rte_duration_ms_t timeout_ms)
{
    rte_status_t status = RTE_STATUS_OK;
    struct timespec deadline;
    size_t done = 0U;
 
    if (!compute_deadline(timeout_ms, &deadline))
    {
        status = RTE_STATUS_INTERNAL_ERROR; /* no deadline - never wait unbounded */
    }
 
    while ((status == RTE_STATUS_OK) && (done < size))
    {
        int wait_ms = posix_remaining_ms(&deadline); /* capped at INT_MAX, never negative from a large timeout */
 
        /* wait_ms == 0 still polls once without blocking, so timeout_ms == 0
         * is a non-blocking try ("at most timeout_ms"), not an immediate TIMEOUT. */
        if (wait_ms < 0)
        {
            status = RTE_STATUS_TIMEOUT;
        }
        else
        {
            struct pollfd pfd;
            int rc;
 
            pfd.fd = fd;
            pfd.events = POLLIN;
            pfd.revents = 0;
            rc = poll(&pfd, 1, wait_ms);
            if (rc == 0)
            {
                status = RTE_STATUS_TIMEOUT;
            }
            else if (rc < 0)
            {
                if (errno != EINTR)
                {
                    status = RTE_STATUS_INTERNAL_ERROR;
                }
            }
            else
            {
                ssize_t n = read(fd, &dst[done], size - done);
 
                if (n < 0)
                {
                    if ((errno != EAGAIN) && (errno != EWOULDBLOCK) && (errno != EINTR))
                    {
                        status = RTE_STATUS_INTERNAL_ERROR;
                    }
                }
                else if (n == 0)
                {
                    /* read() returning 0 means the write end was closed. */
                    status = RTE_STATUS_INTERNAL_ERROR;
                }
                else
                {
                    done += (size_t)n;
                }
            }
        }
    }
 
    return status;
}

static rte_status_t set_nonblocking(const int fds[2])
{
    rte_status_t status = RTE_STATUS_INTERNAL_ERROR;
 
    if (fcntl(fds[0], F_SETFL, O_NONBLOCK) != -1)
    {
        if (fcntl(fds[1], F_SETFL, O_NONBLOCK) != -1)
        {
            status = RTE_STATUS_OK;
        }
    }
    return status;
}
 
#if defined(F_SETPIPE_SZ) && defined(F_GETPIPE_SZ)
static bool required_pipe_bytes(size_t message_size, size_t queue_depth, size_t *out_bytes)
{
    bool ok = false;
    long page = sysconf(_SC_PAGESIZE);
 
    if ((page > 0) && ((size_t)page >= message_size))
    {
        size_t page_bytes = (size_t)page;
        size_t full_slots = queue_depth / (page_bytes / message_size);
 
        if (full_slots <= ((SIZE_MAX / page_bytes) - POSIX_IPC_EXTRA_SLOTS))
        {
            *out_bytes = (full_slots + POSIX_IPC_EXTRA_SLOTS) * page_bytes;
            ok = true;
        }
    }
    return ok;
}

static rte_status_t ensure_pipe_capacity(int write_fd, size_t message_size, size_t queue_depth)
{
    rte_status_t status = RTE_STATUS_RESOURCE_EXHAUSTED;
    size_t needed = 0U;
 
    if (required_pipe_bytes(message_size, queue_depth, &needed) && (needed <= (size_t)INT_MAX))
    {
        if (fcntl(write_fd, F_SETPIPE_SZ, (int)needed) >= 0)
        {
            int actual = fcntl(write_fd, F_GETPIPE_SZ);
 
            if ((actual >= 0) && ((size_t)actual >= needed))
            {
                status = RTE_STATUS_OK;
            }
        }
    }
    return status;
}
#else
static rte_status_t ensure_pipe_capacity(int write_fd, size_t message_size, size_t queue_depth)
{
    rte_status_t status = RTE_STATUS_RESOURCE_EXHAUSTED;
 
    (void)write_fd;
    if ((message_size * queue_depth) <= (size_t)RTE_POSIX_IPC_PIPE_CAPACITY)
    {
        status = RTE_STATUS_OK;
    }
    return status;
}
#endif

static rte_status_t try_send_once(const posix_ipc_state_t *state, const void *message)
{
    rte_status_t status;
    int queued_bytes = 0;
 
    if (ioctl(state->read_fd, FIONREAD, &queued_bytes) != 0)
    {
        status = RTE_STATUS_INTERNAL_ERROR;
    }
    else if (queued_bytes < 0)
    {
        status = RTE_STATUS_INTERNAL_ERROR;
    }
    else if (((size_t)queued_bytes / state->message_size) >= state->queue_depth)
    {
        status = RTE_STATUS_TIMEOUT; /* queue full */
    }
    else
    {
        ssize_t written = write(state->write_fd, message, state->message_size);
 
        if (written < 0)
        {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR))
            {
                status = RTE_STATUS_TIMEOUT;
            }
            else
            {
                status = RTE_STATUS_INTERNAL_ERROR;
            }
        }
        else if ((size_t)written == state->message_size)
        {
            status = RTE_STATUS_OK;
        }
        else
        {
            status = RTE_STATUS_INTERNAL_ERROR;
        }
    }
    return status;
}

static rte_status_t sleep_before_retry(const struct timespec *deadline)
{
    rte_status_t status = RTE_STATUS_INTERNAL_ERROR;
    struct timespec now;
 
    if (clock_gettime(CLOCK_MONOTONIC, &now) == 0)
    {
        int64_t remaining_ns = (((int64_t)deadline->tv_sec - (int64_t)now.tv_sec) * POSIX_IPC_NS_PER_S) +
                               ((int64_t)deadline->tv_nsec - (int64_t)now.tv_nsec);
 
        if (remaining_ns <= 0)
        {
            status = RTE_STATUS_TIMEOUT;
        }
        else
        {
            struct timespec nap;
 
            nap.tv_sec = 0;
            nap.tv_nsec = (long)((remaining_ns < POSIX_IPC_POLL_NS) ? remaining_ns : POSIX_IPC_POLL_NS);
            /* An EINTR only ends the sleep early; the deadline still bounds the loop. */
            (void)nanosleep(&nap, NULL);
            status = RTE_STATUS_OK;
        }
    }
    return status;
}

static rte_status_t send_until_deadline(const posix_ipc_state_t *state, const void *message,
                                        rte_duration_ms_t timeout_ms)
{
    rte_status_t status;
    struct timespec deadline;
 
    if (!compute_deadline(timeout_ms, &deadline))
    {
        status = RTE_STATUS_INTERNAL_ERROR; /* no deadline - never wait unbounded */
    }
    else
    {
        bool trying = true;
 
        status = RTE_STATUS_TIMEOUT;
        while (trying)
        {
            status = try_send_once(state, message);
            if (status == RTE_STATUS_TIMEOUT)
            {
                rte_status_t wait_status = sleep_before_retry(&deadline);
 
                if (wait_status != RTE_STATUS_OK)
                {
                    status = wait_status; /* TIMEOUT at the deadline, INTERNAL_ERROR without a clock */
                    trying = false;
                }
            }
            else
            {
                trying = false;
            }
        }
    }
    return status;
}


static rte_status_t osadapter_create(rte_ipc_storage_t *storage, const rte_ipc_config_t *config,
                                     rte_ipc_handle_t *out_handle)
{
    rte_status_t status;
    int fds[2];
 
    if ((storage == NULL) || (config == NULL) || (out_handle == NULL) || (config->message_size == 0U)
        || (config->queue_depth == 0U))
    {
        status = RTE_STATUS_INVALID_PARAM;
    }
    else if (config->message_size > POSIX_IPC_ATOMIC_MAX)
    {
        status = RTE_STATUS_NOT_SUPPORTED; /* one write() could not queue the message whole */
    }
    else if (config->queue_depth > (SIZE_MAX / config->message_size))
    {
        status = RTE_STATUS_INVALID_PARAM; /* message_size * queue_depth overflows */
    }
    else if (pipe(fds) != 0)
    {
        status = RTE_STATUS_INTERNAL_ERROR;
    }
    else
    {
        status = set_nonblocking(fds);
        if (status == RTE_STATUS_OK)
        {
            status = ensure_pipe_capacity(fds[1], config->message_size, config->queue_depth);
        }
 
        if (status == RTE_STATUS_OK)
        {
            posix_ipc_state_t *state = (posix_ipc_state_t *)(void *)storage;
 
            state->read_fd = fds[0];
            state->write_fd = fds[1];
            state->message_size = config->message_size;
            state->queue_depth = config->queue_depth;
            *out_handle = (rte_ipc_handle_t)(void *)storage;
        }
        else
        {
            (void)close(fds[0]);
            (void)close(fds[1]);
        }
    }
    return status;
}

static rte_status_t backend_send(rte_ipc_handle_t handle, const void *message, size_t message_size,
                                   rte_duration_ms_t timeout_ms)
{
    const posix_ipc_state_t *state = (const posix_ipc_state_t *)(const void *)handle;
    rte_status_t status;
 
    if ((state == NULL) || (message == NULL) || (message_size == 0U))
    {
        status = RTE_STATUS_INVALID_PARAM;
    }
    else if (message_size != state->message_size)
    {
        status = RTE_STATUS_INVALID_PARAM; /* every message has the size fixed at create */
    }
    else
    {
        status = send_until_deadline(state, message, timeout_ms);
    }
    return status;
}

static rte_status_t backend_receive(rte_ipc_handle_t handle, void *out_message, size_t buffer_size,
                                      rte_duration_ms_t timeout_ms)
{
    const posix_ipc_state_t *state = (const posix_ipc_state_t *)(const void *)handle;
    rte_status_t status;
 
    if ((state == NULL) || (out_message == NULL) || (buffer_size == 0U))
    {
        status = RTE_STATUS_INVALID_PARAM;
    }
    else if (buffer_size < state->message_size)
    {
        status = RTE_STATUS_INVALID_PARAM; /* a message would not fit */
    }
    else
    {
        status = read_message(state->read_fd, (uint8_t *)out_message, state->message_size, timeout_ms);
    }
    return status;
}
 
static rte_status_t osadapter_destroy(rte_ipc_handle_t handle)
{
    posix_ipc_state_t *state = (posix_ipc_state_t *)(void *)handle;
 
    if (state == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    (void)close(state->read_fd);
    (void)close(state->write_fd);
    return RTE_STATUS_OK;
}

static const rte_osadapter_ipc_t s_posix_ipc_osadapter = { osadapter_create, backend_send, backend_receive,
                                                          osadapter_destroy };

const rte_osadapter_ipc_t *rte_posix_osadapter_ipc(void)
{
    return &s_posix_ipc_osadapter;
}