{"setnx",setnxCommand,3,REDIS_CMD_DENYOOM,NULL,0,0,0},
{"setex",setexCommand,4,REDIS_CMD_DENYOOM,NULL,0,0,0},
{"append",appendCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1},
- {"substr",substrCommand,4,0,NULL,1,1,1},
{"strlen",strlenCommand,2,0,NULL,1,1,1},
{"del",delCommand,-2,0,NULL,0,0,0},
{"exists",existsCommand,2,0,NULL,1,1,1},
+ {"setbit",setbitCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1},
+ {"getbit",getbitCommand,3,0,NULL,1,1,1},
+ {"setrange",setrangeCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1},
+ {"getrange",getrangeCommand,4,0,NULL,1,1,1},
+ {"substr",getrangeCommand,4,0,NULL,1,1,1},
{"incr",incrCommand,2,REDIS_CMD_DENYOOM,NULL,1,1,1},
{"decr",decrCommand,2,REDIS_CMD_DENYOOM,NULL,1,1,1},
{"mget",mgetCommand,-2,0,NULL,1,-1,1},
{"rpop",rpopCommand,2,0,NULL,1,1,1},
{"lpop",lpopCommand,2,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},
{"llen",llenCommand,2,0,NULL,1,1,1},
{"lindex",lindexCommand,3,0,NULL,1,1,1},
{"lrange",lrangeCommand,4,0,NULL,1,1,1},
{"ltrim",ltrimCommand,4,0,NULL,1,1,1},
{"lrem",lremCommand,4,0,NULL,1,1,1},
- {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_DENYOOM,NULL,1,2,1},
+ {"rpoplpush",rpoplpushCommand,3,REDIS_CMD_DENYOOM,NULL,1,2,1},
{"sadd",saddCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1},
{"srem",sremCommand,3,0,NULL,1,1,1},
{"smove",smoveCommand,4,0,NULL,1,2,1},
}
/* Close connections of timedout clients */
- if ((server.maxidletime && !(loops % 100)) || server.blpop_blocked_clients)
+ if ((server.maxidletime && !(loops % 100)) || server.bpop_blocked_clients)
closeTimedoutClients();
/* Check if a background saving or AOF rewrite in progress terminated */
* for ready file descriptors. */
void beforeSleep(struct aeEventLoop *eventLoop) {
REDIS_NOTUSED(eventLoop);
+ listNode *ln;
+ redisClient *c;
/* Awake clients that got all the swapped keys they requested */
if (server.vm_enabled && listLength(server.io_ready_clients)) {
listIter li;
- listNode *ln;
listRewind(server.io_ready_clients,&li);
while((ln = listNext(&li))) {
- redisClient *c = ln->value;
+ c = ln->value;
struct redisCommand *cmd;
/* Resume the client. */
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 */
flushAppendOnlyFile();
}
server.rdbcompression = 1;
server.activerehashing = 1;
server.maxclients = 0;
- server.blpop_blocked_clients = 0;
+ server.bpop_blocked_clients = 0;
server.maxmemory = 0;
server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU;
server.maxmemory_samples = 3;
server.clients = listCreate();
server.slaves = listCreate();
server.monitors = listCreate();
+ server.unblocked_clients = listCreate();
createSharedObjects();
server.el = aeCreateEventLoop();
server.db = zmalloc(sizeof(redisDb)*server.dbnum);
(float)c_ru.ru_stime.tv_sec+(float)c_ru.ru_stime.tv_usec/1000000,
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
- server.blpop_blocked_clients,
+ server.bpop_blocked_clients,
zmalloc_used_memory(),
hmem,
zmalloc_get_rss(),
int count; /* Total number of MULTI commands */
} 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.
* Clients are taken in a liked list. */
typedef struct redisClient {
long repldboff; /* replication DB file offset */
off_t repldbsize; /* replication DB file size */
multiState mstate; /* MULTI/EXEC state */
- robj **blocking_keys; /* The key we are waiting to terminate a blocking
- * 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. */
+ blockingState bpop; /* blocking state */
list *io_keys; /* Keys this client is waiting to be loaded from the
* swap file in order to continue. */
list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */
int maxmemory_policy;
int maxmemory_samples;
/* Blocked clients */
- unsigned int blpop_blocked_clients;
+ unsigned int bpop_blocked_clients;
unsigned int vm_blocked_clients;
+ list *unblocked_clients;
/* Sort parameters - qsort_r() is only available under BSD so we
* have to take this state global, in order to pass it to sortCompare() */
int sort_desc;
void getCommand(redisClient *c);
void delCommand(redisClient *c);
void existsCommand(redisClient *c);
+void setbitCommand(redisClient *c);
+void getbitCommand(redisClient *c);
+void setrangeCommand(redisClient *c);
+void getrangeCommand(redisClient *c);
void incrCommand(redisClient *c);
void decrCommand(redisClient *c);
void incrbyCommand(redisClient *c);
void flushallCommand(redisClient *c);
void sortCommand(redisClient *c);
void lremCommand(redisClient *c);
- void rpoplpushcommand(redisClient *c);
+ void rpoplpushCommand(redisClient *c);
void infoCommand(redisClient *c);
void mgetCommand(redisClient *c);
void monitorCommand(redisClient *c);
void discardCommand(redisClient *c);
void blpopCommand(redisClient *c);
void brpopCommand(redisClient *c);
+ void brpoplpushCommand(redisClient *c);
void appendCommand(redisClient *c);
-void substrCommand(redisClient *c);
void strlenCommand(redisClient *c);
void zrankCommand(redisClient *c);
void zrevrankCommand(redisClient *c);
projects / redis.git / commitdiff
