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
f858c11d
Commit
f858c11d
authored
Dec 14, 2010
by
antirez
Browse files
Merge remote branch 'pietern/brpoplpush'
parents
8c304be3
a4ce7581
Changes
5
Hide whitespace changes
Inline
Side-by-side
src/networking.c
View file @
f858c11d
...
@@ -41,8 +41,10 @@ redisClient *createClient(int fd) {
...
@@ -41,8 +41,10 @@ redisClient *createClient(int fd) {
c
->
reply
=
listCreate
();
c
->
reply
=
listCreate
();
listSetFreeMethod
(
c
->
reply
,
decrRefCount
);
listSetFreeMethod
(
c
->
reply
,
decrRefCount
);
listSetDupMethod
(
c
->
reply
,
dupClientReplyValue
);
listSetDupMethod
(
c
->
reply
,
dupClientReplyValue
);
c
->
blocking_keys
=
NULL
;
c
->
bpop
.
keys
=
NULL
;
c
->
blocking_keys_num
=
0
;
c
->
bpop
.
count
=
0
;
c
->
bpop
.
timeout
=
0
;
c
->
bpop
.
target
=
NULL
;
c
->
io_keys
=
listCreate
();
c
->
io_keys
=
listCreate
();
c
->
watched_keys
=
listCreate
();
c
->
watched_keys
=
listCreate
();
listSetFreeMethod
(
c
->
io_keys
,
decrRefCount
);
listSetFreeMethod
(
c
->
io_keys
,
decrRefCount
);
...
@@ -699,7 +701,7 @@ void closeTimedoutClients(void) {
...
@@ -699,7 +701,7 @@ void closeTimedoutClients(void) {
redisLog
(
REDIS_VERBOSE
,
"Closing idle client"
);
redisLog
(
REDIS_VERBOSE
,
"Closing idle client"
);
freeClient
(
c
);
freeClient
(
c
);
}
else
if
(
c
->
flags
&
REDIS_BLOCKED
)
{
}
else
if
(
c
->
flags
&
REDIS_BLOCKED
)
{
if
(
c
->
b
lockingto
!=
0
&&
c
->
b
lockingto
<
now
)
{
if
(
c
->
b
pop
.
timeout
!=
0
&&
c
->
b
pop
.
timeout
<
now
)
{
addReply
(
c
,
shared
.
nullmultibulk
);
addReply
(
c
,
shared
.
nullmultibulk
);
unblockClientWaitingData
(
c
);
unblockClientWaitingData
(
c
);
}
}
...
...
src/redis.c
View file @
f858c11d
...
@@ -89,6 +89,7 @@ struct redisCommand readonlyCommandTable[] = {
...
@@ -89,6 +89,7 @@ struct redisCommand readonlyCommandTable[] = {
{
"rpop"
,
rpopCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"rpop"
,
rpopCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"lpop"
,
lpopCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"lpop"
,
lpopCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"brpop"
,
brpopCommand
,
-
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"brpop"
,
brpopCommand
,
-
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"brpoplpush"
,
brpoplpushCommand
,
4
,
REDIS_CMD_DENYOOM
,
NULL
,
1
,
2
,
1
},
{
"blpop"
,
blpopCommand
,
-
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"blpop"
,
blpopCommand
,
-
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"llen"
,
llenCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"llen"
,
llenCommand
,
2
,
0
,
NULL
,
1
,
1
,
1
},
{
"lindex"
,
lindexCommand
,
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"lindex"
,
lindexCommand
,
3
,
0
,
NULL
,
1
,
1
,
1
},
...
@@ -96,7 +97,7 @@ struct redisCommand readonlyCommandTable[] = {
...
@@ -96,7 +97,7 @@ struct redisCommand readonlyCommandTable[] = {
{
"lrange"
,
lrangeCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"lrange"
,
lrangeCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"ltrim"
,
ltrimCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"ltrim"
,
ltrimCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"lrem"
,
lremCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"lrem"
,
lremCommand
,
4
,
0
,
NULL
,
1
,
1
,
1
},
{
"rpoplpush"
,
rpoplpush
c
ommand
,
3
,
REDIS_CMD_DENYOOM
,
NULL
,
1
,
2
,
1
},
{
"rpoplpush"
,
rpoplpush
C
ommand
,
3
,
REDIS_CMD_DENYOOM
,
NULL
,
1
,
2
,
1
},
{
"sadd"
,
saddCommand
,
3
,
REDIS_CMD_DENYOOM
,
NULL
,
1
,
1
,
1
},
{
"sadd"
,
saddCommand
,
3
,
REDIS_CMD_DENYOOM
,
NULL
,
1
,
1
,
1
},
{
"srem"
,
sremCommand
,
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"srem"
,
sremCommand
,
3
,
0
,
NULL
,
1
,
1
,
1
},
{
"smove"
,
smoveCommand
,
4
,
0
,
NULL
,
1
,
2
,
1
},
{
"smove"
,
smoveCommand
,
4
,
0
,
NULL
,
1
,
2
,
1
},
...
@@ -572,7 +573,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
...
@@ -572,7 +573,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
}
}
/* Close connections of timedout clients */
/* Close connections of timedout clients */
if
((
server
.
maxidletime
&&
!
(
loops
%
100
))
||
server
.
b
l
pop_blocked_clients
)
if
((
server
.
maxidletime
&&
!
(
loops
%
100
))
||
server
.
bpop_blocked_clients
)
closeTimedoutClients
();
closeTimedoutClients
();
/* Check if a background saving or AOF rewrite in progress terminated */
/* Check if a background saving or AOF rewrite in progress terminated */
...
@@ -645,15 +646,16 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
...
@@ -645,15 +646,16 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
* for ready file descriptors. */
* for ready file descriptors. */
void
beforeSleep
(
struct
aeEventLoop
*
eventLoop
)
{
void
beforeSleep
(
struct
aeEventLoop
*
eventLoop
)
{
REDIS_NOTUSED
(
eventLoop
);
REDIS_NOTUSED
(
eventLoop
);
listNode
*
ln
;
redisClient
*
c
;
/* Awake clients that got all the swapped keys they requested */
/* Awake clients that got all the swapped keys they requested */
if
(
server
.
vm_enabled
&&
listLength
(
server
.
io_ready_clients
))
{
if
(
server
.
vm_enabled
&&
listLength
(
server
.
io_ready_clients
))
{
listIter
li
;
listIter
li
;
listNode
*
ln
;
listRewind
(
server
.
io_ready_clients
,
&
li
);
listRewind
(
server
.
io_ready_clients
,
&
li
);
while
((
ln
=
listNext
(
&
li
)))
{
while
((
ln
=
listNext
(
&
li
)))
{
redisClient
*
c
=
ln
->
value
;
c
=
ln
->
value
;
struct
redisCommand
*
cmd
;
struct
redisCommand
*
cmd
;
/* Resume the client. */
/* Resume the client. */
...
@@ -671,6 +673,19 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
...
@@ -671,6 +673,19 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
processInputBuffer
(
c
);
processInputBuffer
(
c
);
}
}
}
}
/* Try to process pending commands for clients that were just unblocked. */
while
(
listLength
(
server
.
unblocked_clients
))
{
ln
=
listFirst
(
server
.
unblocked_clients
);
redisAssert
(
ln
!=
NULL
);
c
=
ln
->
value
;
listDelNode
(
server
.
unblocked_clients
,
ln
);
/* Process remaining data in the input buffer. */
if
(
c
->
querybuf
&&
sdslen
(
c
->
querybuf
)
>
0
)
processInputBuffer
(
c
);
}
/* Write the AOF buffer on disk */
/* Write the AOF buffer on disk */
flushAppendOnlyFile
();
flushAppendOnlyFile
();
}
}
...
@@ -758,7 +773,7 @@ void initServerConfig() {
...
@@ -758,7 +773,7 @@ void initServerConfig() {
server
.
rdbcompression
=
1
;
server
.
rdbcompression
=
1
;
server
.
activerehashing
=
1
;
server
.
activerehashing
=
1
;
server
.
maxclients
=
0
;
server
.
maxclients
=
0
;
server
.
b
l
pop_blocked_clients
=
0
;
server
.
bpop_blocked_clients
=
0
;
server
.
maxmemory
=
0
;
server
.
maxmemory
=
0
;
server
.
maxmemory_policy
=
REDIS_MAXMEMORY_VOLATILE_LRU
;
server
.
maxmemory_policy
=
REDIS_MAXMEMORY_VOLATILE_LRU
;
server
.
maxmemory_samples
=
3
;
server
.
maxmemory_samples
=
3
;
...
@@ -817,6 +832,7 @@ void initServer() {
...
@@ -817,6 +832,7 @@ void initServer() {
server
.
clients
=
listCreate
();
server
.
clients
=
listCreate
();
server
.
slaves
=
listCreate
();
server
.
slaves
=
listCreate
();
server
.
monitors
=
listCreate
();
server
.
monitors
=
listCreate
();
server
.
unblocked_clients
=
listCreate
();
createSharedObjects
();
createSharedObjects
();
server
.
el
=
aeCreateEventLoop
();
server
.
el
=
aeCreateEventLoop
();
server
.
db
=
zmalloc
(
sizeof
(
redisDb
)
*
server
.
dbnum
);
server
.
db
=
zmalloc
(
sizeof
(
redisDb
)
*
server
.
dbnum
);
...
@@ -1169,7 +1185,7 @@ sds genRedisInfoString(void) {
...
@@ -1169,7 +1185,7 @@ sds genRedisInfoString(void) {
(
float
)
c_ru
.
ru_stime
.
tv_sec
+
(
float
)
c_ru
.
ru_stime
.
tv_usec
/
1000000
,
(
float
)
c_ru
.
ru_stime
.
tv_sec
+
(
float
)
c_ru
.
ru_stime
.
tv_usec
/
1000000
,
listLength
(
server
.
clients
)
-
listLength
(
server
.
slaves
),
listLength
(
server
.
clients
)
-
listLength
(
server
.
slaves
),
listLength
(
server
.
slaves
),
listLength
(
server
.
slaves
),
server
.
b
l
pop_blocked_clients
,
server
.
bpop_blocked_clients
,
zmalloc_used_memory
(),
zmalloc_used_memory
(),
hmem
,
hmem
,
zmalloc_get_rss
(),
zmalloc_get_rss
(),
...
...
src/redis.h
View file @
f858c11d
...
@@ -293,6 +293,16 @@ typedef struct multiState {
...
@@ -293,6 +293,16 @@ typedef struct multiState {
int
count
;
/* Total number of MULTI commands */
int
count
;
/* Total number of MULTI commands */
}
multiState
;
}
multiState
;
typedef
struct
blockingState
{
robj
**
keys
;
/* The key we are waiting to terminate a blocking
* operation such as BLPOP. Otherwise NULL. */
int
count
;
/* Number of blocking keys */
time_t
timeout
;
/* Blocking operation timeout. If UNIX current time
* is >= timeout then the operation timed out. */
robj
*
target
;
/* The key that should receive the element,
* for BRPOPLPUSH. */
}
blockingState
;
/* With multiplexing we need to take per-clinet state.
/* With multiplexing we need to take per-clinet state.
* Clients are taken in a liked list. */
* Clients are taken in a liked list. */
typedef
struct
redisClient
{
typedef
struct
redisClient
{
...
@@ -316,11 +326,7 @@ typedef struct redisClient {
...
@@ -316,11 +326,7 @@ typedef struct redisClient {
long
repldboff
;
/* replication DB file offset */
long
repldboff
;
/* replication DB file offset */
off_t
repldbsize
;
/* replication DB file size */
off_t
repldbsize
;
/* replication DB file size */
multiState
mstate
;
/* MULTI/EXEC state */
multiState
mstate
;
/* MULTI/EXEC state */
robj
**
blocking_keys
;
/* The key we are waiting to terminate a blocking
blockingState
bpop
;
/* blocking state */
* operation such as BLPOP. Otherwise NULL. */
int
blocking_keys_num
;
/* Number of blocking keys */
time_t
blockingto
;
/* Blocking operation timeout. If UNIX current time
* is >= blockingto then the operation timed out. */
list
*
io_keys
;
/* Keys this client is waiting to be loaded from the
list
*
io_keys
;
/* Keys this client is waiting to be loaded from the
* swap file in order to continue. */
* swap file in order to continue. */
list
*
watched_keys
;
/* Keys WATCHED for MULTI/EXEC CAS */
list
*
watched_keys
;
/* Keys WATCHED for MULTI/EXEC CAS */
...
@@ -427,8 +433,9 @@ struct redisServer {
...
@@ -427,8 +433,9 @@ struct redisServer {
int
maxmemory_policy
;
int
maxmemory_policy
;
int
maxmemory_samples
;
int
maxmemory_samples
;
/* Blocked clients */
/* Blocked clients */
unsigned
int
b
l
pop_blocked_clients
;
unsigned
int
bpop_blocked_clients
;
unsigned
int
vm_blocked_clients
;
unsigned
int
vm_blocked_clients
;
list
*
unblocked_clients
;
/* Sort parameters - qsort_r() is only available under BSD so we
/* Sort parameters - qsort_r() is only available under BSD so we
* have to take this state global, in order to pass it to sortCompare() */
* have to take this state global, in order to pass it to sortCompare() */
int
sort_desc
;
int
sort_desc
;
...
@@ -937,7 +944,7 @@ void flushdbCommand(redisClient *c);
...
@@ -937,7 +944,7 @@ void flushdbCommand(redisClient *c);
void
flushallCommand
(
redisClient
*
c
);
void
flushallCommand
(
redisClient
*
c
);
void
sortCommand
(
redisClient
*
c
);
void
sortCommand
(
redisClient
*
c
);
void
lremCommand
(
redisClient
*
c
);
void
lremCommand
(
redisClient
*
c
);
void
rpoplpush
c
ommand
(
redisClient
*
c
);
void
rpoplpush
C
ommand
(
redisClient
*
c
);
void
infoCommand
(
redisClient
*
c
);
void
infoCommand
(
redisClient
*
c
);
void
mgetCommand
(
redisClient
*
c
);
void
mgetCommand
(
redisClient
*
c
);
void
monitorCommand
(
redisClient
*
c
);
void
monitorCommand
(
redisClient
*
c
);
...
@@ -966,6 +973,7 @@ void execCommand(redisClient *c);
...
@@ -966,6 +973,7 @@ void execCommand(redisClient *c);
void
discardCommand
(
redisClient
*
c
);
void
discardCommand
(
redisClient
*
c
);
void
blpopCommand
(
redisClient
*
c
);
void
blpopCommand
(
redisClient
*
c
);
void
brpopCommand
(
redisClient
*
c
);
void
brpopCommand
(
redisClient
*
c
);
void
brpoplpushCommand
(
redisClient
*
c
);
void
appendCommand
(
redisClient
*
c
);
void
appendCommand
(
redisClient
*
c
);
void
substrCommand
(
redisClient
*
c
);
void
substrCommand
(
redisClient
*
c
);
void
strlenCommand
(
redisClient
*
c
);
void
strlenCommand
(
redisClient
*
c
);
...
...
src/t_list.c
View file @
f858c11d
...
@@ -621,20 +621,38 @@ void lremCommand(redisClient *c) {
...
@@ -621,20 +621,38 @@ void lremCommand(redisClient *c) {
/* This is the semantic of this command:
/* This is the semantic of this command:
* RPOPLPUSH srclist dstlist:
* RPOPLPUSH srclist dstlist:
* IF LLEN(srclist) > 0
*
IF LLEN(srclist) > 0
* element = RPOP srclist
*
element = RPOP srclist
* LPUSH dstlist element
*
LPUSH dstlist element
* RETURN element
*
RETURN element
* ELSE
*
ELSE
* RETURN nil
*
RETURN nil
* END
*
END
* END
* END
*
*
* The idea is to be able to get an element from a list in a reliable way
* The idea is to be able to get an element from a list in a reliable way
* since the element is not just returned but pushed against another list
* since the element is not just returned but pushed against another list
* as well. This command was originally proposed by Ezra Zygmuntowicz.
* as well. This command was originally proposed by Ezra Zygmuntowicz.
*/
*/
void
rpoplpushcommand
(
redisClient
*
c
)
{
void
rpoplpushHandlePush
(
redisClient
*
c
,
robj
*
dstkey
,
robj
*
dstobj
,
robj
*
value
)
{
if
(
!
handleClientsWaitingListPush
(
c
,
dstkey
,
value
))
{
/* Create the list if the key does not exist */
if
(
!
dstobj
)
{
dstobj
=
createZiplistObject
();
dbAdd
(
c
->
db
,
dstkey
,
dstobj
);
}
else
{
touchWatchedKey
(
c
->
db
,
dstkey
);
server
.
dirty
++
;
}
listTypePush
(
dstobj
,
value
,
REDIS_HEAD
);
}
/* Always send the pushed value to the client. */
addReplyBulk
(
c
,
value
);
}
void
rpoplpushCommand
(
redisClient
*
c
)
{
robj
*
sobj
,
*
value
;
robj
*
sobj
,
*
value
;
if
((
sobj
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
1
],
shared
.
nullbulk
))
==
NULL
||
if
((
sobj
=
lookupKeyWriteOrReply
(
c
,
c
->
argv
[
1
],
shared
.
nullbulk
))
==
NULL
||
checkType
(
c
,
sobj
,
REDIS_LIST
))
return
;
checkType
(
c
,
sobj
,
REDIS_LIST
))
return
;
...
@@ -645,20 +663,7 @@ void rpoplpushcommand(redisClient *c) {
...
@@ -645,20 +663,7 @@ void rpoplpushcommand(redisClient *c) {
robj
*
dobj
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
2
]);
robj
*
dobj
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
2
]);
if
(
dobj
&&
checkType
(
c
,
dobj
,
REDIS_LIST
))
return
;
if
(
dobj
&&
checkType
(
c
,
dobj
,
REDIS_LIST
))
return
;
value
=
listTypePop
(
sobj
,
REDIS_TAIL
);
value
=
listTypePop
(
sobj
,
REDIS_TAIL
);
rpoplpushHandlePush
(
c
,
c
->
argv
[
2
],
dobj
,
value
);
/* Add the element to the target list (unless it's directly
* passed to some BLPOP-ing client */
if
(
!
handleClientsWaitingListPush
(
c
,
c
->
argv
[
2
],
value
))
{
/* Create the list if the key does not exist */
if
(
!
dobj
)
{
dobj
=
createZiplistObject
();
dbAdd
(
c
->
db
,
c
->
argv
[
2
],
dobj
);
}
listTypePush
(
dobj
,
value
,
REDIS_HEAD
);
}
/* Send the element to the client as reply as well */
addReplyBulk
(
c
,
value
);
/* listTypePop returns an object with its refcount incremented */
/* listTypePop returns an object with its refcount incremented */
decrRefCount
(
value
);
decrRefCount
(
value
);
...
@@ -705,17 +710,23 @@ void rpoplpushcommand(redisClient *c) {
...
@@ -705,17 +710,23 @@ void rpoplpushcommand(redisClient *c) {
/* Set a client in blocking mode for the specified key, with the specified
/* Set a client in blocking mode for the specified key, with the specified
* timeout */
* timeout */
void
blockForKeys
(
redisClient
*
c
,
robj
**
keys
,
int
numkeys
,
time_t
timeout
)
{
void
blockForKeys
(
redisClient
*
c
,
robj
**
keys
,
int
numkeys
,
time_t
timeout
,
robj
*
target
)
{
dictEntry
*
de
;
dictEntry
*
de
;
list
*
l
;
list
*
l
;
int
j
;
int
j
;
c
->
blocking_keys
=
zmalloc
(
sizeof
(
robj
*
)
*
numkeys
);
c
->
bpop
.
keys
=
zmalloc
(
sizeof
(
robj
*
)
*
numkeys
);
c
->
blocking_keys_num
=
numkeys
;
c
->
bpop
.
count
=
numkeys
;
c
->
blockingto
=
timeout
;
c
->
bpop
.
timeout
=
timeout
;
c
->
bpop
.
target
=
target
;
if
(
target
!=
NULL
)
{
incrRefCount
(
target
);
}
for
(
j
=
0
;
j
<
numkeys
;
j
++
)
{
for
(
j
=
0
;
j
<
numkeys
;
j
++
)
{
/* Add the key in the client structure, to map clients -> keys */
/* Add the key in the client structure, to map clients -> keys */
c
->
b
locking_
keys
[
j
]
=
keys
[
j
];
c
->
b
pop
.
keys
[
j
]
=
keys
[
j
];
incrRefCount
(
keys
[
j
]);
incrRefCount
(
keys
[
j
]);
/* And in the other "side", to map keys -> clients */
/* And in the other "side", to map keys -> clients */
...
@@ -735,7 +746,7 @@ void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout) {
...
@@ -735,7 +746,7 @@ void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout) {
}
}
/* Mark the client as a blocked client */
/* Mark the client as a blocked client */
c
->
flags
|=
REDIS_BLOCKED
;
c
->
flags
|=
REDIS_BLOCKED
;
server
.
b
l
pop_blocked_clients
++
;
server
.
bpop_blocked_clients
++
;
}
}
/* Unblock a client that's waiting in a blocking operation such as BLPOP */
/* Unblock a client that's waiting in a blocking operation such as BLPOP */
...
@@ -744,30 +755,27 @@ void unblockClientWaitingData(redisClient *c) {
...
@@ -744,30 +755,27 @@ void unblockClientWaitingData(redisClient *c) {
list
*
l
;
list
*
l
;
int
j
;
int
j
;
redisAssert
(
c
->
b
locking_
keys
!=
NULL
);
redisAssert
(
c
->
b
pop
.
keys
!=
NULL
);
/* The client may wait for multiple keys, so unblock it for every key. */
/* The client may wait for multiple keys, so unblock it for every key. */
for
(
j
=
0
;
j
<
c
->
b
locking_keys_num
;
j
++
)
{
for
(
j
=
0
;
j
<
c
->
b
pop
.
count
;
j
++
)
{
/* Remove this client from the list of clients waiting for this key. */
/* Remove this client from the list of clients waiting for this key. */
de
=
dictFind
(
c
->
db
->
blocking_keys
,
c
->
b
locking_
keys
[
j
]);
de
=
dictFind
(
c
->
db
->
blocking_keys
,
c
->
b
pop
.
keys
[
j
]);
redisAssert
(
de
!=
NULL
);
redisAssert
(
de
!=
NULL
);
l
=
dictGetEntryVal
(
de
);
l
=
dictGetEntryVal
(
de
);
listDelNode
(
l
,
listSearchKey
(
l
,
c
));
listDelNode
(
l
,
listSearchKey
(
l
,
c
));
/* If the list is empty we need to remove it to avoid wasting memory */
/* If the list is empty we need to remove it to avoid wasting memory */
if
(
listLength
(
l
)
==
0
)
if
(
listLength
(
l
)
==
0
)
dictDelete
(
c
->
db
->
blocking_keys
,
c
->
b
locking_
keys
[
j
]);
dictDelete
(
c
->
db
->
blocking_keys
,
c
->
b
pop
.
keys
[
j
]);
decrRefCount
(
c
->
b
locking_
keys
[
j
]);
decrRefCount
(
c
->
b
pop
.
keys
[
j
]);
}
}
/* Cleanup the client structure */
/* Cleanup the client structure */
zfree
(
c
->
blocking_keys
);
zfree
(
c
->
bpop
.
keys
);
c
->
blocking_keys
=
NULL
;
c
->
bpop
.
keys
=
NULL
;
c
->
bpop
.
target
=
NULL
;
c
->
flags
&=
(
~
REDIS_BLOCKED
);
c
->
flags
&=
(
~
REDIS_BLOCKED
);
server
.
blpop_blocked_clients
--
;
server
.
bpop_blocked_clients
--
;
/* We want to process data if there is some command waiting
listAddNodeTail
(
server
.
unblocked_clients
,
c
);
* in the input buffer. Note that this is safe even if
* unblockClientWaitingData() gets called from freeClient() because
* freeClient() will be smart enough to call this function
* *after* c->querybuf was set to NULL. */
if
(
c
->
querybuf
&&
sdslen
(
c
->
querybuf
)
>
0
)
processInputBuffer
(
c
);
}
}
/* This should be called from any function PUSHing into lists.
/* This should be called from any function PUSHing into lists.
...
@@ -783,39 +791,81 @@ void unblockClientWaitingData(redisClient *c) {
...
@@ -783,39 +791,81 @@ void unblockClientWaitingData(redisClient *c) {
int
handleClientsWaitingListPush
(
redisClient
*
c
,
robj
*
key
,
robj
*
ele
)
{
int
handleClientsWaitingListPush
(
redisClient
*
c
,
robj
*
key
,
robj
*
ele
)
{
struct
dictEntry
*
de
;
struct
dictEntry
*
de
;
redisClient
*
receiver
;
redisClient
*
receiver
;
list
*
l
;
int
numclients
;
list
*
clients
;
listNode
*
ln
;
listNode
*
ln
;
robj
*
dstkey
,
*
dstobj
;
de
=
dictFind
(
c
->
db
->
blocking_keys
,
key
);
de
=
dictFind
(
c
->
db
->
blocking_keys
,
key
);
if
(
de
==
NULL
)
return
0
;
if
(
de
==
NULL
)
return
0
;
l
=
dictGetEntryVal
(
de
);
clients
=
dictGetEntryVal
(
de
);
ln
=
listFirst
(
l
);
numclients
=
listLength
(
clients
);
redisAssert
(
ln
!=
NULL
);
receiver
=
ln
->
value
;
/* Try to handle the push as long as there are clients waiting for a push.
* Note that "numclients" is used because the list of clients waiting for a
* push on "key" is deleted by unblockClient() when empty.
*
* This loop will have more than 1 iteration when there is a BRPOPLPUSH
* that cannot push the target list because it does not contain a list. If
* this happens, it simply tries the next client waiting for a push. */
while
(
numclients
--
)
{
ln
=
listFirst
(
clients
);
redisAssert
(
ln
!=
NULL
);
receiver
=
ln
->
value
;
dstkey
=
receiver
->
bpop
.
target
;
/* This should remove the first element of the "clients" list. */
unblockClientWaitingData
(
receiver
);
redisAssert
(
ln
!=
listFirst
(
clients
));
if
(
dstkey
==
NULL
)
{
/* BRPOP/BLPOP */
addReplyMultiBulkLen
(
receiver
,
2
);
addReplyBulk
(
receiver
,
key
);
addReplyBulk
(
receiver
,
ele
);
return
1
;
}
else
{
/* BRPOPLPUSH */
dstobj
=
lookupKeyWrite
(
receiver
->
db
,
dstkey
);
if
(
dstobj
&&
checkType
(
receiver
,
dstobj
,
REDIS_LIST
))
{
decrRefCount
(
dstkey
);
}
else
{
rpoplpushHandlePush
(
receiver
,
dstkey
,
dstobj
,
ele
);
decrRefCount
(
dstkey
);
return
1
;
}
}
}
addReplyMultiBulkLen
(
receiver
,
2
);
return
0
;
addReplyBulk
(
receiver
,
key
);
}
addReplyBulk
(
receiver
,
ele
);
unblockClientWaitingData
(
receiver
);
int
getTimeoutFromObjectOrReply
(
redisClient
*
c
,
robj
*
object
,
time_t
*
timeout
)
{
return
1
;
long
tval
;
if
(
getLongFromObjectOrReply
(
c
,
object
,
&
tval
,
"timeout is not an integer or out of range"
)
!=
REDIS_OK
)
return
REDIS_ERR
;
if
(
tval
<
0
)
{
addReplyError
(
c
,
"timeout is negative"
);
return
REDIS_ERR
;
}
if
(
tval
>
0
)
tval
+=
time
(
NULL
);
*
timeout
=
tval
;
return
REDIS_OK
;
}
}
/* Blocking RPOP/LPOP */
/* Blocking RPOP/LPOP */
void
blockingPopGenericCommand
(
redisClient
*
c
,
int
where
)
{
void
blockingPopGenericCommand
(
redisClient
*
c
,
int
where
)
{
robj
*
o
;
robj
*
o
;
long
long
lltimeout
;
time_t
timeout
;
time_t
timeout
;
int
j
;
int
j
;
/* Make sure timeout is an integer value */
if
(
getTimeoutFromObjectOrReply
(
c
,
c
->
argv
[
c
->
argc
-
1
],
&
timeout
)
!=
REDIS_OK
)
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
c
->
argc
-
1
],
&
lltimeout
,
"timeout is not an integer"
)
!=
REDIS_OK
)
return
;
/* Make sure the timeout is not negative */
if
(
lltimeout
<
0
)
{
addReplyError
(
c
,
"timeout is negative"
);
return
;
return
;
}
for
(
j
=
1
;
j
<
c
->
argc
-
1
;
j
++
)
{
for
(
j
=
1
;
j
<
c
->
argc
-
1
;
j
++
)
{
o
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
j
]);
o
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
j
]);
...
@@ -845,11 +895,13 @@ void blockingPopGenericCommand(redisClient *c, int where) {
...
@@ -845,11 +895,13 @@ void blockingPopGenericCommand(redisClient *c, int where) {
* because it is... */
* because it is... */
addReplyMultiBulkLen
(
c
,
2
);
addReplyMultiBulkLen
(
c
,
2
);
addReplyBulk
(
c
,
argv
[
1
]);
addReplyBulk
(
c
,
argv
[
1
]);
popGenericCommand
(
c
,
where
);
popGenericCommand
(
c
,
where
);
/* Fix the client structure with the original stuff */
/* Fix the client structure with the original stuff */
c
->
argv
=
orig_argv
;
c
->
argv
=
orig_argv
;
c
->
argc
=
orig_argc
;
c
->
argc
=
orig_argc
;
return
;
return
;
}
}
}
}
...
@@ -864,9 +916,7 @@ void blockingPopGenericCommand(redisClient *c, int where) {
...
@@ -864,9 +916,7 @@ void blockingPopGenericCommand(redisClient *c, int where) {
}
}
/* If the list is empty or the key does not exists we must block */
/* If the list is empty or the key does not exists we must block */
timeout
=
lltimeout
;
blockForKeys
(
c
,
c
->
argv
+
1
,
c
->
argc
-
2
,
timeout
,
NULL
);
if
(
timeout
>
0
)
timeout
+=
time
(
NULL
);
blockForKeys
(
c
,
c
->
argv
+
1
,
c
->
argc
-
2
,
timeout
);
}
}
void
blpopCommand
(
redisClient
*
c
)
{
void
blpopCommand
(
redisClient
*
c
)
{
...
@@ -876,3 +926,34 @@ void blpopCommand(redisClient *c) {
...
@@ -876,3 +926,34 @@ void blpopCommand(redisClient *c) {
void
brpopCommand
(
redisClient
*
c
)
{
void
brpopCommand
(
redisClient
*
c
)
{
blockingPopGenericCommand
(
c
,
REDIS_TAIL
);
blockingPopGenericCommand
(
c
,
REDIS_TAIL
);
}
}
void
brpoplpushCommand
(
redisClient
*
c
)
{
time_t
timeout
;
if
(
getTimeoutFromObjectOrReply
(
c
,
c
->
argv
[
3
],
&
timeout
)
!=
REDIS_OK
)
return
;
robj
*
key
=
lookupKeyWrite
(
c
->
db
,
c
->
argv
[
1
]);
if
(
key
==
NULL
)
{
if
(
c
->
flags
&
REDIS_MULTI
)
{
/* Blocking against an empty list in a multi state
* returns immediately. */
addReply
(
c
,
shared
.
nullmultibulk
);
}
else
{
/* The list is empty and the client blocks. */
blockForKeys
(
c
,
c
->
argv
+
1
,
1
,
timeout
,
c
->
argv
[
2
]);
}
}
else
{
if
(
key
->
type
!=
REDIS_LIST
)
{
addReply
(
c
,
shared
.
wrongtypeerr
);
}
else
{
/* The list exists and has elements, so
* the regular rpoplpushCommand is executed. */
redisAssert
(
listTypeLength
(
key
)
>
0
);
rpoplpushCommand
(
c
);
}
}
}
tests/unit/type/list.tcl
View file @
f858c11d
...
@@ -127,8 +127,141 @@ start_server {
...
@@ -127,8 +127,141 @@ start_server {
assert_equal 0
[
r llen blist1
]
assert_equal 0
[
r llen blist1
]
assert_equal 1
[
r llen blist2
]
assert_equal 1
[
r llen blist2
]
}
}
test
"BRPOPLPUSH -
$type
"
{
r del target
set rd
[
redis_deferring_client
]
create_$type blist
"a b
$large
c d"
$rd brpoplpush blist target 1
assert_equal d
[
$rd
read
]
assert_equal d
[
r rpop target
]
assert_equal
"a b
$large
c"
[
r lrange blist 0 -1
]
}
}
test
"BRPOPLPUSH with zero timeout should block indefinitely"
{
set rd
[
redis_deferring_client
]
r del blist target
$rd brpoplpush blist target 0
after 1000
r rpush blist foo
assert_equal foo
[
$rd
read
]
assert_equal
{
foo
}
[
r lrange target 0 -1
]
}
test
"BRPOPLPUSH with a client BLPOPing the target list"
{
set rd
[
redis_deferring_client
]
set rd2
[
redis_deferring_client
]
r del blist target
$rd2 blpop target 0
$rd brpoplpush blist target 0
after 1000
r rpush blist foo
assert_equal foo
[
$rd
read
]
assert_equal
{
target foo
}
[
$rd2
read
]
assert_equal 0
[
r exists target
]
}
}
test
"BRPOPLPUSH with wrong source type"
{
set rd
[
redis_deferring_client
]
r del blist target
r set blist nolist
$rd brpoplpush blist target 1
assert_error
"ERR*wrong kind*"
{
$rd
read
}
}
test
"BRPOPLPUSH with wrong destination type"
{
set rd
[
redis_deferring_client
]
r del blist target
r set target nolist
r lpush blist foo
$rd brpoplpush blist target 1
assert_error
"ERR*wrong kind*"
{
$rd
read
}
set rd
[
redis_deferring_client
]
r del blist target
r set target nolist
$rd brpoplpush blist target 0
after 1000
r rpush blist foo
assert_error
"ERR*wrong kind*"
{
$rd
read
}
assert_equal
{
foo
}
[
r lrange blist 0 -1
]
}
test
"BRPOPLPUSH with multiple blocked clients"
{
set rd1
[
redis_deferring_client
]
set rd2
[
redis_deferring_client
]
r del blist target1 target2
r set target1 nolist
$rd1 brpoplpush blist target1 0
$rd2 brpoplpush blist target2 0
r lpush blist foo
assert_error
"ERR*wrong kind*"
{
$rd1
read
}
assert_equal
{
foo
}
[
$rd2
read
]
assert_equal
{
foo
}
[
r lrange target2 0 -1
]
}
test
"Linked BRPOPLPUSH"
{
set rd1
[
redis_deferring_client
]
set rd2
[
redis_deferring_client
]
r del list1 list2 list3
$rd1 brpoplpush list1 list2 0
$rd2 brpoplpush list2 list3 0
r rpush list1 foo
assert_equal
{}
[
r lrange list1 0 -1
]
assert_equal
{}
[
r lrange list2 0 -1
]
assert_equal
{
foo
}
[
r lrange list3 0 -1
]
}
test
"Circular BRPOPLPUSH"
{
set rd1
[
redis_deferring_client
]
set rd2
[
redis_deferring_client
]
r del list1 list2
$rd1 brpoplpush list1 list2 0
$rd2 brpoplpush list2 list1 0
r rpush list1 foo
assert_equal
{
foo
}
[
r lrange list1 0 -1
]
assert_equal
{}
[
r lrange list2 0 -1
]
}
test
"Self-referential BRPOPLPUSH"
{
set rd
[
redis_deferring_client
]
r del blist
$rd brpoplpush blist blist 0
r rpush blist foo
assert_equal
{
foo
}
[
r lrange blist 0 -1
]
}
test
"BRPOPLPUSH inside a transaction"
{
r del xlist target
r lpush xlist foo
r lpush xlist bar
r multi
r brpoplpush xlist target 0
r brpoplpush xlist target 0
r brpoplpush xlist target 0
r lrange xlist 0 -1
r lrange target 0 -1
r exec
}
{
foo bar
{}
{}
{
bar foo
}}
foreach
{
pop
}
{
BLPOP BRPOP
}
{
foreach
{
pop
}
{
BLPOP BRPOP
}
{
test
"
$pop:
with single empty list argument"
{
test
"
$pop:
with single empty list argument"
{
set rd
[
redis_deferring_client
]
set rd
[
redis_deferring_client
]
...
...
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