Unverified Commit c1718f9d authored by Wang Yuan's avatar Wang Yuan Committed by GitHub
Browse files

Replication backlog and replicas use one global shared replication buffer (#9166)

## Background
For redis master, one replica uses one copy of replication buffer, that is a big waste of memory,
more replicas more waste, and allocate/free memory for every reply list also cost much.
If we set client-output-buffer-limit small and write traffic is heavy, master may disconnect with
replicas and can't finish synchronization with replica. If we set  client-output-buffer-limit big,
master may be OOM when there are many replicas that separately keep much memory.
Because replication buffers of different replica client are the same, one simple idea is that
all replicas only use one replication buffer, that will effectively save memory.

Since replication backlog content is the same as replicas' output buffer, now we
can discard replication backlog memory and use global shared replication buffer
to implement replication backlog mechanism.

## Implementation
I create one global "replication buffer" which contains content of replication stream.
The structure of "replication buffer" is similar to the reply list that exists in every client.
But the node of list is `replBufBlock`, which has `id, repl_offset, refcount` fields.
```c
/* Replication buffer blocks is the list of replBufBlock.
 *
 * +--------------+       +--------------+       +--------------+
 * | refcount = 1 |  ...  | refcount = 0 |  ...  | refcount = 2 |
 * +--------------+       +--------------+       +--------------+
 *      |                                            /       \
 *      |                                           /         \
 *      |                                          /           \
 *  Repl Backlog                               Replia_A      Replia_B
 * 
 * Each replica or replication backlog increments only the refcount of the
 * 'ref_repl_buf_node' which it points to. So when replica walks to the next
 * node, it should first increase the next node's refcount, and when we trim
 * the replication buffer nodes, we remove node always from the head node which
 * refcount is 0. If the refcount of the head node is not 0, we must stop
 * trimming and never iterate the next node. */

/* Similar with 'clientReplyBlock', it is used for shared buffers between
 * all replica clients and replication backlog. */
typedef struct replBufBlock {
    int refcount;           /* Number of replicas or repl backlog using. */
    long long id;           /* The unique incremental number. */
    long long repl_offset;  /* Start replication offset of the block. */
    size_t size, used;
    char buf[];
} replBufBlock;
```
So now when we feed replication stream into replication backlog and all replicas, we only need
to feed stream into replication buffer `feedReplicationBuffer`. In this function, we set some fields of
replication backlog and replicas to references of the global replication buffer blocks. And we also
need to check replicas' output buffer limit to free if exceeding `client-output-buffer-limit`, and trim
replication backlog if exceeding `repl-backlog-size`.

When sending reply to replicas, we also need to iterate replication buffer blocks and send its
content, when totally sending one block for replica, we decrease current node count and
increase the next current node count, and then free the block which reference is 0 from the
head of replication buffer blocks.

Since now we use linked list to manage replication backlog, it may cost much time for iterating
all linked list nodes to find corresponding replication buffer node. So we create a rax tree to
store some nodes  for index, but to avoid rax tree occupying too much memory, i record
one per 64 nodes for index.

Currently, to make partial resynchronization as possible as much, we always let replication
backlog as the last reference of replication buffer blocks, backlog size may exceeds our setting
if slow replicas that reference vast replication buffer blocks, and this method doesn't increase
memory usage since they share replication buffer. To avoid freezing server for freeing unreferenced
replication buffer blocks when we need to trim backlog for exceeding backlog size setting,
we trim backlog incrementally (free 64 blocks per call now), and make it faster in
`beforeSleep` (free 640 blocks).

### Other changes
- `mem_total_replication_buffers`: we add this field in INFO command, it means the total
  memory of replication buffers used.
- `mem_clients_slaves`:  now even replica is slow to replicate, and its output buffer memory
  is not 0, but it still may be 0, since replication backlog and replicas share one global replication
  buffer, only if replication buffer memory is more than the repl backlog setting size, we consider
  the excess as replicas' memory. Otherwise, we think replication buffer memory is the consumption
  of repl backlog.
- Key eviction
  Since all replicas and replication backlog share global replication buffer, we think only the
  part of exceeding backlog size the extra separate consumption of replicas.
  Because we trim backlog incrementally in the background, backlog size may exceeds our
  setting if slow replicas that reference vast replication buffer blocks disconnect.
  To avoid massive eviction loop, we don't count the delayed freed replication backlog into
  used memory even if there are no replicas, i.e. we also regard this memory as replicas's memory.
- `client-output-buffer-limit` check for replica clients
  It doesn't make sense to set the replica clients output buffer limit lower than the repl-backlog-size
  config (partial sync will succeed and then replica will get disconnected). Such a configuration is
  ignored (the size of repl-backlog-size will be used). This doesn't have memory consumption
  implications since the replica client will share the backlog buffers memory.
- Drop replication backlog after loading data if needed
  We always create replication backlog if server is a master, we need it because we put DELs in
  it when loading expired keys in RDB, but if RDB doesn't have replication info or there is no rdb,
  it is not possible to support partial resynchronization, to avoid extra memory of replication backlog,
  we drop it.
- Multi IO threads
 Since all replicas and replication backlog use global replication buffer,  if I/O threads are enabled,
  to guarantee data accessing thread safe, we must let main thread handle sending the output buffer
  to all replicas. But before, other IO threads could handle sending output buffer of all replicas.

## Other optimizations
This solution resolve some other problem:
- When replicas disconnect with master since of out of output buffer limit, releasing the output
  buffer of replicas may freeze server if we set big `client-output-buffer-limit` for replicas, but now,
  it doesn't cause freezing.
- This implementation may mitigate reply list copy cost time(also freezes server) when one replication
  has huge reply buffer and another replica can copy buffer for full synchronization. now, we just copy
  reference info, it is very light.
- If we set replication backlog size big, it also may cost much time to copy replication backlog into
  replica's output buffer. But this commit eliminates this problem.
- Resizing replication backlog size doesn't empty current replication backlog content.
parent 6b297cd6
...@@ -1836,6 +1836,13 @@ activerehashing yes ...@@ -1836,6 +1836,13 @@ activerehashing yes
# Instead there is a default limit for pubsub and replica clients, since # Instead there is a default limit for pubsub and replica clients, since
# subscribers and replicas receive data in a push fashion. # subscribers and replicas receive data in a push fashion.
# #
# Note that it doesn't make sense to set the replica clients output buffer
# limit lower than the repl-backlog-size config (partial sync will succeed
# and then replica will get disconnected).
# Such a configuration is ignored (the size of repl-backlog-size will be used).
# This doesn't have memory consumption implications since the replica client
# will share the backlog buffers memory.
#
# Both the hard or the soft limit can be disabled by setting them to zero. # Both the hard or the soft limit can be disabled by setting them to zero.
client-output-buffer-limit normal 0 0 0 client-output-buffer-limit normal 0 0 0
client-output-buffer-limit replica 256mb 64mb 60 client-output-buffer-limit replica 256mb 64mb 60
......
...@@ -2399,10 +2399,10 @@ static int updateJemallocBgThread(int val, int prev, const char **err) { ...@@ -2399,10 +2399,10 @@ static int updateJemallocBgThread(int val, int prev, const char **err) {
} }
static int updateReplBacklogSize(long long val, long long prev, const char **err) { static int updateReplBacklogSize(long long val, long long prev, const char **err) {
/* resizeReplicationBacklog sets server.cfg_repl_backlog_size, and relies on /* resizeReplicationBacklog sets server.repl_backlog_size, and relies on
* being able to tell when the size changes, so restore prev before calling it. */ * being able to tell when the size changes, so restore prev before calling it. */
UNUSED(err); UNUSED(err);
server.cfg_repl_backlog_size = prev; server.repl_backlog_size = prev;
resizeReplicationBacklog(val); resizeReplicationBacklog(val);
return 1; return 1;
} }
...@@ -2684,7 +2684,7 @@ standardConfig configs[] = { ...@@ -2684,7 +2684,7 @@ standardConfig configs[] = {
createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL),
createLongLongConfig("proto-max-bulk-len", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1024*1024, LONG_MAX, server.proto_max_bulk_len, 512ll*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */ createLongLongConfig("proto-max-bulk-len", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1024*1024, LONG_MAX, server.proto_max_bulk_len, 512ll*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */
createLongLongConfig("stream-node-max-entries", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.stream_node_max_entries, 100, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("stream-node-max-entries", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.stream_node_max_entries, 100, INTEGER_CONFIG, NULL, NULL),
createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, server.cfg_repl_backlog_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */ createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, server.repl_backlog_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
/* Unsigned Long Long configs */ /* Unsigned Long Long configs */
createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, ULLONG_MAX, server.maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory), createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, ULLONG_MAX, server.maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory),
......
...@@ -325,22 +325,44 @@ unsigned long LFUDecrAndReturn(robj *o) { ...@@ -325,22 +325,44 @@ unsigned long LFUDecrAndReturn(robj *o) {
} }
/* We don't want to count AOF buffers and slaves output buffers as /* We don't want to count AOF buffers and slaves output buffers as
* used memory: the eviction should use mostly data size. This function * used memory: the eviction should use mostly data size, because
* returns the sum of AOF and slaves buffer. */ * it can cause feedback-loop when we push DELs into them, putting
* more and more DELs will make them bigger, if we count them, we
* need to evict more keys, and then generate more DELs, maybe cause
* massive eviction loop, even all keys are evicted.
*
* This function returns the sum of AOF and replication buffer. */
size_t freeMemoryGetNotCountedMemory(void) { size_t freeMemoryGetNotCountedMemory(void) {
size_t overhead = 0; size_t overhead = 0;
int slaves = listLength(server.slaves);
if (slaves) { /* Since all replicas and replication backlog share global replication
listIter li; * buffer, we think only the part of exceeding backlog size is the extra
listNode *ln; * separate consumption of replicas.
*
listRewind(server.slaves,&li); * Note that although the backlog is also initially incrementally grown
while((ln = listNext(&li))) { * (pushing DELs consumes memory), it'll eventually stop growing and
client *slave = listNodeValue(ln); * remain constant in size, so even if its creation will cause some
overhead += getClientOutputBufferMemoryUsage(slave); * eviction, it's capped, and also here to stay (no resonance effect)
*
* Note that, because we trim backlog incrementally in the background,
* backlog size may exceeds our setting if slow replicas that reference
* vast replication buffer blocks disconnect. To avoid massive eviction
* loop, we don't count the delayed freed replication backlog into used
* memory even if there are no replicas, i.e. we still regard this memory
* as replicas'. */
if ((long long)server.repl_buffer_mem > server.repl_backlog_size) {
/* We use list structure to manage replication buffer blocks, so backlog
* also occupies some extra memory, we can't know exact blocks numbers,
* we only get approximate size according to per block size. */
size_t extra_approx_size =
(server.repl_backlog_size/PROTO_REPLY_CHUNK_BYTES + 1) *
(sizeof(replBufBlock)+sizeof(listNode));
size_t counted_mem = server.repl_backlog_size + extra_approx_size;
if (server.repl_buffer_mem > counted_mem) {
overhead += (server.repl_buffer_mem - counted_mem);
} }
} }
if (server.aof_state != AOF_OFF) { if (server.aof_state != AOF_OFF) {
overhead += sdsAllocSize(server.aof_buf)+aofRewriteBufferMemoryUsage(); overhead += sdsAllocSize(server.aof_buf)+aofRewriteBufferMemoryUsage();
} }
......
...@@ -46,6 +46,18 @@ void lazyFreeLuaScripts(void *args[]) { ...@@ -46,6 +46,18 @@ void lazyFreeLuaScripts(void *args[]) {
atomicIncr(lazyfreed_objects,len); atomicIncr(lazyfreed_objects,len);
} }
/* Release replication backlog referencing memory. */
void lazyFreeReplicationBacklogRefMem(void *args[]) {
list *blocks = args[0];
rax *index = args[1];
long long len = listLength(blocks);
len += raxSize(index);
listRelease(blocks);
raxFree(index);
atomicDecr(lazyfree_objects,len);
atomicIncr(lazyfreed_objects,len);
}
/* Return the number of currently pending objects to free. */ /* Return the number of currently pending objects to free. */
size_t lazyfreeGetPendingObjectsCount(void) { size_t lazyfreeGetPendingObjectsCount(void) {
size_t aux; size_t aux;
...@@ -180,3 +192,16 @@ void freeLuaScriptsAsync(dict *lua_scripts) { ...@@ -180,3 +192,16 @@ void freeLuaScriptsAsync(dict *lua_scripts) {
dictRelease(lua_scripts); dictRelease(lua_scripts);
} }
} }
/* Free replication backlog referencing buffer blocks and rax index. */
void freeReplicationBacklogRefMemAsync(list *blocks, rax *index) {
if (listLength(blocks) > LAZYFREE_THRESHOLD ||
raxSize(index) > LAZYFREE_THRESHOLD)
{
atomicIncr(lazyfree_objects,listLength(blocks)+raxSize(index));
bioCreateLazyFreeJob(lazyFreeReplicationBacklogRefMem,2,blocks,index);
} else {
listRelease(blocks);
raxFree(index);
}
}
...@@ -276,7 +276,7 @@ void execCommand(client *c) { ...@@ -276,7 +276,7 @@ void execCommand(client *c) {
* backlog with the final EXEC. */ * backlog with the final EXEC. */
if (server.repl_backlog && was_master && !is_master) { if (server.repl_backlog && was_master && !is_master) {
char *execcmd = "*1\r\n$4\r\nEXEC\r\n"; char *execcmd = "*1\r\n$4\r\nEXEC\r\n";
feedReplicationBacklog(execcmd,strlen(execcmd)); feedReplicationBuffer(execcmd,strlen(execcmd));
} }
afterPropagateExec(); afterPropagateExec();
} }
......
...@@ -140,6 +140,8 @@ client *createClient(connection *conn) { ...@@ -140,6 +140,8 @@ client *createClient(connection *conn) {
c->name = NULL; c->name = NULL;
c->bufpos = 0; c->bufpos = 0;
c->buf_usable_size = zmalloc_usable_size(c)-offsetof(client,buf); c->buf_usable_size = zmalloc_usable_size(c)-offsetof(client,buf);
c->ref_repl_buf_node = NULL;
c->ref_block_pos = 0;
c->qb_pos = 0; c->qb_pos = 0;
c->querybuf = sdsempty(); c->querybuf = sdsempty();
c->pending_querybuf = sdsempty(); c->pending_querybuf = sdsempty();
...@@ -467,7 +469,7 @@ void afterErrorReply(client *c, const char *s, size_t len) { ...@@ -467,7 +469,7 @@ void afterErrorReply(client *c, const char *s, size_t len) {
"to its %s: '%.*s' after processing the command " "to its %s: '%.*s' after processing the command "
"'%s'", from, to, (int)len, s, cmdname); "'%s'", from, to, (int)len, s, cmdname);
if (ctype == CLIENT_TYPE_MASTER && server.repl_backlog && if (ctype == CLIENT_TYPE_MASTER && server.repl_backlog &&
server.repl_backlog_histlen > 0) server.repl_backlog->histlen > 0)
{ {
showLatestBacklog(); showLatestBacklog();
} }
...@@ -985,30 +987,37 @@ void AddReplyFromClient(client *dst, client *src) { ...@@ -985,30 +987,37 @@ void AddReplyFromClient(client *dst, client *src) {
closeClientOnOutputBufferLimitReached(dst, 1); closeClientOnOutputBufferLimitReached(dst, 1);
} }
/* Copy 'src' client output buffers into 'dst' client output buffers. /* Logically copy 'src' replica client buffers info to 'dst' replica.
* The function takes care of freeing the old output buffers of the * Basically increase referenced buffer block node reference count. */
* destination client. */ void copyReplicaOutputBuffer(client *dst, client *src) {
void copyClientOutputBuffer(client *dst, client *src) { serverAssert(src->bufpos == 0 && listLength(src->reply) == 0);
listEmpty(dst->reply);
dst->sentlen = 0;
dst->bufpos = 0;
dst->reply_bytes = 0;
/* First copy src static buffer into dst (either static buffer or reply if (src->ref_repl_buf_node == NULL) return;
* list, maybe clients have different 'usable_buffer_size'). */ dst->ref_repl_buf_node = src->ref_repl_buf_node;
_addReplyToBufferOrList(dst,src->buf,src->bufpos); dst->ref_block_pos = src->ref_block_pos;
((replBufBlock *)listNodeValue(dst->ref_repl_buf_node))->refcount++;
/* Copy src reply list into the dest. */
list* reply = listDup(src->reply);
listJoin(dst->reply,reply);
dst->reply_bytes += src->reply_bytes;
listRelease(reply);
} }
/* Return true if the specified client has pending reply buffers to write to /* Return true if the specified client has pending reply buffers to write to
* the socket. */ * the socket. */
int clientHasPendingReplies(client *c) { int clientHasPendingReplies(client *c) {
return c->bufpos || listLength(c->reply); if (getClientType(c) == CLIENT_TYPE_SLAVE) {
/* Replicas use global shared replication buffer instead of
* private output buffer. */
serverAssert(c->bufpos == 0 && listLength(c->reply) == 0);
if (c->ref_repl_buf_node == NULL) return 0;
/* If the last replication buffer block content is totally sent,
* we have nothing to send. */
listNode *ln = listLast(server.repl_buffer_blocks);
replBufBlock *tail = listNodeValue(ln);
if (ln == c->ref_repl_buf_node &&
c->ref_block_pos == tail->used) return 0;
return 1;
} else {
return c->bufpos || listLength(c->reply);
}
} }
void clientAcceptHandler(connection *conn) { void clientAcceptHandler(connection *conn) {
...@@ -1395,6 +1404,7 @@ void freeClient(client *c) { ...@@ -1395,6 +1404,7 @@ void freeClient(client *c) {
/* Free data structures. */ /* Free data structures. */
listRelease(c->reply); listRelease(c->reply);
freeReplicaReferencedReplBuffer(c);
freeClientArgv(c); freeClientArgv(c);
freeClientOriginalArgv(c); freeClientOriginalArgv(c);
...@@ -1542,6 +1552,77 @@ client *lookupClientByID(uint64_t id) { ...@@ -1542,6 +1552,77 @@ client *lookupClientByID(uint64_t id) {
return (c == raxNotFound) ? NULL : c; return (c == raxNotFound) ? NULL : c;
} }
/* This function does actual writing output buffers to different types of
* clients, it is called by writeToClient.
* If we write successfully, it return C_OK, otherwise, C_ERR is returned,
* And 'nwritten' is a output parameter, it means how many bytes server write
* to client. */
int _writeToClient(client *c, ssize_t *nwritten) {
*nwritten = 0;
if (getClientType(c) == CLIENT_TYPE_SLAVE) {
serverAssert(c->bufpos == 0 && listLength(c->reply) == 0);
replBufBlock *o = listNodeValue(c->ref_repl_buf_node);
serverAssert(o->used >= c->ref_block_pos);
/* Send current block if it is not fully sent. */
if (o->used > c->ref_block_pos) {
*nwritten = connWrite(c->conn, o->buf+c->ref_block_pos,
o->used-c->ref_block_pos);
if (*nwritten <= 0) return C_ERR;
c->ref_block_pos += *nwritten;
}
/* If we fully sent the object on head, go to the next one. */
listNode *next = listNextNode(c->ref_repl_buf_node);
if (next && c->ref_block_pos == o->used) {
o->refcount--;
((replBufBlock *)(listNodeValue(next)))->refcount++;
c->ref_repl_buf_node = next;
c->ref_block_pos = 0;
incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
}
return C_OK;
}
if (c->bufpos > 0) {
*nwritten = connWrite(c->conn,c->buf+c->sentlen,c->bufpos-c->sentlen);
if (*nwritten <= 0) return C_ERR;
c->sentlen += *nwritten;
/* If the buffer was sent, set bufpos to zero to continue with
* the remainder of the reply. */
if ((int)c->sentlen == c->bufpos) {
c->bufpos = 0;
c->sentlen = 0;
}
} else {
clientReplyBlock *o = listNodeValue(listFirst(c->reply));
size_t objlen = o->used;
if (objlen == 0) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
return C_OK;
}
*nwritten = connWrite(c->conn, o->buf + c->sentlen, objlen - c->sentlen);
if (*nwritten <= 0) return C_ERR;
c->sentlen += *nwritten;
/* If we fully sent the object on head go to the next one */
if (c->sentlen == objlen) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0;
/* If there are no longer objects in the list, we expect
* the count of reply bytes to be exactly zero. */
if (listLength(c->reply) == 0)
serverAssert(c->reply_bytes == 0);
}
}
return C_OK;
}
/* Write data in output buffers to client. Return C_OK if the client /* Write data in output buffers to client. Return C_OK if the client
* is still valid after the call, C_ERR if it was freed because of some * is still valid after the call, C_ERR if it was freed because of some
* error. If handler_installed is set, it will attempt to clear the * error. If handler_installed is set, it will attempt to clear the
...@@ -1555,48 +1636,11 @@ int writeToClient(client *c, int handler_installed) { ...@@ -1555,48 +1636,11 @@ int writeToClient(client *c, int handler_installed) {
atomicIncr(server.stat_total_writes_processed, 1); atomicIncr(server.stat_total_writes_processed, 1);
ssize_t nwritten = 0, totwritten = 0; ssize_t nwritten = 0, totwritten = 0;
size_t objlen;
clientReplyBlock *o;
while(clientHasPendingReplies(c)) { while(clientHasPendingReplies(c)) {
if (c->bufpos > 0) { int ret = _writeToClient(c, &nwritten);
nwritten = connWrite(c->conn,c->buf+c->sentlen,c->bufpos-c->sentlen); if (ret == C_ERR) break;
if (nwritten <= 0) break; totwritten += nwritten;
c->sentlen += nwritten;
totwritten += nwritten;
/* If the buffer was sent, set bufpos to zero to continue with
* the remainder of the reply. */
if ((int)c->sentlen == c->bufpos) {
c->bufpos = 0;
c->sentlen = 0;
}
} else {
o = listNodeValue(listFirst(c->reply));
objlen = o->used;
if (objlen == 0) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
continue;
}
nwritten = connWrite(c->conn, o->buf + c->sentlen, objlen - c->sentlen);
if (nwritten <= 0) break;
c->sentlen += nwritten;
totwritten += nwritten;
/* If we fully sent the object on head go to the next one */
if (c->sentlen == objlen) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0;
/* If there are no longer objects in the list, we expect
* the count of reply bytes to be exactly zero. */
if (listLength(c->reply) == 0)
serverAssert(c->reply_bytes == 0);
}
}
/* Note that we avoid to send more than NET_MAX_WRITES_PER_EVENT /* Note that we avoid to send more than NET_MAX_WRITES_PER_EVENT
* bytes, in a single threaded server it's a good idea to serve * bytes, in a single threaded server it's a good idea to serve
* other clients as well, even if a very large request comes from * other clients as well, even if a very large request comes from
...@@ -2077,8 +2121,7 @@ void commandProcessed(client *c) { ...@@ -2077,8 +2121,7 @@ void commandProcessed(client *c) {
if (c->flags & CLIENT_MASTER) { if (c->flags & CLIENT_MASTER) {
long long applied = c->reploff - prev_offset; long long applied = c->reploff - prev_offset;
if (applied) { if (applied) {
replicationFeedSlavesFromMasterStream(server.slaves, replicationFeedStreamFromMasterStream(c->pending_querybuf,applied);
c->pending_querybuf, applied);
sdsrange(c->pending_querybuf,applied,-1); sdsrange(c->pending_querybuf,applied,-1);
} }
} }
...@@ -2399,6 +2442,13 @@ sds catClientInfoString(sds s, client *client) { ...@@ -2399,6 +2442,13 @@ sds catClientInfoString(sds s, client *client) {
/* Compute the total memory consumed by this client. */ /* Compute the total memory consumed by this client. */
size_t obufmem, total_mem = getClientMemoryUsage(client, &obufmem); size_t obufmem, total_mem = getClientMemoryUsage(client, &obufmem);
size_t used_blocks_of_repl_buf = 0;
if (client->ref_repl_buf_node) {
replBufBlock *last = listNodeValue(listLast(server.repl_buffer_blocks));
replBufBlock *cur = listNodeValue(client->ref_repl_buf_node);
used_blocks_of_repl_buf = last->id - cur->id + 1;
}
sds cmdname = client->lastcmd ? getFullCommandName(client->lastcmd) : NULL; sds cmdname = client->lastcmd ? getFullCommandName(client->lastcmd) : NULL;
sds ret = sdscatfmt(s, sds ret = sdscatfmt(s,
"id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U multi-mem=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I resp=%i", "id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U multi-mem=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I resp=%i",
...@@ -2419,7 +2469,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -2419,7 +2469,7 @@ sds catClientInfoString(sds s, client *client) {
(unsigned long long) client->argv_len_sum, (unsigned long long) client->argv_len_sum,
(unsigned long long) client->mstate.argv_len_sums, (unsigned long long) client->mstate.argv_len_sums,
(unsigned long long) client->bufpos, (unsigned long long) client->bufpos,
(unsigned long long) listLength(client->reply), (unsigned long long) listLength(client->reply) + used_blocks_of_repl_buf,
(unsigned long long) obufmem, /* should not include client->buf since we want to see 0 for static clients. */ (unsigned long long) obufmem, /* should not include client->buf since we want to see 0 for static clients. */
(unsigned long long) total_mem, (unsigned long long) total_mem,
events, events,
...@@ -3247,8 +3297,21 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) { ...@@ -3247,8 +3297,21 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) {
* the caller wishes. The main usage of this function currently is * the caller wishes. The main usage of this function currently is
* enforcing the client output length limits. */ * enforcing the client output length limits. */
size_t getClientOutputBufferMemoryUsage(client *c) { size_t getClientOutputBufferMemoryUsage(client *c) {
size_t list_item_size = sizeof(listNode) + sizeof(clientReplyBlock); if (getClientType(c) == CLIENT_TYPE_SLAVE) {
return c->reply_bytes + (list_item_size*listLength(c->reply)); size_t repl_buf_size = 0;
size_t repl_node_num = 0;
size_t repl_node_size = sizeof(listNode) + sizeof(replBufBlock);
if (c->ref_repl_buf_node) {
replBufBlock *last = listNodeValue(listLast(server.repl_buffer_blocks));
replBufBlock *cur = listNodeValue(c->ref_repl_buf_node);
repl_buf_size = last->repl_offset + last->size - cur->repl_offset;
repl_node_num = last->id - cur->id + 1;
}
return repl_buf_size + (repl_node_size*repl_node_num);
} else {
size_t list_item_size = sizeof(listNode) + sizeof(clientReplyBlock);
return c->reply_bytes + (list_item_size*listLength(c->reply));
}
} }
/* Returns the total client's memory usage. /* Returns the total client's memory usage.
...@@ -3332,8 +3395,18 @@ int checkClientOutputBufferLimits(client *c) { ...@@ -3332,8 +3395,18 @@ int checkClientOutputBufferLimits(client *c) {
* like normal clients. */ * like normal clients. */
if (class == CLIENT_TYPE_MASTER) class = CLIENT_TYPE_NORMAL; if (class == CLIENT_TYPE_MASTER) class = CLIENT_TYPE_NORMAL;
/* Note that it doesn't make sense to set the replica clients output buffer
* limit lower than the repl-backlog-size config (partial sync will succeed
* and then replica will get disconnected).
* Such a configuration is ignored (the size of repl-backlog-size will be used).
* This doesn't have memory consumption implications since the replica client
* will share the backlog buffers memory. */
size_t hard_limit_bytes = server.client_obuf_limits[class].hard_limit_bytes;
if (class == CLIENT_TYPE_SLAVE && hard_limit_bytes &&
(long long)hard_limit_bytes < server.repl_backlog_size)
hard_limit_bytes = server.repl_backlog_size;
if (server.client_obuf_limits[class].hard_limit_bytes && if (server.client_obuf_limits[class].hard_limit_bytes &&
used_mem >= server.client_obuf_limits[class].hard_limit_bytes) used_mem >= hard_limit_bytes)
hard = 1; hard = 1;
if (server.client_obuf_limits[class].soft_limit_bytes && if (server.client_obuf_limits[class].soft_limit_bytes &&
used_mem >= server.client_obuf_limits[class].soft_limit_bytes) used_mem >= server.client_obuf_limits[class].soft_limit_bytes)
...@@ -3375,7 +3448,10 @@ int checkClientOutputBufferLimits(client *c) { ...@@ -3375,7 +3448,10 @@ int checkClientOutputBufferLimits(client *c) {
int closeClientOnOutputBufferLimitReached(client *c, int async) { int closeClientOnOutputBufferLimitReached(client *c, int async) {
if (!c->conn) return 0; /* It is unsafe to free fake clients. */ if (!c->conn) return 0; /* It is unsafe to free fake clients. */
serverAssert(c->reply_bytes < SIZE_MAX-(1024*64)); serverAssert(c->reply_bytes < SIZE_MAX-(1024*64));
if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return 0; /* Note that c->reply_bytes is irrelevant for replica clients
* (they use the global repl buffers). */
if ((c->reply_bytes == 0 && getClientType(c) != CLIENT_TYPE_SLAVE) ||
c->flags & CLIENT_CLOSE_ASAP) return 0;
if (checkClientOutputBufferLimits(c)) { if (checkClientOutputBufferLimits(c)) {
sds client = catClientInfoString(sdsempty(),c); sds client = catClientInfoString(sdsempty(),c);
...@@ -3740,6 +3816,15 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -3740,6 +3816,15 @@ int handleClientsWithPendingWritesUsingThreads(void) {
continue; continue;
} }
/* Since all replicas and replication backlog use global replication
* buffer, to guarantee data accessing thread safe, we must put all
* replicas client into io_threads_list[0] i.e. main thread handles
* sending the output buffer of all replicas. */
if (getClientType(c) == CLIENT_TYPE_SLAVE) {
listAddNodeTail(io_threads_list[0],c);
continue;
}
int target_id = item_id % server.io_threads_num; int target_id = item_id % server.io_threads_num;
listAddNodeTail(io_threads_list[target_id],c); listAddNodeTail(io_threads_list[target_id],c);
item_id++; item_id++;
......
...@@ -1172,20 +1172,34 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -1172,20 +1172,34 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mem_total += server.initial_memory_usage; mem_total += server.initial_memory_usage;
mem = 0; /* Replication backlog and replicas share one global replication buffer,
if (server.repl_backlog) * only if replication buffer memory is more than the repl backlog setting,
mem += zmalloc_size(server.repl_backlog); * we consider the excess as replicas' memory. Otherwise, replication buffer
mh->repl_backlog = mem; * memory is the consumption of repl backlog. */
mem_total += mem; if (listLength(server.slaves) &&
(long long)server.repl_buffer_mem > server.repl_backlog_size)
{
mh->clients_slaves = server.repl_buffer_mem - server.repl_backlog_size;
mh->repl_backlog = server.repl_backlog_size;
} else {
mh->clients_slaves = 0;
mh->repl_backlog = server.repl_buffer_mem;
}
if (server.repl_backlog) {
/* The approximate memory of rax tree for indexed blocks. */
mh->repl_backlog +=
server.repl_backlog->blocks_index->numnodes * sizeof(raxNode) +
raxSize(server.repl_backlog->blocks_index) * sizeof(void*);
}
mem_total += mh->repl_backlog;
mem_total += mh->clients_slaves;
/* Computing the memory used by the clients would be O(N) if done /* Computing the memory used by the clients would be O(N) if done
* here online. We use our values computed incrementally by * here online. We use our values computed incrementally by
* updateClientMemUsage(). */ * updateClientMemUsage(). */
mh->clients_slaves = server.stat_clients_type_memory[CLIENT_TYPE_SLAVE];
mh->clients_normal = server.stat_clients_type_memory[CLIENT_TYPE_MASTER]+ mh->clients_normal = server.stat_clients_type_memory[CLIENT_TYPE_MASTER]+
server.stat_clients_type_memory[CLIENT_TYPE_PUBSUB]+ server.stat_clients_type_memory[CLIENT_TYPE_PUBSUB]+
server.stat_clients_type_memory[CLIENT_TYPE_NORMAL]; server.stat_clients_type_memory[CLIENT_TYPE_NORMAL];
mem_total += mh->clients_slaves;
mem_total += mh->clients_normal; mem_total += mh->clients_normal;
mem = 0; mem = 0;
...@@ -1312,7 +1326,7 @@ sds getMemoryDoctorReport(void) { ...@@ -1312,7 +1326,7 @@ sds getMemoryDoctorReport(void) {
} }
/* Slaves using more than 10 MB each? */ /* Slaves using more than 10 MB each? */
if (numslaves > 0 && mh->clients_slaves / numslaves > (1024*1024*10)) { if (numslaves > 0 && mh->clients_slaves > (1024*1024*10)) {
big_slave_buf = 1; big_slave_buf = 1;
num_reports++; num_reports++;
} }
......
This diff is collapsed.
...@@ -3450,6 +3450,11 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -3450,6 +3450,11 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Close clients that need to be closed asynchronous */ /* Close clients that need to be closed asynchronous */
freeClientsInAsyncFreeQueue(); freeClientsInAsyncFreeQueue();
/* Incrementally trim replication backlog, 10 times the normal speed is
* to free replication backlog as much as possible. */
if (server.repl_backlog)
incrementalTrimReplicationBacklog(10*REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
/* Disconnect some clients if they are consuming too much memory. */ /* Disconnect some clients if they are consuming too much memory. */
evictClients(); evictClients();
...@@ -3717,9 +3722,6 @@ void initServerConfig(void) { ...@@ -3717,9 +3722,6 @@ void initServerConfig(void) {
/* Replication partial resync backlog */ /* Replication partial resync backlog */
server.repl_backlog = NULL; server.repl_backlog = NULL;
server.repl_backlog_histlen = 0;
server.repl_backlog_idx = 0;
server.repl_backlog_off = 0;
server.repl_no_slaves_since = time(NULL); server.repl_no_slaves_since = time(NULL);
/* Failover related */ /* Failover related */
...@@ -4171,6 +4173,7 @@ void initServer(void) { ...@@ -4171,6 +4173,7 @@ void initServer(void) {
server.blocked_last_cron = 0; server.blocked_last_cron = 0;
server.blocking_op_nesting = 0; server.blocking_op_nesting = 0;
server.thp_enabled = 0; server.thp_enabled = 0;
resetReplicationBuffer();
if ((server.tls_port || server.tls_replication || server.tls_cluster) if ((server.tls_port || server.tls_replication || server.tls_cluster)
&& tlsConfigure(&server.tls_ctx_config) == C_ERR) { && tlsConfigure(&server.tls_ctx_config) == C_ERR) {
...@@ -6330,6 +6333,7 @@ sds genRedisInfoString(const char *section) { ...@@ -6330,6 +6333,7 @@ sds genRedisInfoString(const char *section) {
"mem_fragmentation_bytes:%zd\r\n" "mem_fragmentation_bytes:%zd\r\n"
"mem_not_counted_for_evict:%zu\r\n" "mem_not_counted_for_evict:%zu\r\n"
"mem_replication_backlog:%zu\r\n" "mem_replication_backlog:%zu\r\n"
"mem_total_replication_buffers:%zu\r\n"
"mem_clients_slaves:%zu\r\n" "mem_clients_slaves:%zu\r\n"
"mem_clients_normal:%zu\r\n" "mem_clients_normal:%zu\r\n"
"mem_aof_buffer:%zu\r\n" "mem_aof_buffer:%zu\r\n"
...@@ -6374,6 +6378,7 @@ sds genRedisInfoString(const char *section) { ...@@ -6374,6 +6378,7 @@ sds genRedisInfoString(const char *section) {
mh->total_frag_bytes, mh->total_frag_bytes,
freeMemoryGetNotCountedMemory(), freeMemoryGetNotCountedMemory(),
mh->repl_backlog, mh->repl_backlog,
server.repl_buffer_mem,
mh->clients_slaves, mh->clients_slaves,
mh->clients_normal, mh->clients_normal,
mh->aof_buffer, mh->aof_buffer,
...@@ -6762,8 +6767,8 @@ sds genRedisInfoString(const char *section) { ...@@ -6762,8 +6767,8 @@ sds genRedisInfoString(const char *section) {
server.second_replid_offset, server.second_replid_offset,
server.repl_backlog != NULL, server.repl_backlog != NULL,
server.repl_backlog_size, server.repl_backlog_size,
server.repl_backlog_off, server.repl_backlog ? server.repl_backlog->offset : 0,
server.repl_backlog_histlen); server.repl_backlog ? server.repl_backlog->histlen : 0);
} }
/* CPU */ /* CPU */
...@@ -7515,15 +7520,19 @@ void dismissMemoryInChild(void) { ...@@ -7515,15 +7520,19 @@ void dismissMemoryInChild(void) {
/* Currently we use zmadvise_dontneed only when we use jemalloc with Linux. /* Currently we use zmadvise_dontneed only when we use jemalloc with Linux.
* so we avoid these pointless loops when they're not going to do anything. */ * so we avoid these pointless loops when they're not going to do anything. */
#if defined(USE_JEMALLOC) && defined(__linux__) #if defined(USE_JEMALLOC) && defined(__linux__)
listIter li;
listNode *ln;
/* Dismiss replication backlog. */ /* Dismiss replication buffer. We don't need to separately dismiss replication
if (server.repl_backlog != NULL) { * backlog and replica' output buffer, because they just reference the global
dismissMemory(server.repl_backlog, server.repl_backlog_size); * replication buffer but don't cost real memory. */
listRewind(server.repl_buffer_blocks, &li);
while((ln = listNext(&li))) {
replBufBlock *o = listNodeValue(ln);
dismissMemory(o, o->size);
} }
/* Dismiss all clients memory. */ /* Dismiss all clients memory. */
listIter li;
listNode *ln;
listRewind(server.clients, &li); listRewind(server.clients, &li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
client *c = listNodeValue(ln); client *c = listNodeValue(ln);
...@@ -7592,15 +7601,34 @@ void loadDataFromDisk(void) { ...@@ -7592,15 +7601,34 @@ void loadDataFromDisk(void) {
server.second_replid_offset = rsi.repl_offset+1; server.second_replid_offset = rsi.repl_offset+1;
/* Rebase master_repl_offset from rsi.repl_offset. */ /* Rebase master_repl_offset from rsi.repl_offset. */
server.master_repl_offset += rsi.repl_offset; server.master_repl_offset += rsi.repl_offset;
server.repl_backlog_off = server.master_repl_offset - serverAssert(server.repl_backlog);
server.repl_backlog_histlen + 1; server.repl_backlog->offset = server.master_repl_offset -
server.repl_backlog->histlen + 1;
server.repl_no_slaves_since = time(NULL); server.repl_no_slaves_since = time(NULL);
/* Rebase replication buffer blocks' offset since the previous
* setting offset starts from 0. */
listIter li;
listNode *ln;
listRewind(server.repl_buffer_blocks, &li);
while ((ln = listNext(&li))) {
replBufBlock *o = listNodeValue(ln);
o->repl_offset += rsi.repl_offset;
}
} }
} }
} else if (errno != ENOENT) { } else if (errno != ENOENT) {
serverLog(LL_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno)); serverLog(LL_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno));
exit(1); exit(1);
} }
/* We always create replication backlog if server is a master, we need
* it because we put DELs in it when loading expired keys in RDB, but
* if RDB doesn't have replication info or there is no rdb, it is not
* possible to support partial resynchronization, to avoid extra memory
* of replication backlog, we drop it. */
if (server.master_repl_offset == 0 && server.repl_backlog)
freeReplicationBacklog();
} }
} }
......
...@@ -377,6 +377,13 @@ typedef enum { ...@@ -377,6 +377,13 @@ typedef enum {
/* Synchronous read timeout - slave side */ /* Synchronous read timeout - slave side */
#define CONFIG_REPL_SYNCIO_TIMEOUT 5 #define CONFIG_REPL_SYNCIO_TIMEOUT 5
/* The default number of replication backlog blocks to trim per call. */
#define REPL_BACKLOG_TRIM_BLOCKS_PER_CALL 64
/* In order to quickly find the requested offset for PSYNC requests,
* we index some nodes in the replication buffer linked list into a rax. */
#define REPL_BACKLOG_INDEX_PER_BLOCKS 64
/* List related stuff */ /* List related stuff */
#define LIST_HEAD 0 #define LIST_HEAD 0
#define LIST_TAIL 1 #define LIST_TAIL 1
...@@ -767,6 +774,33 @@ typedef struct clientReplyBlock { ...@@ -767,6 +774,33 @@ typedef struct clientReplyBlock {
char buf[]; char buf[];
} clientReplyBlock; } clientReplyBlock;
/* Replication buffer blocks is the list of replBufBlock.
*
* +--------------+ +--------------+ +--------------+
* | refcount = 1 | ... | refcount = 0 | ... | refcount = 2 |
* +--------------+ +--------------+ +--------------+
* | / \
* | / \
* | / \
* Repl Backlog Replia_A Replia_B
*
* Each replica or replication backlog increments only the refcount of the
* 'ref_repl_buf_node' which it points to. So when replica walks to the next
* node, it should first increase the next node's refcount, and when we trim
* the replication buffer nodes, we remove node always from the head node which
* refcount is 0. If the refcount of the head node is not 0, we must stop
* trimming and never iterate the next node. */
/* Similar with 'clientReplyBlock', it is used for shared buffers between
* all replica clients and replication backlog. */
typedef struct replBufBlock {
int refcount; /* Number of replicas or repl backlog using. */
long long id; /* The unique incremental number. */
long long repl_offset; /* Start replication offset of the block. */
size_t size, used;
char buf[];
} replBufBlock;
/* Redis database representation. There are multiple databases identified /* Redis database representation. There are multiple databases identified
* by integers from 0 (the default database) up to the max configured * by integers from 0 (the default database) up to the max configured
* database. The database number is the 'id' field in the structure. */ * database. The database number is the 'id' field in the structure. */
...@@ -929,6 +963,24 @@ typedef struct { ...@@ -929,6 +963,24 @@ typedef struct {
need more reserved IDs use UINT64_MAX-1, need more reserved IDs use UINT64_MAX-1,
-2, ... and so forth. */ -2, ... and so forth. */
/* Replication backlog is not separate memory, it just is one consumer of
* the global replication buffer. This structure records the reference of
* replication buffers. Since the replication buffer block list may be very long,
* it would cost much time to search replication offset on partial resync, so
* we use one rax tree to index some blocks every REPL_BACKLOG_INDEX_PER_BLOCKS
* to make searching offset from replication buffer blocks list faster. */
typedef struct replBacklog {
listNode *ref_repl_buf_node; /* Referenced node of replication buffer blocks,
* see the definition of replBufBlock. */
size_t unindexed_count; /* The count from last creating index block. */
rax *blocks_index; /* The index of reocrded blocks of replication
* buffer for quickly searching replication
* offset on partial resynchronization. */
long long histlen; /* Backlog actual data length */
long long offset; /* Replication "master offset" of first
* byte in the replication backlog buffer.*/
} replBacklog;
typedef struct { typedef struct {
list *clients; list *clients;
size_t mem_usage_sum; size_t mem_usage_sum;
...@@ -1029,6 +1081,11 @@ typedef struct client { ...@@ -1029,6 +1081,11 @@ typedef struct client {
listNode *mem_usage_bucket_node; listNode *mem_usage_bucket_node;
clientMemUsageBucket *mem_usage_bucket; clientMemUsageBucket *mem_usage_bucket;
listNode *ref_repl_buf_node; /* Referenced node of replication buffer blocks,
* see the definition of replBufBlock. */
size_t ref_block_pos; /* Access position of referenced buffer block,
* i.e. the next offset to send. */
/* Response buffer */ /* Response buffer */
int bufpos; int bufpos;
size_t buf_usable_size; /* Usable size of buffer. */ size_t buf_usable_size; /* Usable size of buffer. */
...@@ -1528,14 +1585,8 @@ struct redisServer { ...@@ -1528,14 +1585,8 @@ struct redisServer {
long long second_replid_offset; /* Accept offsets up to this for replid2. */ long long second_replid_offset; /* Accept offsets up to this for replid2. */
int slaveseldb; /* Last SELECTed DB in replication output */ int slaveseldb; /* Last SELECTed DB in replication output */
int repl_ping_slave_period; /* Master pings the slave every N seconds */ int repl_ping_slave_period; /* Master pings the slave every N seconds */
char *repl_backlog; /* Replication backlog for partial syncs */ replBacklog *repl_backlog; /* Replication backlog for partial syncs */
long long repl_backlog_size; /* Backlog circular buffer size */ long long repl_backlog_size; /* Backlog circular buffer size */
long long cfg_repl_backlog_size;/* Backlog circular buffer size in config */
long long repl_backlog_histlen; /* Backlog actual data length */
long long repl_backlog_idx; /* Backlog circular buffer current offset,
that is the next byte will'll write to.*/
long long repl_backlog_off; /* Replication "master offset" of first
byte in the replication backlog buffer.*/
time_t repl_backlog_time_limit; /* Time without slaves after the backlog time_t repl_backlog_time_limit; /* Time without slaves after the backlog
gets released. */ gets released. */
time_t repl_no_slaves_since; /* We have no slaves since that time. time_t repl_no_slaves_since; /* We have no slaves since that time.
...@@ -1547,6 +1598,9 @@ struct redisServer { ...@@ -1547,6 +1598,9 @@ struct redisServer {
int repl_diskless_load; /* Slave parse RDB directly from the socket. int repl_diskless_load; /* Slave parse RDB directly from the socket.
* see REPL_DISKLESS_LOAD_* enum */ * see REPL_DISKLESS_LOAD_* enum */
int repl_diskless_sync_delay; /* Delay to start a diskless repl BGSAVE. */ int repl_diskless_sync_delay; /* Delay to start a diskless repl BGSAVE. */
size_t repl_buffer_mem; /* The memory of replication buffer. */
list *repl_buffer_blocks; /* Replication buffers blocks list
* (serving replica clients and repl backlog) */
/* Replication (slave) */ /* Replication (slave) */
char *masteruser; /* AUTH with this user and masterauth with master */ char *masteruser; /* AUTH with this user and masterauth with master */
sds masterauth; /* AUTH with this password with master */ sds masterauth; /* AUTH with this password with master */
...@@ -2031,6 +2085,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask); ...@@ -2031,6 +2085,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptTLSHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(connection *conn); void readQueryFromClient(connection *conn);
int prepareClientToWrite(client *c);
void addReplyNull(client *c); void addReplyNull(client *c);
void addReplyNullArray(client *c); void addReplyNullArray(client *c);
void addReplyBool(client *c, int b); void addReplyBool(client *c, int b);
...@@ -2063,8 +2118,8 @@ void addReplyPushLen(client *c, long length); ...@@ -2063,8 +2118,8 @@ void addReplyPushLen(client *c, long length);
void addReplyHelp(client *c, const char **help); void addReplyHelp(client *c, const char **help);
void addReplySubcommandSyntaxError(client *c); void addReplySubcommandSyntaxError(client *c);
void addReplyLoadedModules(client *c); void addReplyLoadedModules(client *c);
void copyReplicaOutputBuffer(client *dst, client *src);
void addListRangeReply(client *c, robj *o, long start, long end, int reverse); void addListRangeReply(client *c, robj *o, long start, long end, int reverse);
void copyClientOutputBuffer(client *dst, client *src);
size_t sdsZmallocSize(sds s); size_t sdsZmallocSize(sds s);
size_t getStringObjectSdsUsedMemory(robj *o); size_t getStringObjectSdsUsedMemory(robj *o);
void freeClientReplyValue(void *o); void freeClientReplyValue(void *o);
...@@ -2238,7 +2293,10 @@ ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout); ...@@ -2238,7 +2293,10 @@ ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout);
/* Replication */ /* Replication */
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc); void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
void replicationFeedSlavesFromMasterStream(list *slaves, char *buf, size_t buflen); void replicationFeedStreamFromMasterStream(char *buf, size_t buflen);
void resetReplicationBuffer(void);
void feedReplicationBuffer(char *buf, size_t len);
void freeReplicaReferencedReplBuffer(client *replica);
void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc); void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc);
void updateSlavesWaitingBgsave(int bgsaveerr, int type); void updateSlavesWaitingBgsave(int bgsaveerr, int type);
void replicationCron(void); void replicationCron(void);
...@@ -2264,8 +2322,11 @@ int replicationSetupSlaveForFullResync(client *slave, long long offset); ...@@ -2264,8 +2322,11 @@ int replicationSetupSlaveForFullResync(client *slave, long long offset);
void changeReplicationId(void); void changeReplicationId(void);
void clearReplicationId2(void); void clearReplicationId2(void);
void createReplicationBacklog(void); void createReplicationBacklog(void);
void freeReplicationBacklog(void);
void replicationCacheMasterUsingMyself(void); void replicationCacheMasterUsingMyself(void);
void feedReplicationBacklog(void *ptr, size_t len); void feedReplicationBacklog(void *ptr, size_t len);
void incrementalTrimReplicationBacklog(size_t blocks);
int canFeedReplicaReplBuffer(client *replica);
void showLatestBacklog(void); void showLatestBacklog(void);
void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask); void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
void rdbPipeWriteHandlerConnRemoved(struct connection *conn); void rdbPipeWriteHandlerConnRemoved(struct connection *conn);
...@@ -2613,7 +2674,7 @@ size_t lazyfreeGetPendingObjectsCount(void); ...@@ -2613,7 +2674,7 @@ size_t lazyfreeGetPendingObjectsCount(void);
size_t lazyfreeGetFreedObjectsCount(void); size_t lazyfreeGetFreedObjectsCount(void);
void lazyfreeResetStats(void); void lazyfreeResetStats(void);
void freeObjAsync(robj *key, robj *obj, int dbid); void freeObjAsync(robj *key, robj *obj, int dbid);
void freeReplicationBacklogRefMemAsync(list *blocks, rax *index);
/* API to get key arguments from commands */ /* API to get key arguments from commands */
int *getKeysPrepareResult(getKeysResult *result, int numkeys); int *getKeysPrepareResult(getKeysResult *result, int numkeys);
......
...@@ -118,7 +118,8 @@ start_server {} { ...@@ -118,7 +118,8 @@ start_server {} {
$master config rewrite $master config rewrite
$master debug set-active-expire 0 $master debug set-active-expire 0
for {set j 0} {$j < 1024} {incr j} { # Make sure replication backlog is full and will be trimmed.
for {set j 0} {$j < 2048} {incr j} {
$master select [expr $j%16] $master select [expr $j%16]
$master set $j somevalue px 10 $master set $j somevalue px 10
} }
...@@ -149,7 +150,7 @@ start_server {} { ...@@ -149,7 +150,7 @@ start_server {} {
assert {[status $master repl_backlog_first_byte_offset] > [status $master second_repl_offset]} assert {[status $master repl_backlog_first_byte_offset] > [status $master second_repl_offset]}
assert {[status $master sync_partial_ok] == 0} assert {[status $master sync_partial_ok] == 0}
assert {[status $master sync_full] == 1} assert {[status $master sync_full] == 1}
assert {[status $master rdb_last_load_keys_expired] == 1024} assert {[status $master rdb_last_load_keys_expired] == 2048}
assert {[status $replica sync_full] == 1} assert {[status $replica sync_full] == 1}
set digest [$master debug digest] set digest [$master debug digest]
......
# This test group aims to test that all replicas share one global replication buffer,
# two replicas don't make replication buffer size double, and when there is no replica,
# replica buffer will shrink.
start_server {tags {"repl external:skip"}} {
start_server {} {
start_server {} {
start_server {} {
set replica1 [srv -3 client]
set replica2 [srv -2 client]
set replica3 [srv -1 client]
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
$master config set save ""
$master config set repl-backlog-size 16384
$master config set client-output-buffer-limit "replica 0 0 0"
# Make sure replica3 is synchronized with master
$replica3 replicaof $master_host $master_port
wait_for_sync $replica3
# Generating RDB will take some 100 seconds
$master config set rdb-key-save-delay 1000000
populate 100 "" 16
# Make sure replica1 and replica2 are waiting bgsave
$replica1 replicaof $master_host $master_port
$replica2 replicaof $master_host $master_port
wait_for_condition 50 100 {
([s rdb_bgsave_in_progress] == 1) &&
[lindex [$replica1 role] 3] eq {sync} &&
[lindex [$replica2 role] 3] eq {sync}
} else {
fail "fail to sync with replicas"
}
test {All replicas share one global replication buffer} {
set before_used [s used_memory]
populate 1024 "" 1024 ; # Write extra 1M data
# New data uses 1M memory, but all replicas use only one
# replication buffer, so all replicas output memory is not
# more than double of replication buffer.
set repl_buf_mem [s mem_total_replication_buffers]
set extra_mem [expr {[s used_memory]-$before_used-1024*1024}]
assert {$extra_mem < 2*$repl_buf_mem}
# Kill replica1, replication_buffer will not become smaller
catch {$replica1 shutdown nosave}
wait_for_condition 50 100 {
[s connected_slaves] eq {2}
} else {
fail "replica doesn't disconnect with master"
}
assert_equal $repl_buf_mem [s mem_total_replication_buffers]
}
test {Replication buffer will become smaller when no replica uses} {
# Make sure replica3 catch up with the master
wait_for_ofs_sync $master $replica3
set repl_buf_mem [s mem_total_replication_buffers]
# Kill replica2, replication_buffer will become smaller
catch {$replica2 shutdown nosave}
wait_for_condition 50 100 {
[s connected_slaves] eq {1}
} else {
fail "replica2 doesn't disconnect with master"
}
assert {[expr $repl_buf_mem - 1024*1024] > [s mem_total_replication_buffers]}
}
}
}
}
}
# This test group aims to test replication backlog size can outgrow the backlog
# limit config if there is a slow replica which keep massive replication buffers,
# and replicas could use this replication buffer (beyond backlog config) for
# partial re-synchronization. Of course, replication backlog memory also can
# become smaller when master disconnects with slow replicas since output buffer
# limit is reached.
start_server {tags {"repl external:skip"}} {
start_server {} {
start_server {} {
set replica1 [srv -2 client]
set replica1_pid [s -2 process_id]
set replica2 [srv -1 client]
set replica2_pid [s -1 process_id]
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
$master config set save ""
$master config set repl-backlog-size 16384
$master config set client-output-buffer-limit "replica 0 0 0"
$replica1 replicaof $master_host $master_port
wait_for_sync $replica1
test {Replication backlog size can outgrow the backlog limit config} {
# Generating RDB will take 1000 seconds
$master config set rdb-key-save-delay 1000000
populate 1000 master 10000
$replica2 replicaof $master_host $master_port
# Make sure replica2 is waiting bgsave
wait_for_condition 5000 100 {
([s rdb_bgsave_in_progress] == 1) &&
[lindex [$replica2 role] 3] eq {sync}
} else {
fail "fail to sync with replicas"
}
# Replication actual backlog grow more than backlog setting since
# the slow replica2 kept replication buffer.
populate 10000 master 10000
assert {[s repl_backlog_histlen] > [expr 10000*10000]}
}
# Wait replica1 catch up with the master
wait_for_condition 1000 100 {
[s -2 master_repl_offset] eq [s master_repl_offset]
} else {
fail "Replica offset didn't catch up with the master after too long time"
}
test {Replica could use replication buffer (beyond backlog config) for partial resynchronization} {
# replica1 disconnects with master
$replica1 replicaof [srv -1 host] [srv -1 port]
# Write a mass of data that exceeds repl-backlog-size
populate 10000 master 10000
# replica1 reconnects with master
$replica1 replicaof $master_host $master_port
wait_for_condition 1000 100 {
[s -2 master_repl_offset] eq [s master_repl_offset]
} else {
fail "Replica offset didn't catch up with the master after too long time"
}
# replica2 still waits for bgsave ending
assert {[s rdb_bgsave_in_progress] eq {1} && [lindex [$replica2 role] 3] eq {sync}}
# master accepted replica1 partial resync
assert_equal [s sync_partial_ok] {1}
assert_equal [$master debug digest] [$replica1 debug digest]
}
test {Replication backlog memory will become smaller if disconnecting with replica} {
assert {[s repl_backlog_histlen] > [expr 2*10000*10000]}
assert_equal [s connected_slaves] {2}
exec kill -SIGSTOP $replica2_pid
r config set client-output-buffer-limit "replica 128k 0 0"
# trigger output buffer limit check
r set key [string repeat A [expr 64*1024]]
# master will close replica2's connection since replica2's output
# buffer limit is reached, so there only is replica1.
wait_for_condition 100 100 {
[s connected_slaves] eq {1}
} else {
fail "master didn't disconnect with replica2"
}
# Since we trim replication backlog inrementally, replication backlog
# memory may take time to be reclaimed.
wait_for_condition 1000 100 {
[s repl_backlog_histlen] < [expr 10000*10000]
} else {
fail "Replication backlog memory is not smaller"
}
exec kill -SIGCONT $replica2_pid
}
}
}
}
test {Partial resynchronization is successful even client-output-buffer-limit is less than repl-backlog-size} {
start_server {tags {"repl external:skip"}} {
start_server {} {
r config set save ""
r config set repl-backlog-size 100mb
r config set client-output-buffer-limit "replica 512k 0 0"
set replica [srv -1 client]
$replica replicaof [srv 0 host] [srv 0 port]
wait_for_sync $replica
set big_str [string repeat A [expr 10*1024*1024]] ;# 10mb big string
r multi
r client kill type replica
r set key $big_str
r set key $big_str
r debug sleep 2 ;# wait for replica reconnecting
r exec
# When replica reconnects with master, master accepts partial resync,
# and don't close replica client even client output buffer limit is
# reached.
r set key $big_str ;# trigger output buffer limit check
wait_for_ofs_sync r $replica
# master accepted replica partial resync
assert_equal [s sync_full] {1}
assert_equal [s sync_partial_ok] {1}
r multi
r set key $big_str
r set key $big_str
r exec
# replica's reply buffer size is more than client-output-buffer-limit but
# doesn't exceed repl-backlog-size, we don't close replica client.
wait_for_condition 1000 100 {
[s -1 master_repl_offset] eq [s master_repl_offset]
} else {
fail "Replica offset didn't catch up with the master after too long time"
}
assert_equal [s sync_full] {1}
assert_equal [s sync_partial_ok] {1}
}
}
}
...@@ -527,8 +527,11 @@ test {diskless loading short read} { ...@@ -527,8 +527,11 @@ test {diskless loading short read} {
$master multi $master multi
$master client kill type replica $master client kill type replica
$master set asdf asdf $master set asdf asdf
# the side effect of resizing the backlog is that it is flushed (16k is the min size) # fill replication backlog with new content
$master config set repl-backlog-size [expr {16384 + $i}] $master config set repl-backlog-size 16384
for {set keyid 0} {$keyid < 10} {incr keyid} {
$master set "$keyid string_$keyid" [string repeat A 16384]
}
$master exec $master exec
} }
# wait for loading to stop (fail) # wait for loading to stop (fail)
......
...@@ -43,6 +43,7 @@ set ::all_tests { ...@@ -43,6 +43,7 @@ set ::all_tests {
integration/replication-3 integration/replication-3
integration/replication-4 integration/replication-4
integration/replication-psync integration/replication-psync
integration/replication-buffer
integration/aof integration/aof
integration/rdb integration/rdb
integration/corrupt-dump integration/corrupt-dump
......
...@@ -355,7 +355,7 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline} ...@@ -355,7 +355,7 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline}
$rd_master setrange key:0 0 [string repeat A $payload_len] $rd_master setrange key:0 0 [string repeat A $payload_len]
} }
for {set k 0} {$k < $cmd_count} {incr k} { for {set k 0} {$k < $cmd_count} {incr k} {
#$rd_master read $rd_master read
} }
} else { } else {
for {set k 0} {$k < $cmd_count} {incr k} { for {set k 0} {$k < $cmd_count} {incr k} {
...@@ -382,12 +382,14 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline} ...@@ -382,12 +382,14 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline}
assert {$delta < $delta_max && $delta > -$delta_max} assert {$delta < $delta_max && $delta > -$delta_max}
$master client kill type slave $master client kill type slave
set killed_used [s -1 used_memory] set info_str [$master info memory]
set killed_used [getInfoProperty $info_str used_memory]
set killed_mem_not_counted_for_evict [getInfoProperty $info_str mem_not_counted_for_evict]
set killed_slave_buf [s -1 mem_clients_slaves] set killed_slave_buf [s -1 mem_clients_slaves]
set killed_mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
# we need to exclude replies buffer and query buffer of slave from used memory after kill slave # we need to exclude replies buffer and query buffer of slave from used memory after kill slave
set killed_used_no_repl [expr {$killed_used - $killed_mem_not_counted_for_evict - [slave_query_buffer $master]}] set killed_used_no_repl [expr {$killed_used - $killed_mem_not_counted_for_evict - [slave_query_buffer $master]}]
set delta_no_repl [expr {$killed_used_no_repl - $used_no_repl}] set delta_no_repl [expr {$killed_used_no_repl - $used_no_repl}]
assert {[$master dbsize] == 100}
assert {$killed_slave_buf == 0} assert {$killed_slave_buf == 0}
assert {$delta_no_repl > -$delta_max && $delta_no_repl < $delta_max} assert {$delta_no_repl > -$delta_max && $delta_no_repl < $delta_max}
......
...@@ -107,8 +107,11 @@ tags "modules" { ...@@ -107,8 +107,11 @@ tags "modules" {
$master multi $master multi
$master client kill type replica $master client kill type replica
$master set asdf asdf $master set asdf asdf
# the side effect of resizing the backlog is that it is flushed (16k is the min size) # fill replication backlog with new content
$master config set repl-backlog-size [expr {16384 + $i}] $master config set repl-backlog-size 16384
for {set keyid 0} {$keyid < 10} {incr keyid} {
$master set "$keyid string_$keyid" [string repeat A 16384]
}
$master exec $master exec
} }
# wait for loading to stop (fail) # wait for loading to stop (fail)
......
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