Draft: Namespaces

src/acl.cpp

@@ -248,6 +248,7 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     if (raxFind(Users,(unsigned char*)name,namelen) != raxNotFound) return NULL;
     user *u = (user*)zmalloc(sizeof(*u), MALLOC_LOCAL);
     u->name = sdsnewlen(name,namelen);
+    u->ns_name = sdsnew("::");
     u->flags = USER_FLAG_DISABLED | g_pserver->acl_pubsub_default;
     u->allowed_subcommands = NULL;
     u->passwords = listCreate();
@@ -288,6 +289,7 @@ user *ACLCreateUnlinkedUser(void) {
  * will not remove the user from the Users global radix tree. */
 void ACLFreeUser(user *u) {
     sdsfree(u->name);
+    sdsfree(u->ns_name);
     listRelease(u->passwords);
     listRelease(u->patterns);
     listRelease(u->channels);
@@ -338,6 +340,8 @@ void ACLCopyUser(user *dst, user *src) {
     memcpy(dst->allowed_commands,src->allowed_commands,
            sizeof(dst->allowed_commands));
     dst->flags = src->flags;
+    sdsfree(dst->ns_name);
+    dst->ns_name = sdsdup(src->ns_name);
     ACLResetSubcommands(dst);
     /* Copy the allowed subcommands array of array of SDS strings. */
     if (src->allowed_subcommands) {
@@ -669,6 +673,13 @@ sds ACLDescribeUser(user *u) {
     sds rules = ACLDescribeUserCommandRules(u);
     res = sdscatsds(res,rules);
     sdsfree(rules);
+
+    /* namespace */
+    if (strcmp(u->ns_name, DefaultUser->ns_name) != 0) {
+        res = sdscatlen(res, " ", 1);
+        res = sdscatsds(res, u->ns_name);
+    }
+
     return res;
 }
 
@@ -801,6 +812,10 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  * reset        Performs the following actions: resetpass, resetkeys, off,
  *              -@all. The user returns to the same state it has immediately
  *              after its creation.
+ * ::<ns>       Sets the namespace for the user.
+ *              :: is the default namespace
+ *              ::auto will generate a new namespace with a random uuid
+ *              (only if the current user is still in the default namespace)
  *
  * The 'op' string must be null terminated. The 'oplen' argument should
  * specify the length of the 'op' string in case the caller requires to pass
@@ -1000,6 +1015,11 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
             errno = ENOENT;
             return C_ERR;
         }
+    } else if (op[0] == ':' && op[1] == ':') {
+        //TODO: implement ::auto
+        sds ns = sdsnewlen(op,oplen);
+        sdsfree(u->ns_name);
+        u->ns_name = ns;
     } else if (!strcasecmp(op,"reset")) {
         serverAssert(ACLSetUser(u,"resetpass",-1) == C_OK);
         serverAssert(ACLSetUser(u,"resetkeys",-1) == C_OK);
@@ -1114,6 +1134,8 @@ int ACLAuthenticateUser(client *c, robj *username, robj *password) {
     if (ACLCheckUserCredentials(username,password) == C_OK) {
         c->authenticated = 1;
         c->user = ACLGetUserByName((sds)ptrFromObj(username),sdslen((sds)ptrFromObj(username)));
+        c->ns = getNamespace(c->user->ns_name);
+        selectDbNamespaced(c, c->db->mapped_id);
         moduleNotifyUserChanged(c);
         return C_OK;
     } else {
@@ -1969,7 +1991,7 @@ void aclCommand(client *c) {
             return;
         }
 
-        addReplyMapLen(c,5);
+        addReplyMapLen(c,6);
 
         /* Flags */
         addReplyBulkCString(c,"flags");
@@ -2030,6 +2052,10 @@ void aclCommand(client *c) {
                 addReplyBulkCBuffer(c,thispat,sdslen(thispat));
             }
         }
+
+        /* namespace */
+        addReplyBulkCString(c,"namespace");
+        addReplyBulkCString(c, u->ns_name);
     } else if ((!strcasecmp(sub,"list") || !strcasecmp(sub,"users")) &&
                c->argc == 2)
     {

src/aof.cpp

@@ -731,10 +731,14 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
 
     /* The DB this command was targeting is not the same as the last command
      * we appended. To issue a SELECT command is needed. */
-    if (dictid != g_pserver->aof_selected_db) {
-        char seldb[64];
-
+    if (dictid != -1 && dictid != g_pserver->aof_selected_db) {
+        char seldb[64], mappeddb[64];
         snprintf(seldb,sizeof(seldb),"%d",dictid);
+        snprintf(mappeddb,sizeof(seldb),"%d",g_pserver->db[dictid].mapped_id);
+
+        buf = sdscatprintf(buf, "*4\r\n$8\r\nALLOCATE\r\n$%lu\r\n%s\r\n$%lu\r\n%s\r\n$%lu\r\n%s\r\n",
+            sdslen(g_pserver->db[dictid].ns->name), g_pserver->db[dictid].ns->name, strlen(seldb), seldb, strlen(mappeddb), mappeddb);
+
         buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
             (unsigned long)strlen(seldb),seldb);
         g_pserver->aof_selected_db = dictid;
@@ -767,7 +771,8 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
 struct client *createAOFClient(void) {
     struct client *c =(client*) zmalloc(sizeof(*c), MALLOC_LOCAL);
 
-    selectDb(c,0);
+    c->ns = g_pserver->default_namespace;
+    selectDbNamespaced(c, 0);
     c->id = CLIENT_ID_AOF; /* So modules can identify it's the AOF client. */
     c->conn = NULL;
     c->iel = IDX_EVENT_LOOP_MAIN;
@@ -1546,12 +1551,22 @@ int rewriteAppendOnlyFileRio(rio *aof) {
     long long updated_time = 0;
 
     for (j = 0; j < cserver.dbnum; j++) {
+        char allocatecmd[] = "*4\r\n$8\r\nALLOCATE\r\n";
         char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
         redisDb *db = g_pserver->db+j;
         dict *d = db->dict;
         if (dictSize(d) == 0) continue;
+        serverAssert(db->ns);
+        serverAssert(db->mapped_id >= 0);
+
         di = dictGetSafeIterator(d);
 
+        /* ALLOCATE DB */
+        if (rioWrite(aof,allocatecmd,sizeof(allocatecmd)-1) == 0) goto werr;
+        if (rioWriteBulkString(aof,db->ns->name, sdslen(db->ns->name)) == 0) goto werr;
+        if (rioWriteBulkLongLong(aof,db->id) == 0) goto werr;
+        if (rioWriteBulkLongLong(aof,db->mapped_id) == 0) goto werr;
+
         /* SELECT the new DB */
         if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
         if (rioWriteBulkLongLong(aof,j) == 0) goto werr;

src/cluster.cpp

@@ -2196,24 +2196,32 @@ int clusterProcessPacket(clusterLink *link) {
         }
     } else if (type == CLUSTERMSG_TYPE_PUBLISH) {
         robj *channel, *message;
-        uint32_t channel_len, message_len;
+        sds ns_name;
+        redisNamespace *ns;
+        uint32_t ns_len, channel_len, message_len;
+
+        ns_len = ntohl(hdr->data.publish.msg.ns_len);
+        ns_name = sdsnewlen((char*)hdr->data.publish.msg.bulk_data,ns_len);
+        ns = getNamespace(ns_name);
 
         /* Don't bother creating useless objects if there are no
          * Pub/Sub subscribers. */
-        if (dictSize(g_pserver->pubsub_channels) ||
-           dictSize(g_pserver->pubsub_patterns))
+        if (!ns ||
+           dictSize(ns->pubsub_channels) ||
+           dictSize(ns->pubsub_patterns))
         {
             channel_len = ntohl(hdr->data.publish.msg.channel_len);
             message_len = ntohl(hdr->data.publish.msg.message_len);
             channel = createStringObject(
-                        (char*)hdr->data.publish.msg.bulk_data,channel_len);
+                        (char*)hdr->data.publish.msg.bulk_data+ns_len,channel_len);
             message = createStringObject(
-                        (char*)hdr->data.publish.msg.bulk_data+channel_len,
+                        (char*)hdr->data.publish.msg.bulk_data+ns_len+channel_len,
                         message_len);
-            pubsubPublishMessage(channel,message);
+            pubsubPublishMessage(ns,channel,message);
             decrRefCount(channel);
             decrRefCount(message);
         }
+        sdsfree(ns_name);
     } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST) {
         if (!sender) return 1;  /* We don't know that node. */
         clusterSendFailoverAuthIfNeeded(sender,hdr);
@@ -2754,22 +2762,24 @@ void clusterBroadcastPong(int target) {
 /* Send a PUBLISH message.
  *
  * If link is NULL, then the message is broadcasted to the whole cluster. */
-void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
+void clusterSendPublish(redisNamespace *ns, clusterLink *link, robj *channel, robj *message) {
     unsigned char *payload;
     clusterMsg buf[1];
     clusterMsg *hdr = (clusterMsg*) buf;
     uint32_t totlen;
-    uint32_t channel_len, message_len;
+    uint32_t channel_len, message_len, ns_len;
 
     channel = getDecodedObject(channel);
     message = getDecodedObject(message);
     channel_len = sdslen(szFromObj(channel));
     message_len = sdslen(szFromObj(message));
+    ns_len = sdslen(ns->name);
 
     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_PUBLISH);
     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
-    totlen += sizeof(clusterMsgDataPublish) - 8 + channel_len + message_len;
+    totlen += sizeof(clusterMsgDataPublish) - 8 + channel_len + message_len + ns_len;
 
+    hdr->data.publish.msg.ns_len = htonl(ns_len);
     hdr->data.publish.msg.channel_len = htonl(channel_len);
     hdr->data.publish.msg.message_len = htonl(message_len);
     hdr->totlen = htonl(totlen);
@@ -2782,7 +2792,8 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
         memcpy(payload,hdr,sizeof(*hdr));
         hdr = (clusterMsg*) payload;
     }
-    memcpy(hdr->data.publish.msg.bulk_data,ptrFromObj(channel),sdslen(szFromObj(channel)));
+    memcpy(hdr->data.publish.msg.bulk_data,ns->name,ns_len);
+    memcpy(hdr->data.publish.msg.bulk_data+ns_len,ptrFromObj(channel),sdslen(szFromObj(channel)));
     memcpy(hdr->data.publish.msg.bulk_data+sdslen(szFromObj(channel)),
         ptrFromObj(message),sdslen(szFromObj(message)));
 
@@ -2888,8 +2899,8 @@ int clusterSendModuleMessageToTarget(const char *target, uint64_t module_id, uin
  * cluster. In the future we'll try to get smarter and avoiding propagating those
  * messages to hosts without receives for a given channel.
  * -------------------------------------------------------------------------- */
-void clusterPropagatePublish(robj *channel, robj *message) {
-    clusterSendPublish(NULL, channel, message);
+void clusterPropagatePublish(redisNamespace *ns, robj *channel, robj *message) {
+    clusterSendPublish(ns, NULL, channel, message);
 }
 
 /* -----------------------------------------------------------------------------

src/cluster.h

@@ -209,6 +209,7 @@ typedef struct {
 } clusterMsgDataFail;
 
 typedef struct {
+    uint32_t ns_len;
     uint32_t channel_len;
     uint32_t message_len;
     unsigned char bulk_data[8]; /* 8 bytes just as placeholder. */

src/config.cpp

@@ -542,7 +542,11 @@ void loadServerConfigFromString(char *config) {
             if (port < 0 || port > 65535 || *ptr != '\0') {
                 err= "Invalid master port"; goto loaderr;
             }
-            replicationAddMaster(argv[1], port);
+
+            redisMasterConnInfo *masterInfo = (redisMasterConnInfo*)zmalloc(sizeof(redisMasterConnInfo), MALLOC_LOCAL);
+            masterInfo->ip = sdsnew(argv[1]);
+            masterInfo->port = port;
+            listAddNodeTail(g_pserver->repl_init_masters, masterInfo);
         } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
             if (strlen(argv[1]) > CONFIG_AUTHPASS_MAX_LEN) {
                 err = "Password is longer than CONFIG_AUTHPASS_MAX_LEN";
@@ -2685,6 +2689,7 @@ standardConfig configs[] = {
 
     /* Integer configs */
     createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, cserver.dbnum, 16, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("databases-per-namespace", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, cserver.ns_dbnum, INT_MAX, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->port, 6379, INTEGER_CONFIG, NULL, updatePort), /* TCP port. */
     createIntConfig("auto-aof-rewrite-percentage", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->aof_rewrite_perc, 100, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("cluster-replica-validity-factor", "cluster-slave-validity-factor", MODIFIABLE_CONFIG, 0, INT_MAX, g_pserver->cluster_slave_validity_factor, 10, INTEGER_CONFIG, NULL, NULL), /* Slave max data age factor. */