Commit f5ca1f9e authored by Oran Agra's avatar Oran Agra
Browse files

Merge unstable into 6.2

parents 92bde124 61d3fdb4
......@@ -249,7 +249,7 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o)
}
streamIteratorStop(&si);
} else if (o->type == OBJ_MODULE) {
RedisModuleDigest md;
RedisModuleDigest md = {{0},{0}};
moduleValue *mv = o->ptr;
moduleType *mt = mv->type;
moduleInitDigestContext(md);
......@@ -441,7 +441,7 @@ void debugCommand(client *c) {
" conflicting keys will generate an exception and kill the server."
" * NOSAVE: the database will be loaded from an existing RDB file.",
" Examples:",
" * DEBUG RELOAD: verify that the server is able to persist, flsuh and reload",
" * DEBUG RELOAD: verify that the server is able to persist, flush and reload",
" the database.",
" * DEBUG RELOAD NOSAVE: replace the current database with the contents of an",
" existing RDB file.",
......@@ -473,7 +473,7 @@ NULL
} else if (!strcasecmp(c->argv[1]->ptr,"segfault")) {
*((char*)-1) = 'x';
} else if (!strcasecmp(c->argv[1]->ptr,"panic")) {
serverPanic("DEBUG PANIC called at Unix time %ld", time(NULL));
serverPanic("DEBUG PANIC called at Unix time %lld", (long long)time(NULL));
} else if (!strcasecmp(c->argv[1]->ptr,"restart") ||
!strcasecmp(c->argv[1]->ptr,"crash-and-recover"))
{
......@@ -1809,7 +1809,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
serverLog(LL_WARNING,
"Accessing address: %p", (void*)info->si_addr);
}
if (info->si_pid != -1) {
if (info->si_code <= SI_USER && info->si_pid != -1) {
serverLog(LL_WARNING, "Killed by PID: %ld, UID: %d", (long) info->si_pid, info->si_uid);
}
......
......@@ -347,7 +347,9 @@ long activeDefragSdsListAndDict(list *l, dict *d, int dict_val_type) {
if ((newsds = activeDefragSds(sdsele))) {
/* When defragging an sds value, we need to update the dict key */
uint64_t hash = dictGetHash(d, newsds);
replaceSatelliteDictKeyPtrAndOrDefragDictEntry(d, sdsele, newsds, hash, &defragged);
dictEntry **deref = dictFindEntryRefByPtrAndHash(d, sdsele, hash);
if (deref)
(*deref)->key = newsds;
ln->value = newsds;
defragged++;
}
......
......@@ -45,11 +45,7 @@
#include "dict.h"
#include "zmalloc.h"
#ifndef DICT_BENCHMARK_MAIN
#include "redisassert.h"
#else
#include <assert.h>
#endif
/* Using dictEnableResize() / dictDisableResize() we make possible to
* enable/disable resizing of the hash table as needed. This is very important
......@@ -1175,20 +1171,18 @@ void dictGetStats(char *buf, size_t bufsize, dict *d) {
/* ------------------------------- Benchmark ---------------------------------*/
#ifdef DICT_BENCHMARK_MAIN
#include "sds.h"
#ifdef REDIS_TEST
uint64_t hashCallback(const void *key) {
return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
return dictGenHashFunction((unsigned char*)key, strlen((char*)key));
}
int compareCallback(void *privdata, const void *key1, const void *key2) {
int l1,l2;
DICT_NOTUSED(privdata);
l1 = sdslen((sds)key1);
l2 = sdslen((sds)key2);
l1 = strlen((char*)key1);
l2 = strlen((char*)key2);
if (l1 != l2) return 0;
return memcmp(key1, key2, l1) == 0;
}
......@@ -1196,7 +1190,19 @@ int compareCallback(void *privdata, const void *key1, const void *key2) {
void freeCallback(void *privdata, void *val) {
DICT_NOTUSED(privdata);
sdsfree(val);
zfree(val);
}
char *stringFromLongLong(long long value) {
char buf[32];
int len;
char *s;
len = sprintf(buf,"%lld",value);
s = zmalloc(len+1);
memcpy(s, buf, len);
s[len] = '\0';
return s;
}
dictType BenchmarkDictType = {
......@@ -1215,22 +1221,26 @@ dictType BenchmarkDictType = {
printf(msg ": %ld items in %lld ms\n", count, elapsed); \
} while(0)
/* dict-benchmark [count] */
int main(int argc, char **argv) {
/* ./redis-server test dict [<count> | --accurate] */
int dictTest(int argc, char **argv, int accurate) {
long j;
long long start, elapsed;
dict *dict = dictCreate(&BenchmarkDictType,NULL);
long count = 0;
if (argc == 2) {
count = strtol(argv[1],NULL,10);
} else {
if (argc == 4) {
if (accurate) {
count = 5000000;
} else {
count = strtol(argv[3],NULL,10);
}
} else {
count = 5000;
}
start_benchmark();
for (j = 0; j < count; j++) {
int retval = dictAdd(dict,sdsfromlonglong(j),(void*)j);
int retval = dictAdd(dict,stringFromLongLong(j),(void*)j);
assert(retval == DICT_OK);
}
end_benchmark("Inserting");
......@@ -1243,28 +1253,28 @@ int main(int argc, char **argv) {
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(j);
char *key = stringFromLongLong(j);
dictEntry *de = dictFind(dict,key);
assert(de != NULL);
sdsfree(key);
zfree(key);
}
end_benchmark("Linear access of existing elements");
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(j);
char *key = stringFromLongLong(j);
dictEntry *de = dictFind(dict,key);
assert(de != NULL);
sdsfree(key);
zfree(key);
}
end_benchmark("Linear access of existing elements (2nd round)");
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(rand() % count);
char *key = stringFromLongLong(rand() % count);
dictEntry *de = dictFind(dict,key);
assert(de != NULL);
sdsfree(key);
zfree(key);
}
end_benchmark("Random access of existing elements");
......@@ -1277,17 +1287,17 @@ int main(int argc, char **argv) {
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(rand() % count);
char *key = stringFromLongLong(rand() % count);
key[0] = 'X';
dictEntry *de = dictFind(dict,key);
assert(de == NULL);
sdsfree(key);
zfree(key);
}
end_benchmark("Accessing missing");
start_benchmark();
for (j = 0; j < count; j++) {
sds key = sdsfromlonglong(j);
char *key = stringFromLongLong(j);
int retval = dictDelete(dict,key);
assert(retval == DICT_OK);
key[0] += 17; /* Change first number to letter. */
......@@ -1295,5 +1305,7 @@ int main(int argc, char **argv) {
assert(retval == DICT_OK);
}
end_benchmark("Removing and adding");
dictRelease(dict);
return 0;
}
#endif
......@@ -201,4 +201,8 @@ extern dictType dictTypeHeapStringCopyKey;
extern dictType dictTypeHeapStrings;
extern dictType dictTypeHeapStringCopyKeyValue;
#ifdef REDIS_TEST
int dictTest(int argc, char *argv[], int accurate);
#endif
#endif /* __DICT_H */
......@@ -105,11 +105,12 @@ uint64_t intrev64(uint64_t v) {
#include <stdio.h>
#define UNUSED(x) (void)(x)
int endianconvTest(int argc, char *argv[]) {
int endianconvTest(int argc, char *argv[], int accurate) {
char buf[32];
UNUSED(argc);
UNUSED(argv);
UNUSED(accurate);
sprintf(buf,"ciaoroma");
memrev16(buf);
......
......@@ -72,7 +72,7 @@ uint64_t intrev64(uint64_t v);
#endif
#ifdef REDIS_TEST
int endianconvTest(int argc, char *argv[]);
int endianconvTest(int argc, char *argv[], int accurate);
#endif
#endif
......@@ -184,7 +184,7 @@ struct commandHelp {
8,
"6.2.0" },
{ "CLIENT KILL",
"[ip:port] [ID client-id] [TYPE normal|master|slave|pubsub] [USER username] [ADDR ip:port] [SKIPME yes/no]",
"[ip:port] [ID client-id] [TYPE normal|master|slave|pubsub] [USER username] [ADDR ip:port] [LADDR ip:port] [SKIPME yes/no]",
"Kill the connection of a client",
8,
"2.4.0" },
......
......@@ -284,7 +284,7 @@ size_t intsetBlobLen(intset *is) {
return sizeof(intset)+intrev32ifbe(is->length)*intrev32ifbe(is->encoding);
}
/* Validate the integrity of the data stracture.
/* Validate the integrity of the data structure.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we make sure there are no duplicate or out of order records. */
int intsetValidateIntegrity(const unsigned char *p, size_t size, int deep) {
......@@ -392,7 +392,7 @@ static void checkConsistency(intset *is) {
}
#define UNUSED(x) (void)(x)
int intsetTest(int argc, char **argv) {
int intsetTest(int argc, char **argv, int accurate) {
uint8_t success;
int i;
intset *is;
......@@ -400,6 +400,7 @@ int intsetTest(int argc, char **argv) {
UNUSED(argc);
UNUSED(argv);
UNUSED(accurate);
printf("Value encodings: "); {
assert(_intsetValueEncoding(-32768) == INTSET_ENC_INT16);
......@@ -424,6 +425,7 @@ int intsetTest(int argc, char **argv) {
is = intsetAdd(is,4,&success); assert(success);
is = intsetAdd(is,4,&success); assert(!success);
ok();
zfree(is);
}
printf("Large number of random adds: "); {
......@@ -436,6 +438,7 @@ int intsetTest(int argc, char **argv) {
assert(intrev32ifbe(is->length) == inserts);
checkConsistency(is);
ok();
zfree(is);
}
printf("Upgrade from int16 to int32: "); {
......@@ -447,6 +450,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,32));
assert(intsetFind(is,65535));
checkConsistency(is);
zfree(is);
is = intsetNew();
is = intsetAdd(is,32,NULL);
......@@ -457,6 +461,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,-65535));
checkConsistency(is);
ok();
zfree(is);
}
printf("Upgrade from int16 to int64: "); {
......@@ -468,6 +473,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,32));
assert(intsetFind(is,4294967295));
checkConsistency(is);
zfree(is);
is = intsetNew();
is = intsetAdd(is,32,NULL);
......@@ -478,6 +484,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,-4294967295));
checkConsistency(is);
ok();
zfree(is);
}
printf("Upgrade from int32 to int64: "); {
......@@ -489,6 +496,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,65535));
assert(intsetFind(is,4294967295));
checkConsistency(is);
zfree(is);
is = intsetNew();
is = intsetAdd(is,65535,NULL);
......@@ -499,6 +507,7 @@ int intsetTest(int argc, char **argv) {
assert(intsetFind(is,-4294967295));
checkConsistency(is);
ok();
zfree(is);
}
printf("Stress lookups: "); {
......@@ -512,6 +521,7 @@ int intsetTest(int argc, char **argv) {
for (i = 0; i < num; i++) intsetSearch(is,rand() % ((1<<bits)-1),NULL);
printf("%ld lookups, %ld element set, %lldusec\n",
num,size,usec()-start);
zfree(is);
}
printf("Stress add+delete: "); {
......@@ -528,6 +538,7 @@ int intsetTest(int argc, char **argv) {
}
checkConsistency(is);
ok();
zfree(is);
}
return 0;
......
......@@ -49,7 +49,7 @@ size_t intsetBlobLen(intset *is);
int intsetValidateIntegrity(const unsigned char *is, size_t size, int deep);
#ifdef REDIS_TEST
int intsetTest(int argc, char *argv[]);
int intsetTest(int argc, char *argv[], int accurate);
#endif
#endif // __INTSET_H
......@@ -256,7 +256,7 @@ sds createLatencyReport(void) {
if (dictSize(server.latency_events) == 0 &&
server.latency_monitor_threshold == 0)
{
report = sdscat(report,"I'm sorry, Dave, I can't do that. Latency monitoring is disabled in this Redis instance. You may use \"CONFIG SET latency-monitor-threshold <milliseconds>.\" in order to enable it. If we weren't in a deep space mission I'd suggest to take a look at http://redis.io/topics/latency-monitor.\n");
report = sdscat(report,"I'm sorry, Dave, I can't do that. Latency monitoring is disabled in this Redis instance. You may use \"CONFIG SET latency-monitor-threshold <milliseconds>.\" in order to enable it. If we weren't in a deep space mission I'd suggest to take a look at https://redis.io/topics/latency-monitor.\n");
return report;
}
......@@ -426,7 +426,7 @@ sds createLatencyReport(void) {
}
if (advise_slowlog_inspect) {
report = sdscat(report,"- Check your Slow Log to understand what are the commands you are running which are too slow to execute. Please check http://redis.io/commands/slowlog for more information.\n");
report = sdscat(report,"- Check your Slow Log to understand what are the commands you are running which are too slow to execute. Please check https://redis.io/commands/slowlog for more information.\n");
}
/* Intrinsic latency. */
......
......@@ -908,7 +908,7 @@ int lpValidateNext(unsigned char *lp, unsigned char **pp, size_t lpbytes) {
#undef OUT_OF_RANGE
}
/* Validate the integrity of the data stracture.
/* Validate the integrity of the data structure.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int lpValidateIntegrity(unsigned char *lp, size_t size, int deep){
......
This diff is collapsed.
# coding: utf-8
# gendoc.rb -- Converts the top-comments inside module.c to modules API
# reference documentation in markdown format.
......@@ -21,15 +22,48 @@ def markdown(s)
l = l.gsub(/(?<![`A-z])[a-z_]+\(\)/, '`\0`')
# Add backquotes around macro and var names containing underscores.
l = l.gsub(/(?<![`A-z\*])[A-Za-z]+_[A-Za-z0-9_]+/){|x| "`#{x}`"}
# Link URLs preceded by space (i.e. when not already linked)
l = l.gsub(/ (https?:\/\/[A-Za-z0-9_\/\.\-]+[A-Za-z0-9\/])/,
' [\1](\1)')
# Link URLs preceded by space or newline (not already linked)
l = l.gsub(/(^| )(https?:\/\/[A-Za-z0-9_\/\.\-]+[A-Za-z0-9\/])/,
'\1[\2](\2)')
# Replace double-dash with unicode ndash
l = l.gsub(/ -- /, ' – ')
end
# Link function names to their definition within the page
l = l.gsub(/`(RedisModule_[A-z0-9]+)[()]*`/) {|x|
$index[$1] ? "[#{x}](\##{$1})" : x
}
newlines << l
}
return newlines.join("\n")
end
# Linebreak a prototype longer than 80 characters on the commas, but only
# between balanced parentheses so that we don't linebreak args which are
# function pointers, and then aligning each arg under each other.
def linebreak_proto(proto, indent)
if proto.bytesize <= 80
return proto
end
parts = proto.split(/,\s*/);
if parts.length == 1
return proto;
end
align_pos = proto.index("(") + 1;
align = " " * align_pos
result = parts.shift;
bracket_balance = 0;
parts.each{|part|
if bracket_balance == 0
result += ",\n" + indent + align
else
result += ", "
end
result += part
bracket_balance += part.count("(") - part.count(")")
}
return result;
end
# Given the source code array and the index at which an exported symbol was
# detected, extracts and outputs the documentation.
def docufy(src,i)
......@@ -38,7 +72,11 @@ def docufy(src,i)
name = name.sub("RM_","RedisModule_")
proto = src[i].sub("{","").strip+";\n"
proto = proto.sub("RM_","RedisModule_")
puts "## `#{name}`\n\n"
proto = linebreak_proto(proto, " ");
# Add a link target with the function name. (We don't trust the exact id of
# the generated one, which depends on the Markdown implementation.)
puts "<span id=\"#{name}\"></span>\n\n"
puts "### `#{name}`\n\n"
puts " #{proto}\n"
comment = ""
while true
......@@ -50,13 +88,87 @@ def docufy(src,i)
puts comment+"\n\n"
end
# Print a comment from line until */ is found, as markdown.
def section_doc(src, i)
name = get_section_heading(src, i)
comment = "<span id=\"#{section_name_to_id(name)}\"></span>\n\n"
while true
# append line, except if it's a horizontal divider
comment = comment + src[i] if src[i] !~ /^[\/ ]?\*{1,2} ?-{50,}/
break if src[i] =~ /\*\//
i = i+1
end
comment = markdown(comment)
puts comment+"\n\n"
end
# generates an id suitable for links within the page
def section_name_to_id(name)
return "section-" +
name.strip.downcase.gsub(/[^a-z0-9]+/, '-').gsub(/^-+|-+$/, '')
end
# Returns the name of the first section heading in the comment block for which
# is_section_doc(src, i) is true
def get_section_heading(src, i)
if src[i] =~ /^\/\*\*? \#+ *(.*)/
heading = $1
elsif src[i+1] =~ /^ ?\* \#+ *(.*)/
heading = $1
end
return heading.gsub(' -- ', ' – ')
end
# Returns true if the line is the start of a generic documentation section. Such
# section must start with the # symbol, i.e. a markdown heading, on the first or
# the second line.
def is_section_doc(src, i)
return src[i] =~ /^\/\*\*? \#/ ||
(src[i] =~ /^\/\*/ && src[i+1] =~ /^ ?\* \#/)
end
def is_func_line(src, i)
line = src[i]
return line =~ /RM_/ &&
line[0] != ' ' && line[0] != '#' && line[0] != '/' &&
src[i-1] =~ /\*\//
end
puts "# Modules API reference\n\n"
puts "<!-- This file is generated from module.c using gendoc.rb -->\n\n"
src = File.open("../module.c").to_a
src.each_with_index{|line,i|
if line =~ /RM_/ && line[0] != ' ' && line[0] != '#' && line[0] != '/'
if src[i-1] =~ /\*\//
docufy(src,i)
src = File.open(File.dirname(__FILE__) ++ "/../module.c").to_a
# Build function index
$index = {}
src.each_with_index do |line,i|
if is_func_line(src, i)
line =~ /RM_([A-z0-9]+)/
name = "RedisModule_#{$1}"
$index[name] = true
end
end
# Print TOC
puts "## Sections\n\n"
src.each_with_index do |_line,i|
if is_section_doc(src, i)
name = get_section_heading(src, i)
puts "* [#{name}](\##{section_name_to_id(name)})\n"
end
}
end
puts "* [Function index](#section-function-index)\n\n"
# Docufy: Print function prototype and markdown docs
src.each_with_index do |_line,i|
if is_func_line(src, i)
docufy(src, i)
elsif is_section_doc(src, i)
section_doc(src, i)
end
end
# Print function index
puts "<span id=\"section-function-index\"></span>\n\n"
puts "## Function index\n\n"
$index.keys.sort.each{|x| puts "* [`#{x}`](\##{x})\n"}
puts "\n"
......@@ -113,34 +113,34 @@ void discardCommand(client *c) {
addReply(c,shared.ok);
}
void beforePropagateMultiOrExec(int multi) {
if (multi) {
void beforePropagateMulti() {
/* Propagating MULTI */
serverAssert(!server.propagate_in_transaction);
server.propagate_in_transaction = 1;
} else {
}
void afterPropagateExec() {
/* Propagating EXEC */
serverAssert(server.propagate_in_transaction == 1);
server.propagate_in_transaction = 0;
}
}
/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
* implementation for more information. */
void execCommandPropagateMulti(int dbid) {
beforePropagateMultiOrExec(1);
beforePropagateMulti();
propagate(server.multiCommand,dbid,&shared.multi,1,
PROPAGATE_AOF|PROPAGATE_REPL);
}
void execCommandPropagateExec(int dbid) {
beforePropagateMultiOrExec(0);
propagate(server.execCommand,dbid,&shared.exec,1,
PROPAGATE_AOF|PROPAGATE_REPL);
afterPropagateExec();
}
/* Aborts a transaction, with a specific error message.
* The transaction is always aboarted with -EXECABORT so that the client knows
* The transaction is always aborted with -EXECABORT so that the client knows
* the server exited the multi state, but the actual reason for the abort is
* included too.
* Note: 'error' may or may not end with \r\n. see addReplyErrorFormat. */
......@@ -202,11 +202,9 @@ void execCommand(client *c) {
c->cmd = c->mstate.commands[j].cmd;
/* ACL permissions are also checked at the time of execution in case
* they were changed after the commands were ququed. */
* they were changed after the commands were queued. */
int acl_errpos;
int acl_retval = ACLCheckCommandPerm(c,&acl_errpos);
if (acl_retval == ACL_OK && c->cmd->proc == publishCommand)
acl_retval = ACLCheckPubsubPerm(c,1,1,0,&acl_errpos);
int acl_retval = ACLCheckAllPerm(c,&acl_errpos);
if (acl_retval != ACL_OK) {
char *reason;
switch (acl_retval) {
......@@ -217,7 +215,8 @@ void execCommand(client *c) {
reason = "no permission to touch the specified keys";
break;
case ACL_DENIED_CHANNEL:
reason = "no permission to publish to the specified channel";
reason = "no permission to access one of the channels used "
"as arguments";
break;
default:
reason = "no permission";
......@@ -254,7 +253,6 @@ void execCommand(client *c) {
if (server.propagate_in_transaction) {
int is_master = server.masterhost == NULL;
server.dirty++;
beforePropagateMultiOrExec(0);
/* If inside the MULTI/EXEC block this instance was suddenly
* switched from master to slave (using the SLAVEOF command), the
* initial MULTI was propagated into the replication backlog, but the
......@@ -264,6 +262,7 @@ void execCommand(client *c) {
char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
feedReplicationBacklog(execcmd,strlen(execcmd));
}
afterPropagateExec();
}
server.in_exec = 0;
......
......@@ -154,6 +154,7 @@ client *createClient(connection *conn) {
c->read_reploff = 0;
c->repl_ack_off = 0;
c->repl_ack_time = 0;
c->repl_last_partial_write = 0;
c->slave_listening_port = 0;
c->slave_addr = NULL;
c->slave_capa = SLAVE_CAPA_NONE;
......@@ -331,8 +332,9 @@ void _addReplyProtoToList(client *c, const char *s, size_t len) {
memcpy(tail->buf, s, len);
listAddNodeTail(c->reply, tail);
c->reply_bytes += tail->size;
}
asyncCloseClientOnOutputBufferLimitReached(c);
}
}
/* -----------------------------------------------------------------------------
......@@ -562,7 +564,7 @@ void *addReplyDeferredLen(client *c) {
void setDeferredReply(client *c, void *node, const char *s, size_t length) {
listNode *ln = (listNode*)node;
clientReplyBlock *next;
clientReplyBlock *next, *prev;
/* Abort when *node is NULL: when the client should not accept writes
* we return NULL in addReplyDeferredLen() */
......@@ -571,14 +573,31 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) {
/* Normally we fill this dummy NULL node, added by addReplyDeferredLen(),
* with a new buffer structure containing the protocol needed to specify
* the length of the array following. However sometimes when there is
* little memory to move, we may instead remove this NULL node, and prefix
* our protocol in the node immediately after to it, in order to save a
* write(2) syscall later. Conditions needed to do it:
* the length of the array following. However sometimes there might be room
* in the previous/next node so we can instead remove this NULL node, and
* suffix/prefix our data in the node immediately before/after it, in order
* to save a write(2) syscall later. Conditions needed to do it:
*
* - The prev node is non-NULL and has space in it or
* - The next node is non-NULL,
* - It has enough room already allocated
* - And not too large (avoid large memmove) */
if (ln->prev != NULL && (prev = listNodeValue(ln->prev)) &&
prev->size - prev->used > 0)
{
size_t len_to_copy = prev->size - prev->used;
if (len_to_copy > length)
len_to_copy = length;
memcpy(prev->buf + prev->used, s, len_to_copy);
prev->used += len_to_copy;
length -= len_to_copy;
if (length == 0) {
listDelNode(c->reply, ln);
return;
}
s += len_to_copy;
}
if (ln->next != NULL && (next = listNodeValue(ln->next)) &&
next->size - next->used >= length &&
next->used < PROTO_REPLY_CHUNK_BYTES * 4)
......@@ -596,8 +615,9 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) {
memcpy(buf->buf, s, length);
listNodeValue(ln) = buf;
c->reply_bytes += buf->size;
}
asyncCloseClientOnOutputBufferLimitReached(c);
}
}
/* Populate the length object and try gluing it to the next chunk. */
......@@ -1531,9 +1551,7 @@ int writeToClient(client *c, int handler_installed) {
}
atomicIncr(server.stat_net_output_bytes, totwritten);
if (nwritten == -1) {
if (connGetState(c->conn) == CONN_STATE_CONNECTED) {
nwritten = 0;
} else {
if (connGetState(c->conn) != CONN_STATE_CONNECTED) {
serverLog(LL_VERBOSE,
"Error writing to client: %s", connGetLastError(c->conn));
freeClientAsync(c);
......@@ -1645,6 +1663,9 @@ void resetClient(client *c) {
c->flags |= CLIENT_REPLY_SKIP;
c->flags &= ~CLIENT_REPLY_SKIP_NEXT;
}
/* Always clear the prevent logging field. */
c->flags &= ~CLIENT_PREVENT_LOGGING;
}
/* This function is used when we want to re-enter the event loop but there
......@@ -1954,13 +1975,10 @@ void commandProcessed(client *c) {
c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
}
/* Don't reset the client structure for clients blocked in a
* module blocking command, so that the reply callback will
* still be able to access the client argv and argc field.
* The client will be reset in unblockClientFromModule(). */
if (!(c->flags & CLIENT_BLOCKED) ||
(c->btype != BLOCKED_MODULE && c->btype != BLOCKED_PAUSE))
{
/* Don't reset the client structure for blocked clients, so that the reply
* callback will still be able to access the client argv and argc fields.
* The client will be reset in unblockClient(). */
if (!(c->flags & CLIENT_BLOCKED)) {
resetClient(c);
}
......@@ -1990,12 +2008,20 @@ void commandProcessed(client *c) {
* of processing the command, otherwise C_OK is returned. */
int processCommandAndResetClient(client *c) {
int deadclient = 0;
client *old_client = server.current_client;
server.current_client = c;
if (processCommand(c) == C_OK) {
commandProcessed(c);
}
if (server.current_client == NULL) deadclient = 1;
server.current_client = NULL;
/*
* Restore the old client, this is needed because when a script
* times out, we will get into this code from processEventsWhileBlocked.
* Which will cause to set the server.current_client. If not restored
* we will return 1 to our caller which will falsely indicate the client
* is dead and will stop reading from its buffer.
*/
server.current_client = old_client;
/* performEvictions may flush slave output buffers. This may
* result in a slave, that may be the active client, to be
* freed. */
......@@ -2941,6 +2967,7 @@ void helloCommand(client *c) {
int moreargs = (c->argc-1) - j;
const char *opt = c->argv[j]->ptr;
if (!strcasecmp(opt,"AUTH") && moreargs >= 2) {
preventCommandLogging(c);
if (ACLAuthenticateUser(c, c->argv[j+1], c->argv[j+2]) == C_ERR) {
addReplyError(c,"-WRONGPASS invalid username-password pair or user is disabled.");
return;
......@@ -3007,7 +3034,7 @@ void securityWarningCommand(client *c) {
static time_t logged_time;
time_t now = time(NULL);
if (labs(now-logged_time) > 60) {
if (llabs(now-logged_time) > 60) {
serverLog(LL_WARNING,"Possible SECURITY ATTACK detected. It looks like somebody is sending POST or Host: commands to Redis. This is likely due to an attacker attempting to use Cross Protocol Scripting to compromise your Redis instance. Connection aborted.");
logged_time = now;
}
......@@ -3304,6 +3331,8 @@ int areClientsPaused(void) {
* if it has. Also returns true if clients are now paused and false
* otherwise. */
int checkClientPauseTimeoutAndReturnIfPaused(void) {
if (!areClientsPaused())
return 0;
if (server.client_pause_end_time < server.mstime) {
unpauseClients();
}
......@@ -3332,7 +3361,7 @@ void processEventsWhileBlocked(void) {
/* Note: when we are processing events while blocked (for instance during
* busy Lua scripts), we set a global flag. When such flag is set, we
* avoid handling the read part of clients using threaded I/O.
* See https://github.com/antirez/redis/issues/6988 for more info. */
* See https://github.com/redis/redis/issues/6988 for more info. */
ProcessingEventsWhileBlocked = 1;
while (iterations--) {
long long startval = server.events_processed_while_blocked;
......
......@@ -56,6 +56,7 @@ int keyspaceEventsStringToFlags(char *classes) {
case 'E': flags |= NOTIFY_KEYEVENT; break;
case 't': flags |= NOTIFY_STREAM; break;
case 'm': flags |= NOTIFY_KEY_MISS; break;
case 'd': flags |= NOTIFY_MODULE; break;
default: return -1;
}
}
......@@ -82,6 +83,7 @@ sds keyspaceEventsFlagsToString(int flags) {
if (flags & NOTIFY_EXPIRED) res = sdscatlen(res,"x",1);
if (flags & NOTIFY_EVICTED) res = sdscatlen(res,"e",1);
if (flags & NOTIFY_STREAM) res = sdscatlen(res,"t",1);
if (flags & NOTIFY_MODULE) res = sdscatlen(res,"d",1);
}
if (flags & NOTIFY_KEYSPACE) res = sdscatlen(res,"K",1);
if (flags & NOTIFY_KEYEVENT) res = sdscatlen(res,"E",1);
......
......@@ -727,7 +727,11 @@ int getRangeLongFromObjectOrReply(client *c, robj *o, long min, long max, long *
}
int getPositiveLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg) {
if (msg) {
return getRangeLongFromObjectOrReply(c, o, 0, LONG_MAX, target, msg);
} else {
return getRangeLongFromObjectOrReply(c, o, 0, LONG_MAX, target, "value is out of range, must be positive");
}
}
int getIntFromObjectOrReply(client *c, robj *o, int *target, const char *msg) {
......
......@@ -331,21 +331,6 @@ int pubsubPublishMessage(robj *channel, robj *message) {
return receivers;
}
/* This wraps handling ACL channel permissions for the given client. */
int pubsubCheckACLPermissionsOrReply(client *c, int idx, int count, int literal) {
/* Check if the user can run the command according to the current
* ACLs. */
int acl_chanpos;
int acl_retval = ACLCheckPubsubPerm(c,idx,count,literal,&acl_chanpos);
if (acl_retval == ACL_DENIED_CHANNEL) {
addACLLogEntry(c,acl_retval,acl_chanpos,NULL);
addReplyError(c,
"-NOPERM this user has no permissions to access "
"one of the channels used as arguments");
}
return acl_retval;
}
/*-----------------------------------------------------------------------------
* Pubsub commands implementation
*----------------------------------------------------------------------------*/
......@@ -353,7 +338,6 @@ int pubsubCheckACLPermissionsOrReply(client *c, int idx, int count, int literal)
/* SUBSCRIBE channel [channel ...] */
void subscribeCommand(client *c) {
int j;
if (pubsubCheckACLPermissionsOrReply(c,1,c->argc-1,0) != ACL_OK) return;
if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
/**
* A client that has CLIENT_DENY_BLOCKING flag on
......@@ -387,7 +371,6 @@ void unsubscribeCommand(client *c) {
/* PSUBSCRIBE pattern [pattern ...] */
void psubscribeCommand(client *c) {
int j;
if (pubsubCheckACLPermissionsOrReply(c,1,c->argc-1,1) != ACL_OK) return;
if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
/**
* A client that has CLIENT_DENY_BLOCKING flag on
......@@ -420,7 +403,6 @@ void punsubscribeCommand(client *c) {
/* PUBLISH <channel> <message> */
void publishCommand(client *c) {
if (pubsubCheckACLPermissionsOrReply(c,1,1,0) != ACL_OK) return;
int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
if (server.cluster_enabled)
clusterPropagatePublish(c->argv[1],c->argv[2]);
......
This diff is collapsed.
......@@ -199,7 +199,7 @@ quicklistNode *quicklistBookmarkFind(quicklist *ql, const char *name);
void quicklistBookmarksClear(quicklist *ql);
#ifdef REDIS_TEST
int quicklistTest(int argc, char *argv[]);
int quicklistTest(int argc, char *argv[], int accurate);
#endif
/* Directions for iterators */
......
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