rte_posix_osadapter_channel_service.c

Go to the documentation of this file.

#define _POSIX_C_SOURCE 200809L

 
#include "rte/posix_osadapter/rte_posix_osadapter_channel_service.h"
 
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netdb.h>
#include <netinet/in.h>
#include <poll.h>
#include <stdio.h>
#include <time.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
 
#include "rte/oal/memory/rte_mem_util.h"
#include "rte_posix_osadapter_poll.h"
#include "rte_posix_osadapter_sockbuf.h"
 
typedef struct
{
    int fd;
    rte_netlink_config_t config;
    struct sockaddr_in peer_addr;
    socklen_t peer_addr_len;
    bool has_peer_addr;
    bool initialized;
    uint64_t last_rx_ms;    /* connect role: when a datagram last arrived (0 = none yet) */
    uint64_t last_hello_ms; /* connect role: when a hello was last sent */
} rte_posix_channel_state_t;
 
typedef char rte_posix_channel_state_fits_[(sizeof(rte_posix_channel_state_t) <=
                                             sizeof(rte_channel_service_storage_t)) ? 1 : -1];
 
/* A connect-role end that has learned its peer sends no hello any more, so a listen-role peer that restarted (its
 * socket is new and knows nobody) would stay silent towards it for good: the listener only learns a peer from a
 * datagram it receives. After this long without any datagram the connect end says hello again; a healthy listener
 * just acknowledges it. */
#define RTE_POSIX_CHANNEL_REHELLO_MS 2000U
 
static uint64_t channel_now_ms(void)
{
    struct timespec ts;
 
    (void)clock_gettime(CLOCK_MONOTONIC, &ts);
    return ((uint64_t)ts.tv_sec * 1000U) + ((uint64_t)ts.tv_nsec / 1000000U);
}
 
static rte_posix_channel_resolve_fn s_resolver;
static void *s_resolver_context;
 
static rte_posix_channel_state_t *channel_state(rte_channel_service_storage_t *storage)
{
    return (rte_posix_channel_state_t *)(void *)storage;
}
 
static rte_status_t set_nonblocking(int fd)
{
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0)
    {
        return RTE_STATUS_INTERNAL_ERROR;
    }
    if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0)
    {
        return RTE_STATUS_INTERNAL_ERROR;
    }
    return RTE_STATUS_OK;
}
 
static rte_status_t init_listen_socket(rte_posix_channel_state_t *state)
{
    int fd;
    int reuse = 1;
    struct sockaddr_in bind_addr;
 
    if (state->fd >= 0)
    {
        return RTE_STATUS_OK;
    }
 
    fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd >= 0)
    {
        /* Not inherited across execv(): the in-process REBOOT re-execs this binary, and an inherited listening socket
         * kept receiving datagrams for its port that nobody read (Docker: sockets piled up with every reboot). */
        (void)fcntl(fd, F_SETFD, FD_CLOEXEC);
    }
    if (fd < 0)
    {
        return RTE_STATUS_INTERNAL_ERROR;
    }
    (void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
#ifdef SO_REUSEPORT
    (void)setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &reuse, sizeof(reuse));
#endif
 
    rte_mem_set(&bind_addr, 0, sizeof(bind_addr));
    bind_addr.sin_family = AF_INET;
    bind_addr.sin_port = htons(state->config.port);
    if (state->config.host == NULL)
    {
        bind_addr.sin_addr.s_addr = INADDR_ANY;
    }
    else if (inet_pton(AF_INET, state->config.host, &bind_addr.sin_addr) != 1)
    {
        (void)close(fd);
        return RTE_STATUS_INVALID_PARAM;
    }
 
    if (bind(fd, (struct sockaddr *)&bind_addr, sizeof(bind_addr)) != 0)
    {
        (void)close(fd);
        return RTE_STATUS_INTERNAL_ERROR;
    }
    /* REQ-OAL-NETLINK-015: configured buffer sizes on every (re)init; both 0 -> no call at all. */
    {
        const rte_status_t buf_status = posix_socket_apply_buffers(fd, &state->config);
 
        if (buf_status != RTE_STATUS_OK)
        {
            (void)close(fd);
            return buf_status;
        }
    }
    if (set_nonblocking(fd) != RTE_STATUS_OK)
    {
        (void)close(fd);
        return RTE_STATUS_INTERNAL_ERROR;
    }
    state->fd = fd;
    return RTE_STATUS_OK;
}
 
static rte_status_t init_connect_socket(rte_posix_channel_state_t *state)
{
    struct addrinfo hints;
    struct addrinfo *resolved = NULL;
    char port_str[8];
    int fd;
 
    if (state->fd >= 0)
    {
        return RTE_STATUS_OK;
    }
 
    (void)snprintf(port_str, sizeof(port_str), "%u", (unsigned int)state->config.port);
    rte_mem_set(&hints, 0, sizeof(hints));
    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_DGRAM;
 
    if (getaddrinfo(state->config.host, port_str, &hints, &resolved) != 0)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    fd = socket(resolved->ai_family, resolved->ai_socktype, resolved->ai_protocol);
    if (fd >= 0)
    {
        /* Not inherited across execv(): the in-process REBOOT re-execs this binary, and an inherited listening socket
         * kept receiving datagrams for its port that nobody read (Docker: sockets piled up with every reboot). */
        (void)fcntl(fd, F_SETFD, FD_CLOEXEC);
    }
    if (fd < 0)
    {
        perror("init_connect_socket: socket");
        freeaddrinfo(resolved);
        return RTE_STATUS_INTERNAL_ERROR;
    }
    /* REQ-OAL-NETLINK-015: configured buffer sizes on every (re)init; both 0 -> no call at all. */
    {
        const rte_status_t buf_status = posix_socket_apply_buffers(fd, &state->config);
 
        if (buf_status != RTE_STATUS_OK)
        {
            (void)close(fd);
            freeaddrinfo(resolved);
            return buf_status;
        }
    }
    if (set_nonblocking(fd) != RTE_STATUS_OK)
    {
        perror("init_connect_socket: set_nonblocking");
        (void)close(fd);
        freeaddrinfo(resolved);
        return RTE_STATUS_INTERNAL_ERROR;
    }
 
    if (connect(fd, resolved->ai_addr, resolved->ai_addrlen) != 0)
    {
        perror("init_connect_socket: connect");
        (void)close(fd);
        freeaddrinfo(resolved);
        return RTE_STATUS_INTERNAL_ERROR;
    }
    freeaddrinfo(resolved);
    state->fd = fd;
    {
        uint8_t hello = 0xA5U;
        (void)send(state->fd, &hello, 1, 0);
    }
    return RTE_STATUS_OK;
}
 
static rte_status_t osadapter_setup(rte_channel_service_storage_t *storage, const char *channel_name)
{
    rte_posix_channel_state_t *state;
    rte_status_t status;
 
    if ((storage == NULL) || (channel_name == NULL) || (s_resolver == NULL))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    state = channel_state(storage);
    rte_mem_set(state, 0, sizeof(*state));
    state->fd = -1;
    status = s_resolver(channel_name, &state->config, s_resolver_context);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
    state->initialized = true;
    if (state->config.role == RTE_NETLINK_ROLE_LISTEN)
    {
        return init_listen_socket(state);
    }
    return init_connect_socket(state);
}
 
/* Shared by osadapter_read and osadapter_read_ex: recvfrom()/recv() already return the true
 * datagram length in n before either wrapper discards it - out_actual_size, when non-NULL, is
 * set to that value on RTE_STATUS_OK. Behaviour is otherwise identical to the old single
 * osadapter_read function this replaces (same hello/ack handshake handling, same status
 * mapping), verified by construction rather than duplicated by hand. */
static rte_status_t do_read(rte_channel_service_storage_t *storage, void *data, size_t data_size,
                             rte_duration_ms_t timeout_ms, size_t *out_actual_size)
{
    rte_posix_channel_state_t *state;
    ssize_t n;
 
    if ((storage == NULL) || (data == NULL) || (data_size == 0U))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    state = channel_state(storage);
    if (!state->initialized)
    {
        return RTE_STATUS_NOT_INITIALIZED;
    }
    if (state->fd < 0)
    {
        if (state->config.role == RTE_NETLINK_ROLE_LISTEN)
        {
            if (init_listen_socket(state) != RTE_STATUS_OK)
            {
                return RTE_STATUS_TIMEOUT;
            }
        }
        else
        {
            if (init_connect_socket(state) != RTE_STATUS_OK)
            {
                return RTE_STATUS_TIMEOUT;
            }
        }
    }
 
    if (timeout_ms > 0U)
    {
        struct pollfd pfd;
        int rc;
 
        pfd.fd = state->fd;
        pfd.events = POLLIN;
        pfd.revents = 0;
        rc = poll(&pfd, 1, posix_poll_timeout_ms(timeout_ms));
        if (rc == 0)
        {
            return RTE_STATUS_TIMEOUT;
        }
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                return RTE_STATUS_TIMEOUT;
            }
            return RTE_STATUS_INTERNAL_ERROR;
        }
    }
 
    if (state->config.role == RTE_NETLINK_ROLE_LISTEN)
    {
        while (1)
        {
            struct sockaddr_in src_addr;
            socklen_t addr_len = (socklen_t)sizeof(src_addr);
            rte_mem_set(&src_addr, 0, sizeof(src_addr));
 
            n = recvfrom(state->fd, data, data_size, 0, (struct sockaddr *)&src_addr, &addr_len);
            if (n > 0)
            {
                state->peer_addr = src_addr;
                state->peer_addr_len = addr_len;
                state->has_peer_addr = true;
 
                if ((n == 1) && (((const uint8_t *)data)[0] == 0xA5U))
                {
                    uint8_t ack = 0x5AU;
                    (void)sendto(state->fd, &ack, 1, 0, (const struct sockaddr *)&src_addr, addr_len);
                    continue;
                }
            }
            break;
        }
    }
    else
    {
        while (1)
        {
            const uint64_t now_ms = channel_now_ms();
 
            if (!state->has_peer_addr)
            {
                uint8_t hello = 0xA5U;
                (void)send(state->fd, &hello, 1, 0);
                state->last_hello_ms = now_ms;
            }
            else
            {
                if (state->last_rx_ms == 0U)
                {
                    state->last_rx_ms = now_ms;
                }
                if (((now_ms - state->last_rx_ms) >= RTE_POSIX_CHANNEL_REHELLO_MS) &&
                    ((now_ms - state->last_hello_ms) >= RTE_POSIX_CHANNEL_REHELLO_MS))
                {
                    uint8_t hello = 0xA5U;
                    (void)send(state->fd, &hello, 1, 0);
                    state->last_hello_ms = now_ms;
                }
            }
            n = recv(state->fd, data, data_size, 0);
            if (n > 0)
            {
                state->last_rx_ms = now_ms;
                if ((n == 1) && (((const uint8_t *)data)[0] == 0x5AU))
                {
                    state->has_peer_addr = true;
                    continue;
                }
                if ((n == 1) && (((const uint8_t *)data)[0] == 0xA5U))
                {
                    uint8_t ack = 0x5AU;
                    (void)send(state->fd, &ack, 1, 0);
                    state->has_peer_addr = true;
                    continue;
                }
            }
            break;
        }
    }
 
    if (n < 0)
    {
        if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR) || (errno == ECONNREFUSED) ||
            (errno == ENOTCONN) || (errno == EHOSTUNREACH) || (errno == EPIPE))
        {
            return RTE_STATUS_TIMEOUT;
        }
        return RTE_STATUS_INTERNAL_ERROR;
    }
    if (n == 0)
    {
        return RTE_STATUS_TIMEOUT;
    }
    if (out_actual_size != NULL)
    {
        *out_actual_size = (size_t)n;
    }
    return RTE_STATUS_OK;
}
 
static rte_status_t osadapter_read(rte_channel_service_storage_t *storage, void *data, size_t data_size,
                                    rte_duration_ms_t timeout_ms)
{
    return do_read(storage, data, data_size, timeout_ms, NULL);
}
 
static rte_status_t osadapter_read_ex(rte_channel_service_storage_t *storage, void *data, size_t data_size,
                                       rte_duration_ms_t timeout_ms, size_t *out_actual_size)
{
    if (out_actual_size == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    return do_read(storage, data, data_size, timeout_ms, out_actual_size);
}
 
static rte_status_t backend_send(rte_channel_service_storage_t *storage, const void *data, size_t data_size,
                                  rte_duration_ms_t timeout_ms)
{
    rte_posix_channel_state_t *state;
    ssize_t n;
 
    if ((storage == NULL) || (data == NULL) || (data_size == 0U))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    state = channel_state(storage);
    if (!state->initialized)
    {
        return RTE_STATUS_NOT_INITIALIZED;
    }
    if (state->fd < 0)
    {
        if (state->config.role == RTE_NETLINK_ROLE_LISTEN)
        {
            if (init_listen_socket(state) != RTE_STATUS_OK)
            {
                return RTE_STATUS_TIMEOUT;
            }
        }
        else
        {
            if (init_connect_socket(state) != RTE_STATUS_OK)
            {
                return RTE_STATUS_TIMEOUT;
            }
        }
    }
 
    if (timeout_ms > 0U)
    {
        struct pollfd pfd;
        int rc;
 
        pfd.fd = state->fd;
        pfd.events = POLLOUT;
        pfd.revents = 0;
        rc = poll(&pfd, 1, posix_poll_timeout_ms(timeout_ms));
        if (rc == 0)
        {
            return RTE_STATUS_TIMEOUT;
        }
        if (rc < 0)
        {
            if (errno == EINTR)
            {
                return RTE_STATUS_TIMEOUT;
            }
            return RTE_STATUS_INTERNAL_ERROR;
        }
    }
 
    if (state->config.role == RTE_NETLINK_ROLE_LISTEN)
    {
        if (!state->has_peer_addr)
        {
            /* REQ-CHANSVC-001 (Platform_RTE, OSA rule R5): no peer has spoken to this listen socket yet, so there is
             * no address to send to. Nothing is sent, and the caller is told so instead of getting OK. */
            return RTE_STATUS_NOT_CONNECTED;
        }
        n = sendto(state->fd, data, data_size, 0, (const struct sockaddr *)&state->peer_addr, state->peer_addr_len);
    }
    else
    {
        n = send(state->fd, data, data_size, 0);
    }
 
    if (n < 0)
    {
        if ((errno == EAGAIN) || (errno == EWOULDBLOCK) || (errno == EINTR) || (errno == ECONNREFUSED) ||
            (errno == ENOTCONN) || (errno == EHOSTUNREACH) || (errno == EADDRNOTAVAIL) || (errno == EPIPE))
        {
            return RTE_STATUS_TIMEOUT;
        }
        return RTE_STATUS_INTERNAL_ERROR;
    }
    return RTE_STATUS_OK;
}
 
static rte_status_t osadapter_close(rte_channel_service_storage_t *storage)
{
    rte_posix_channel_state_t *state;
 
    if (storage == NULL)
    {
        return RTE_STATUS_OK;
    }
    state = channel_state(storage);
    if (!state->initialized)
    {
        return RTE_STATUS_OK;
    }
    state->initialized = false;
    if (state->fd >= 0)
    {
        (void)close(state->fd);
        state->fd = -1;
    }
    state->has_peer_addr = false;
    return RTE_STATUS_OK;
}
 
static const rte_osadapter_channel_service_t s_osadapter = {
    osadapter_setup,
    osadapter_read,
    backend_send,
    osadapter_close,
    osadapter_read_ex
};
 
rte_status_t rte_posix_osadapter_channel_service_register_resolver(rte_posix_channel_resolve_fn resolver, void *context)
{
    if (resolver == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    s_resolver = resolver;
    s_resolver_context = context;
    return rte_osadapter_channel_service_register(&s_osadapter);
}
 
const rte_osadapter_channel_service_t *rte_posix_osadapter_channel_service(void)
{
    return &s_osadapter;
}