Commit ee22a8bd authored by antirez's avatar antirez
Browse files

Revert "Fix #7306 less aggressively."

This reverts commit b407590c.
parent 0471dcaf
...@@ -1039,12 +1039,9 @@ static void freeClientArgv(client *c) { ...@@ -1039,12 +1039,9 @@ static void freeClientArgv(client *c) {
* when we resync with our own master and want to force all our slaves to * when we resync with our own master and want to force all our slaves to
* resync with us as well. */ * resync with us as well. */
void disconnectSlaves(void) { void disconnectSlaves(void) {
listIter li; while (listLength(server.slaves)) {
listNode *ln;
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
listNode *ln = listFirst(server.slaves); listNode *ln = listFirst(server.slaves);
freeClientAsync((client*)ln->value); freeClient((client*)ln->value);
} }
} }
......
...@@ -39,7 +39,7 @@ ...@@ -39,7 +39,7 @@
#include <sys/socket.h> #include <sys/socket.h>
#include <sys/stat.h> #include <sys/stat.h>
long long adjustMeaningfulReplOffset(int *adjusted); long long adjustMeaningfulReplOffset();
void replicationDiscardCachedMaster(void); void replicationDiscardCachedMaster(void);
void replicationResurrectCachedMaster(connection *conn); void replicationResurrectCachedMaster(connection *conn);
void replicationSendAck(void); void replicationSendAck(void);
...@@ -2084,12 +2084,19 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) { ...@@ -2084,12 +2084,19 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) {
* new one. */ * new one. */
memcpy(server.replid,new,sizeof(server.replid)); memcpy(server.replid,new,sizeof(server.replid));
memcpy(server.cached_master->replid,new,sizeof(server.replid)); memcpy(server.cached_master->replid,new,sizeof(server.replid));
/* Disconnect all the sub-slaves: they need to be notified. */
disconnectSlaves();
} }
} }
/* Disconnect all the sub-replicas: they need to be notified always because
* in case the master has last replicated some non-meaningful commands
* (e.g. PINGs), the primary replica will request the PSYNC offset for the
* last known meaningful command. This means the master will again replicate
* the non-meaningful commands. If the sub-replicas still remains connected,
* they will receive those commands a second time and increment the master
* replication offset to be higher than the master's offset forever.
*/
disconnectSlaves();
/* Setup the replication to continue. */ /* Setup the replication to continue. */
sdsfree(reply); sdsfree(reply);
replicationResurrectCachedMaster(conn); replicationResurrectCachedMaster(conn);
...@@ -2758,8 +2765,7 @@ void replicationCacheMaster(client *c) { ...@@ -2758,8 +2765,7 @@ void replicationCacheMaster(client *c) {
/* Adjust reploff and read_reploff to the last meaningful offset we /* Adjust reploff and read_reploff to the last meaningful offset we
* executed. This is the offset the replica will use for future PSYNC. */ * executed. This is the offset the replica will use for future PSYNC. */
int offset_adjusted; server.master->reploff = adjustMeaningfulReplOffset();
server.master->reploff = adjustMeaningfulReplOffset(&offset_adjusted);
server.master->read_reploff = server.master->reploff; server.master->read_reploff = server.master->reploff;
if (c->flags & CLIENT_MULTI) discardTransaction(c); if (c->flags & CLIENT_MULTI) discardTransaction(c);
listEmpty(c->reply); listEmpty(c->reply);
...@@ -2782,13 +2788,6 @@ void replicationCacheMaster(client *c) { ...@@ -2782,13 +2788,6 @@ void replicationCacheMaster(client *c) {
* so make sure to adjust the replication state. This function will * so make sure to adjust the replication state. This function will
* also set server.master to NULL. */ * also set server.master to NULL. */
replicationHandleMasterDisconnection(); replicationHandleMasterDisconnection();
/* If we trimmed this replica backlog, we need to disconnect our chained
* replicas (if any), otherwise they may have the PINGs we removed
* from the stream and their offset would no longer match: upon
* disconnection they will also trim the final PINGs and will be able
* to incrementally sync without issues. */
if (offset_adjusted) disconnectSlaves();
} }
/* If the "meaningful" offset, that is the offset without the final PINGs /* If the "meaningful" offset, that is the offset without the final PINGs
...@@ -2798,13 +2797,8 @@ void replicationCacheMaster(client *c) { ...@@ -2798,13 +2797,8 @@ void replicationCacheMaster(client *c) {
* offset because of the PINGs and will not be able to incrementally * offset because of the PINGs and will not be able to incrementally
* PSYNC with the new master. * PSYNC with the new master.
* This function trims the replication backlog when needed, and returns * This function trims the replication backlog when needed, and returns
* the offset to be used for future partial sync. * the offset to be used for future partial sync. */
* long long adjustMeaningfulReplOffset() {
* If the integer 'adjusted' was passed by reference, it is set to 1
* if the function call actually modified the offset and the replication
* backlog, otherwise it is set to 0. It can be NULL if the caller is
* not interested in getting this info. */
long long adjustMeaningfulReplOffset(int *adjusted) {
if (server.master_repl_offset > server.master_repl_meaningful_offset) { if (server.master_repl_offset > server.master_repl_meaningful_offset) {
long long delta = server.master_repl_offset - long long delta = server.master_repl_offset -
server.master_repl_meaningful_offset; server.master_repl_meaningful_offset;
...@@ -2824,9 +2818,6 @@ long long adjustMeaningfulReplOffset(int *adjusted) { ...@@ -2824,9 +2818,6 @@ long long adjustMeaningfulReplOffset(int *adjusted) {
(server.repl_backlog_idx + (server.repl_backlog_size - delta)) % (server.repl_backlog_idx + (server.repl_backlog_size - delta)) %
server.repl_backlog_size; server.repl_backlog_size;
} }
if (adjusted) *adjusted = 1;
} else {
if (adjusted) *adjusted = 0;
} }
return server.master_repl_offset; return server.master_repl_offset;
} }
...@@ -2850,7 +2841,7 @@ void replicationCacheMasterUsingMyself(void) { ...@@ -2850,7 +2841,7 @@ void replicationCacheMasterUsingMyself(void) {
* by replicationCreateMasterClient(). We'll later set the created * by replicationCreateMasterClient(). We'll later set the created
* master as server.cached_master, so the replica will use such * master as server.cached_master, so the replica will use such
* offset for PSYNC. */ * offset for PSYNC. */
server.master_initial_offset = adjustMeaningfulReplOffset(NULL); server.master_initial_offset = adjustMeaningfulReplOffset();
/* The master client we create can be set to any DBID, because /* The master client we create can be set to any DBID, because
* the new master will start its replication stream with SELECT. */ * the new master will start its replication stream with SELECT. */
......
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