Commit 0db3b0a0 authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

Merge remote-tracking branch 'upstream/unstable' into tls

parents c469f6ad b8e02f2b
......@@ -121,21 +121,34 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len
if (r == NULL)
return NULL;
buf = malloc(len+1);
if (buf == NULL) {
freeReplyObject(r);
return NULL;
}
assert(task->type == REDIS_REPLY_ERROR ||
task->type == REDIS_REPLY_STATUS ||
task->type == REDIS_REPLY_STRING);
task->type == REDIS_REPLY_STRING ||
task->type == REDIS_REPLY_VERB);
/* Copy string value */
memcpy(buf,str,len);
buf[len] = '\0';
if (task->type == REDIS_REPLY_VERB) {
buf = malloc(len-4+1); /* Skip 4 bytes of verbatim type header. */
if (buf == NULL) {
freeReplyObject(r);
return NULL;
}
memcpy(r->vtype,str,3);
r->vtype[3] = '\0';
memcpy(buf,str+4,len-4);
buf[len-4] = '\0';
r->len = len-4;
} else {
buf = malloc(len+1);
if (buf == NULL) {
freeReplyObject(r);
return NULL;
}
memcpy(buf,str,len);
buf[len] = '\0';
r->len = len;
}
r->str = buf;
r->len = len;
if (task->parent) {
parent = task->parent->obj;
......
......@@ -102,6 +102,8 @@ typedef struct redisReply {
size_t len; /* Length of string */
char *str; /* Used for REDIS_REPLY_ERROR, REDIS_REPLY_STRING
and REDIS_REPLY_DOUBLE (in additionl to dval). */
char vtype[4]; /* Used for REDIS_REPLY_VERB, contains the null
terminated 3 character content type, such as "txt". */
size_t elements; /* number of elements, for REDIS_REPLY_ARRAY */
struct redisReply **element; /* elements vector for REDIS_REPLY_ARRAY */
} redisReply;
......
......@@ -380,10 +380,18 @@ static int processBulkItem(redisReader *r) {
/* Only continue when the buffer contains the entire bulk item. */
bytelen += len+2; /* include \r\n */
if (r->pos+bytelen <= r->len) {
if ((cur->type == REDIS_REPLY_VERB && len < 4) ||
(cur->type == REDIS_REPLY_VERB && s[5] != ':'))
{
__redisReaderSetError(r,REDIS_ERR_PROTOCOL,
"Verbatim string 4 bytes of content type are "
"missing or incorrectly encoded.");
return REDIS_ERR;
}
if (r->fn && r->fn->createString)
obj = r->fn->createString(cur,s+2,len);
else
obj = (void*)REDIS_REPLY_STRING;
obj = (void*)(long)cur->type;
success = 1;
}
}
......@@ -524,6 +532,9 @@ static int processItem(redisReader *r) {
case '#':
cur->type = REDIS_REPLY_BOOL;
break;
case '=':
cur->type = REDIS_REPLY_VERB;
break;
default:
__redisReaderSetErrorProtocolByte(r,*p);
return REDIS_ERR;
......@@ -544,6 +555,7 @@ static int processItem(redisReader *r) {
case REDIS_REPLY_BOOL:
return processLineItem(r);
case REDIS_REPLY_STRING:
case REDIS_REPLY_VERB:
return processBulkItem(r);
case REDIS_REPLY_ARRAY:
case REDIS_REPLY_MAP:
......
......@@ -56,12 +56,12 @@
#define REDIS_REPLY_ERROR 6
#define REDIS_REPLY_DOUBLE 7
#define REDIS_REPLY_BOOL 8
#define REDIS_REPLY_VERB 9
#define REDIS_REPLY_MAP 9
#define REDIS_REPLY_SET 10
#define REDIS_REPLY_ATTR 11
#define REDIS_REPLY_PUSH 12
#define REDIS_REPLY_BIGNUM 13
#define REDIS_REPLY_VERB 14
#define REDIS_READER_MAX_BUF (1024*16) /* Default max unused reader buffer. */
......
......@@ -787,7 +787,13 @@ background_thread_stats_read(tsdn_t *tsdn, background_thread_stats_t *stats) {
nstime_init(&stats->run_interval, 0);
for (unsigned i = 0; i < max_background_threads; i++) {
background_thread_info_t *info = &background_thread_info[i];
malloc_mutex_lock(tsdn, &info->mtx);
if (malloc_mutex_trylock(tsdn, &info->mtx)) {
/*
* Each background thread run may take a long time;
* avoid waiting on the stats if the thread is active.
*/
continue;
}
if (info->state != background_thread_stopped) {
num_runs += info->tot_n_runs;
nstime_add(&stats->run_interval, &info->tot_sleep_time);
......
......@@ -13,4 +13,4 @@ then
fi
make -C tests/modules && \
$TCLSH tests/test_helper.tcl --single unit/moduleapi/commandfilter --single unit/moduleapi/testrdb "${@}"
$TCLSH tests/test_helper.tcl --single unit/moduleapi/commandfilter --single unit/moduleapi/fork --single unit/moduleapi/testrdb --single unit/moduleapi/infotest --single unit/moduleapi/propagate "${@}"
......@@ -172,7 +172,7 @@ 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 localtime.o lolwut.o lolwut5.o acl.o gopher.o tracking.o connection.o tls.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 connection.o tls.o sha256.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
......
......@@ -28,6 +28,7 @@
*/
#include "server.h"
#include "sha256.h"
#include <fcntl.h>
/* =============================================================================
......@@ -93,6 +94,9 @@ void ACLResetSubcommandsForCommand(user *u, unsigned long id);
void ACLResetSubcommands(user *u);
void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub);
/* The length of the string representation of a hashed password. */
#define HASH_PASSWORD_LEN SHA256_BLOCK_SIZE*2
/* =============================================================================
* Helper functions for the rest of the ACL implementation
* ==========================================================================*/
......@@ -139,6 +143,25 @@ int time_independent_strcmp(char *a, char *b) {
return diff; /* If zero strings are the same. */
}
/* Given an SDS string, returns the SHA256 hex representation as a
* new SDS string. */
sds ACLHashPassword(unsigned char *cleartext, size_t len) {
SHA256_CTX ctx;
unsigned char hash[SHA256_BLOCK_SIZE];
char hex[HASH_PASSWORD_LEN];
char *cset = "0123456789abcdef";
sha256_init(&ctx);
sha256_update(&ctx,(unsigned char*)cleartext,len);
sha256_final(&ctx,hash);
for (int j = 0; j < SHA256_BLOCK_SIZE; j++) {
hex[j*2] = cset[((hash[j]&0xF0)>>4)];
hex[j*2+1] = cset[(hash[j]&0xF)];
}
return sdsnewlen(hex,HASH_PASSWORD_LEN);
}
/* =============================================================================
* Low level ACL API
* ==========================================================================*/
......@@ -502,7 +525,7 @@ sds ACLDescribeUser(user *u) {
listRewind(u->passwords,&li);
while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln);
res = sdscatlen(res,">",1);
res = sdscatlen(res,"#",1);
res = sdscatsds(res,thispass);
res = sdscatlen(res," ",1);
}
......@@ -629,7 +652,14 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
* ><password> Add this password to the list of valid password for the user.
* For example >mypass will add "mypass" to the list.
* This directive clears the "nopass" flag (see later).
* #<hash> Add this password hash to the list of valid hashes for
* the user. This is useful if you have previously computed
* the hash, and don't want to store it in plaintext.
* This directive clears the "nopass" flag (see later).
* <<password> Remove this password from the list of valid passwords.
* !<hash> Remove this hashed password from the list of valid passwords.
* This is useful when you want to remove a password just by
* hash without knowing its plaintext version at all.
* nopass All the set passwords of the user are removed, and the user
* is flagged as requiring no password: it means that every
* password will work against this user. If this directive is
......@@ -665,6 +695,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
* EEXIST: You are adding a key pattern after "*" was already added. This is
* almost surely an error on the user side.
* ENODEV: The password you are trying to remove from the user does not exist.
* EBADMSG: The hash you are trying to add is not a valid hash.
*/
int ACLSetUser(user *u, const char *op, ssize_t oplen) {
if (oplen == -1) oplen = strlen(op);
......@@ -700,14 +731,48 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
} else if (!strcasecmp(op,"resetpass")) {
u->flags &= ~USER_FLAG_NOPASS;
listEmpty(u->passwords);
} else if (op[0] == '>') {
sds newpass = sdsnewlen(op+1,oplen-1);
} else if (op[0] == '>' || op[0] == '#') {
sds newpass;
if (op[0] == '>') {
newpass = ACLHashPassword((unsigned char*)op+1,oplen-1);
} else {
if (oplen != HASH_PASSWORD_LEN + 1) {
errno = EBADMSG;
return C_ERR;
}
/* Password hashes can only be characters that represent
* hexadecimal values, which are numbers and lowercase
* characters 'a' through 'f'.
*/
for(int i = 1; i < HASH_PASSWORD_LEN + 1; i++) {
char c = op[i];
if ((c < 'a' || c > 'f') && (c < '0' || c > '9')) {
errno = EBADMSG;
return C_ERR;
}
}
newpass = sdsnewlen(op+1,oplen-1);
}
listNode *ln = listSearchKey(u->passwords,newpass);
/* Avoid re-adding the same password multiple times. */
if (ln == NULL) listAddNodeTail(u->passwords,newpass);
if (ln == NULL)
listAddNodeTail(u->passwords,newpass);
else
sdsfree(newpass);
u->flags &= ~USER_FLAG_NOPASS;
} else if (op[0] == '<') {
sds delpass = sdsnewlen(op+1,oplen-1);
} else if (op[0] == '<' || op[0] == '!') {
sds delpass;
if (op[0] == '<') {
delpass = ACLHashPassword((unsigned char*)op+1,oplen-1);
} else {
if (oplen != HASH_PASSWORD_LEN + 1) {
errno = EBADMSG;
return C_ERR;
}
delpass = sdsnewlen(op+1,oplen-1);
}
listNode *ln = listSearchKey(u->passwords,delpass);
sdsfree(delpass);
if (ln) {
......@@ -724,7 +789,10 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
sds newpat = sdsnewlen(op+1,oplen-1);
listNode *ln = listSearchKey(u->patterns,newpat);
/* Avoid re-adding the same pattern multiple times. */
if (ln == NULL) listAddNodeTail(u->patterns,newpat);
if (ln == NULL)
listAddNodeTail(u->patterns,newpat);
else
sdsfree(newpat);
u->flags &= ~USER_FLAG_ALLKEYS;
} else if (op[0] == '+' && op[1] != '@') {
if (strchr(op,'|') == NULL) {
......@@ -822,6 +890,9 @@ char *ACLSetUserStringError(void) {
else if (errno == ENODEV)
errmsg = "The password you are trying to remove from the user does "
"not exist";
else if (errno == EBADMSG)
errmsg = "The password hash must be exactly 64 characters and contain "
"only lowercase hexadecimal characters";
return errmsg;
}
......@@ -879,11 +950,15 @@ int ACLCheckUserCredentials(robj *username, robj *password) {
listIter li;
listNode *ln;
listRewind(u->passwords,&li);
sds hashed = ACLHashPassword(password->ptr,sdslen(password->ptr));
while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln);
if (!time_independent_strcmp(password->ptr, thispass))
if (!time_independent_strcmp(hashed, thispass)) {
sdsfree(hashed);
return C_OK;
}
}
sdsfree(hashed);
/* If we reached this point, no password matched. */
errno = EINVAL;
......
......@@ -264,9 +264,9 @@ int startAppendOnly(void) {
strerror(errno));
return C_ERR;
}
if (server.rdb_child_pid != -1) {
if (hasActiveChildProcess() && server.aof_child_pid == -1) {
server.aof_rewrite_scheduled = 1;
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.");
serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. 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 because it is not
......@@ -399,7 +399,7 @@ void flushAppendOnlyFile(int force) {
* useful for graphing / monitoring purposes. */
if (sync_in_progress) {
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
} else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
} else if (hasActiveChildProcess()) {
latencyAddSampleIfNeeded("aof-write-active-child",latency);
} else {
latencyAddSampleIfNeeded("aof-write-alone",latency);
......@@ -495,9 +495,8 @@ void flushAppendOnlyFile(int force) {
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 &&
(server.aof_child_pid != -1 || server.rdb_child_pid != -1))
return;
if (server.aof_no_fsync_on_rewrite && hasActiveChildProcess())
return;
/* Perform the fsync if needed. */
if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
......@@ -1569,39 +1568,24 @@ void aofClosePipes(void) {
*/
int rewriteAppendOnlyFileBackground(void) {
pid_t childpid;
long long start;
if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
if (aofCreatePipes() != C_OK) return C_ERR;
openChildInfoPipe();
start = ustime();
if ((childpid = fork()) == 0) {
if ((childpid = redisFork()) == 0) {
char tmpfile[256];
/* Child */
closeListeningSockets(0);
redisSetProcTitle("redis-aof-rewrite");
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"AOF rewrite: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
server.child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_AOF);
sendChildCOWInfo(CHILD_INFO_TYPE_AOF, "AOF rewrite");
exitFromChild(0);
} else {
exitFromChild(1);
}
} else {
/* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
if (childpid == -1) {
closeChildInfoPipe();
serverLog(LL_WARNING,
......@@ -1615,7 +1599,6 @@ int rewriteAppendOnlyFileBackground(void) {
server.aof_rewrite_scheduled = 0;
server.aof_rewrite_time_start = time(NULL);
server.aof_child_pid = childpid;
updateDictResizePolicy();
/* We set appendseldb to -1 in order to force the next call to the
* feedAppendOnlyFile() to issue a SELECT command, so the differences
* accumulated by the parent into server.aof_rewrite_buf will start
......@@ -1630,13 +1613,14 @@ int rewriteAppendOnlyFileBackground(void) {
void bgrewriteaofCommand(client *c) {
if (server.aof_child_pid != -1) {
addReplyError(c,"Background append only file rewriting already in progress");
} else if (server.rdb_child_pid != -1) {
} else if (hasActiveChildProcess()) {
server.aof_rewrite_scheduled = 1;
addReplyStatus(c,"Background append only file rewriting scheduled");
} else if (rewriteAppendOnlyFileBackground() == C_OK) {
addReplyStatus(c,"Background append only file rewriting started");
} else {
addReply(c,shared.err);
addReplyError(c,"Can't execute an AOF background rewriting. "
"Please check the server logs for more information.");
}
}
......
......@@ -80,6 +80,8 @@ void receiveChildInfo(void) {
server.stat_rdb_cow_bytes = server.child_info_data.cow_size;
} else if (server.child_info_data.process_type == CHILD_INFO_TYPE_AOF) {
server.stat_aof_cow_bytes = server.child_info_data.cow_size;
} else if (server.child_info_data.process_type == CHILD_INFO_TYPE_MODULE) {
server.stat_module_cow_bytes = server.child_info_data.cow_size;
}
}
}
......@@ -2210,7 +2210,7 @@ void clusterLinkConnectHandler(connection *conn) {
* full length of the packet. When a whole packet is in memory this function
* will call the function to process the packet. And so forth. */
void clusterReadHandler(connection *conn) {
char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
ssize_t nread;
clusterMsg *hdr;
clusterLink *link = connGetPrivateData(conn);
......@@ -2585,7 +2585,8 @@ void clusterBroadcastPong(int target) {
*
* If link is NULL, then the message is broadcasted to the whole cluster. */
void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
unsigned char buf[sizeof(clusterMsg)], *payload;
unsigned char *payload;
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
uint32_t channel_len, message_len;
......@@ -2605,7 +2606,7 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
/* Try to use the local buffer if possible */
if (totlen < sizeof(buf)) {
payload = buf;
payload = (unsigned char*)buf;
} else {
payload = zmalloc(totlen);
memcpy(payload,hdr,sizeof(*hdr));
......@@ -2622,7 +2623,7 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
decrRefCount(channel);
decrRefCount(message);
if (payload != buf) zfree(payload);
if (payload != (unsigned char*)buf) zfree(payload);
}
/* Send a FAIL message to all the nodes we are able to contact.
......@@ -2631,7 +2632,7 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
* we switch the node state to CLUSTER_NODE_FAIL and ask all the other
* nodes to do the same ASAP. */
void clusterSendFail(char *nodename) {
unsigned char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL);
......@@ -2643,7 +2644,7 @@ void clusterSendFail(char *nodename) {
* slots configuration. The node name, slots bitmap, and configEpoch info
* are included. */
void clusterSendUpdate(clusterLink *link, clusterNode *node) {
unsigned char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
if (link == NULL) return;
......@@ -2651,7 +2652,7 @@ void clusterSendUpdate(clusterLink *link, clusterNode *node) {
memcpy(hdr->data.update.nodecfg.nodename,node->name,CLUSTER_NAMELEN);
hdr->data.update.nodecfg.configEpoch = htonu64(node->configEpoch);
memcpy(hdr->data.update.nodecfg.slots,node->slots,sizeof(node->slots));
clusterSendMessage(link,buf,ntohl(hdr->totlen));
clusterSendMessage(link,(unsigned char*)buf,ntohl(hdr->totlen));
}
/* Send a MODULE message.
......@@ -2659,7 +2660,8 @@ void clusterSendUpdate(clusterLink *link, clusterNode *node) {
* If link is NULL, then the message is broadcasted to the whole cluster. */
void clusterSendModule(clusterLink *link, uint64_t module_id, uint8_t type,
unsigned char *payload, uint32_t len) {
unsigned char buf[sizeof(clusterMsg)], *heapbuf;
unsigned char *heapbuf;
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
......@@ -2674,7 +2676,7 @@ void clusterSendModule(clusterLink *link, uint64_t module_id, uint8_t type,
/* Try to use the local buffer if possible */
if (totlen < sizeof(buf)) {
heapbuf = buf;
heapbuf = (unsigned char*)buf;
} else {
heapbuf = zmalloc(totlen);
memcpy(heapbuf,hdr,sizeof(*hdr));
......@@ -2687,7 +2689,7 @@ void clusterSendModule(clusterLink *link, uint64_t module_id, uint8_t type,
else
clusterBroadcastMessage(heapbuf,totlen);
if (heapbuf != buf) zfree(heapbuf);
if (heapbuf != (unsigned char*)buf) zfree(heapbuf);
}
/* This function gets a cluster node ID string as target, the same way the nodes
......@@ -2731,7 +2733,7 @@ void clusterPropagatePublish(robj *channel, robj *message) {
* Note that we send the failover request to everybody, master and slave nodes,
* but only the masters are supposed to reply to our query. */
void clusterRequestFailoverAuth(void) {
unsigned char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
......@@ -2747,7 +2749,7 @@ void clusterRequestFailoverAuth(void) {
/* Send a FAILOVER_AUTH_ACK message to the specified node. */
void clusterSendFailoverAuth(clusterNode *node) {
unsigned char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
......@@ -2755,12 +2757,12 @@ void clusterSendFailoverAuth(clusterNode *node) {
clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK);
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
hdr->totlen = htonl(totlen);
clusterSendMessage(node->link,buf,totlen);
clusterSendMessage(node->link,(unsigned char*)buf,totlen);
}
/* Send a MFSTART message to the specified node. */
void clusterSendMFStart(clusterNode *node) {
unsigned char buf[sizeof(clusterMsg)];
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
......@@ -2768,7 +2770,7 @@ void clusterSendMFStart(clusterNode *node) {
clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_MFSTART);
totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
hdr->totlen = htonl(totlen);
clusterSendMessage(node->link,buf,totlen);
clusterSendMessage(node->link,(unsigned char*)buf,totlen);
}
/* Vote for the node asking for our vote if there are the conditions. */
......@@ -4292,7 +4294,9 @@ NULL
}
} else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
/* CLUSTER NODES */
addReplyBulkSds(c,clusterGenNodesDescription(0));
sds nodes = clusterGenNodesDescription(0);
addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
sdsfree(nodes);
} else if (!strcasecmp(c->argv[1]->ptr,"myid") && c->argc == 2) {
/* CLUSTER MYID */
addReplyBulkCBuffer(c,myself->name, CLUSTER_NAMELEN);
......@@ -4534,10 +4538,8 @@ NULL
"cluster_stats_messages_received:%lld\r\n", tot_msg_received);
/* Produce the reply protocol. */
addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
(unsigned long)sdslen(info)));
addReplySds(c,info);
addReply(c,shared.crlf);
addReplyVerbatim(c,info,sdslen(info),"txt");
sdsfree(info);
} else if (!strcasecmp(c->argv[1]->ptr,"saveconfig") && c->argc == 2) {
int retval = clusterSaveConfig(1);
......
......@@ -144,6 +144,7 @@ configYesNo configs_yesno[] = {
{"replica-serve-stale-data","slave-serve-stale-data",&server.repl_serve_stale_data,1,CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA},
{"replica-read-only","slave-read-only",&server.repl_slave_ro,1,CONFIG_DEFAULT_SLAVE_READ_ONLY},
{"replica-ignore-maxmemory","slave-ignore-maxmemory",&server.repl_slave_ignore_maxmemory,1,CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY},
{"jemalloc-bg-thread",NULL,&server.jemalloc_bg_thread,1,1},
{NULL, NULL, 0, 0}
};
......@@ -438,6 +439,7 @@ void loadServerConfigFromString(char *config) {
server.repl_diskless_load = configEnumGetValue(repl_diskless_load_enum,argv[1]);
if (server.repl_diskless_load == INT_MIN) {
err = "argument must be 'disabled', 'on-empty-db', 'swapdb' or 'flushdb'";
goto loaderr;
}
} else if (!strcasecmp(argv[0],"repl-diskless-sync-delay") && argc==2) {
server.repl_diskless_sync_delay = atoi(argv[1]);
......@@ -678,6 +680,10 @@ void loadServerConfigFromString(char *config) {
server.lua_time_limit = strtoll(argv[1],NULL,10);
} else if (!strcasecmp(argv[0],"lua-replicate-commands") && argc == 2) {
server.lua_always_replicate_commands = yesnotoi(argv[1]);
if (server.lua_always_replicate_commands == -1) {
err = "argument must be 'yes' or 'no'";
goto loaderr;
}
} else if (!strcasecmp(argv[0],"slowlog-log-slower-than") &&
argc == 2)
{
......
......@@ -60,10 +60,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
/* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */
if (server.rdb_child_pid == -1 &&
server.aof_child_pid == -1 &&
!(flags & LOOKUP_NOTOUCH))
{
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
......@@ -460,6 +457,13 @@ void flushdbCommand(client *c) {
if (getFlushCommandFlags(c,&flags) == C_ERR) return;
server.dirty += emptyDb(c->db->id,flags,NULL);
addReply(c,shared.ok);
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
}
/* FLUSHALL [ASYNC]
......@@ -482,6 +486,13 @@ void flushallCommand(client *c) {
server.dirty = saved_dirty;
}
server.dirty++;
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
}
/* This command implements DEL and LAZYDEL. */
......
......@@ -297,6 +297,56 @@ void computeDatasetDigest(unsigned char *final) {
}
}
#ifdef USE_JEMALLOC
void mallctl_int(client *c, robj **argv, int argc) {
int ret;
/* start with the biggest size (int64), and if that fails, try smaller sizes (int32, bool) */
int64_t old = 0, val;
if (argc > 1) {
long long ll;
if (getLongLongFromObjectOrReply(c, argv[1], &ll, NULL) != C_OK)
return;
val = ll;
}
size_t sz = sizeof(old);
while (sz > 0) {
if ((ret=je_mallctl(argv[0]->ptr, &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
if (ret==EINVAL) {
/* size might be wrong, try a smaller one */
sz /= 2;
#if BYTE_ORDER == BIG_ENDIAN
val <<= 8*sz;
#endif
continue;
}
addReplyErrorFormat(c,"%s", strerror(ret));
return;
} else {
#if BYTE_ORDER == BIG_ENDIAN
old >>= 64 - 8*sz;
#endif
addReplyLongLong(c, old);
return;
}
}
addReplyErrorFormat(c,"%s", strerror(EINVAL));
}
void mallctl_string(client *c, robj **argv, int argc) {
int ret;
char *old;
size_t sz = sizeof(old);
/* for strings, it seems we need to first get the old value, before overriding it. */
if ((ret=je_mallctl(argv[0]->ptr, &old, &sz, NULL, 0))) {
addReplyErrorFormat(c,"%s", strerror(ret));
return;
}
addReplyBulkCString(c, old);
if(argc > 1)
je_mallctl(argv[0]->ptr, NULL, 0, &argv[1]->ptr, sizeof(char*));
}
#endif
void debugCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = {
......@@ -324,6 +374,10 @@ void debugCommand(client *c) {
"STRUCTSIZE -- Return the size of different Redis core C structures.",
"ZIPLIST <key> -- Show low level info about the ziplist encoding.",
"STRINGMATCH-TEST -- Run a fuzz tester against the stringmatchlen() function.",
#ifdef USE_JEMALLOC
"MALLCTL <key> [<val>] -- Get or set a malloc tunning integer.",
"MALLCTL-STR <key> [<val>] -- Get or set a malloc tunning string.",
#endif
NULL
};
addReplyHelp(c, help);
......@@ -644,7 +698,8 @@ NULL
dictGetStats(buf,sizeof(buf),server.db[dbid].expires);
stats = sdscat(stats,buf);
addReplyBulkSds(c,stats);
addReplyVerbatim(c,stats,sdslen(stats),"txt");
sdsfree(stats);
} else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc == 3) {
robj *o;
dict *ht = NULL;
......@@ -671,7 +726,7 @@ NULL
} else {
char buf[4096];
dictGetStats(buf,sizeof(buf),ht);
addReplyBulkCString(c,buf);
addReplyVerbatim(c,buf,strlen(buf),"txt");
}
} else if (!strcasecmp(c->argv[1]->ptr,"change-repl-id") && c->argc == 2) {
serverLog(LL_WARNING,"Changing replication IDs after receiving DEBUG change-repl-id");
......@@ -682,6 +737,14 @@ NULL
{
stringmatchlen_fuzz_test();
addReplyStatus(c,"Apparently Redis did not crash: test passed");
#ifdef USE_JEMALLOC
} else if(!strcasecmp(c->argv[1]->ptr,"mallctl") && c->argc >= 3) {
mallctl_int(c, c->argv+2, c->argc-2);
return;
} else if(!strcasecmp(c->argv[1]->ptr,"mallctl-str") && c->argc >= 3) {
mallctl_string(c, c->argv+2, c->argc-2);
return;
#endif
} else {
addReplySubcommandSyntaxError(c);
return;
......@@ -1117,6 +1180,33 @@ void logRegisters(ucontext_t *uc) {
(unsigned long) uc->uc_mcontext.mc_cs
);
logStackContent((void**)uc->uc_mcontext.mc_rsp);
#elif defined(__aarch64__) /* Linux AArch64 */
serverLog(LL_WARNING,
"\n"
"X18:%016lx X19:%016lx\nX20:%016lx X21:%016lx\n"
"X22:%016lx X23:%016lx\nX24:%016lx X25:%016lx\n"
"X26:%016lx X27:%016lx\nX28:%016lx X29:%016lx\n"
"X30:%016lx\n"
"pc:%016lx sp:%016lx\npstate:%016lx fault_address:%016lx\n",
(unsigned long) uc->uc_mcontext.regs[18],
(unsigned long) uc->uc_mcontext.regs[19],
(unsigned long) uc->uc_mcontext.regs[20],
(unsigned long) uc->uc_mcontext.regs[21],
(unsigned long) uc->uc_mcontext.regs[22],
(unsigned long) uc->uc_mcontext.regs[23],
(unsigned long) uc->uc_mcontext.regs[24],
(unsigned long) uc->uc_mcontext.regs[25],
(unsigned long) uc->uc_mcontext.regs[26],
(unsigned long) uc->uc_mcontext.regs[27],
(unsigned long) uc->uc_mcontext.regs[28],
(unsigned long) uc->uc_mcontext.regs[29],
(unsigned long) uc->uc_mcontext.regs[30],
(unsigned long) uc->uc_mcontext.pc,
(unsigned long) uc->uc_mcontext.sp,
(unsigned long) uc->uc_mcontext.pstate,
(unsigned long) uc->uc_mcontext.fault_address
);
logStackContent((void**)uc->uc_mcontext.sp);
#else
serverLog(LL_WARNING,
" Dumping of registers not supported for this OS/arch");
......@@ -1344,6 +1434,12 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
/* Log dump of processor registers */
logRegisters(uc);
/* Log Modules INFO */
serverLogRaw(LL_WARNING|LL_RAW, "\n------ MODULES INFO OUTPUT ------\n");
infostring = modulesCollectInfo(sdsempty(), NULL, 1, 0);
serverLogRaw(LL_WARNING|LL_RAW, infostring);
sdsfree(infostring);
#if defined(HAVE_PROC_MAPS)
/* Test memory */
serverLogRaw(LL_WARNING|LL_RAW, "\n------ FAST MEMORY TEST ------\n");
......
......@@ -374,7 +374,7 @@ long activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
if ((newele = activeDefragStringOb(ele, &defragged)))
de->v.val = newele, defragged++;
} else if (dict_val_type == DEFRAG_SDS_DICT_VAL_VOID_PTR) {
void *newptr, *ptr = ln->value;
void *newptr, *ptr = dictGetVal(de);
if ((newptr = activeDefragAlloc(ptr)))
ln->value = newptr, defragged++;
}
......@@ -1039,7 +1039,7 @@ void activeDefragCycle(void) {
mstime_t latency;
int quit = 0;
if (server.aof_child_pid!=-1 || server.rdb_child_pid!=-1)
if (hasActiveChildProcess())
return; /* Defragging memory while there's a fork will just do damage. */
/* Once a second, check if we the fragmentation justfies starting a scan
......
......@@ -444,6 +444,7 @@ int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *lev
* Otehrwise if we are over the memory limit, but not enough memory
* was freed to return back under the limit, the function returns C_ERR. */
int freeMemoryIfNeeded(void) {
int keys_freed = 0;
/* By default replicas should ignore maxmemory
* and just be masters exact copies. */
if (server.masterhost && server.repl_slave_ignore_maxmemory) return C_OK;
......@@ -467,7 +468,7 @@ int freeMemoryIfNeeded(void) {
latencyStartMonitor(latency);
while (mem_freed < mem_tofree) {
int j, k, i, keys_freed = 0;
int j, k, i;
static unsigned int next_db = 0;
sds bestkey = NULL;
int bestdbid;
......@@ -598,9 +599,7 @@ int freeMemoryIfNeeded(void) {
mem_freed = mem_tofree;
}
}
}
if (!keys_freed) {
} else {
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("eviction-cycle",latency);
goto cant_free; /* nothing to free... */
......
......@@ -466,7 +466,7 @@ void georadiusGeneric(client *c, int flags) {
/* Look up the requested zset */
robj *zobj = NULL;
if ((zobj = lookupKeyReadOrReply(c, key, shared.null[c->resp])) == NULL ||
if ((zobj = lookupKeyReadOrReply(c, key, shared.emptyarray)) == NULL ||
checkType(c, zobj, OBJ_ZSET)) {
return;
}
......@@ -566,7 +566,7 @@ void georadiusGeneric(client *c, int flags) {
/* If no matching results, the user gets an empty reply. */
if (ga->used == 0 && storekey == NULL) {
addReplyNull(c);
addReply(c,shared.emptyarray);
geoArrayFree(ga);
return;
}
......@@ -734,14 +734,14 @@ void geohashCommand(client *c) {
r[1].max = 90;
geohashEncode(&r[0],&r[1],xy[0],xy[1],26,&hash);
char buf[12];
char buf[11];
int i;
for (i = 0; i < 11; i++) {
for (i = 0; i < 10; i++) {
int idx = (hash.bits >> (52-((i+1)*5))) & 0x1f;
buf[i] = geoalphabet[idx];
}
buf[11] = '\0';
addReplyBulkCBuffer(c,buf,11);
buf[10] = '\0';
addReplyBulkCBuffer(c,buf,10);
}
}
}
......
......@@ -1242,7 +1242,7 @@ void pfcountCommand(client *c) {
if (o == NULL) continue; /* Assume empty HLL for non existing var.*/
if (isHLLObjectOrReply(c,o) != C_OK) return;
/* Merge with this HLL with our 'max' HHL by setting max[i]
/* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */
if (hllMerge(registers,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
......@@ -1329,7 +1329,7 @@ void pfmergeCommand(client *c) {
hdr = o->ptr;
if (hdr->encoding == HLL_DENSE) use_dense = 1;
/* Merge with this HLL with our 'max' HHL by setting max[i]
/* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */
if (hllMerge(max,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
......
......@@ -599,7 +599,7 @@ NULL
event = dictGetKey(de);
graph = latencyCommandGenSparkeline(event,ts);
addReplyBulkCString(c,graph);
addReplyVerbatim(c,graph,sdslen(graph),"txt");
sdsfree(graph);
} else if (!strcasecmp(c->argv[1]->ptr,"latest") && c->argc == 2) {
/* LATENCY LATEST */
......@@ -608,7 +608,7 @@ NULL
/* LATENCY DOCTOR */
sds report = createLatencyReport();
addReplyBulkCBuffer(c,report,sdslen(report));
addReplyVerbatim(c,report,sdslen(report),"txt");
sdsfree(report);
} else if (!strcasecmp(c->argv[1]->ptr,"reset") && c->argc >= 2) {
/* LATENCY RESET */
......
......@@ -34,8 +34,11 @@
*/
#include "server.h"
#include "lolwut.h"
#include <math.h>
void lolwut5Command(client *c);
void lolwut6Command(client *c);
/* The default target for LOLWUT if no matching version was found.
* This is what unstable versions of Redis will display. */
......@@ -43,14 +46,143 @@ void lolwutUnstableCommand(client *c) {
sds rendered = sdsnew("Redis ver. ");
rendered = sdscat(rendered,REDIS_VERSION);
rendered = sdscatlen(rendered,"\n",1);
addReplyBulkSds(c,rendered);
addReplyVerbatim(c,rendered,sdslen(rendered),"txt");
sdsfree(rendered);
}
/* LOLWUT [VERSION <version>] [... version specific arguments ...] */
void lolwutCommand(client *c) {
char *v = REDIS_VERSION;
if ((v[0] == '5' && v[1] == '.') ||
char verstr[64];
if (c->argc >= 3 && !strcasecmp(c->argv[1]->ptr,"version")) {
long ver;
if (getLongFromObjectOrReply(c,c->argv[2],&ver,NULL) != C_OK) return;
snprintf(verstr,sizeof(verstr),"%u.0.0",(unsigned int)ver);
v = verstr;
/* Adjust argv/argc to filter the "VERSION ..." option, since the
* specific LOLWUT version implementations don't know about it
* and expect their arguments. */
c->argv += 2;
c->argc -= 2;
}
if ((v[0] == '5' && v[1] == '.' && v[2] != '9') ||
(v[0] == '4' && v[1] == '.' && v[2] == '9'))
lolwut5Command(c);
else if ((v[0] == '6' && v[1] == '.' && v[2] != '9') ||
(v[0] == '5' && v[1] == '.' && v[2] == '9'))
lolwut6Command(c);
else
lolwutUnstableCommand(c);
/* Fix back argc/argv in case of VERSION argument. */
if (v == verstr) {
c->argv -= 2;
c->argc += 2;
}
}
/* ========================== LOLWUT Canvase ===============================
* Many LOWUT versions will likely print some computer art to the screen.
* This is the case with LOLWUT 5 and LOLWUT 6, so here there is a generic
* canvas implementation that can be reused. */
/* Allocate and return a new canvas of the specified size. */
lwCanvas *lwCreateCanvas(int width, int height, int bgcolor) {
lwCanvas *canvas = zmalloc(sizeof(*canvas));
canvas->width = width;
canvas->height = height;
canvas->pixels = zmalloc(width*height);
memset(canvas->pixels,bgcolor,width*height);
return canvas;
}
/* Free the canvas created by lwCreateCanvas(). */
void lwFreeCanvas(lwCanvas *canvas) {
zfree(canvas->pixels);
zfree(canvas);
}
/* Set a pixel to the specified color. Color is 0 or 1, where zero means no
* dot will be displyed, and 1 means dot will be displayed.
* Coordinates are arranged so that left-top corner is 0,0. You can write
* out of the size of the canvas without issues. */
void lwDrawPixel(lwCanvas *canvas, int x, int y, int color) {
if (x < 0 || x >= canvas->width ||
y < 0 || y >= canvas->height) return;
canvas->pixels[x+y*canvas->width] = color;
}
/* Return the value of the specified pixel on the canvas. */
int lwGetPixel(lwCanvas *canvas, int x, int y) {
if (x < 0 || x >= canvas->width ||
y < 0 || y >= canvas->height) return 0;
return canvas->pixels[x+y*canvas->width];
}
/* Draw a line from x1,y1 to x2,y2 using the Bresenham algorithm. */
void lwDrawLine(lwCanvas *canvas, int x1, int y1, int x2, int y2, int color) {
int dx = abs(x2-x1);
int dy = abs(y2-y1);
int sx = (x1 < x2) ? 1 : -1;
int sy = (y1 < y2) ? 1 : -1;
int err = dx-dy, e2;
while(1) {
lwDrawPixel(canvas,x1,y1,color);
if (x1 == x2 && y1 == y2) break;
e2 = err*2;
if (e2 > -dy) {
err -= dy;
x1 += sx;
}
if (e2 < dx) {
err += dx;
y1 += sy;
}
}
}
/* Draw a square centered at the specified x,y coordinates, with the specified
* rotation angle and size. In order to write a rotated square, we use the
* trivial fact that the parametric equation:
*
* x = sin(k)
* y = cos(k)
*
* Describes a circle for values going from 0 to 2*PI. So basically if we start
* at 45 degrees, that is k = PI/4, with the first point, and then we find
* the other three points incrementing K by PI/2 (90 degrees), we'll have the
* points of the square. In order to rotate the square, we just start with
* k = PI/4 + rotation_angle, and we are done.
*
* Of course the vanilla equations above will describe the square inside a
* circle of radius 1, so in order to draw larger squares we'll have to
* multiply the obtained coordinates, and then translate them. However this
* is much simpler than implementing the abstract concept of 2D shape and then
* performing the rotation/translation transformation, so for LOLWUT it's
* a good approach. */
void lwDrawSquare(lwCanvas *canvas, int x, int y, float size, float angle, int color) {
int px[4], py[4];
/* Adjust the desired size according to the fact that the square inscribed
* into a circle of radius 1 has the side of length SQRT(2). This way
* size becomes a simple multiplication factor we can use with our
* coordinates to magnify them. */
size /= 1.4142135623;
size = round(size);
/* Compute the four points. */
float k = M_PI/4 + angle;
for (int j = 0; j < 4; j++) {
px[j] = round(sin(k) * size + x);
py[j] = round(cos(k) * size + y);
k += M_PI/2;
}
/* Draw the square. */
for (int j = 0; j < 4; j++)
lwDrawLine(canvas,px[j],py[j],px[(j+1)%4],py[(j+1)%4],color);
}
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