Redis-Cluster-Fast
view release on metacpan or search on metacpan
my $redis = Redis::Cluster::Fast->new(
startup_nodes => [
'localhost:9000',
'localhost:9001',
'localhost:9002',
'localhost:9003',
'localhost:9004',
'localhost:9005',
],
connect_timeout => 0.05,
command_timeout => 0.05,
max_retry_count => 10,
);
$redis->set('test', 123);
# '123'
my $str = $redis->get('test');
$redis->mset('{my}foo', 'hoge', '{my}bar', 'fuga');
You can then perform a blocking wait for those responses later, if needed.
Executes one iteration of the event loop to process any pending commands that have not yet been sent
and any incoming responses from Redis.
If there are events that can be triggered immediately, they will all be processed.
In other words, if there are unsent commands, they will be pipelined and sent,
and if there are already-received responses, their corresponding callbacks will be executed.
If there are no events that can be triggered immediately: there are neither unsent commands nor any Redis responses available to read,
but unprocessed callbacks remain, then this method will block for up to `command_timeout` while waiting for a response from Redis.
When a timeout occurs, an error will be propagated to the corresponding callback(s).
The return value can be either 1 for success (e.g., commands sent or responses read),
0 for no callbacks remained, or undef for other errors.
### Notes
- Be aware that the timeout check will only be triggered when there are neither unsent commands nor Redis responses available to read.
If a timeout occurs, all remaining commands on that node will time out as well.
- Internally, this method calls `event_base_loop(..., EVLOOP_ONCE)`, which
deps/hiredis-cluster/hircluster.c view on Meta::CPAN
}
if (ip == NULL || port <= 0) {
__redisClusterSetError(cc, REDIS_ERR_OTHER, "Ip or port error!");
goto error;
}
redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, ip, port);
options.connect_timeout = cc->connect_timeout;
options.command_timeout = cc->command_timeout;
c = redisConnectWithOptions(&options);
if (c == NULL) {
__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
}
if (cc->on_connect) {
cc->on_connect(c, c->err ? REDIS_ERR : REDIS_OK);
}
deps/hiredis-cluster/hircluster.c view on Meta::CPAN
if (cc->event_callback) {
cc->event_callback(cc, HIRCLUSTER_EVENT_FREE_CONTEXT,
cc->event_privdata);
}
if (cc->connect_timeout) {
hi_free(cc->connect_timeout);
cc->connect_timeout = NULL;
}
if (cc->command_timeout) {
hi_free(cc->command_timeout);
cc->command_timeout = NULL;
}
if (cc->table != NULL) {
hi_free(cc->table);
cc->table = NULL;
}
if (cc->nodes != NULL) {
/* Clear cc->nodes before releasing the dict since the release procedure
might access cc->nodes. When a node and its hiredis context are freed
deps/hiredis-cluster/hircluster.c view on Meta::CPAN
return REDIS_OK;
}
int redisClusterSetOptionTimeout(redisClusterContext *cc,
const struct timeval tv) {
if (cc == NULL) {
return REDIS_ERR;
}
if (cc->command_timeout == NULL ||
cc->command_timeout->tv_sec != tv.tv_sec ||
cc->command_timeout->tv_usec != tv.tv_usec) {
if (cc->command_timeout == NULL) {
cc->command_timeout = hi_malloc(sizeof(struct timeval));
if (cc->command_timeout == NULL) {
__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
return REDIS_ERR;
}
}
memcpy(cc->command_timeout, &tv, sizeof(struct timeval));
/* Set timeout on already connected nodes */
if (cc->nodes && dictSize(cc->nodes) > 0) {
dictEntry *de;
redisClusterNode *node;
dictIterator di;
dictInitIterator(&di, cc->nodes);
while ((de = dictNext(&di)) != NULL) {
deps/hiredis-cluster/hircluster.c view on Meta::CPAN
return c;
}
if (node->host == NULL || node->port <= 0) {
return NULL;
}
redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, node->host, node->port);
options.connect_timeout = cc->connect_timeout;
options.command_timeout = cc->command_timeout;
c = redisConnectWithOptions(&options);
if (c == NULL) {
__redisClusterSetError(cc, REDIS_ERR_OOM, "Out of memory");
return NULL;
}
if (cc->on_connect) {
cc->on_connect(c, c->err ? REDIS_ERR : REDIS_OK);
}
deps/hiredis-cluster/hircluster.c view on Meta::CPAN
if (node->host == NULL || node->port <= 0) {
__redisClusterAsyncSetError(acc, REDIS_ERR_OTHER,
"node host or port is error");
return NULL;
}
redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, node->host, node->port);
options.connect_timeout = acc->cc->connect_timeout;
options.command_timeout = acc->cc->command_timeout;
node->lastConnectionAttempt = hi_usec_now();
ac = redisAsyncConnectWithOptions(&options);
if (ac == NULL) {
__redisClusterAsyncSetError(acc, REDIS_ERR_OOM, "Out of memory");
return NULL;
}
if (ac->err) {
deps/hiredis-cluster/hircluster.h view on Meta::CPAN
} copen_slot;
/* Context for accessing a Redis Cluster */
typedef struct redisClusterContext {
int err; /* Error flags, 0 when there is no error */
char errstr[128]; /* String representation of error when applicable */
/* Configurations */
int flags; /* Configuration flags */
struct timeval *connect_timeout; /* TCP connect timeout */
struct timeval *command_timeout; /* Receive and send timeout */
int max_retry_count; /* Allowed retry attempts */
char *username; /* Authenticate using user */
char *password; /* Authentication password */
struct dict *nodes; /* Known redisClusterNode's */
uint64_t route_version; /* Increased when the node lookup table changes */
redisClusterNode **table; /* redisClusterNode lookup table */
struct hilist *requests; /* Outstanding commands (Pipelining) */
deps/hiredis-cluster/tests/ct_connection.c view on Meta::CPAN
assert(cc->err == REDIS_ERR_IO);
assert(strcmp(cc->errstr, "Connection timed out") == 0);
assert(connect_success_counter == 0);
assert(connect_failure_counter == 1);
reset_counters();
redisClusterFree(cc);
}
/* Connect using a pre-configured command timeout */
void test_command_timeout(void) {
struct timeval timeout = {0, 10000};
redisClusterContext *cc = redisClusterContextInit();
assert(cc);
redisClusterSetOptionAddNodes(cc, CLUSTER_NODE);
redisClusterSetOptionTimeout(cc, timeout);
int status = redisClusterConnect2(cc);
ASSERT_MSG(status == REDIS_OK, cc->errstr);
deps/hiredis-cluster/tests/ct_connection.c view on Meta::CPAN
if (reply && reply->type == REDIS_REPLY_STATUS)
break;
}
CHECK_REPLY_OK(cc, reply);
freeReplyObject(reply);
redisClusterFree(cc);
}
/* Connect and configure a command timeout while connected. */
void test_command_timeout_set_while_connected(void) {
struct timeval timeout = {0, 10000};
redisClusterContext *cc = redisClusterContextInit();
assert(cc);
redisClusterSetOptionAddNodes(cc, CLUSTER_NODE);
int status = redisClusterConnect2(cc);
ASSERT_MSG(status == REDIS_OK, cc->errstr);
redisClusterNodeIterator ni;
deps/hiredis-cluster/tests/ct_connection.c view on Meta::CPAN
assert(acc->cc->err == REDIS_ERR_IO);
assert(strcmp(acc->cc->errstr, "Connection timed out") == 0);
event_base_dispatch(base);
redisClusterAsyncFree(acc);
event_base_free(base);
}
/* Connect using a pre-configured command timeout */
void test_async_command_timeout(void) {
struct timeval timeout = {0, 10000};
redisClusterAsyncContext *acc = redisClusterAsyncContextInit();
assert(acc);
redisClusterSetOptionAddNodes(acc->cc, CLUSTER_NODE);
redisClusterSetOptionTimeout(acc->cc, timeout);
struct event_base *base = event_base_new();
redisClusterLibeventAttach(acc, base);
deps/hiredis-cluster/tests/ct_connection.c view on Meta::CPAN
int main(void) {
test_password_ok();
test_password_wrong();
test_password_missing();
test_username_ok();
test_username_disabled();
test_multicluster();
test_connect_timeout();
test_command_timeout();
test_command_timeout_set_while_connected();
test_async_password_ok();
test_async_password_wrong();
test_async_password_missing();
test_async_username_ok();
test_async_multicluster();
test_async_connect_timeout();
test_async_command_timeout();
return 0;
}
deps/hiredis/README.md view on Meta::CPAN
Version 1.0.0 marks the first stable release of Hiredis.
It includes some minor breaking changes, mostly to make the exposed API more uniform and self-explanatory.
It also bundles the updated `sds` library, to sync up with upstream and Redis.
For code changes see the [Changelog](CHANGELOG.md).
_Note: As described below, a few member names have been changed but most applications should be able to upgrade with minor code changes and recompiling._
## IMPORTANT: Breaking changes from `0.14.1` -> `1.0.0`
* `redisContext` has two additional members (`free_privdata`, and `privctx`).
* `redisOptions.timeout` has been renamed to `redisOptions.connect_timeout`, and we've added `redisOptions.command_timeout`.
* `redisReplyObjectFunctions.createArray` now takes `size_t` instead of `int` for its length parameter.
## IMPORTANT: Breaking changes when upgrading from 0.13.x -> 0.14.x
Bulk and multi-bulk lengths less than -1 or greater than `LLONG_MAX` are now
protocol errors. This is consistent with the RESP specification. On 32-bit
platforms, the upper bound is lowered to `SIZE_MAX`.
Change `redisReply.len` to `size_t`, as it denotes the the size of a string
deps/hiredis/async.c view on Meta::CPAN
redisCallback cb;
/* must not be called from a callback */
assert(!(c->flags & REDIS_IN_CALLBACK));
if ((c->flags & REDIS_CONNECTED)) {
if (ac->replies.head == NULL && ac->sub.replies.head == NULL) {
/* Nothing to do - just an idle timeout */
return;
}
if (!ac->c.command_timeout ||
(!ac->c.command_timeout->tv_sec && !ac->c.command_timeout->tv_usec)) {
/* A belated connect timeout arriving, ignore */
return;
}
}
if (!c->err) {
__redisSetError(c, REDIS_ERR_TIMEOUT, "Timeout");
__redisAsyncCopyError(ac);
}
deps/hiredis/async.c view on Meta::CPAN
return status;
}
redisAsyncPushFn *redisAsyncSetPushCallback(redisAsyncContext *ac, redisAsyncPushFn *fn) {
redisAsyncPushFn *old = ac->push_cb;
ac->push_cb = fn;
return old;
}
int redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv) {
if (!ac->c.command_timeout) {
ac->c.command_timeout = hi_calloc(1, sizeof(tv));
if (ac->c.command_timeout == NULL) {
__redisSetError(&ac->c, REDIS_ERR_OOM, "Out of memory");
__redisAsyncCopyError(ac);
return REDIS_ERR;
}
}
if (tv.tv_sec != ac->c.command_timeout->tv_sec ||
tv.tv_usec != ac->c.command_timeout->tv_usec)
{
*ac->c.command_timeout = tv;
}
return REDIS_OK;
}
deps/hiredis/async_private.h view on Meta::CPAN
static inline void refreshTimeout(redisAsyncContext *ctx) {
#define REDIS_TIMER_ISSET(tvp) \
(tvp && ((tvp)->tv_sec || (tvp)->tv_usec))
#define REDIS_EL_TIMER(ac, tvp) \
if ((ac)->ev.scheduleTimer && REDIS_TIMER_ISSET(tvp)) { \
(ac)->ev.scheduleTimer((ac)->ev.data, *(tvp)); \
}
if (ctx->c.flags & REDIS_CONNECTED) {
REDIS_EL_TIMER(ctx, ctx->c.command_timeout);
} else {
REDIS_EL_TIMER(ctx, ctx->c.connect_timeout);
}
}
void __redisAsyncDisconnect(redisAsyncContext *ac);
void redisProcessCallbacks(redisAsyncContext *ac);
#endif /* __HIREDIS_ASYNC_PRIVATE_H */
deps/hiredis/hiredis.c view on Meta::CPAN
if (c->funcs && c->funcs->close) {
c->funcs->close(c);
}
sdsfree(c->obuf);
redisReaderFree(c->reader);
hi_free(c->tcp.host);
hi_free(c->tcp.source_addr);
hi_free(c->unix_sock.path);
hi_free(c->connect_timeout);
hi_free(c->command_timeout);
hi_free(c->saddr);
if (c->privdata && c->free_privdata)
c->free_privdata(c->privdata);
if (c->funcs && c->funcs->free_privctx)
c->funcs->free_privctx(c->privctx);
memset(c, 0xff, sizeof(*c));
hi_free(c);
deps/hiredis/hiredis.c view on Meta::CPAN
c->connect_timeout, c->tcp.source_addr);
} else if (c->connection_type == REDIS_CONN_UNIX) {
ret = redisContextConnectUnix(c, c->unix_sock.path, c->connect_timeout);
} else {
/* Something bad happened here and shouldn't have. There isn't
enough information in the context to reconnect. */
__redisSetError(c,REDIS_ERR_OTHER,"Not enough information to reconnect");
ret = REDIS_ERR;
}
if (c->command_timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *c->command_timeout);
}
return ret;
}
redisContext *redisConnectWithOptions(const redisOptions *options) {
redisContext *c = redisContextInit();
if (c == NULL) {
return NULL;
}
deps/hiredis/hiredis.c view on Meta::CPAN
* as a default unless specifically flagged that we don't want one. */
if (options->push_cb != NULL)
redisSetPushCallback(c, options->push_cb);
else if (!(options->options & REDIS_OPT_NO_PUSH_AUTOFREE))
redisSetPushCallback(c, redisPushAutoFree);
c->privdata = options->privdata;
c->free_privdata = options->free_privdata;
if (redisContextUpdateConnectTimeout(c, options->connect_timeout) != REDIS_OK ||
redisContextUpdateCommandTimeout(c, options->command_timeout) != REDIS_OK) {
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return c;
}
if (options->type == REDIS_CONN_TCP) {
redisContextConnectBindTcp(c, options->endpoint.tcp.ip,
options->endpoint.tcp.port, options->connect_timeout,
options->endpoint.tcp.source_addr);
} else if (options->type == REDIS_CONN_UNIX) {
redisContextConnectUnix(c, options->endpoint.unix_socket,
options->connect_timeout);
} else if (options->type == REDIS_CONN_USERFD) {
c->fd = options->endpoint.fd;
c->flags |= REDIS_CONNECTED;
} else {
redisFree(c);
return NULL;
}
if (c->err == 0 && c->fd != REDIS_INVALID_FD &&
options->command_timeout != NULL && (c->flags & REDIS_BLOCK))
{
redisContextSetTimeout(c, *options->command_timeout);
}
return c;
}
/* Connect to a Redis instance. On error the field error in the returned
* context will be set to the return value of the error function.
* When no set of reply functions is given, the default set will be used. */
redisContext *redisConnect(const char *ip, int port) {
redisOptions options = {0};
deps/hiredis/hiredis.h view on Meta::CPAN
* the type of connection to use. This also indicates which
* `endpoint` member field to use
*/
int type;
/* bit field of REDIS_OPT_xxx */
int options;
/* timeout value for connect operation. If NULL, no timeout is used */
const struct timeval *connect_timeout;
/* timeout value for commands. If NULL, no timeout is used. This can be
* updated at runtime with redisSetTimeout/redisAsyncSetTimeout. */
const struct timeval *command_timeout;
union {
/** use this field for tcp/ip connections */
struct {
const char *source_addr;
const char *ip;
int port;
} tcp;
/** use this field for unix domain sockets */
const char *unix_socket;
/**
deps/hiredis/hiredis.h view on Meta::CPAN
int err; /* Error flags, 0 when there is no error */
char errstr[128]; /* String representation of error when applicable */
redisFD fd;
int flags;
char *obuf; /* Write buffer */
redisReader *reader; /* Protocol reader */
enum redisConnectionType connection_type;
struct timeval *connect_timeout;
struct timeval *command_timeout;
struct {
char *host;
char *source_addr;
int port;
} tcp;
struct {
char *path;
} unix_sock;
deps/hiredis/net.c view on Meta::CPAN
if (c->connect_timeout == NULL)
return REDIS_ERR;
}
memcpy(c->connect_timeout, timeout, sizeof(*c->connect_timeout));
return REDIS_OK;
}
int redisContextUpdateCommandTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->command_timeout == timeout)
return REDIS_OK;
/* Allocate context timeval if we need to */
if (c->command_timeout == NULL) {
c->command_timeout = hi_malloc(sizeof(*c->command_timeout));
if (c->command_timeout == NULL)
return REDIS_ERR;
}
memcpy(c->command_timeout, timeout, sizeof(*c->command_timeout));
return REDIS_OK;
}
static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
const struct timeval *timeout,
const char *source_addr) {
redisFD s;
int rv, n;
char _port[6]; /* strlen("65535"); */
struct addrinfo hints, *servinfo, *bservinfo, *p, *b;
deps/hiredis/test.c view on Meta::CPAN
test("Returns error when the port is not open: ");
c = redisConnect((char*)"localhost", 1);
test_cond(c->err == REDIS_ERR_IO &&
strcmp(c->errstr,"Connection refused") == 0);
redisFree(c);
/* Verify we don't regress from the fix in PR #1180 */
test("We don't clobber connection exception with setsockopt error: ");
tv = (struct timeval){.tv_sec = 0, .tv_usec = 500000};
opt.command_timeout = opt.connect_timeout = &tv;
REDIS_OPTIONS_SET_TCP(&opt, "localhost", 10337);
c = redisConnectWithOptions(&opt);
test_cond(c->err == REDIS_ERR_IO &&
strcmp(c->errstr, "Connection refused") == 0);
redisFree(c);
test("Returns error when the unix_sock socket path doesn't accept connections: ");
c = redisConnectUnix((char*)"/tmp/idontexist.sock");
test_cond(c->err == REDIS_ERR_IO); /* Don't care about the message... */
redisFree(c);
deps/hiredis/test.c view on Meta::CPAN
/* Start event dispatching loop */
test_cond(event_base_dispatch(base) == 0);
event_free(timeout);
event_base_free(base);
/* Verify test checkpoints */
assert(state.checkpoint == 6);
}
/* Subscribe callback for test_command_timeout_during_pubsub:
* - a subscribe response triggers a published message
* - the published message triggers a command that times out
* - the command timeout triggers a disconnect */
void subscribe_with_timeout_cb(redisAsyncContext *ac, void *r, void *privdata) {
redisReply *reply = r;
TestState *state = privdata;
/* The non-clean disconnect should trigger the
* subscription callback with a NULL reply. */
if (reply == NULL) {
deps/hiredis/test.c view on Meta::CPAN
/* Send a command that will trigger a timeout */
redisAsyncCommand(ac,null_cb,state,"DEBUG SLEEP 3");
redisAsyncCommand(ac,null_cb,state,"LPUSH mylist foo");
} else {
printf("Unexpected pubsub command: %s\n", reply->element[0]->str);
exit(1);
}
}
static void test_command_timeout_during_pubsub(struct config config) {
test("Command timeout during Pub/Sub: ");
/* Setup event dispatcher with a testcase timeout */
base = event_base_new();
struct event *timeout = evtimer_new(base,timeout_cb,NULL);
assert(timeout != NULL);
evtimer_assign(timeout,base,timeout_cb,NULL);
struct timeval timeout_tv = {.tv_sec = 10};
evtimer_add(timeout,&timeout_tv);
/* Connect */
redisOptions options = get_redis_tcp_options(config);
redisAsyncContext *ac = redisAsyncConnectWithOptions(&options);
assert(ac != NULL && ac->err == 0);
redisLibeventAttach(ac,base);
/* Configure a command timout */
struct timeval command_timeout = {.tv_sec = 2};
redisAsyncSetTimeout(ac,command_timeout);
/* Not expecting any push messages in this test */
redisAsyncSetPushCallback(ac,unexpected_push_cb);
/* Switch protocol */
redisAsyncCommand(ac,NULL,NULL,"HELLO 3");
/* Start subscribe */
TestState state = {.options = &options, .resp3 = 1};
redisAsyncCommand(ac,subscribe_with_timeout_cb,&state,"subscribe mychannel");
deps/hiredis/test.c view on Meta::CPAN
int major;
redisContext *c = do_connect(cfg);
get_redis_version(c, &major, NULL);
disconnect(c, 0);
test_pubsub_handling(cfg);
test_pubsub_multiple_channels(cfg);
test_monitor(cfg);
if (major >= 6) {
test_pubsub_handling_resp3(cfg);
test_command_timeout_during_pubsub(cfg);
}
#endif /* HIREDIS_TEST_ASYNC */
cfg.type = CONN_TCP;
printf("\nTesting asynchronous API using polling_adapter TCP (%s:%d):\n", cfg.tcp.host, cfg.tcp.port);
test_async_polling(cfg);
if (test_unix_socket) {
cfg.type = CONN_UNIX;
printf("\nTesting asynchronous API using polling_adapter UNIX (%s):\n", cfg.unix_sock.path);
test_async_polling(cfg);
lib/Redis/Cluster/Fast.pm view on Meta::CPAN
croak 'need startup_nodes' unless defined $args{startup_nodes} && @{$args{startup_nodes}};
if (my $servers = join(',', @{$args{startup_nodes}})) {
$self->__set_servers($servers);
}
my $connect_timeout = $args{connect_timeout};
$connect_timeout = DEFAULT_CONNECT_TIMEOUT unless defined $connect_timeout;
$self->__set_connect_timeout($connect_timeout);
my $command_timeout = $args{command_timeout};
$command_timeout = DEFAULT_COMMAND_TIMEOUT unless defined $command_timeout;
$self->__set_command_timeout($command_timeout);
my $discovery_timeout = $args{cluster_discovery_retry_timeout};
$discovery_timeout = DEFAULT_CLUSTER_DISCOVERY_RETRY_TIMEOUT unless defined $discovery_timeout;
$self->__set_cluster_discovery_retry_timeout($discovery_timeout);
my $max_retry = $args{max_retry_count};
$max_retry = DEFAULT_MAX_RETRY_COUNT unless defined $max_retry;
$self->__set_max_retry($max_retry);
$self->__set_route_use_slots($args{route_use_slots} ? 1 : 0);
lib/Redis/Cluster/Fast.pm view on Meta::CPAN
my $redis = Redis::Cluster::Fast->new(
startup_nodes => [
'localhost:9000',
'localhost:9001',
'localhost:9002',
'localhost:9003',
'localhost:9004',
'localhost:9005',
],
connect_timeout => 0.05,
command_timeout => 0.05,
max_retry_count => 10,
);
$redis->set('test', 123);
# '123'
my $str = $redis->get('test');
$redis->mset('{my}foo', 'hoge', '{my}bar', 'fuga');
lib/Redis/Cluster/Fast.pm view on Meta::CPAN
=head3 startup_nodes
Specifies the list of Redis Cluster nodes.
=head3 connect_timeout
A fractional seconds. (default: 1.0)
Connection timeout to connect to a Redis node.
=head3 command_timeout
A fractional seconds. (default: 1.0)
Specifies the timeout value for each read/write event to execute a Redis Command.
=head3 max_retry_count
A integer value. (default: 5)
The client will retry calling the Redis Command only if it successfully get one of the following error responses.
lib/Redis/Cluster/Fast.pm view on Meta::CPAN
You can then perform a blocking wait for those responses later, if needed.
Executes one iteration of the event loop to process any pending commands that have not yet been sent
and any incoming responses from Redis.
If there are events that can be triggered immediately, they will all be processed.
In other words, if there are unsent commands, they will be pipelined and sent,
and if there are already-received responses, their corresponding callbacks will be executed.
If there are no events that can be triggered immediately: there are neither unsent commands nor any Redis responses available to read,
but unprocessed callbacks remain, then this method will block for up to C<command_timeout> while waiting for a response from Redis.
When a timeout occurs, an error will be propagated to the corresponding callback(s).
The return value can be either 1 for success (e.g., commands sent or responses read),
0 for no callbacks remained, or undef for other errors.
=head3 Notes
=over 4
=item *
src/Fast.xs view on Meta::CPAN
typedef struct redis_cluster_fast_s {
redisClusterAsyncContext *acc;
struct event_base *cluster_event_base;
char *hostnames;
int debug;
int max_retry;
int use_cluster_slots;
int event_ready;
struct timeval connect_timeout;
struct timeval command_timeout;
int64_t discovery_timeout_usec;
pid_t pid;
int64_t pipeline_callback_remain;
} redis_cluster_fast_t, *Redis__Cluster__Fast;
int64_t get_usec_timestamp(void) {
struct timespec ts;
int status;
status = clock_gettime(CLOCK_MONOTONIC, &ts);
if (status < 0) {
src/Fast.xs view on Meta::CPAN
self->pipeline_callback_remain = 0;
self->pid = getpid();
self->acc = redisClusterAsyncContextInit();
if (redisClusterSetOptionAddNodes(self->acc->cc, self->hostnames) != REDIS_OK) {
return newSVpvf("failed to add nodes: %s", self->acc->cc->errstr);
}
if (redisClusterSetOptionConnectTimeout(self->acc->cc, self->connect_timeout) != REDIS_OK) {
return newSVpvf("failed to set connect timeout: %s", self->acc->cc->errstr);
}
if (redisClusterSetOptionTimeout(self->acc->cc, self->command_timeout) != REDIS_OK) {
return newSVpvf("failed to set command timeout: %s", self->acc->cc->errstr);
}
if (redisClusterSetOptionMaxRetry(self->acc->cc, self->max_retry) != REDIS_OK) {
return newSVpvf("%s", "failed to set max retry");
}
if (self->use_cluster_slots) {
DEBUG_MSG("%s", "use cluster slots");
if (redisClusterSetOptionRouteUseSlots(self->acc->cc) != REDIS_OK) {
return newSVpvf("%s", "failed to set redisClusterSetOptionRouteUseSlots");
src/Fast.xs view on Meta::CPAN
struct timeval timeout;
CODE:
second = (int) (double_sec);
micro_second = (int) (fmod(double_sec * ONE_SECOND_TO_MICRO, ONE_SECOND_TO_MICRO) + 0.999);
timeout.tv_sec = second;
timeout.tv_usec = micro_second;
self->connect_timeout = timeout;
DEBUG_MSG("connect timeout %d, %d", second, micro_second);
void
__set_command_timeout(Redis::Cluster::Fast self, double double_sec)
PREINIT:
int second, micro_second;
struct timeval timeout;
CODE:
second = (int) (double_sec);
micro_second = (int) (fmod(double_sec * ONE_SECOND_TO_MICRO, ONE_SECOND_TO_MICRO) + 0.999);
timeout.tv_sec = second;
timeout.tv_usec = micro_second;
self->command_timeout = timeout;
DEBUG_MSG("command timeout %d, %d", second, micro_second);
void
__set_max_retry(Redis::Cluster::Fast self, int max_retry)
CODE:
self->max_retry = max_retry;
DEBUG_MSG("max_retry %d", max_retry);
void
__set_route_use_slots(Redis::Cluster::Fast self, int use_slot)
xt/04_timeout.t view on Meta::CPAN
use warnings FATAL => 'all';
use Test::More;
use lib './xt/lib';
use Test::Docker::RedisCluster qw/get_startup_nodes/;
use Redis::Cluster::Fast;
my $redis = Redis::Cluster::Fast->new(
startup_nodes => get_startup_nodes,
connect_timeout => 0.05,
command_timeout => 0.05,
max_retry_count => 3,
);
my $lua = <<EOF;
local tmp = KEYS[1]
redis.call("SET", tmp, "1")
redis.call("EXPIRE", tmp, ARGV[1])
for i = 0, ARGV[1] do
xt/05_valgrind.t view on Meta::CPAN
use Test::Valgrind (extra_supps => [ './xt/lib/memcheck-extra.supp' ]);
};
plan skip_all => 'Test::Valgrind is required to test your distribution with valgrind' if $@;
use Test::Docker::RedisCluster qw/get_startup_nodes/; # Valgrind check this module too
use Redis::Cluster::Fast;
my $redis = Redis::Cluster::Fast->new(
startup_nodes => get_startup_nodes,
connect_timeout => 0.5,
command_timeout => 0.5,
max_retry_count => 10,
);
$redis->del('valgrind');
$redis->set('valgrind', 123);
eval {
# wide character
$redis->set('euro', "\x{20ac}");
};
xt/09_valgrind_srandom.t view on Meta::CPAN
plan skip_all => 'Test::Valgrind is required to test your distribution with valgrind' if $@;
use Test::Docker::RedisCluster qw/get_startup_nodes/; # Valgrind check this module too
use Redis::Cluster::Fast;
Redis::Cluster::Fast::srandom(1111);
my $redis = Redis::Cluster::Fast->new(
startup_nodes => get_startup_nodes,
connect_timeout => 0.5,
command_timeout => 0.5,
max_retry_count => 10,
);
$redis->del('valgrind');
$redis->set('valgrind', 123);
eval {
# wide character
$redis->set('euro', "\x{20ac}");
};
xt/10_timeout_srandom.t view on Meta::CPAN
use lib './xt/lib';
use Test::Docker::RedisCluster qw/get_startup_nodes/;
use Redis::Cluster::Fast;
Redis::Cluster::Fast::srandom(1111);
my $redis = Redis::Cluster::Fast->new(
startup_nodes => get_startup_nodes,
connect_timeout => 0.05,
command_timeout => 0.05,
max_retry_count => 3,
);
my $lua = <<EOF;
local tmp = KEYS[1]
redis.call("SET", tmp, "1")
redis.call("EXPIRE", tmp, ARGV[1])
for i = 0, ARGV[1] do
( run in 0.839 second using v1.01-cache-2.11-cpan-a5abf4f5562 )