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
43222f25
Commit
43222f25
authored
May 29, 2011
by
antirez
Browse files
Merge remote branch 'pietern/unstable-replnonblock' into unstable
parents
18f63d8d
890a2ed9
Changes
3
Show whitespace changes
Inline
Side-by-side
src/redis.c
View file @
43222f25
...
...
@@ -871,6 +871,7 @@ void initServerConfig() {
server
.
masterport
=
6379
;
server
.
master
=
NULL
;
server
.
replstate
=
REDIS_REPL_NONE
;
server
.
repl_syncio_timeout
=
REDIS_REPL_SYNCIO_TIMEOUT
;
server
.
repl_serve_stale_data
=
1
;
/* Double constants initialization */
...
...
src/redis.h
View file @
43222f25
...
...
@@ -155,8 +155,12 @@
/* Slave replication state - slave side */
#define REDIS_REPL_NONE 0
/* No active replication */
#define REDIS_REPL_CONNECT 1
/* Must connect to master */
#define REDIS_REPL_TRANSFER 2
/* Receiving .rdb from master */
#define REDIS_REPL_CONNECTED 3
/* Connected to master */
#define REDIS_REPL_CONNECTING 2
/* Connecting to master */
#define REDIS_REPL_TRANSFER 3
/* Receiving .rdb from master */
#define REDIS_REPL_CONNECTED 4
/* Connected to master */
/* Synchronous read timeout - slave side */
#define REDIS_REPL_SYNCIO_TIMEOUT 5
/* Slave replication state - from the point of view of master
* Note that in SEND_BULK and ONLINE state the slave receives new updates
...
...
@@ -585,6 +589,7 @@ struct redisServer {
char
*
masterhost
;
int
masterport
;
redisClient
*
master
;
/* client that is master for this slave */
int
repl_syncio_timeout
;
/* timeout for synchronous I/O calls */
int
replstate
;
/* replication status if the instance is a slave */
off_t
repl_transfer_left
;
/* bytes left reading .rdb */
int
repl_transfer_s
;
/* slave -> master SYNC socket */
...
...
@@ -888,7 +893,6 @@ int fwriteBulkCount(FILE *fp, char prefix, int count);
/* Replication */
void
replicationFeedSlaves
(
list
*
slaves
,
int
dictid
,
robj
**
argv
,
int
argc
);
void
replicationFeedMonitors
(
list
*
monitors
,
int
dictid
,
robj
**
argv
,
int
argc
);
int
syncWithMaster
(
void
);
void
updateSlavesWaitingBgsave
(
int
bgsaveerr
);
void
replicationCron
(
void
);
...
...
src/replication.c
View file @
43222f25
...
...
@@ -315,19 +315,18 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int 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
)
{
if
(
syncReadLine
(
fd
,
buf
,
1024
,
server
.
repl_syncio_timeout
)
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"I/O error reading bulk count from MASTER: %s"
,
strerror
(
errno
));
replicationAbortSyncTransfer
();
return
;
goto
error
;
}
if
(
buf
[
0
]
==
'-'
)
{
redisLog
(
REDIS_WARNING
,
"MASTER aborted replication with an error: %s"
,
buf
+
1
);
replicationAbortSyncTransfer
();
return
;
goto
error
;
}
else
if
(
buf
[
0
]
==
'\0'
)
{
/* At this stage just a newline works as a PING in order to take
* the connection live. So we refresh our last interaction
...
...
@@ -336,8 +335,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
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
;
goto
error
;
}
server
.
repl_transfer_left
=
strtol
(
buf
+
1
,
NULL
,
10
);
redisLog
(
REDIS_NOTICE
,
...
...
@@ -359,8 +357,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server
.
repl_transfer_lastio
=
time
(
NULL
);
if
(
write
(
server
.
repl_transfer_fd
,
buf
,
nread
)
!=
nread
)
{
redisLog
(
REDIS_WARNING
,
"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s"
,
strerror
(
errno
));
replicationAbortSyncTransfer
();
return
;
goto
error
;
}
server
.
repl_transfer_left
-=
nread
;
/* Check if the transfer is now complete */
...
...
@@ -391,48 +388,54 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server
.
replstate
=
REDIS_REPL_CONNECTED
;
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync: Finished with success"
);
}
return
;
error:
replicationAbortSyncTransfer
();
return
;
}
int
syncWithMaster
(
void
)
{
char
buf
[
1024
],
tmpfile
[
256
],
authcmd
[
1024
];
int
fd
=
anetTcpConnect
(
NULL
,
server
.
masterhost
,
server
.
masterport
);
void
syncWithMaster
(
aeEventLoop
*
el
,
int
fd
,
void
*
privdata
,
int
mask
)
{
char
buf
[
1024
],
tmpfile
[
256
];
int
dfd
,
maxtries
=
5
;
REDIS_NOTUSED
(
el
);
REDIS_NOTUSED
(
privdata
);
REDIS_NOTUSED
(
mask
);
if
(
fd
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"Unable to connect to MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
}
/* This event should only be triggered once since it is used to have a
* non-blocking connect(2) to the master. It has been triggered when this
* function is called, so we can delete it. */
aeDeleteFileEvent
(
server
.
el
,
fd
,
AE_WRITABLE
);
/* AUTH with the master if required. */
if
(
server
.
masterauth
)
{
snprintf
(
authcmd
,
1024
,
"AUTH %s
\r\n
"
,
server
.
masterauth
);
if
(
syncWrite
(
fd
,
authcmd
,
strlen
(
server
.
masterauth
)
+
7
,
5
)
==
-
1
)
{
close
(
fd
);
char
authcmd
[
1024
];
size_t
authlen
;
authlen
=
snprintf
(
authcmd
,
sizeof
(
authcmd
),
"AUTH %s
\r\n
"
,
server
.
masterauth
);
if
(
syncWrite
(
fd
,
authcmd
,
authlen
,
server
.
repl_syncio_timeout
)
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"Unable to AUTH to MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
goto
error
;
}
/* Read the AUTH result. */
if
(
syncReadLine
(
fd
,
buf
,
1024
,
3600
)
==
-
1
)
{
close
(
fd
);
if
(
syncReadLine
(
fd
,
buf
,
1024
,
server
.
repl_syncio_timeout
)
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"I/O error reading auth result from MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
goto
error
;
}
if
(
buf
[
0
]
!=
'+'
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"Cannot AUTH to MASTER, is the masterauth password correct?"
);
return
REDIS_ERR
;
goto
error
;
}
}
/* Issue the SYNC command */
if
(
syncWrite
(
fd
,
"SYNC
\r\n
"
,
7
,
5
)
==
-
1
)
{
close
(
fd
);
if
(
syncWrite
(
fd
,
"SYNC
\r\n
"
,
7
,
server
.
repl_syncio_timeout
)
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"I/O error writing to MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
goto
error
;
}
/* Prepare a suitable temp file for bulk transfer */
...
...
@@ -444,25 +447,51 @@ int syncWithMaster(void) {
sleep
(
1
);
}
if
(
dfd
==
-
1
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
goto
error
;
}
/* Setup the non blocking download of the bulk file. */
if
(
aeCreateFileEvent
(
server
.
el
,
fd
,
AE_READABLE
,
readSyncBulkPayload
,
NULL
)
if
(
aeCreateFileEvent
(
server
.
el
,
fd
,
AE_READABLE
,
readSyncBulkPayload
,
NULL
)
==
AE_ERR
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"Can't create readable event for SYNC"
);
return
REDIS_ERR
;
goto
error
;
}
server
.
replstate
=
REDIS_REPL_TRANSFER
;
server
.
repl_transfer_left
=
-
1
;
server
.
repl_transfer_s
=
fd
;
server
.
repl_transfer_fd
=
dfd
;
server
.
repl_transfer_lastio
=
time
(
NULL
);
server
.
repl_transfer_tmpfile
=
zstrdup
(
tmpfile
);
return
;
error:
server
.
replstate
=
REDIS_REPL_CONNECT
;
close
(
fd
);
return
;
}
int
connectWithMaster
(
void
)
{
int
fd
;
fd
=
anetTcpNonBlockConnect
(
NULL
,
server
.
masterhost
,
server
.
masterport
);
if
(
fd
==
-
1
)
{
redisLog
(
REDIS_WARNING
,
"Unable to connect to MASTER: %s"
,
strerror
(
errno
));
return
REDIS_ERR
;
}
if
(
aeCreateFileEvent
(
server
.
el
,
fd
,
AE_WRITABLE
,
syncWithMaster
,
NULL
)
==
AE_ERR
)
{
close
(
fd
);
redisLog
(
REDIS_WARNING
,
"Can't create readable event for SYNC"
);
return
REDIS_ERR
;
}
server
.
repl_transfer_s
=
fd
;
server
.
replstate
=
REDIS_REPL_CONNECTING
;
return
REDIS_OK
;
}
...
...
@@ -517,8 +546,8 @@ void replicationCron(void) {
/* Check if we should connect to a MASTER */
if
(
server
.
replstate
==
REDIS_REPL_CONNECT
)
{
redisLog
(
REDIS_NOTICE
,
"Connecting to MASTER..."
);
if
(
sync
WithMaster
()
==
REDIS_OK
)
{
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync started
: SYNC sent
"
);
if
(
connect
WithMaster
()
==
REDIS_OK
)
{
redisLog
(
REDIS_NOTICE
,
"MASTER <-> SLAVE sync started"
);
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