view release on metacpan or  search on metacpan

etcd_common.h  view on Meta::CPAN

#ifndef ETCD_COMMON_H
#define ETCD_COMMON_H

#include <EV/EVAPI.h>
#include <pthread.h>

#define RECONNECT_BACKOFF_SECONDS(attempt) \
    ((attempt) * 0.5 > 5.0 ? 5.0 : (attempt) * 0.5)

/* Retry pace after a member answered without a leader */
#define NO_LEADER_RETRY_SECONDS 1.0

/* Perl before 5.42 crashes running a %SIG handler on a thread without a Perl
 * context; threads inherit the starter's mask, so gRPC's and ours block all */
#define WITH_SIGNALS_BLOCKED(stmt) \
    do { \
        sigset_t _ws_all, _ws_old; \
        sigfillset(&_ws_all); \
        pthread_sigmask(SIG_SETMASK, &_ws_all, &_ws_old); \
        stmt; \
        pthread_sigmask(SIG_SETMASK, &_ws_old, NULL); \
    } while (0)

#include <grpc/grpc.h>
#ifdef HAVE_GRPC_CREDENTIALS_H
#include <grpc/credentials.h>
#else
#include <grpc/grpc_security.h>
#endif
#include <grpc/byte_buffer.h>
#include <grpc/byte_buffer_reader.h>

#include "kv.pb-c.h"
#include "rpc.pb-c.h"
#include "lock.pb-c.h"
#include "election.pb-c.h"

/* etcd's default MaxRequestBytes is 1.5 MiB */
#define ETCD_MAX_KEY_SIZE   (1024 * 1024)
#define ETCD_MAX_VALUE_SIZE (1024 * 1024)

#define VALIDATE_KEY_SIZE(key_len) \
    do { \
        if ((key_len) > ETCD_MAX_KEY_SIZE) { \
            croak("key too large: %zu bytes (max %d)", (size_t)(key_len), ETCD_MAX_KEY_SIZE); \
        } \
    } while (0)

#define VALIDATE_VALUE_SIZE(value_len) \
    do { \
        if ((value_len) > ETCD_MAX_VALUE_SIZE) { \
            croak("value too large: %zu bytes (max %d)", (size_t)(value_len), ETCD_MAX_VALUE_SIZE); \
        } \
    } while (0)

#define ETCD_MAX_USERNAME_SIZE  256
#define ETCD_MAX_PASSWORD_SIZE  4096

/* protobuf-c packs proto3 strings with strlen: a NUL would cut them short */
#define VALIDATE_NO_NUL(str, len, what) \
    do { \
        if (memchr((str), '\0', (len))) \
            croak("%s contains a NUL byte", what); \
    } while (0)

#define VALIDATE_NAME(str, len) \
    do { \
        if ((len) > ETCD_MAX_USERNAME_SIZE) { \
            croak("name too large: %zu bytes (max %d)", (size_t)(len), ETCD_MAX_USERNAME_SIZE); \
        } \
        VALIDATE_NO_NUL(str, len, "name"); \
    } while (0)

#define VALIDATE_PASSWORD(str, len) \
    do { \
        if ((len) > ETCD_MAX_PASSWORD_SIZE) { \
            croak("password too large: %zu bytes (max %d)", (size_t)(len), ETCD_MAX_PASSWORD_SIZE); \
        } \
        VALIDATE_NO_NUL(str, len, "password"); \
    } while (0)

#define ETCD_MAX_URL_SIZE  2048

#define VALIDATE_URL_SIZE(len) \
    do { \
        if ((len) > ETCD_MAX_URL_SIZE) { \

etcd_common.h  view on Meta::CPAN

    ev_timer reconnect_timer;
    ev_timer renew_timer;
    int client_owns;           /* dual ownership, see watch_call_t */
    int perl_owns;
} keepalive_call_t;

typedef struct observe_params {
    char *name;
    size_t name_len;
} observe_params_t;

typedef struct observe_call {
    call_base_t base;  /* Must be first */
    grpc_call *call;
    SV *callback;
    grpc_metadata_array initial_metadata;
    grpc_metadata_array trailing_metadata;
    grpc_byte_buffer *recv_buffer;
    grpc_status_code status;
    grpc_slice status_details;
    int active;
    struct ev_etcd_struct *client;
    struct observe_call *next;
    int auto_reconnect;
    int established;
    int reconnect_attempt;
    unsigned attempt_epoch;
    ev_timer reconnect_timer;
    observe_params_t params;
    int client_owns;           /* dual ownership, see watch_call_t */
    int perl_owns;
} observe_call_t;

typedef struct ev_etcd_struct {
    grpc_channel *channel;
    channel_ref_t *channel_ref;
    grpc_completion_queue *cq;

    /* cq_thread hands completions to the EV thread via event_queue + cq_async */
    pthread_t cq_thread;
    pthread_mutex_t queue_mutex; /* Protects event_queue */
    ev_async cq_async;
    queued_event_t *event_queue;
    queued_event_t *event_queue_tail;

    pending_call_t *pending_calls;
    watch_call_t *watches;
    keepalive_call_t *keepalives;
    observe_call_t *observes;
    int active;
    int in_callback;  /* Guard against freeing client during event processing */
    char *auth_token;
    size_t auth_token_len;
    int timeout_seconds;

    char **endpoints;
    int endpoint_count;
    int current_endpoint;
    unsigned channel_gen;  /* bumped on every endpoint switch */

    grpc_channel_credentials *creds;  /* NULL = insecure */
    char *tls_server_name;
    int keepalive_ms;          /* 0 = no keepalive pings */
    int keepalive_timeout_ms;

    int max_retries;
    /* Bumped by each answer without a leader: voids reconnects counted before it */
    unsigned no_leader_epoch;

    ev_timer health_timer;
    int is_healthy;
    SV *health_callback;
    pid_t owner_pid;
    struct ev_etcd_struct *next_live;  /* this process's live clients */
} ev_etcd_t;

static inline void link_pending_call(ev_etcd_t *client, pending_call_t *pc) {
    pc->next = client->pending_calls;
    if (pc->next)
        pc->next->pprev = &pc->next;
    pc->pprev = &client->pending_calls;
    client->pending_calls = pc;
}

static inline void unlink_pending_call(pending_call_t *pc) {
    if (!pc->pprev)
        return;
    *pc->pprev = pc->next;
    if (pc->next)
        pc->next->pprev = pc->pprev;
    pc->pprev = NULL;
}

typedef ev_etcd_t *EV__Etcd;
typedef watch_call_t *EV__Etcd__Watch;
typedef keepalive_call_t *EV__Etcd__Keepalive;
typedef observe_call_t *EV__Etcd__Observe;

static inline void init_call_base(call_base_t *base, call_type_t type) {
    base->type = type;
}

static inline void etcd_call_acquire(ev_etcd_t *client, call_base_t *base) {
    base->channel_gen = client->channel_gen;
    base->channel_ref = client->channel_ref;
    base->channel_ref->refs++;
}

#define VALIDATE_CALLBACK(cb) \
    do { \
        if (!SvROK(cb) || SvTYPE(SvRV(cb)) != SVt_PVCV) { \
            croak("callback must be a code reference"); \
        } \
    } while (0)

/* 32-bit IV perls go through NV, exact only to 2^53: fine for revisions,
 * lossy for lease, member and cluster ids, which use all 63/64 bits */
#if IVSIZE >= 8
#  define newSVi64(v) newSViv((IV)(v))
#  define newSVu64(v) newSVuv((UV)(v))
#  define SvI64(sv)   ((int64_t)SvIV(sv))