UBS Atomic 对外接口文档 ​

#include "ub_dist_tx_res.h"
#include <stdio.h>

static int app_logger(int level, const char *file, const char *func, uint32_t line, const char *message)
{
    printf("[%d] %s:%u %s: %s\n", level, file, line, func, message);
    return 0;
}

void setup_log(void)
{
    ub_atomic_register_log_func(app_logger);
    (void)ub_atomic_set_log_level(LOG_LEVEL_INFO);
}

ub_comm_queue_heartbeat_config_t req = {
    .heartbeat_interval_ms = 100,
    .check_interval_ms = 100,
    .timeout_ms = 1000,
};
ub_comm_queue_heartbeat_config_t eff = {0};
int ret = ub_comm_queue_config_heartbeat(&handle, &req, &eff);

#include "ub_dist_comm_queue.h"
#include <stdint.h>
#include <stdlib.h>

static void on_msg(const message_t *msg, void *ctx)
{
    (void)ctx;
    /* 处理 msg->body，长度为 msg->header.body_length */
}

int comm_example(void)
{
    void *init_mem = calloc(1, 4096);
    void *ring_mem0 = calloc(1, 1 << 20);
    void *ring_mem1 = calloc(1, 1 << 20);
    ub_ring_region_info_t entries[2] = {
        {.region = {.size = 1 << 20, .ptr = ring_mem0}, .node_id = 0},
        {.region = {.size = 1 << 20, .ptr = ring_mem1}, .node_id = 1},
    };
    ub_ring_region_map_t map = {.entries = entries, .count = 2};
    ub_ring_desc_t desc = {.ring_capacity = 1024, .max_msg_size = 256, .priority = 1};
    ub_comm_conf_t conf = {.cpu_id = -1, .max_nodes = 2, .current_node_id = 0, .num_rings = 1, .ring_descs = &desc};
    ub_shm_area_t init_region = {.size = 4096, .ptr = init_mem};
    ub_shm_comm_t handle = NULL;

    int ret = ub_comm_queue_init(&handle, &init_region, &map, &conf);
    if (ret != 0) {
        return ret;
    }
    ret = ub_comm_queue_register_process_func(&handle, 1, UB_FUNC_SYNC, on_msg, NULL);
    if (ret != 0) {
        (void)ub_comm_queue_deinit(&handle);
        return ret;
    }

    char body[] = "hello";
    message_t msg = {
        .header = {.src_thread_id = 1001, .body_length = (uint32_t)sizeof(body), .dest_node_id = 1,
                   .src_node_id = 0, .msg_type = 1, .priority = 1},
        .body = body,
    };
    ret = ub_comm_queue_send(&handle, &msg);
    (void)ub_comm_queue_deinit(&handle);
    return ret == UB_COMM_SEND_CONGESTED ? 0 : ret;
}

#include <stdbool.h>
#include "ub_dist_lock.h"
#include <stdlib.h>

void rw_lock_example(void)
{
    ub_rw_lock_t *lock = (ub_rw_lock_t *)calloc(1, UB_RW_LOCK_SIZE);
    ub_location_t loc = {.tid = 1001, .node_id = 1};
    ub_lock_config_t config = {.lease_time = 60000, .heartbeat_timeout = 500};
    ub_lock_policy_t policy = {.timeout_ts = 10000, .allow_delay_release = false, .recursive = false};

    ub_rw_lock_create(lock, &config, &loc);
    if (ub_rw_lock_x_lock(lock, &policy, &loc) == UB_LOCK_SUCCESS) {
        /* 临界区 */
        (void)ub_rw_lock_x_unlock(lock, &policy, &loc);
    }
    ub_rw_lock_free(lock, &loc);
    free(lock);
}

void mutex_lock_example(void)
{
    ub_mutex_lock_t *lock = (ub_mutex_lock_t *)calloc(1, UB_MUTEX_LOCK_SIZE);
    ub_location_t loc = {.tid = 1001, .node_id = 1};

    ub_mutex_lock_create(lock);
    if (ub_mutex_lock(lock, 10000, &loc) == UB_LOCK_SUCCESS) {
        /* 临界区 */
        (void)ub_mutex_unlock(lock, &loc);
    }
    ub_mutex_lock_free(lock);
    free(lock);
}

void spin_lock_example(void)
{
    ub_spin_lock_t *lock = (ub_spin_lock_t *)calloc(1, UB_SPIN_LOCK_SIZE);
    ub_location_t loc = {.tid = 1001, .node_id = 1};

    ub_spin_lock_init(lock);
    if (ub_spin_lock(lock, 10000, &loc) == UB_LOCK_SUCCESS) {
        /* 临界区（应极短） */
        (void)ub_spin_unlock(lock, &loc);
    }
    free(lock);
}

#include "ub_dist_tx_res.h"
#include <stdint.h>

int tx_res_example(void)
{
    uint64_t value = 0;
    uint64_t old_value = 0;

    if (ub_dist_tx_res_init(&value) != UB_RES_OK) {
        return -1;
    }
    if (ub_dist_tx_res_set(&value, 99) != UB_RES_OK) {
        return -1;
    }
    if (ub_dist_tx_res_fetch_add(&value, 1, &old_value) != UB_RES_OK) {
        return -1;
    }
    /* old_value == 99, value == 100 */

    /* fetch_xor: 0x64 ^ 0x0F = 0x6B, 旧值 0x64 */
    if (ub_dist_tx_res_fetch_xor(&value, 0x0F, &old_value) != UB_RES_OK) {
        return -1;
    }

    /* compare_exchange: 期望 0x6B 匹配，替换为 200 */
    uint64_t expected = 0x6B;
    int success = 0;
    if (ub_dist_tx_res_compare_exchange(&value, &expected, 200, &success) != UB_RES_OK) {
        return -1;
    }
    /* success == 1, value == 200 */

    /* fence: release + acquire 配对保证可见性 */
    ub_dist_tx_res_fence(UB_FENCE_RELEASE);
    ub_dist_tx_res_fence(UB_FENCE_ACQUIRE);

    return ub_dist_tx_res_get(&value, &old_value);
}