Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
26b33669
Commit
26b33669
authored
Nov 04, 2010
by
antirez
Browse files
non blocking slave replication is now more non blocking than the first implementation...
parent
62ec599c
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/replication.c
View file @
26b33669
...
@@ -306,12 +306,41 @@ void replicationAbortSyncTransfer(void) {
...
@@ -306,12 +306,41 @@ void replicationAbortSyncTransfer(void) {
/* Asynchronously read the SYNC payload we receive from a master */
/* Asynchronously read the SYNC payload we receive from a master */
void
readSyncBulkPayload
(
aeEventLoop
*
el
,
int
fd
,
void
*
privdata
,
int
mask
)
{
void
readSyncBulkPayload
(
aeEventLoop
*
el
,
int
fd
,
void
*
privdata
,
int
mask
)
{
unsigned
char
buf
[
4096
];
char
buf
[
4096
];
ssize_t
nread
,
readlen
;
ssize_t
nread
,
readlen
;
REDIS_NOTUSED
(
el
);
REDIS_NOTUSED
(
el
);
REDIS_NOTUSED
(
privdata
);
REDIS_NOTUSED
(
privdata
);
REDIS_NOTUSED
(
mask
);
REDIS_NOTUSED
(
mask
);
/* If repl_transfer_left == -1 we still have to read the bulk length
* from the master reply. */
if
(
server
.
repl_transfer_left
==
-
1
)
{
if
(
syncReadLine
(
fd
,
buf
,
1024
,
3600
)
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"I/O error reading bulk count from MASTER: %s"
,
strerror
(
errno
));
replicationAbortSyncTransfer
();
return
;
}
if
(
buf
[
0
]
==
'-'
)
{
redisLog
(
REDIS_WARNING
,
"MASTER aborted replication with an error: %s"
,
buf
+
1
);
replicationAbortSyncTransfer
();
return
;
}
else
if
(
buf
[
0
]
!=
'$'
)
{
redisLog
(
REDIS_WARNING
,
"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?"
);
replicationAbortSyncTransfer
();
return
;
}
server
.
repl_transfer_left
=
strtol
(
buf
+
1
,
NULL
,
10
);
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync: receiving %ld bytes from master"
,
server
.
repl_transfer_left
);
return
;
}
/* Read bulk data */
readlen
=
(
server
.
repl_transfer_left
<
(
signed
)
sizeof
(
buf
))
?
readlen
=
(
server
.
repl_transfer_left
<
(
signed
)
sizeof
(
buf
))
?
server
.
repl_transfer_left
:
(
signed
)
sizeof
(
buf
);
server
.
repl_transfer_left
:
(
signed
)
sizeof
(
buf
);
nread
=
read
(
fd
,
buf
,
readlen
);
nread
=
read
(
fd
,
buf
,
readlen
);
...
@@ -349,12 +378,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
...
@@ -349,12 +378,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server
.
master
->
flags
|=
REDIS_MASTER
;
server
.
master
->
flags
|=
REDIS_MASTER
;
server
.
master
->
authenticated
=
1
;
server
.
master
->
authenticated
=
1
;
server
.
replstate
=
REDIS_REPL_CONNECTED
;
server
.
replstate
=
REDIS_REPL_CONNECTED
;
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync: Finished with success"
);
}
}
}
}
int
syncWithMaster
(
void
)
{
int
syncWithMaster
(
void
)
{
char
buf
[
1024
],
tmpfile
[
256
],
authcmd
[
1024
];
char
buf
[
1024
],
tmpfile
[
256
],
authcmd
[
1024
];
off_t
dumpsize
;
int
fd
=
anetTcpConnect
(
NULL
,
server
.
masterhost
,
server
.
masterport
);
int
fd
=
anetTcpConnect
(
NULL
,
server
.
masterhost
,
server
.
masterport
);
int
dfd
,
maxtries
=
5
;
int
dfd
,
maxtries
=
5
;
...
@@ -394,26 +423,8 @@ int syncWithMaster(void) {
...
@@ -394,26 +423,8 @@ int syncWithMaster(void) {
strerror
(
errno
));
strerror
(
errno
));
return
REDIS_ERR
;
return
REDIS_ERR
;
}
}
/* Read the bulk write count */
if
(
syncReadLine
(
fd
,
buf
,
1024
,
3600
)
==
-
1
)
{
/* Prepare a suitable temp file for bulk transfer */
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"I/O error reading bulk count from MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
}
if
(
buf
[
0
]
==
'-'
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"MASTER aborted replication with an error: %s"
,
buf
+
1
);
return
REDIS_ERR
;
}
else
if
(
buf
[
0
]
!=
'$'
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?"
);
return
REDIS_ERR
;
}
dumpsize
=
strtol
(
buf
+
1
,
NULL
,
10
);
redisLog
(
REDIS_NOTICE
,
"Receiving %ld bytes data dump from MASTER"
,
dumpsize
);
/* Read the bulk write data on a temp file */
while
(
maxtries
--
)
{
while
(
maxtries
--
)
{
snprintf
(
tmpfile
,
256
,
snprintf
(
tmpfile
,
256
,
"temp-%d.%ld.rdb"
,(
int
)
time
(
NULL
),(
long
int
)
getpid
());
"temp-%d.%ld.rdb"
,(
int
)
time
(
NULL
),(
long
int
)
getpid
());
...
@@ -436,7 +447,7 @@ int syncWithMaster(void) {
...
@@ -436,7 +447,7 @@ int syncWithMaster(void) {
return
REDIS_ERR
;
return
REDIS_ERR
;
}
}
server
.
replstate
=
REDIS_REPL_TRANSFER
;
server
.
replstate
=
REDIS_REPL_TRANSFER
;
server
.
repl_transfer_left
=
dumpsize
;
server
.
repl_transfer_left
=
-
1
;
server
.
repl_transfer_s
=
fd
;
server
.
repl_transfer_s
=
fd
;
server
.
repl_transfer_fd
=
dfd
;
server
.
repl_transfer_fd
=
dfd
;
server
.
repl_transfer_lastio
=
time
(
NULL
);
server
.
repl_transfer_lastio
=
time
(
NULL
);
...
@@ -487,7 +498,7 @@ void replicationCron(void) {
...
@@ -487,7 +498,7 @@ void replicationCron(void) {
if
(
server
.
replstate
==
REDIS_REPL_CONNECT
)
{
if
(
server
.
replstate
==
REDIS_REPL_CONNECT
)
{
redisLog
(
REDIS_NOTICE
,
"Connecting to MASTER..."
);
redisLog
(
REDIS_NOTICE
,
"Connecting to MASTER..."
);
if
(
syncWithMaster
()
==
REDIS_OK
)
{
if
(
syncWithMaster
()
==
REDIS_OK
)
{
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync s
ucceeded
"
);
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync s
tarted: SYNC sent
"
);
if
(
server
.
appendonly
)
rewriteAppendOnlyFileBackground
();
if
(
server
.
appendonly
)
rewriteAppendOnlyFileBackground
();
}
}
}
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment