view release on metacpan or  search on metacpan

etcd_common.c  view on Meta::CPAN

        case GRPC_STATUS_DEADLINE_EXCEEDED:
            return 1;
        default:
            return 0;
    }
}

/* Statuses a reconnect cannot fix; streams retry on any other end */
int is_permanent_status(grpc_status_code code) {
    switch (code) {
        case GRPC_STATUS_INVALID_ARGUMENT:
        case GRPC_STATUS_NOT_FOUND:
        case GRPC_STATUS_ALREADY_EXISTS:
        case GRPC_STATUS_PERMISSION_DENIED:
        case GRPC_STATUS_FAILED_PRECONDITION:
        case GRPC_STATUS_OUT_OF_RANGE:
        case GRPC_STATUS_UNIMPLEMENTED:
        case GRPC_STATUS_UNAUTHENTICATED:
            return 1;
        default:
            return 0;
    }
}

int etcd_is_no_leader(grpc_status_code status, grpc_slice status_details) {
    return status == GRPC_STATUS_UNAVAILABLE
        && grpc_slice_eq(status_details, grpc_slice_from_static_string("etcdserver: no leader"));
}

static void set_int_arg(grpc_arg *arg, const char *key, int value) {
    arg->type = GRPC_ARG_INTEGER;
    arg->key = (char *)key;
    arg->value.integer = value;
}

channel_ref_t *etcd_create_channel(ev_etcd_t *client, const char *target) {
    grpc_arg arg[9];
    grpc_channel_args args = { 0, arg };
    /* gRPC caps messages at 4 MiB; a watch replay or a range can be larger */
    set_int_arg(&arg[args.num_args++], "grpc.max_receive_message_length", -1);
    /* gRPC's default is 120 s, shared by every channel to the target */
    set_int_arg(&arg[args.num_args++], "grpc.max_reconnect_backoff_ms", 5000);
    if (client->tls_server_name) {
        arg[args.num_args].type = GRPC_ARG_STRING;
        arg[args.num_args].key = (char *)"grpc.ssl_target_name_override";
        arg[args.num_args++].value.string = client->tls_server_name;
    }
    if (client->keepalive_ms) {
        /* etcd GOAWAYs pings without calls; gRPC's own throttles would hide a dead
         * idle watch, and newer gRPC times pings out by ping_timeout_ms */
        set_int_arg(&arg[args.num_args++], "grpc.keepalive_time_ms", client->keepalive_ms);
        set_int_arg(&arg[args.num_args++], "grpc.keepalive_timeout_ms", client->keepalive_timeout_ms);
        set_int_arg(&arg[args.num_args++], "grpc.keepalive_permit_without_calls", 0);
        set_int_arg(&arg[args.num_args++], "grpc.http2.max_pings_without_data", 0);
        set_int_arg(&arg[args.num_args++], "grpc.http2.min_time_between_pings_ms", client->keepalive_ms);
        set_int_arg(&arg[args.num_args++], "grpc.http2.ping_timeout_ms", client->keepalive_timeout_ms);
    }
    /* grpc_channel_create arrived in gRPC 1.45 */
    grpc_channel *channel;
#ifdef HAVE_GRPC_NEW_CHANNEL_API
    grpc_channel_credentials *creds = client->creds
        ? client->creds : grpc_insecure_credentials_create();
    WITH_SIGNALS_BLOCKED(channel = grpc_channel_create(target, creds, &args));
    if (!client->creds) grpc_channel_credentials_release(creds);
#else
    WITH_SIGNALS_BLOCKED(channel = client->creds
        ? grpc_secure_channel_create(client->creds, target, &args, NULL)
        : grpc_insecure_channel_create(target, &args, NULL));
#endif
    if (!channel) return NULL;

    channel_ref_t *ref;
    Newx(ref, 1, channel_ref_t);
    ref->channel = channel;
    ref->refs = 1;
    return ref;
}

void etcd_channel_release(channel_ref_t *ref, int destroy) {
    if (ref && !--ref->refs) {
        /* A forked child discards only our wrapper, without touching gRPC. */
        if (destroy) grpc_channel_destroy(ref->channel);
        Safefree(ref);
    }
}

static void release_send_metadata(call_base_t *base) {
    if (base->owns_auth_value) {
        grpc_slice_unref(base->send_md[0].value);
        base->owns_auth_value = 0;
    }
}

void etcd_call_release(call_base_t *base) {
    etcd_channel_release(base->channel_ref, 1);
    base->channel_ref = NULL;
    release_send_metadata(base);
}

static void cancel_blocking_calls(ev_etcd_t *client, unsigned channel_gen, const char *message) {
    for (pending_call_t *pc = client->pending_calls; pc; pc = pc->next) {
        if (pc->call && pc->base.channel_gen == channel_gen
            && (pc->base.type == CALL_TYPE_LOCK || pc->base.type == CALL_TYPE_ELECTION_CAMPAIGN))
            grpc_call_cancel_with_status(pc->call, GRPC_STATUS_UNAVAILABLE, message, NULL);
    }
}

static void check_leader_failure(ev_etcd_t *client, unsigned channel_gen,
                                grpc_status_code status, grpc_slice status_details) {
    if (etcd_is_no_leader(status, status_details)) {
        /* Retired channels too: another failure may have moved the client already */
        cancel_blocking_calls(client, channel_gen, "etcdserver: no leader");
    }
}

void etcd_rotate_endpoint(ev_etcd_t *client) {
    int next = client->current_endpoint;
    if (client->endpoint_count > 1)
        next = (next + 1) % client->endpoint_count;

    channel_ref_t *ref = etcd_create_channel(client, client->endpoints[next]);
    if (!ref) return;

    channel_ref_t *old = client->channel_ref;