CLIENT UNBLOCK implemented.

src/networking.c

@@ -75,6 +75,8 @@ void linkClient(client *c) {
      * this way removing the client in unlinkClient() will not require
      * a linear scan, but just a constant time operation. */
     c->client_list_node = listLast(server.clients);
+    uint64_t id = htonu64(c->id);
+    raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);
 }
 
 client *createClient(int fd) {
@@ -720,6 +722,8 @@ void unlinkClient(client *c) {
     if (c->fd != -1) {
         /* Remove from the list of active clients. */
         if (c->client_list_node) {
+            uint64_t id = htonu64(c->id);
+            raxRemove(server.clients_index,(unsigned char*)&id,sizeof(id),NULL);
             listDelNode(server.clients,c->client_list_node);
             c->client_list_node = NULL;
         }
@@ -864,6 +868,15 @@ void freeClientsInAsyncFreeQueue(void) {
     }
 }
 
+/* Return a client by ID, or NULL if the client ID is not in the set
+ * of registered clients. Note that "fake clients", created with -1 as FD,
+ * are not registered clients. */
+client *lookupClientByID(uint64_t id) {
+    id = htonu64(id);
+    client *c = raxFind(server.clients_index,(unsigned char*)&id,sizeof(id));
+    return (c == raxNotFound) ? NULL : c;
+}
+
 /* Write data in output buffers to client. Return C_OK if the client
  * is still valid after the call, C_ERR if it was freed. */
 int writeToClient(int fd, client *c, int handler_installed) {
@@ -1553,6 +1566,7 @@ void clientCommand(client *c) {
 
     if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
         const char *help[] = {
+"id      -- Return the ID of the current connection.",
 "getname -- Return the name of the current connection.",
 "kill <ip:port> -- Kill connection made from <ip:port>.",
 "kill <option> <value> [option value ...] -- Kill connections. Options are:",
@@ -1566,6 +1580,9 @@ void clientCommand(client *c) {
 NULL
         };
         addReplyHelp(c, help);
+    } else if (!strcasecmp(c->argv[1]->ptr,"id") && c->argc == 2) {
+        /* CLIENT ID */
+        addReplyLongLong(c,c->id);
     } else if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) {
         /* CLIENT LIST */
         sds o = getAllClientsInfoString();
@@ -1671,6 +1688,19 @@ NULL
         /* If this client has to be closed, flag it as CLOSE_AFTER_REPLY
          * only after we queued the reply to its output buffers. */
         if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY;
+    } else if (!strcasecmp(c->argv[1]->ptr,"unblock") && c->argc == 3) {
+        /* CLIENT UNBLOCK <id> */
+        long long id;
+        if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL)
+            != C_OK) return;
+        struct client *target = lookupClientByID(id);
+        if (target && target->flags & CLIENT_BLOCKED) {
+            replyToBlockedClientTimedOut(target);
+            unblockClient(target);
+            addReply(c,shared.cone);
+        } else {
+            addReply(c,shared.czero);
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) {
         int j, len = sdslen(c->argv[2]->ptr);
         char *p = c->argv[2]->ptr;

src/server.c

@@ -1890,6 +1890,7 @@ void initServer(void) {
     server.pid = getpid();
     server.current_client = NULL;
     server.clients = listCreate();
+    server.clients_index = raxNew();
     server.clients_to_close = listCreate();
     server.slaves = listCreate();
     server.monitors = listCreate();

src/server.h

@@ -952,6 +952,7 @@ struct redisServer {
     list *clients_pending_write; /* There is to write or install handler. */
     list *slaves, *monitors;    /* List of slaves and MONITORs */
     client *current_client; /* Current client, only used on crash report */
+    rax *clients_index;         /* Active clients dictionary by client ID. */
     int clients_paused;         /* True if clients are currently paused */
     mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
     char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */