Query the archive to provide a complete KEYS list.

[redis.git] / src / migrate.c

diff --git a/src/migrate.c b/src/migrate.c
index a00f0e637d7877cbc0c1f6ab5acb4deab75dd355..a998c5f08841e65703949bdbd626b66e4b2cab20 100644 (file)
--- a/src/migrate.c
+++ b/src/migrate.c
@@ -252,41 +252,60 @@ socket_rd_err:
     return;
 }
 
     return;
 }
 

-void mdb(void) {
-    int ret;
+void stopKeyArchive(void) {
+    redisAssert(env != NULL);

 
 

-    if (env != NULL)
-        return;
+    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);
 
     ret = mdb_env_create(&env);

-    redisAssert(ret == 0);
+    if (ret != 0) return ret;

 
     ret = mdb_env_set_mapsize(env, server.mdb_mapsize);
 
     ret = mdb_env_set_mapsize(env, server.mdb_mapsize);

-    redisAssert(ret == 0);
+    if (ret != 0) return ret;

 
     ret = mdb_env_set_maxdbs(env, 1);
 
     ret = mdb_env_set_maxdbs(env, 1);

-    redisAssert(ret == 0);
+    if (ret != 0) return ret;

 
 

-    mkdir(server.mdb_environment, 0644);
+    mkdir(server.mdb_environment, 0755);

 
 

-    ret = mdb_env_open(env, server.mdb_environment, MDB_FIXEDMAP | MDB_NOSYNC, 0664);
-    redisAssert(ret == 0);
+    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);
 
     MDB_txn *txn;
     ret = mdb_txn_begin(env, NULL, 0, &txn);

-    redisAssert(ret == 0);
+    if (ret != 0) return ret;

 
 

-    ret = mdb_open(txn, NULL, 0, &dbi);
-    redisAssert(ret == 0);
+    ret = mdb_dbi_open(txn, NULL, 0, &dbi);
+    if (ret != 0) return ret;

 
     mdb_txn_commit(txn);
 
     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;
 }
 
 int archive(redisDb *db, robj *key) {
     if (server.mdb_state == REDIS_MDB_OFF)
         return 1;

-
-    mdb();
+    redisAssert(env != NULL);

 
     MDB_val kval;
     kval.mv_data = key->ptr;
 
     MDB_val kval;
     kval.mv_data = key->ptr;


@@ -298,7 +317,7 @@ int archive(redisDb *db, robj *key) {
         return 0;
 
     if (object->archived != 0)
         return 0;
 
     if (object->archived != 0)

-        return 0;
+        return 1;

 
     rio payload;
     createDumpPayload(&payload, object);
 
     rio payload;
     createDumpPayload(&payload, object);


@@ -336,8 +355,6 @@ robj *recover(redisDb *db, robj *key) {
 
     int ret;
 
 
     int ret;
 

-    mdb();
-

     MDB_val kval;
     kval.mv_data = key->ptr;
     kval.mv_size = sdslen((sds)key->ptr);
     MDB_val kval;
     kval.mv_data = key->ptr;
     kval.mv_size = sdslen((sds)key->ptr);


@@ -389,3 +406,58 @@ recover_err:
     sdsfree(sval);
     return NULL;
 }
     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;
+}