as_event_internal.h

Go to the documentation of this file.

/*
 * Copyright 2008-2023 Aerospike, Inc.
 *
 * Portions may be licensed to Aerospike, Inc. under one or more contributor
 * license agreements.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
#pragma once

#include <aerospike/as_admin.h>
#include <aerospike/as_cluster.h>
#include <aerospike/as_listener.h>
#include <aerospike/as_queue.h>
#include <aerospike/as_proto.h>
#include <aerospike/as_socket.h>
#include <citrusleaf/cf_ll.h>
#include <pthread.h>

#if defined(AS_USE_LIBEV)
#include <ev.h>
#elif defined(AS_USE_LIBUV)
#include <uv.h>
struct as_uv_tls;
#elif defined(AS_USE_LIBEVENT)
#include <event2/event.h>
#else
#endif

#ifdef __cplusplus
extern "C" {
#endif

/******************************************************************************
 * TYPES
 *****************************************************************************/
    
#define AS_ASYNC_STATE_UNREGISTERED 0
#define AS_ASYNC_STATE_REGISTERED 1
#define AS_ASYNC_STATE_DELAY_QUEUE 2
#define AS_ASYNC_STATE_CONNECT 3
#define AS_ASYNC_STATE_TLS_CONNECT 4
#define AS_ASYNC_STATE_AUTH_WRITE 5
#define AS_ASYNC_STATE_AUTH_READ_HEADER 6
#define AS_ASYNC_STATE_AUTH_READ_BODY 7
#define AS_ASYNC_STATE_COMMAND_WRITE 8
#define AS_ASYNC_STATE_COMMAND_READ_HEADER 9
#define AS_ASYNC_STATE_COMMAND_READ_BODY 10
#define AS_ASYNC_STATE_QUEUE_ERROR 11
#define AS_ASYNC_STATE_RETRY 12

#define AS_ASYNC_FLAGS_DESERIALIZE 1
#define AS_ASYNC_FLAGS_READ 2
#define AS_ASYNC_FLAGS_HAS_TIMER 4
#define AS_ASYNC_FLAGS_USING_SOCKET_TIMER 8
#define AS_ASYNC_FLAGS_EVENT_RECEIVED 16
#define AS_ASYNC_FLAGS_FREE_BUF 32
#define AS_ASYNC_FLAGS_LINEARIZE 64
#define AS_ASYNC_FLAGS_HEAP_REC 128

#define AS_ASYNC_AUTH_RETURN_CODE 1

#define AS_EVENT_CONNECTION_COMPLETE 0
#define AS_EVENT_CONNECTION_PENDING 1
#define AS_EVENT_CONNECTION_ERROR 2

#define AS_EVENT_QUEUE_INITIAL_CAPACITY 256
    
struct as_event_command;
struct as_event_executor;

typedef struct {
#if defined(AS_USE_LIBEV)
    struct ev_io watcher;
    as_socket socket;
#elif defined(AS_USE_LIBUV)
    uv_tcp_t socket;
    struct as_uv_tls* tls;
    // Reuse memory for requests, because only one request is active at a time.
    union {
        uv_connect_t connect;
        uv_write_t write;
    } req;
    uint64_t last_used;
#elif defined(AS_USE_LIBEVENT)
    struct event watcher;
    as_socket socket;
#else
#endif
    int watching;
    bool pipeline;
} as_event_connection;

typedef struct {
    as_event_connection base;
    struct as_event_command* cmd;
} as_async_connection;

typedef struct {
    as_pipe_listener listener;
    void* udata;
} as_queued_pipe_cb;

typedef void (*as_event_executable) (as_event_loop* event_loop, void* udata);
typedef bool (*as_event_parse_results_fn) (struct as_event_command* cmd);
typedef void (*as_event_executor_complete_fn) (struct as_event_executor* executor);

typedef struct as_event_command {
#if defined(AS_USE_LIBEV)
    struct ev_timer timer;
#elif defined(AS_USE_LIBUV)
    uv_timer_t timer;
#elif defined(AS_USE_LIBEVENT)
    struct event timer;
#else
#endif
    uint64_t total_deadline;
    uint32_t socket_timeout;
    uint32_t max_retries;
    uint32_t iteration;
    as_policy_replica replica;
    as_event_loop* event_loop;
    as_event_state* event_state;
    as_event_connection* conn;
    as_cluster* cluster;
    as_node* node;
    const char* ns;
    void* partition;  // as_partition* or as_partition_shm*
    void* udata;
    as_event_parse_results_fn parse_results;
    as_pipe_listener pipe_listener;
    cf_ll_element pipe_link;
    
    uint8_t* buf;
    uint32_t command_sent_counter;
    uint32_t write_offset;
    uint32_t write_len;
    uint32_t read_capacity;
    uint32_t len;
    uint32_t pos;

    uint8_t type;
    uint8_t proto_type;
    uint8_t proto_type_rcv;
    uint8_t state;
    uint8_t flags;
    uint8_t replica_size;
    uint8_t replica_index;
    uint8_t replica_index_sc; // Used in batch only.
} as_event_command;

typedef struct {
    as_event_executable executable;
    void* udata;
} as_event_commander;

typedef struct as_event_executor {
    pthread_mutex_t lock;
    struct as_event_command** commands;
    as_event_loop* event_loop;
    as_event_executor_complete_fn complete_fn;
    void* udata;
    as_error* err;
    char* ns;
    uint64_t cluster_key;
    uint32_t max_concurrent;
    uint32_t max;
    uint32_t count;
    uint32_t queued;
    bool notify;
    bool valid;
} as_event_executor;

/******************************************************************************
 * COMMON FUNCTIONS
 *****************************************************************************/

as_status
as_event_command_execute(as_event_command* cmd, as_error* err);

void
as_event_command_schedule(as_event_command* cmd);

bool
as_event_proto_parse(as_event_command* cmd, as_proto* proto);

bool
as_event_proto_parse_auth(as_event_command* cmd, as_proto* proto);

bool
as_event_decompress(as_event_command* cmd);

void
as_event_process_timer(as_event_command* cmd);

void
as_event_socket_timeout(as_event_command* cmd);

void
as_event_total_timeout(as_event_command* cmd);

bool
as_event_command_retry(as_event_command* cmd, bool timeout);
    
void
as_event_execute_retry(as_event_command* cmd);

void
as_event_query_complete(as_event_command* cmd);

void
as_event_batch_complete(as_event_command* cmd);

void
as_event_executor_error(as_event_executor* executor, as_error* err, uint32_t command_count);

void
as_event_executor_cancel(as_event_executor* executor, uint32_t queued_count);

void
as_event_executor_complete(as_event_executor* executor);

void
as_event_error_callback(as_event_command* cmd, as_error* err);

void
as_event_notify_error(as_event_command* cmd, as_error* err);

void
as_event_parse_error(as_event_command* cmd, as_error* err);

void
as_event_socket_error(as_event_command* cmd, as_error* err);

void
as_event_response_error(as_event_command* cmd, as_error* err);

bool
as_event_command_parse_result(as_event_command* cmd);
    
bool
as_event_command_parse_header(as_event_command* cmd);

bool
as_event_command_parse_success_failure(as_event_command* cmd);

bool
as_event_command_parse_info(as_event_command* cmd);

void
as_event_command_free(as_event_command* cmd);

void
as_event_connector_success(as_event_command* cmd);

void
as_event_create_connections(as_node* node, as_async_conn_pool* pools);

void
as_event_close_cluster(as_cluster* cluster);

/******************************************************************************
 * IMPLEMENTATION SPECIFIC FUNCTIONS
 *****************************************************************************/

bool
as_event_create_loop(as_event_loop* event_loop);

void
as_event_register_external_loop(as_event_loop* event_loop);

/**
 * Schedule execution of function on specified event loop.
 * Command is placed on event loop queue and is never executed directly.
 */
bool
as_event_execute(as_event_loop* event_loop, as_event_executable executable, void* udata);

void
as_event_command_write_start(as_event_command* cmd);

void
as_event_connect(as_event_command* cmd, as_async_conn_pool* pool);

void
as_event_node_destroy(as_node* node);

/******************************************************************************
 * LIBEV INLINE FUNCTIONS
 *****************************************************************************/

#if defined(AS_USE_LIBEV)

void as_ev_timer_cb(struct ev_loop* loop, ev_timer* timer, int revents);
void as_ev_repeat_cb(struct ev_loop* loop, ev_timer* timer, int revents);

static inline bool
as_event_conn_current_trim(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_trim(conn->socket.last_used, max_socket_idle_ns);
}

static inline bool
as_event_conn_current_tran(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_tran(conn->socket.last_used, max_socket_idle_ns);
}

static inline int
as_event_conn_validate(as_event_connection* conn)
{
    return as_socket_validate_fd(conn->socket.fd);
}

static inline void
as_event_close_connection(as_event_connection* conn)
{
    as_socket_close(&conn->socket);
    cf_free(conn);
}

static inline void
as_event_set_conn_last_used(as_event_connection* conn)
{
    conn->socket.last_used = cf_getns();
}

static inline void
as_event_timer_once(as_event_command* cmd, uint64_t timeout)
{
    ev_timer_init(&cmd->timer, as_ev_timer_cb, (double)timeout / 1000.0, 0.0);
    cmd->timer.data = cmd;
    ev_timer_start(cmd->event_loop->loop, &cmd->timer);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER;
}

static inline void
as_event_timer_repeat(as_event_command* cmd, uint64_t repeat)
{
    ev_init(&cmd->timer, as_ev_repeat_cb);
    cmd->timer.repeat = (double)repeat / 1000.0;
    cmd->timer.data = cmd;
    ev_timer_again(cmd->event_loop->loop, &cmd->timer);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER | AS_ASYNC_FLAGS_USING_SOCKET_TIMER;
}

static inline void
as_event_timer_again(as_event_command* cmd)
{
    ev_timer_again(cmd->event_loop->loop, &cmd->timer);
}

static inline void
as_event_timer_stop(as_event_command* cmd)
{
    if (cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER) {
        ev_timer_stop(cmd->event_loop->loop, &cmd->timer);
    }
}

static inline void
as_event_stop_watcher(as_event_command* cmd, as_event_connection* conn)
{
    ev_io_stop(cmd->event_loop->loop, &conn->watcher);
    conn->watching = 0;
}

static inline void
as_event_stop_read(as_event_connection* conn)
{
    // This method only needed for libuv pipelined connections.
}

static inline void
as_event_command_release(as_event_command* cmd)
{
    as_event_command_free(cmd);
}

/******************************************************************************
 * LIBUV INLINE FUNCTIONS
 *****************************************************************************/

#elif defined(AS_USE_LIBUV)

void as_uv_timer_cb(uv_timer_t* timer);
void as_uv_repeat_cb(uv_timer_t* timer);
void as_event_close_connection(as_event_connection* conn);

static inline bool
as_event_conn_current_trim(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_trim(conn->last_used, max_socket_idle_ns);
}

static inline bool
as_event_conn_current_tran(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_tran(conn->last_used, max_socket_idle_ns);
}

static inline int
as_event_conn_validate(as_event_connection* conn)
{
    // Libuv does not have a peek function, so use fd directly.
    uv_os_fd_t fd;
    
    if (uv_fileno((uv_handle_t*)&conn->socket, &fd) == 0) {
        return as_socket_validate_fd((as_socket_fd)fd);
    }
    return -1;
}

static inline void
as_event_set_conn_last_used(as_event_connection* conn)
{
    conn->last_used = cf_getns();
}

static inline void
as_event_timer_once(as_event_command* cmd, uint64_t timeout)
{
    if (!(cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER)) {
        uv_timer_init(cmd->event_loop->loop, &cmd->timer);
        cmd->timer.data = cmd;
    }
    uv_timer_start(&cmd->timer, as_uv_timer_cb, timeout, 0);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER;
}

static inline void
as_event_timer_repeat(as_event_command* cmd, uint64_t repeat)
{
    if (!(cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER)) {
        uv_timer_init(cmd->event_loop->loop, &cmd->timer);
        cmd->timer.data = cmd;
    }
    uv_timer_start(&cmd->timer, as_uv_repeat_cb, repeat, repeat);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER | AS_ASYNC_FLAGS_USING_SOCKET_TIMER;
}

static inline void
as_event_timer_again(as_event_command* cmd)
{
    // libuv socket timers automatically repeat.
}

static inline void
as_event_timer_stop(as_event_command* cmd)
{
    if (cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER) {
        uv_timer_stop(&cmd->timer);
    }
}

static inline void
as_event_stop_watcher(as_event_command* cmd, as_event_connection* conn)
{
    // uv_read_stop() will handle case where read is already stopped.
    // Do not set watching to zero because conn is still initialized and active.
    // libuv works differently here.
    uv_read_stop((uv_stream_t*)conn);
}

static inline void
as_event_stop_read(as_event_connection* conn)
{
    uv_read_stop((uv_stream_t*)conn);
}

void
as_uv_timer_closed(uv_handle_t* handle);

static inline void
as_event_command_release(as_event_command* cmd)
{
    if (cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER) {
        // libuv requires that cmd can't be freed until timer is closed.
        uv_close((uv_handle_t*)&cmd->timer, as_uv_timer_closed);
    }
    else {
        as_event_command_free(cmd);
    }
}

/******************************************************************************
 * LIBEVENT INLINE FUNCTIONS
 *****************************************************************************/

#elif defined(AS_USE_LIBEVENT)

void as_libevent_timer_cb(evutil_socket_t sock, short events, void* udata);
void as_libevent_repeat_cb(evutil_socket_t sock, short events, void* udata);

static inline bool
as_event_conn_current_trim(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_trim(conn->socket.last_used, max_socket_idle_ns);
}

static inline bool
as_event_conn_current_tran(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return as_socket_current_tran(conn->socket.last_used, max_socket_idle_ns);
}

static inline int
as_event_conn_validate(as_event_connection* conn)
{
    return as_socket_validate_fd(conn->socket.fd);
}

static inline void
as_event_close_connection(as_event_connection* conn)
{
    as_socket_close(&conn->socket);
    cf_free(conn);
}

static inline void
as_event_set_conn_last_used(as_event_connection* conn)
{
    conn->socket.last_used = cf_getns();
}

static inline void
as_event_timer_once(as_event_command* cmd, uint64_t timeout)
{
    evtimer_assign(&cmd->timer, cmd->event_loop->loop, as_libevent_timer_cb, cmd);
    struct timeval tv;
    tv.tv_sec = (uint32_t)timeout / 1000;
    tv.tv_usec = ((uint32_t)timeout % 1000) * 1000;
    evtimer_add(&cmd->timer, &tv);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER;
}

static inline void
as_event_timer_repeat(as_event_command* cmd, uint64_t repeat)
{
    event_assign(&cmd->timer, cmd->event_loop->loop, -1, EV_PERSIST, as_libevent_repeat_cb, cmd);
    struct timeval tv;
    tv.tv_sec = (uint32_t)repeat / 1000;
    tv.tv_usec = ((uint32_t)repeat % 1000) * 1000;
    evtimer_add(&cmd->timer, &tv);
    cmd->flags |= AS_ASYNC_FLAGS_HAS_TIMER | AS_ASYNC_FLAGS_USING_SOCKET_TIMER;
}

static inline void
as_event_timer_again(as_event_command* cmd)
{
    // libevent socket timers automatically repeat.
}

static inline void
as_event_timer_stop(as_event_command* cmd)
{
    if (cmd->flags & AS_ASYNC_FLAGS_HAS_TIMER) {
        evtimer_del(&cmd->timer);
    }
}

static inline void
as_event_stop_watcher(as_event_command* cmd, as_event_connection* conn)
{
    event_del(&conn->watcher);
    conn->watching = 0;
}

static inline void
as_event_stop_read(as_event_connection* conn)
{
    // This method only needed for libuv pipelined connections.
}

static inline void
as_event_command_release(as_event_command* cmd)
{
    as_event_command_free(cmd);
}

/******************************************************************************
 * EVENT_LIB NOT DEFINED INLINE FUNCTIONS
 *****************************************************************************/

#else

static inline bool
as_event_conn_current_trim(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return false;
}

static inline bool
as_event_conn_current_tran(as_event_connection* conn, uint64_t max_socket_idle_ns)
{
    return false;
}

static inline int
as_event_conn_validate(as_event_connection* conn)
{
    return -1;
}

static inline void
as_event_close_connection(as_event_connection* conn)
{
}

static inline void
as_event_set_conn_last_used(as_event_connection* conn)
{
}

static inline void
as_event_timer_once(as_event_command* cmd, uint64_t timeout)
{
}

static inline void
as_event_timer_repeat(as_event_command* cmd, uint64_t repeat)
{
}

static inline void
as_event_timer_again(as_event_command* cmd)
{
}

static inline void
as_event_timer_stop(as_event_command* cmd)
{
}

static inline void
as_event_stop_watcher(as_event_command* cmd, as_event_connection* conn)
{
}

static inline void
as_event_stop_read(as_event_connection* conn)
{
}

static inline void
as_event_command_release(as_event_command* cmd)
{
}

#endif
    
/******************************************************************************
 * COMMON INLINE FUNCTIONS
 *****************************************************************************/

static inline as_event_loop*
as_event_assign(as_event_loop* event_loop)
{
    // Assign event loop using round robin distribution if not specified.
    return event_loop ? event_loop : as_event_loop_get();
}

static inline void
as_event_set_auth_write(as_event_command* cmd, as_session* session)
{
    // Authentication write buffer is always located after command write buffer.
    uint8_t* buf = (uint8_t*)cmd + cmd->write_offset + cmd->write_len;
    uint32_t len = as_authenticate_set(cmd->cluster, session, buf);
    cmd->len = cmd->write_len + len;
    cmd->pos = cmd->write_len;
}

static inline void
as_event_set_auth_read_header(as_event_command* cmd)
{
    // Authenticate read buffer uses the standard read buffer (buf).
    cmd->len = sizeof(as_proto);
    cmd->pos = 0;
    cmd->state = AS_ASYNC_STATE_AUTH_READ_HEADER;
}
    
static inline bool
as_event_set_auth_parse_header(as_event_command* cmd)
{
    // Authenticate read buffer uses the standard read buffer (buf).
    as_proto* proto = (as_proto*)cmd->buf;

    if (! as_event_proto_parse_auth(cmd, proto)) {
        return false;
    }

    cmd->len = (uint32_t)proto->sz;
    cmd->pos = 0;
    cmd->state = AS_ASYNC_STATE_AUTH_READ_BODY;
    return true;
}

static inline void
as_event_set_write(as_event_command* cmd)
{
    cmd->len = cmd->write_len;
    cmd->pos = 0;
}

static inline void
as_async_conn_pool_init(as_async_conn_pool* pool, uint32_t min_size, uint32_t max_size)
{
    as_queue_init(&pool->queue, sizeof(void*), max_size);
    pool->min_size = min_size;
    pool->limit = max_size;
    pool->opened = 0;
    pool->closed = 0;
}

static inline bool
as_async_conn_pool_incr_total(as_async_conn_pool* pool)
{
    if (pool->queue.total >= pool->limit) {
        return false;
    }
    pool->queue.total++;
    return true;
}

static inline bool
as_async_conn_pool_push_head(as_async_conn_pool* pool, as_event_connection* conn)
{
    if (pool->queue.total > pool->limit) {
        return false;
    }
    return as_queue_push_head(&pool->queue, &conn);
}

static inline bool
as_async_conn_pool_push(as_async_conn_pool* pool, as_event_connection* conn)
{
    if (pool->queue.total > pool->limit) {
        return false;
    }
    return as_queue_push(&pool->queue, &conn);
}

static inline void
as_event_release_connection(as_event_connection* conn, as_async_conn_pool* pool)
{
    as_event_close_connection(conn);
    as_queue_decr_total(&pool->queue);
    pool->closed++;
}

static inline void
as_event_release_async_connection(as_event_command* cmd)
{
    as_async_conn_pool* pool = &cmd->node->async_conn_pools[cmd->event_loop->index];
    as_event_release_connection(cmd->conn, pool);
    as_node_incr_error_count(cmd->node);
}

static inline void
as_event_decr_conn(as_event_command* cmd)
{
    as_async_conn_pool* pool = cmd->pipe_listener != NULL ?
        &cmd->node->pipe_conn_pools[cmd->event_loop->index] :
        &cmd->node->async_conn_pools[cmd->event_loop->index];

    as_queue_decr_total(&pool->queue);
}

static inline void
as_event_connection_timeout(as_event_command* cmd, as_async_conn_pool* pool)
{
    as_event_connection* conn = cmd->conn;

    if (conn) {
        if (conn->watching > 0) {
            as_event_stop_watcher(cmd, conn);
            as_event_release_connection(conn, pool);
            as_node_incr_error_count(cmd->node);
        }
        else {
            cf_free(conn);
            as_queue_decr_total(&pool->queue);
            pool->closed++;
        }
    }
}

static inline bool
as_event_socket_retry(as_event_command* cmd)
{
    if (cmd->pipe_listener) {
        return false;
    }

    as_event_stop_watcher(cmd, cmd->conn);
    as_event_release_async_connection(cmd);
    return as_event_command_retry(cmd, false);
}

static inline void
as_event_command_destroy(as_event_command* cmd)
{
    // Use this function to free batch/scan/query commands that were never started.
    as_node_release(cmd->node);
    cf_free(cmd);
}

static inline void
as_event_loop_destroy(as_event_loop* event_loop)
{
    as_queue_destroy(&event_loop->queue);
    as_queue_destroy(&event_loop->delay_queue);
    as_queue_destroy(&event_loop->pipe_cb_queue);
    pthread_mutex_destroy(&event_loop->lock);
}

#ifdef __cplusplus
} // end extern "C"
#endif