X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/e95740392b54635ec83994261af021015afc6551..64b2834657e4c215dc0489417177a37c80d9ac46:/src/migrate.c?ds=sidebyside

diff --git a/src/migrate.c b/src/migrate.c
index d3b92859..6d8aad7f 100644
--- a/src/migrate.c
+++ b/src/migrate.c
@@ -1,10 +1,16 @@
 #include "redis.h"
 #include "endianconv.h"
 
+#include <sys/stat.h>
+#include <lmdb.h>
+
 /* -----------------------------------------------------------------------------
  * DUMP, RESTORE and MIGRATE commands
  * -------------------------------------------------------------------------- */
 
+MDB_env *env;
+MDB_dbi dbi;
+
 /* Generates a DUMP-format representation of the object 'o', adding it to the
  * io stream pointed by 'rio'. This function can't fail. */
 void createDumpPayload(rio *payload, robj *o) {
@@ -130,7 +136,7 @@ void migrateCommand(redisClient *c) {
     int fd;
     long timeout;
     long dbid;
-    long long ttl, expireat;
+    long long ttl = 0, expireat;
     robj *o;
     rio cmd, payload;
 
@@ -139,7 +145,7 @@ void migrateCommand(redisClient *c) {
         return;
     if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
         return;
-    if (timeout <= 0) timeout = 1;
+    if (timeout <= 0) timeout = 1000;
 
     /* Check if the key is here. If not we reply with success as there is
      * nothing to migrate (for instance the key expired in the meantime), but
@@ -157,7 +163,8 @@ void migrateCommand(redisClient *c) {
             server.neterr);
         return;
     }
-    if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
+    if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {
+        close(fd);
         addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n"));
         return;
     }
@@ -177,7 +184,7 @@ void migrateCommand(redisClient *c) {
     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
     redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
-    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,(expireat==-1) ? 0 : ttl));
+    redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));
 
     /* Finally the last argument that is the serailized object payload
      * in the DUMP format. */
@@ -244,3 +251,174 @@ socket_rd_err:
     close(fd);
     return;
 }
+
+void stopKeyArchive(void) {
+    redisAssert(env != NULL);
+
+    mdb_dbi_close(env, dbi);
+    mdb_env_close(env);
+    env = NULL;
+
+    server.mdb_state = REDIS_MDB_OFF;
+}
+
+int startKeyArchive(void) {
+    redisAssert(env == NULL);
+
+    int ret;
+
+    ret = mdb_env_create(&env);
+    if (ret != 0) return ret;
+
+    ret = mdb_env_set_mapsize(env, server.mdb_mapsize);
+    if (ret != 0) return ret;
+
+    ret = mdb_env_set_maxdbs(env, 1);
+    if (ret != 0) return ret;
+
+    mkdir(server.mdb_environment, 0755);
+
+    ret = mdb_env_open(env, server.mdb_environment, MDB_FIXEDMAP | MDB_NOSYNC, 0644);
+    if (ret != 0) return ret;
+
+    MDB_txn *txn;
+    ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0) return ret;
+
+    ret = mdb_dbi_open(txn, NULL, 0, &dbi);
+    if (ret != 0) return ret;
+
+    mdb_txn_commit(txn);
+
+    server.mdb_state = REDIS_MDB_ON;
+    return 0;
+}
+
+int archive(redisDb *db, robj *key) {
+    if (server.mdb_state == REDIS_MDB_OFF)
+        return 1;
+    redisAssert(env != NULL);
+
+    MDB_val kval;
+    kval.mv_data = key->ptr;
+    kval.mv_size = sdslen((sds)key->ptr);
+
+    robj *object;
+    object = lookupKey(db, key);
+    if (object == NULL)
+        return 0;
+
+    if (object->archived != 0)
+        return 1;
+
+    rio payload;
+    createDumpPayload(&payload, object);
+
+    MDB_val dval;
+    dval.mv_size = sdslen(payload.io.buffer.ptr);
+    dval.mv_data = payload.io.buffer.ptr;
+
+    int ret;
+
+    MDB_txn *txn;
+    ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0)
+        goto archive_err;
+
+    ret = mdb_put(txn, dbi, &kval, &dval, 0);
+    if (ret != 0) {
+        mdb_txn_abort(txn);
+        goto archive_err;
+    }
+
+    mdb_txn_commit(txn);
+    sdsfree(payload.io.buffer.ptr);
+    return 1;
+
+archive_err:
+    sdsfree(payload.io.buffer.ptr);
+    redisAssert(0);
+    return 0;
+}
+
+robj *recover(redisDb *db, robj *key) {
+    if (server.mdb_state == REDIS_MDB_OFF)
+        return NULL;
+
+    int ret;
+
+    MDB_val kval;
+    kval.mv_data = key->ptr;
+    kval.mv_size = sdslen((sds)key->ptr);
+
+    MDB_txn *txn;
+    ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0)
+        return NULL;
+
+    MDB_cursor *cursor;
+    ret = mdb_cursor_open(txn, dbi, &cursor);
+    if (ret != 0) {
+        mdb_txn_abort(txn);
+        return NULL;
+    }
+
+    MDB_val pval;
+    ret = mdb_cursor_get(cursor, &kval, &pval, MDB_SET);
+    if (ret != 0) {
+        mdb_txn_abort(txn);
+        return NULL;
+    }
+
+    sds sval = sdsnewlen(pval.mv_data, pval.mv_size);
+    mdb_cursor_close(cursor);
+    mdb_txn_abort(txn);
+
+    rio payload;
+    rioInitWithBuffer(&payload, sval);
+
+    int type = rdbLoadObjectType(&payload);
+    if (type == -1)
+        goto recover_err;
+
+    robj *object = rdbLoadObject(type, &payload);
+    if (object == NULL)
+        goto recover_err;
+
+    object->archived = 1;
+
+    dbAdd(db, key, object);
+    signalModifiedKey(db, key);
+    server.dirty++;
+
+    sdsfree(sval);
+    return object;
+
+recover_err:
+    sdsfree(sval);
+    return NULL;
+}
+
+void purge(robj *key) {
+    if (server.mdb_state == REDIS_MDB_OFF)
+        return;
+
+    int ret;
+
+    MDB_val kval;
+    kval.mv_data = key->ptr;
+    kval.mv_size = sdslen((sds)key->ptr);
+
+    MDB_txn *txn;
+    ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0)
+        return;
+
+    ret = mdb_del(txn, dbi, &kval, NULL);
+    if (ret != 0) {
+        mdb_txn_abort(txn);
+        return;
+    }
+
+    mdb_txn_commit(txn);
+}