-
Notifications
You must be signed in to change notification settings - Fork 694
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Offload replication writes to IO threads #1485
base: unstable
Are you sure you want to change the base?
Conversation
Signed-off-by: Uri Yagelnik <[email protected]>
Signed-off-by: Uri Yagelnik <[email protected]>
Signed-off-by: Uri Yagelnik <[email protected]>
3aee49b
to
846c816
Compare
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## unstable #1485 +/- ##
============================================
+ Coverage 70.78% 70.85% +0.07%
============================================
Files 119 119
Lines 64691 64928 +237
============================================
+ Hits 45790 46005 +215
- Misses 18901 18923 +22
|
listNode *last_node; | ||
size_t bufpos; | ||
|
||
serverAssert(c->bufpos == 0 && listLength(c->reply) == 0); | ||
while (clientHasPendingReplies(c)) { | ||
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) { | ||
c->write_flags |= WRITE_FLAGS_WRITE_ERROR; | ||
return; | ||
} | ||
c->nwritten += nwritten; | ||
c->ref_block_pos += nwritten; | ||
/* Determine the last block and buffer position based on thread context */ | ||
if (inMainThread()) { | ||
last_node = listLast(server.repl_buffer_blocks); | ||
if (!last_node) return; | ||
bufpos = ((replBufBlock *)listNodeValue(last_node))->used; | ||
} else { | ||
last_node = c->io_last_reply_block; | ||
serverAssert(last_node != NULL); | ||
bufpos = c->io_last_bufpos; | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Consider simplifying this code to reduce duplication and improve clarity, eg:
listNode *last_node = inMainThread() ? listLast(server.repl_buffer_blocks) : c->io_last_reply_block;
if (!last_node) return;
size_t bufpos = inMainThread() ?
((replBufBlock *)listNodeValue(last_node))->used : c->io_last_bufpos;
This PR offloads the write to replica clients to IO threads.
Main Changes
Implementation Details
In order to offload the writes,
writeToReplica
has been split into 2 parts:Additional Changes
writeToReplica
we now usewritev
in case more than 1 buffer exists.nwritten
field tossize_t
since with a replica thenwritten
can theoretically exceedint
size (not subject toNET_MAX_WRITES_PER_EVENT
limit).memchr
instead ofstrchr
:strchr
to look for the next\r
memchr
as it's more secure and resolves the issueTesting
Related issue: #761