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
968ce964
Commit
968ce964
authored
Feb 26, 2015
by
Salvatore Sanfilippo
Browse files
Merge pull request #2425 from twang817/migrateCacheDbid
Add last_dbid to migrateCachedSocket to avoid redundant SELECT
parents
2c27b887
97c4167a
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/cluster.c
View file @
968ce964
...
@@ -4365,11 +4365,12 @@ void restoreCommand(redisClient *c) {
...
@@ -4365,11 +4365,12 @@ void restoreCommand(redisClient *c) {
typedef
struct
migrateCachedSocket
{
typedef
struct
migrateCachedSocket
{
int
fd
;
int
fd
;
long
last_dbid
;
time_t
last_use_time
;
time_t
last_use_time
;
}
migrateCachedSocket
;
}
migrateCachedSocket
;
/* Return a
TCP socket connected with the target instance, possibly returning
/* Return a
migrateCachedSocket containing a TCP socket connected with the
* a cached one.
*
target instance, possibly returning
a cached one.
*
*
* This function is responsible of sending errors to the client if a
* This function is responsible of sending errors to the client if a
* connection can't be established. In this case -1 is returned.
* connection can't be established. In this case -1 is returned.
...
@@ -4379,7 +4380,7 @@ typedef struct migrateCachedSocket {
...
@@ -4379,7 +4380,7 @@ typedef struct migrateCachedSocket {
* If the caller detects an error while using the socket, migrateCloseSocket()
* If the caller detects an error while using the socket, migrateCloseSocket()
* should be called so that the connection will be created from scratch
* should be called so that the connection will be created from scratch
* the next time. */
* the next time. */
int
migrateGetSocket
(
redisClient
*
c
,
robj
*
host
,
robj
*
port
,
long
timeout
)
{
migrateCachedSocket
*
migrateGetSocket
(
redisClient
*
c
,
robj
*
host
,
robj
*
port
,
long
timeout
)
{
int
fd
;
int
fd
;
sds
name
=
sdsempty
();
sds
name
=
sdsempty
();
migrateCachedSocket
*
cs
;
migrateCachedSocket
*
cs
;
...
@@ -4392,7 +4393,7 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
...
@@ -4392,7 +4393,7 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
if
(
cs
)
{
if
(
cs
)
{
sdsfree
(
name
);
sdsfree
(
name
);
cs
->
last_use_time
=
server
.
unixtime
;
cs
->
last_use_time
=
server
.
unixtime
;
return
cs
->
fd
;
return
cs
;
}
}
/* No cached socket, create one. */
/* No cached socket, create one. */
...
@@ -4412,7 +4413,7 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
...
@@ -4412,7 +4413,7 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
sdsfree
(
name
);
sdsfree
(
name
);
addReplyErrorFormat
(
c
,
"Can't connect to target node: %s"
,
addReplyErrorFormat
(
c
,
"Can't connect to target node: %s"
,
server
.
neterr
);
server
.
neterr
);
return
-
1
;
return
NULL
;
}
}
anetEnableTcpNoDelay
(
server
.
neterr
,
fd
);
anetEnableTcpNoDelay
(
server
.
neterr
,
fd
);
...
@@ -4422,15 +4423,16 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
...
@@ -4422,15 +4423,16 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
addReplySds
(
c
,
addReplySds
(
c
,
sdsnew
(
"-IOERR error or timeout connecting to the client
\r\n
"
));
sdsnew
(
"-IOERR error or timeout connecting to the client
\r\n
"
));
close
(
fd
);
close
(
fd
);
return
-
1
;
return
NULL
;
}
}
/* Add to the cache and return it to the caller. */
/* Add to the cache and return it to the caller. */
cs
=
zmalloc
(
sizeof
(
*
cs
));
cs
=
zmalloc
(
sizeof
(
*
cs
));
cs
->
fd
=
fd
;
cs
->
fd
=
fd
;
cs
->
last_dbid
=
-
1
;
cs
->
last_use_time
=
server
.
unixtime
;
cs
->
last_use_time
=
server
.
unixtime
;
dictAdd
(
server
.
migrate_cached_sockets
,
name
,
cs
);
dictAdd
(
server
.
migrate_cached_sockets
,
name
,
cs
);
return
fd
;
return
cs
;
}
}
/* Free a migrate cached connection. */
/* Free a migrate cached connection. */
...
@@ -4471,7 +4473,8 @@ void migrateCloseTimedoutSockets(void) {
...
@@ -4471,7 +4473,8 @@ void migrateCloseTimedoutSockets(void) {
/* MIGRATE host port key dbid timeout [COPY | REPLACE] */
/* MIGRATE host port key dbid timeout [COPY | REPLACE] */
void
migrateCommand
(
redisClient
*
c
)
{
void
migrateCommand
(
redisClient
*
c
)
{
int
fd
,
copy
,
replace
,
j
;
migrateCachedSocket
*
cs
;
int
copy
,
replace
,
j
;
long
timeout
;
long
timeout
;
long
dbid
;
long
dbid
;
long
long
ttl
,
expireat
;
long
long
ttl
,
expireat
;
...
@@ -4481,6 +4484,7 @@ void migrateCommand(redisClient *c) {
...
@@ -4481,6 +4484,7 @@ void migrateCommand(redisClient *c) {
try_again:
try_again:
/* Initialization */
/* Initialization */
cs
=
NULL
;
copy
=
0
;
copy
=
0
;
replace
=
0
;
replace
=
0
;
ttl
=
0
;
ttl
=
0
;
...
@@ -4513,14 +4517,17 @@ try_again:
...
@@ -4513,14 +4517,17 @@ try_again:
}
}
/* Connect */
/* Connect */
fd
=
migrateGetSocket
(
c
,
c
->
argv
[
1
],
c
->
argv
[
2
],
timeout
);
cs
=
migrateGetSocket
(
c
,
c
->
argv
[
1
],
c
->
argv
[
2
],
timeout
);
if
(
fd
==
-
1
)
return
;
/* error sent to the client by migrateGetSocket() */
if
(
cs
==
NULL
)
return
;
/* error sent to the client by migrateGetSocket() */
/* Create RESTORE payload and generate the protocol to call the command. */
rioInitWithBuffer
(
&
cmd
,
sdsempty
());
rioInitWithBuffer
(
&
cmd
,
sdsempty
());
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkCount
(
&
cmd
,
'*'
,
2
));
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkString
(
&
cmd
,
"SELECT"
,
6
));
/* Create RESTORE payload and generate the protocol to call the command. */
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkLongLong
(
&
cmd
,
dbid
));
if
(
cs
->
last_dbid
!=
dbid
)
{
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkCount
(
&
cmd
,
'*'
,
2
));
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkString
(
&
cmd
,
"SELECT"
,
6
));
redisAssertWithInfo
(
c
,
NULL
,
rioWriteBulkLongLong
(
&
cmd
,
dbid
));
}
expireat
=
getExpire
(
c
->
db
,
c
->
argv
[
3
]);
expireat
=
getExpire
(
c
->
db
,
c
->
argv
[
3
]);
if
(
expireat
!=
-
1
)
{
if
(
expireat
!=
-
1
)
{
...
@@ -4559,7 +4566,7 @@ try_again:
...
@@ -4559,7 +4566,7 @@ try_again:
while
((
towrite
=
sdslen
(
buf
)
-
pos
)
>
0
)
{
while
((
towrite
=
sdslen
(
buf
)
-
pos
)
>
0
)
{
towrite
=
(
towrite
>
(
64
*
1024
)
?
(
64
*
1024
)
:
towrite
);
towrite
=
(
towrite
>
(
64
*
1024
)
?
(
64
*
1024
)
:
towrite
);
nwritten
=
syncWrite
(
fd
,
buf
+
pos
,
towrite
,
timeout
);
nwritten
=
syncWrite
(
cs
->
fd
,
buf
+
pos
,
towrite
,
timeout
);
if
(
nwritten
!=
(
signed
)
towrite
)
goto
socket_wr_err
;
if
(
nwritten
!=
(
signed
)
towrite
)
goto
socket_wr_err
;
pos
+=
nwritten
;
pos
+=
nwritten
;
}
}
...
@@ -4571,14 +4578,18 @@ try_again:
...
@@ -4571,14 +4578,18 @@ try_again:
char
buf2
[
1024
];
char
buf2
[
1024
];
/* Read the two replies */
/* Read the two replies */
if
(
syncReadLine
(
fd
,
buf1
,
sizeof
(
buf1
),
timeout
)
<=
0
)
if
(
cs
->
last_dbid
!=
dbid
&&
syncReadLine
(
cs
->
fd
,
buf1
,
sizeof
(
buf1
),
timeout
)
<=
0
)
goto
socket_rd_err
;
goto
socket_rd_err
;
if
(
syncReadLine
(
fd
,
buf2
,
sizeof
(
buf2
),
timeout
)
<=
0
)
if
(
syncReadLine
(
cs
->
fd
,
buf2
,
sizeof
(
buf2
),
timeout
)
<=
0
)
goto
socket_rd_err
;
goto
socket_rd_err
;
if
(
buf1
[
0
]
==
'-'
||
buf2
[
0
]
==
'-'
)
{
if
((
cs
->
last_dbid
!=
dbid
&&
buf1
[
0
]
==
'-'
)
||
buf2
[
0
]
==
'-'
)
{
/* If we got an error at all, assume that the last_dbid is no longer valid */
cs
->
last_dbid
=
-
1
;
addReplyErrorFormat
(
c
,
"Target instance replied with error: %s"
,
addReplyErrorFormat
(
c
,
"Target instance replied with error: %s"
,
(
buf1
[
0
]
==
'-'
)
?
buf1
+
1
:
buf2
+
1
);
(
cs
->
last_dbid
!=
dbid
&&
buf1
[
0
]
==
'-'
)
?
buf1
+
1
:
buf2
+
1
);
}
else
{
}
else
{
/* Update the last_dbid in migrateCachedSocket */
cs
->
last_dbid
=
dbid
;
robj
*
aux
;
robj
*
aux
;
if
(
!
copy
)
{
if
(
!
copy
)
{
...
@@ -4589,10 +4600,12 @@ try_again:
...
@@ -4589,10 +4600,12 @@ try_again:
addReply
(
c
,
shared
.
ok
);
addReply
(
c
,
shared
.
ok
);
server
.
dirty
++
;
server
.
dirty
++
;
/* Translate MIGRATE as DEL for replication/AOF. */
if
(
!
copy
)
{
aux
=
createStringObject
(
"DEL"
,
3
);
/* Translate MIGRATE as DEL for replication/AOF. */
rewriteClientCommandVector
(
c
,
2
,
aux
,
c
->
argv
[
3
]);
aux
=
createStringObject
(
"DEL"
,
3
);
decrRefCount
(
aux
);
rewriteClientCommandVector
(
c
,
2
,
aux
,
c
->
argv
[
3
]);
decrRefCount
(
aux
);
}
}
}
}
}
...
...
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