Unverified Commit 5cbc73c6 authored by jared's avatar jared Committed by GitHub
Browse files

Merge pull request #1 from antirez/unstable

merge redis
parents a16aa03a d5536e04
......@@ -3,7 +3,9 @@
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <netdb.h>
#include <assert.h>
#include <unistd.h>
#include <signal.h>
......@@ -91,7 +93,7 @@ static int disconnect(redisContext *c, int keep_fd) {
return -1;
}
static redisContext *connect(struct config config) {
static redisContext *do_connect(struct config config) {
redisContext *c = NULL;
if (config.type == CONN_TCP) {
......@@ -248,7 +250,7 @@ static void test_append_formatted_commands(struct config config) {
char *cmd;
int len;
c = connect(config);
c = do_connect(config);
test("Append format command: ");
......@@ -302,6 +304,82 @@ static void test_reply_reader(void) {
strncasecmp(reader->errstr,"No support for",14) == 0);
redisReaderFree(reader);
test("Correctly parses LLONG_MAX: ");
reader = redisReaderCreate();
redisReaderFeed(reader, ":9223372036854775807\r\n",22);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_OK &&
((redisReply*)reply)->type == REDIS_REPLY_INTEGER &&
((redisReply*)reply)->integer == LLONG_MAX);
freeReplyObject(reply);
redisReaderFree(reader);
test("Set error when > LLONG_MAX: ");
reader = redisReaderCreate();
redisReaderFeed(reader, ":9223372036854775808\r\n",22);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Bad integer value") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
test("Correctly parses LLONG_MIN: ");
reader = redisReaderCreate();
redisReaderFeed(reader, ":-9223372036854775808\r\n",23);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_OK &&
((redisReply*)reply)->type == REDIS_REPLY_INTEGER &&
((redisReply*)reply)->integer == LLONG_MIN);
freeReplyObject(reply);
redisReaderFree(reader);
test("Set error when < LLONG_MIN: ");
reader = redisReaderCreate();
redisReaderFeed(reader, ":-9223372036854775809\r\n",23);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Bad integer value") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
test("Set error when array < -1: ");
reader = redisReaderCreate();
redisReaderFeed(reader, "*-2\r\n+asdf\r\n",12);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Multi-bulk length out of range") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
test("Set error when bulk < -1: ");
reader = redisReaderCreate();
redisReaderFeed(reader, "$-2\r\nasdf\r\n",11);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Bulk string length out of range") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
test("Set error when array > INT_MAX: ");
reader = redisReaderCreate();
redisReaderFeed(reader, "*9223372036854775807\r\n+asdf\r\n",29);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Multi-bulk length out of range") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
#if LLONG_MAX > SIZE_MAX
test("Set error when bulk > SIZE_MAX: ");
reader = redisReaderCreate();
redisReaderFeed(reader, "$9223372036854775807\r\nasdf\r\n",28);
ret = redisReaderGetReply(reader,&reply);
test_cond(ret == REDIS_ERR &&
strcasecmp(reader->errstr,"Bulk string length out of range") == 0);
freeReplyObject(reply);
redisReaderFree(reader);
#endif
test("Works with NULL functions for reply: ");
reader = redisReaderCreate();
reader->fn = NULL;
......@@ -358,18 +436,32 @@ static void test_free_null(void) {
static void test_blocking_connection_errors(void) {
redisContext *c;
struct addrinfo hints = {.ai_family = AF_INET};
struct addrinfo *ai_tmp = NULL;
const char *bad_domain = "idontexist.com";
int rv = getaddrinfo(bad_domain, "6379", &hints, &ai_tmp);
if (rv != 0) {
// Address does *not* exist
test("Returns error when host cannot be resolved: ");
c = redisConnect((char*)"idontexist.test", 6379);
test_cond(c->err == REDIS_ERR_OTHER &&
(strcmp(c->errstr,"Name or service not known") == 0 ||
strcmp(c->errstr,"Can't resolve: idontexist.test") == 0 ||
strcmp(c->errstr,"nodename nor servname provided, or not known") == 0 ||
strcmp(c->errstr,"No address associated with hostname") == 0 ||
strcmp(c->errstr,"Temporary failure in name resolution") == 0 ||
strcmp(c->errstr,"hostname nor servname provided, or not known") == 0 ||
strcmp(c->errstr,"no address associated with name") == 0));
// First see if this domain name *actually* resolves to NXDOMAIN
c = redisConnect("dontexist.com", 6379);
test_cond(
c->err == REDIS_ERR_OTHER &&
(strcmp(c->errstr, "Name or service not known") == 0 ||
strcmp(c->errstr, "Can't resolve: sadkfjaskfjsa.com") == 0 ||
strcmp(c->errstr,
"nodename nor servname provided, or not known") == 0 ||
strcmp(c->errstr, "No address associated with hostname") == 0 ||
strcmp(c->errstr, "Temporary failure in name resolution") == 0 ||
strcmp(c->errstr,
"hostname nor servname provided, or not known") == 0 ||
strcmp(c->errstr, "no address associated with name") == 0));
redisFree(c);
} else {
printf("Skipping NXDOMAIN test. Found evil ISP!\n");
freeaddrinfo(ai_tmp);
}
test("Returns error when the port is not open: ");
c = redisConnect((char*)"localhost", 1);
......@@ -387,7 +479,7 @@ static void test_blocking_connection(struct config config) {
redisContext *c;
redisReply *reply;
c = connect(config);
c = do_connect(config);
test("Is able to deliver commands: ");
reply = redisCommand(c,"PING");
......@@ -468,7 +560,7 @@ static void test_blocking_connection_timeouts(struct config config) {
const char *cmd = "DEBUG SLEEP 3\r\n";
struct timeval tv;
c = connect(config);
c = do_connect(config);
test("Successfully completes a command when the timeout is not exceeded: ");
reply = redisCommand(c,"SET foo fast");
freeReplyObject(reply);
......@@ -480,7 +572,7 @@ static void test_blocking_connection_timeouts(struct config config) {
freeReplyObject(reply);
disconnect(c, 0);
c = connect(config);
c = do_connect(config);
test("Does not return a reply when the command times out: ");
s = write(c->fd, cmd, strlen(cmd));
tv.tv_sec = 0;
......@@ -514,7 +606,7 @@ static void test_blocking_io_errors(struct config config) {
int major, minor;
/* Connect to target given by config. */
c = connect(config);
c = do_connect(config);
{
/* Find out Redis version to determine the path for the next test */
const char *field = "redis_version:";
......@@ -549,7 +641,7 @@ static void test_blocking_io_errors(struct config config) {
strcmp(c->errstr,"Server closed the connection") == 0);
redisFree(c);
c = connect(config);
c = do_connect(config);
test("Returns I/O error on socket timeout: ");
struct timeval tv = { 0, 1000 };
assert(redisSetTimeout(c,tv) == REDIS_OK);
......@@ -583,7 +675,7 @@ static void test_invalid_timeout_errors(struct config config) {
}
static void test_throughput(struct config config) {
redisContext *c = connect(config);
redisContext *c = do_connect(config);
redisReply **replies;
int i, num;
long long t1, t2;
......@@ -616,6 +708,17 @@ static void test_throughput(struct config config) {
free(replies);
printf("\t(%dx LRANGE with 500 elements: %.3fs)\n", num, (t2-t1)/1000000.0);
replies = malloc(sizeof(redisReply*)*num);
t1 = usec();
for (i = 0; i < num; i++) {
replies[i] = redisCommand(c, "INCRBY incrkey %d", 1000000);
assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_INTEGER);
}
t2 = usec();
for (i = 0; i < num; i++) freeReplyObject(replies[i]);
free(replies);
printf("\t(%dx INCRBY: %.3fs)\n", num, (t2-t1)/1000000.0);
num = 10000;
replies = malloc(sizeof(redisReply*)*num);
for (i = 0; i < num; i++)
......@@ -644,6 +747,19 @@ static void test_throughput(struct config config) {
free(replies);
printf("\t(%dx LRANGE with 500 elements (pipelined): %.3fs)\n", num, (t2-t1)/1000000.0);
replies = malloc(sizeof(redisReply*)*num);
for (i = 0; i < num; i++)
redisAppendCommand(c,"INCRBY incrkey %d", 1000000);
t1 = usec();
for (i = 0; i < num; i++) {
assert(redisGetReply(c, (void*)&replies[i]) == REDIS_OK);
assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_INTEGER);
}
t2 = usec();
for (i = 0; i < num; i++) freeReplyObject(replies[i]);
free(replies);
printf("\t(%dx INCRBY (pipelined): %.3fs)\n", num, (t2-t1)/1000000.0);
disconnect(c, 0);
}
......
......@@ -215,4 +215,32 @@ ixalloc(tsdn_t *tsdn, void *ptr, size_t oldsize, size_t size, size_t extra,
return arena_ralloc_no_move(tsdn, ptr, oldsize, size, extra, zero);
}
JEMALLOC_ALWAYS_INLINE int
iget_defrag_hint(tsdn_t *tsdn, void* ptr, int *bin_util, int *run_util) {
int defrag = 0;
rtree_ctx_t rtree_ctx_fallback;
rtree_ctx_t *rtree_ctx = tsdn_rtree_ctx(tsdn, &rtree_ctx_fallback);
szind_t szind;
bool is_slab;
rtree_szind_slab_read(tsdn, &extents_rtree, rtree_ctx, (uintptr_t)ptr, true, &szind, &is_slab);
if (likely(is_slab)) {
/* Small allocation. */
extent_t *slab = iealloc(tsdn, ptr);
arena_t *arena = extent_arena_get(slab);
szind_t binind = extent_szind_get(slab);
bin_t *bin = &arena->bins[binind];
malloc_mutex_lock(tsdn, &bin->lock);
/* don't bother moving allocations from the slab currently used for new allocations */
if (slab != bin->slabcur) {
const bin_info_t *bin_info = &bin_infos[binind];
size_t availregs = bin_info->nregs * bin->stats.curslabs;
*bin_util = ((long long)bin->stats.curregs<<16) / availregs;
*run_util = ((long long)(bin_info->nregs - extent_nfree_get(slab))<<16) / bin_info->nregs;
defrag = 1;
}
malloc_mutex_unlock(tsdn, &bin->lock);
}
return defrag;
}
#endif /* JEMALLOC_INTERNAL_INLINES_C_H */
......@@ -120,3 +120,7 @@
# define JEMALLOC_RESTRICT_RETURN
# define JEMALLOC_ALLOCATOR
#endif
/* This version of Jemalloc, modified for Redis, has the je_get_defrag_hint()
* function. */
#define JEMALLOC_FRAG_HINT
......@@ -3324,3 +3324,14 @@ jemalloc_postfork_child(void) {
}
/******************************************************************************/
/* Helps the application decide if a pointer is worth re-allocating in order to reduce fragmentation.
* returns 0 if the allocation is in the currently active run,
* or when it is not causing any frag issue (large or huge bin)
* returns the bin utilization and run utilization both in fixed point 16:16.
* If the application decides to re-allocate it should use MALLOCX_TCACHE_NONE when doing so. */
JEMALLOC_EXPORT int JEMALLOC_NOTHROW
get_defrag_hint(void* ptr, int *bin_util, int *run_util) {
assert(ptr != NULL);
return iget_defrag_hint(TSDN_NULL, ptr, bin_util, run_util);
}
This diff is collapsed.
......@@ -11,4 +11,4 @@ then
echo "You need tcl 8.5 or newer in order to run the Redis test"
exit 1
fi
$TCLSH tests/test_helper.tcl $*
$TCLSH tests/test_helper.tcl "${@}"
#!/bin/sh
TCL_VERSIONS="8.5 8.6"
TCLSH=""
for VERSION in $TCL_VERSIONS; do
TCL=`which tclsh$VERSION 2>/dev/null` && TCLSH=$TCL
done
if [ -z $TCLSH ]
then
echo "You need tcl 8.5 or newer in order to run the Redis test"
exit 1
fi
make -C tests/modules && \
$TCLSH tests/test_helper.tcl --single unit/moduleapi/commandfilter --single unit/moduleapi/testrdb "${@}"
......@@ -20,6 +20,21 @@
# The port that this sentinel instance will run on
port 26379
# By default Redis Sentinel does not run as a daemon. Use 'yes' if you need it.
# Note that Redis will write a pid file in /var/run/redis-sentinel.pid when
# daemonized.
daemonize no
# When running daemonized, Redis Sentinel writes a pid file in
# /var/run/redis-sentinel.pid by default. You can specify a custom pid file
# location here.
pidfile /var/run/redis-sentinel.pid
# Specify the log file name. Also the empty string can be used to force
# Sentinel to log on the standard output. Note that if you use standard
# output for logging but daemonize, logs will be sent to /dev/null
logfile ""
# sentinel announce-ip <ip>
# sentinel announce-port <port>
#
......@@ -58,11 +73,11 @@ dir /tmp
# be elected by the majority of the known Sentinels in order to
# start a failover, so no failover can be performed in minority.
#
# Slaves are auto-discovered, so you don't need to specify slaves in
# Replicas are auto-discovered, so you don't need to specify replicas in
# any way. Sentinel itself will rewrite this configuration file adding
# the slaves using additional configuration options.
# the replicas using additional configuration options.
# Also note that the configuration file is rewritten when a
# slave is promoted to master.
# replica is promoted to master.
#
# Note: master name should not include special characters or spaces.
# The valid charset is A-z 0-9 and the three characters ".-_".
......@@ -70,11 +85,11 @@ sentinel monitor mymaster 127.0.0.1 6379 2
# sentinel auth-pass <master-name> <password>
#
# Set the password to use to authenticate with the master and slaves.
# Set the password to use to authenticate with the master and replicas.
# Useful if there is a password set in the Redis instances to monitor.
#
# Note that the master password is also used for slaves, so it is not
# possible to set a different password in masters and slaves instances
# Note that the master password is also used for replicas, so it is not
# possible to set a different password in masters and replicas instances
# if you want to be able to monitor these instances with Sentinel.
#
# However you can have Redis instances without the authentication enabled
......@@ -89,7 +104,7 @@ sentinel monitor mymaster 127.0.0.1 6379 2
# sentinel down-after-milliseconds <master-name> <milliseconds>
#
# Number of milliseconds the master (or any attached slave or sentinel) should
# Number of milliseconds the master (or any attached replica or sentinel) should
# be unreachable (as in, not acceptable reply to PING, continuously, for the
# specified period) in order to consider it in S_DOWN state (Subjectively
# Down).
......@@ -97,11 +112,11 @@ sentinel monitor mymaster 127.0.0.1 6379 2
# Default is 30 seconds.
sentinel down-after-milliseconds mymaster 30000
# sentinel parallel-syncs <master-name> <numslaves>
# sentinel parallel-syncs <master-name> <numreplicas>
#
# How many slaves we can reconfigure to point to the new slave simultaneously
# during the failover. Use a low number if you use the slaves to serve query
# to avoid that all the slaves will be unreachable at about the same
# How many replicas we can reconfigure to point to the new replica simultaneously
# during the failover. Use a low number if you use the replicas to serve query
# to avoid that all the replicas will be unreachable at about the same
# time while performing the synchronization with the master.
sentinel parallel-syncs mymaster 1
......@@ -113,18 +128,18 @@ sentinel parallel-syncs mymaster 1
# already tried against the same master by a given Sentinel, is two
# times the failover timeout.
#
# - The time needed for a slave replicating to a wrong master according
# - The time needed for a replica replicating to a wrong master according
# to a Sentinel current configuration, to be forced to replicate
# with the right master, is exactly the failover timeout (counting since
# the moment a Sentinel detected the misconfiguration).
#
# - The time needed to cancel a failover that is already in progress but
# did not produced any configuration change (SLAVEOF NO ONE yet not
# acknowledged by the promoted slave).
# acknowledged by the promoted replica).
#
# - The maximum time a failover in progress waits for all the slaves to be
# reconfigured as slaves of the new master. However even after this time
# the slaves will be reconfigured by the Sentinels anyway, but not with
# - The maximum time a failover in progress waits for all the replicas to be
# reconfigured as replicas of the new master. However even after this time
# the replicas will be reconfigured by the Sentinels anyway, but not with
# the exact parallel-syncs progression as specified.
#
# Default is 3 minutes.
......@@ -185,7 +200,7 @@ sentinel failover-timeout mymaster 180000
# <role> is either "leader" or "observer"
#
# The arguments from-ip, from-port, to-ip, to-port are used to communicate
# the old address of the master and the new address of the elected slave
# the old address of the master and the new address of the elected replica
# (now a master).
#
# This script should be resistant to multiple invocations.
......@@ -203,3 +218,27 @@ sentinel failover-timeout mymaster 180000
sentinel deny-scripts-reconfig yes
# REDIS COMMANDS RENAMING
#
# Sometimes the Redis server has certain commands, that are needed for Sentinel
# to work correctly, renamed to unguessable strings. This is often the case
# of CONFIG and SLAVEOF in the context of providers that provide Redis as
# a service, and don't want the customers to reconfigure the instances outside
# of the administration console.
#
# In such case it is possible to tell Sentinel to use different command names
# instead of the normal ones. For example if the master "mymaster", and the
# associated replicas, have "CONFIG" all renamed to "GUESSME", I could use:
#
# SENTINEL rename-command mymaster CONFIG GUESSME
#
# After such configuration is set, every time Sentinel would use CONFIG it will
# use GUESSME instead. Note that there is no actual need to respect the command
# case, so writing "config guessme" is the same in the example above.
#
# SENTINEL SET can also be used in order to perform this configuration at runtime.
#
# In order to set a command back to its original name (undo the renaming), it
# is possible to just rename a command to itsef:
#
# SENTINEL rename-command mymaster CONFIG CONFIG
......@@ -20,7 +20,12 @@ DEPENDENCY_TARGETS=hiredis linenoise lua
NODEPS:=clean distclean
# Default settings
STD=-std=c99 -pedantic -DREDIS_STATIC=''
STD=-std=c11 -pedantic -DREDIS_STATIC=''
ifneq (,$(findstring clang,$(CC)))
ifneq (,$(findstring FreeBSD,$(uname_S)))
STD+=-Wno-c11-extensions
endif
endif
WARN=-Wall -W -Wno-missing-field-initializers
OPT=$(OPTIMIZATION)
......@@ -41,6 +46,10 @@ endif
# To get ARM stack traces if Redis crashes we need a special C flag.
ifneq (,$(filter aarch64 armv,$(uname_M)))
CFLAGS+=-funwind-tables
else
ifneq (,$(findstring armv,$(uname_M)))
CFLAGS+=-funwind-tables
endif
endif
# Backwards compatibility for selecting an allocator
......@@ -93,10 +102,20 @@ else
ifeq ($(uname_S),OpenBSD)
# OpenBSD
FINAL_LIBS+= -lpthread
ifeq ($(USE_BACKTRACE),yes)
FINAL_CFLAGS+= -DUSE_BACKTRACE -I/usr/local/include
FINAL_LDFLAGS+= -L/usr/local/lib
FINAL_LIBS+= -lexecinfo
endif
else
ifeq ($(uname_S),FreeBSD)
# FreeBSD
FINAL_LIBS+= -lpthread
FINAL_LIBS+= -lpthread -lexecinfo
else
ifeq ($(uname_S),DragonFly)
# FreeBSD
FINAL_LIBS+= -lpthread -lexecinfo
else
# All the other OSes (notably Linux)
FINAL_LDFLAGS+= -rdynamic
......@@ -106,6 +125,7 @@ endif
endif
endif
endif
endif
# Include paths to dependencies
FINAL_CFLAGS+= -I../deps/hiredis -I../deps/linenoise -I../deps/lua/src
......@@ -144,11 +164,11 @@ endif
REDIS_SERVER_NAME=redis-server
REDIS_SENTINEL_NAME=redis-sentinel
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o acl.o gopher.o tracking.o
REDIS_CLI_NAME=redis-cli
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o
REDIS_BENCHMARK_NAME=redis-benchmark
REDIS_BENCHMARK_OBJ=ae.o anet.o redis-benchmark.o adlist.o zmalloc.o redis-benchmark.o
REDIS_BENCHMARK_OBJ=ae.o anet.o redis-benchmark.o adlist.o dict.o zmalloc.o siphash.o redis-benchmark.o
REDIS_CHECK_RDB_NAME=redis-check-rdb
REDIS_CHECK_AOF_NAME=redis-check-aof
......@@ -290,3 +310,6 @@ install: all
$(REDIS_INSTALL) $(REDIS_CHECK_RDB_NAME) $(INSTALL_BIN)
$(REDIS_INSTALL) $(REDIS_CHECK_AOF_NAME) $(INSTALL_BIN)
@ln -sf $(REDIS_SERVER_NAME) $(INSTALL_BIN)/$(REDIS_SENTINEL_NAME)
uninstall:
rm -f $(INSTALL_BIN)/{$(REDIS_SERVER_NAME),$(REDIS_BENCHMARK_NAME),$(REDIS_CLI_NAME),$(REDIS_CHECK_RDB_NAME),$(REDIS_CHECK_AOF_NAME),$(REDIS_SENTINEL_NAME)}
This diff is collapsed.
......@@ -351,8 +351,8 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
* if flags has AE_FILE_EVENTS set, file events are processed.
* if flags has AE_TIME_EVENTS set, time events are processed.
* if flags has AE_DONT_WAIT set the function returns ASAP until all
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
* the events that's possible to process without to wait are processed.
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
*
* The function returns the number of events processed. */
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
......
......@@ -193,6 +193,20 @@ int anetSendTimeout(char *err, int fd, long long ms) {
return ANET_OK;
}
/* Set the socket receive timeout (SO_RCVTIMEO socket option) to the specified
* number of milliseconds, or disable it if the 'ms' argument is zero. */
int anetRecvTimeout(char *err, int fd, long long ms) {
struct timeval tv;
tv.tv_sec = ms/1000;
tv.tv_usec = (ms%1000)*1000;
if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) == -1) {
anetSetError(err, "setsockopt SO_RCVTIMEO: %s", strerror(errno));
return ANET_ERR;
}
return ANET_OK;
}
/* anetGenericResolve() is called by anetResolve() and anetResolveIP() to
* do the actual work. It resolves the hostname "host" and set the string
* representation of the IP address into the buffer pointed by "ipbuf".
......
......@@ -70,6 +70,7 @@ int anetEnableTcpNoDelay(char *err, int fd);
int anetDisableTcpNoDelay(char *err, int fd);
int anetTcpKeepAlive(char *err, int fd);
int anetSendTimeout(char *err, int fd, long long ms);
int anetRecvTimeout(char *err, int fd, long long ms);
int anetPeerToString(int fd, char *ip, size_t ip_len, int *port);
int anetKeepAlive(char *err, int fd, int interval);
int anetSockName(int fd, char *ip, size_t ip_len, int *port);
......
......@@ -197,6 +197,12 @@ ssize_t aofRewriteBufferWrite(int fd) {
* AOF file implementation
* ------------------------------------------------------------------------- */
/* Return true if an AOf fsync is currently already in progress in a
* BIO thread. */
int aofFsyncInProgress(void) {
return bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
}
/* Starts a background task that performs fsync() against the specified
* file descriptor (the one of the AOF file) in another thread. */
void aof_background_fsync(int fd) {
......@@ -204,7 +210,7 @@ void aof_background_fsync(int fd) {
}
/* Kills an AOFRW child process if exists */
static void killAppendOnlyChild(void) {
void killAppendOnlyChild(void) {
int statloc;
/* No AOFRW child? return. */
if (server.aof_child_pid == -1) return;
......@@ -221,6 +227,8 @@ static void killAppendOnlyChild(void) {
server.aof_rewrite_time_start = -1;
/* Close pipes used for IPC between the two processes. */
aofClosePipes();
closeChildInfoPipe();
updateDictResizePolicy();
}
/* Called when the user switches from "appendonly yes" to "appendonly no"
......@@ -261,7 +269,7 @@ int startAppendOnly(void) {
serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible.");
} else {
/* If there is a pending AOF rewrite, we need to switch it off and
* start a new one: the old one cannot be reused becuase it is not
* start a new one: the old one cannot be reused because it is not
* accumulating the AOF buffer. */
if (server.aof_child_pid != -1) {
serverLog(LL_WARNING,"AOF was enabled but there is already an AOF rewriting in background. Stopping background AOF and starting a rewrite now.");
......@@ -295,9 +303,7 @@ ssize_t aofWrite(int fd, const char *buf, size_t len) {
nwritten = write(fd, buf, len);
if (nwritten < 0) {
if (errno == EINTR) {
continue;
}
if (errno == EINTR) continue;
return totwritten ? totwritten : -1;
}
......@@ -333,10 +339,24 @@ void flushAppendOnlyFile(int force) {
int sync_in_progress = 0;
mstime_t latency;
if (sdslen(server.aof_buf) == 0) return;
if (sdslen(server.aof_buf) == 0) {
/* Check if we need to do fsync even the aof buffer is empty,
* because previously in AOF_FSYNC_EVERYSEC mode, fsync is
* called only when aof buffer is not empty, so if users
* stop write commands before fsync called in one second,
* the data in page cache cannot be flushed in time. */
if (server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.aof_fsync_offset != server.aof_current_size &&
server.unixtime > server.aof_last_fsync &&
!(sync_in_progress = aofFsyncInProgress())) {
goto try_fsync;
} else {
return;
}
}
if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
sync_in_progress = bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
sync_in_progress = aofFsyncInProgress();
if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
/* With this append fsync policy we do background fsyncing.
......@@ -468,6 +488,7 @@ void flushAppendOnlyFile(int force) {
server.aof_buf = sdsempty();
}
try_fsync:
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
* children doing I/O in the background. */
if (server.aof_no_fsync_on_rewrite &&
......@@ -482,10 +503,14 @@ void flushAppendOnlyFile(int force) {
redis_fsync(server.aof_fd); /* Let's try to get this data on the disk */
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("aof-fsync-always",latency);
server.aof_fsync_offset = server.aof_current_size;
server.aof_last_fsync = server.unixtime;
} else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
server.unixtime > server.aof_last_fsync)) {
if (!sync_in_progress) aof_background_fsync(server.aof_fd);
if (!sync_in_progress) {
aof_background_fsync(server.aof_fd);
server.aof_fsync_offset = server.aof_current_size;
}
server.aof_last_fsync = server.unixtime;
}
}
......@@ -645,7 +670,9 @@ struct client *createFakeClient(void) {
c->obuf_soft_limit_reached_time = 0;
c->watched_keys = listCreate();
c->peerid = NULL;
listSetFreeMethod(c->reply,decrRefCountVoid);
c->resp = 2;
c->user = NULL;
listSetFreeMethod(c->reply,freeClientReplyValue);
listSetDupMethod(c->reply,dupClientReplyValue);
initClientMultiState(c);
return c;
......@@ -677,18 +704,20 @@ int loadAppendOnlyFile(char *filename) {
int old_aof_state = server.aof_state;
long loops = 0;
off_t valid_up_to = 0; /* Offset of latest well-formed command loaded. */
off_t valid_before_multi = 0; /* Offset before MULTI command loaded. */
if (fp == NULL) {
serverLog(LL_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
exit(1);
}
/* Handle a zero-length AOF file as a special case. An emtpy AOF file
/* Handle a zero-length AOF file as a special case. An empty AOF file
* is a valid AOF because an empty server with AOF enabled will create
* a zero length file at startup, that will remain like that if no write
* operation is received. */
if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
server.aof_current_size = 0;
server.aof_fsync_offset = server.aof_current_size;
fclose(fp);
return C_ERR;
}
......@@ -698,7 +727,7 @@ int loadAppendOnlyFile(char *filename) {
server.aof_state = AOF_OFF;
fakeClient = createFakeClient();
startLoading(fp);
startLoadingFile(fp, filename);
/* Check if this AOF file has an RDB preamble. In that case we need to
* load the RDB file and later continue loading the AOF tail. */
......@@ -777,16 +806,28 @@ int loadAppendOnlyFile(char *filename) {
/* Command lookup */
cmd = lookupCommand(argv[0]->ptr);
if (!cmd) {
serverLog(LL_WARNING,"Unknown command '%s' reading the append only file", (char*)argv[0]->ptr);
serverLog(LL_WARNING,
"Unknown command '%s' reading the append only file",
(char*)argv[0]->ptr);
exit(1);
}
if (cmd == server.multiCommand) valid_before_multi = valid_up_to;
/* Run the command in the context of a fake client */
fakeClient->cmd = cmd;
if (fakeClient->flags & CLIENT_MULTI &&
fakeClient->cmd->proc != execCommand)
{
queueMultiCommand(fakeClient);
} else {
cmd->proc(fakeClient);
}
/* The fake client should not have a reply */
serverAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0);
serverAssert(fakeClient->bufpos == 0 &&
listLength(fakeClient->reply) == 0);
/* The fake client should never get blocked */
serverAssert((fakeClient->flags & CLIENT_BLOCKED) == 0);
......@@ -798,8 +839,15 @@ int loadAppendOnlyFile(char *filename) {
}
/* This point can only be reached when EOF is reached without errors.
* If the client is in the middle of a MULTI/EXEC, log error and quit. */
if (fakeClient->flags & CLIENT_MULTI) goto uxeof;
* If the client is in the middle of a MULTI/EXEC, handle it as it was
* a short read, even if technically the protocol is correct: we want
* to remove the unprocessed tail and continue. */
if (fakeClient->flags & CLIENT_MULTI) {
serverLog(LL_WARNING,
"Revert incomplete MULTI/EXEC transaction in AOF file");
valid_up_to = valid_before_multi;
goto uxeof;
}
loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
fclose(fp);
......@@ -808,6 +856,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
stopLoading();
aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size;
server.aof_fsync_offset = server.aof_current_size;
return C_OK;
readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
......@@ -1119,6 +1168,7 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
streamID id;
int64_t numfields;
if (s->length) {
/* Reconstruct the stream data using XADD commands. */
while(streamIteratorGetID(&si,&id,&numfields)) {
/* Emit a two elements array for each item. The first is
......@@ -1137,6 +1187,27 @@ int rewriteStreamObject(rio *r, robj *key, robj *o) {
if (rioWriteBulkString(r,(char*)value,value_len) == 0) return 0;
}
}
} else {
/* Use the XADD MAXLEN 0 trick to generate an empty stream if
* the key we are serializing is an empty string, which is possible
* for the Stream type. */
if (rioWriteBulkCount(r,'*',7) == 0) return 0;
if (rioWriteBulkString(r,"XADD",4) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0;
if (rioWriteBulkString(r,"MAXLEN",6) == 0) return 0;
if (rioWriteBulkString(r,"0",1) == 0) return 0;
if (rioWriteBulkStreamID(r,&s->last_id) == 0) return 0;
if (rioWriteBulkString(r,"x",1) == 0) return 0;
if (rioWriteBulkString(r,"y",1) == 0) return 0;
}
/* Append XSETID after XADD, make sure lastid is correct,
* in case of XDEL lastid. */
if (rioWriteBulkCount(r,'*',3) == 0) return 0;
if (rioWriteBulkString(r,"XSETID",6) == 0) return 0;
if (rioWriteBulkObject(r,key) == 0) return 0;
if (rioWriteBulkStreamID(r,&s->last_id) == 0) return 0;
/* Create all the stream consumer groups. */
if (s->cgroups) {
......@@ -1193,7 +1264,7 @@ int rewriteModuleObject(rio *r, robj *key, robj *o) {
RedisModuleIO io;
moduleValue *mv = o->ptr;
moduleType *mt = mv->type;
moduleInitIOContext(io,mt,r);
moduleInitIOContext(io,mt,r,key);
mt->aof_rewrite(&io,key,mv->value);
if (io.ctx) {
moduleFreeContext(io.ctx);
......@@ -1565,6 +1636,9 @@ void aofRemoveTempFile(pid_t childpid) {
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
unlink(tmpfile);
snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
unlink(tmpfile);
}
/* Update the server.aof_current_size field explicitly using stat(2)
......@@ -1692,6 +1766,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size;
server.aof_current_size = server.aof_current_size;
/* Clear regular AOF buffer since its contents was just written to
* the new AOF from the background rewrite buffer. */
......@@ -1713,7 +1788,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
"Background AOF rewrite signal handler took %lldus", ustime()-now);
} else if (!bysignal && exitcode != 0) {
/* SIGUSR1 is whitelisted, so we have a way to kill a child without
* tirggering an error conditon. */
* tirggering an error condition. */
if (bysignal != SIGUSR1)
server.aof_lastbgrewrite_status = C_ERR;
serverLog(LL_WARNING,
......
/* This file implements atomic counters using __atomic or __sync macros if
* available, otherwise synchronizing different threads using a mutex.
*
* The exported interaface is composed of three macros:
* The exported interface is composed of three macros:
*
* atomicIncr(var,count) -- Increment the atomic counter
* atomicGetIncr(var,oldvalue_var,count) -- Get and increment the atomic counter
......@@ -16,7 +16,7 @@
* pthread_mutex_t myvar_mutex;
* atomicSet(myvar,12345);
*
* If atomic primitives are availble (tested in config.h) the mutex
* If atomic primitives are available (tested in config.h) the mutex
* is not used.
*
* Never use return value from the macros, instead use the AtomicGetIncr()
......
......@@ -17,7 +17,7 @@
*
* The design is trivial, we have a structure representing a job to perform
* and a different thread and job queue for every job type.
* Every thread wait for new jobs in its queue, and process every job
* Every thread waits for new jobs in its queue, and process every job
* sequentially.
*
* Jobs of the same type are guaranteed to be processed from the least
......@@ -204,14 +204,14 @@ void *bioProcessBackgroundJobs(void *arg) {
}
zfree(job);
/* Unblock threads blocked on bioWaitStepOfType() if any. */
pthread_cond_broadcast(&bio_step_cond[type]);
/* Lock again before reiterating the loop, if there are no longer
* jobs to process we'll block again in pthread_cond_wait(). */
pthread_mutex_lock(&bio_mutex[type]);
listDelNode(bio_jobs[type],ln);
bio_pending[type]--;
/* Unblock threads blocked on bioWaitStepOfType() if any. */
pthread_cond_broadcast(&bio_step_cond[type]);
}
}
......
......@@ -918,7 +918,7 @@ void bitfieldCommand(client *c) {
struct bitfieldOp *ops = NULL; /* Array of ops to execute at end. */
int owtype = BFOVERFLOW_WRAP; /* Overflow type. */
int readonly = 1;
size_t higest_write_offset = 0;
size_t highest_write_offset = 0;
for (j = 2; j < c->argc; j++) {
int remargs = c->argc-j-1; /* Remaining args other than current. */
......@@ -968,8 +968,8 @@ void bitfieldCommand(client *c) {
if (opcode != BITFIELDOP_GET) {
readonly = 0;
if (higest_write_offset < bitoffset + bits - 1)
higest_write_offset = bitoffset + bits - 1;
if (highest_write_offset < bitoffset + bits - 1)
highest_write_offset = bitoffset + bits - 1;
/* INCRBY and SET require another argument. */
if (getLongLongFromObjectOrReply(c,c->argv[j+3],&i64,NULL) != C_OK){
zfree(ops);
......@@ -994,15 +994,21 @@ void bitfieldCommand(client *c) {
/* Lookup for read is ok if key doesn't exit, but errors
* if it's not a string. */
o = lookupKeyRead(c->db,c->argv[1]);
if (o != NULL && checkType(c,o,OBJ_STRING)) return;
if (o != NULL && checkType(c,o,OBJ_STRING)) {
zfree(ops);
return;
}
} else {
/* Lookup by making room up to the farest bit reached by
* this operation. */
if ((o = lookupStringForBitCommand(c,
higest_write_offset)) == NULL) return;
highest_write_offset)) == NULL) {
zfree(ops);
return;
}
}
addReplyMultiBulkLen(c,numops);
addReplyArrayLen(c,numops);
/* Actually process the operations. */
for (j = 0; j < numops; j++) {
......@@ -1047,7 +1053,7 @@ void bitfieldCommand(client *c) {
setSignedBitfield(o->ptr,thisop->offset,
thisop->bits,newval);
} else {
addReply(c,shared.nullbulk);
addReplyNull(c);
}
} else {
uint64_t oldval, newval, wrapped, retval;
......@@ -1076,7 +1082,7 @@ void bitfieldCommand(client *c) {
setUnsignedBitfield(o->ptr,thisop->offset,
thisop->bits,newval);
} else {
addReply(c,shared.nullbulk);
addReplyNull(c);
}
}
changes++;
......
This diff is collapsed.
This diff is collapsed.
......@@ -100,6 +100,13 @@ typedef struct clusterLink {
#define CLUSTERMSG_TYPE_MODULE 9 /* Module cluster API message. */
#define CLUSTERMSG_TYPE_COUNT 10 /* Total number of message types. */
/* Flags that a module can set in order to prevent certain Redis Cluster
* features to be enabled. Useful when implementing a different distributed
* system on top of Redis Cluster message bus, using modules. */
#define CLUSTER_MODULE_FLAG_NONE 0
#define CLUSTER_MODULE_FLAG_NO_FAILOVER (1<<1)
#define CLUSTER_MODULE_FLAG_NO_REDIRECTION (1<<2)
/* This structure represent elements of node->fail_reports. */
typedef struct clusterNodeFailReport {
struct clusterNode *node; /* Node reporting the failure condition. */
......@@ -243,7 +250,7 @@ union clusterMsgData {
#define CLUSTER_PROTO_VER 1 /* Cluster bus protocol version. */
typedef struct {
char sig[4]; /* Siganture "RCmb" (Redis Cluster message bus). */
char sig[4]; /* Signature "RCmb" (Redis Cluster message bus). */
uint32_t totlen; /* Total length of this message */
uint16_t ver; /* Protocol version, currently set to 1. */
uint16_t port; /* TCP base port number. */
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment