X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/ae3aeca8288bbb22c6ef2eac51827409e19e41df..bfbf8c888bff858fbde7b8a213c2cc3d01eedd5f:/src/migrate.c

diff --git a/src/migrate.c b/src/migrate.c
index 86dfe165..a998c5f0 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) {
@@ -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,7 @@ 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;
@@ -245,3 +251,213 @@ socket_rd_err:
     close(fd);
     return;
 }
+
+void stopKeyArchive(void) {
+    redisAssert(env != NULL);
+
+    MDB_txn *txn;
+    int ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0)
+        mdb_txn_abort(txn);
+    else {
+        mdb_dbi_close(env, dbi);
+        mdb_txn_commit(txn);
+    }
+
+    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);
+}
+
+int rummage(redisClient *c, unsigned long *numkeys) {
+    if (server.mdb_state == REDIS_MDB_OFF)
+        return REDIS_OK;
+
+    int ret;
+
+    MDB_txn *txn;
+    ret = mdb_txn_begin(env, NULL, 0, &txn);
+    if (ret != 0)
+        return REDIS_ERR;
+
+    MDB_cursor *cursor;
+    ret = mdb_cursor_open(txn, dbi, &cursor);
+    if (ret != 0) {
+        mdb_txn_abort(txn);
+        return REDIS_ERR;
+    }
+
+    MDB_val kval;
+    while ((ret = mdb_cursor_get(cursor, &kval, NULL, MDB_NEXT)) == 0) {
+        robj *key = createStringObject(kval.mv_data, kval.mv_size);
+        addReplyBulk(c, key);
+        ++*numkeys;
+        decrRefCount(key);
+    }
+
+    mdb_cursor_close(cursor);
+    mdb_txn_abort(txn);
+    return REDIS_OK;
+}