finish implementation of Backup -> Copy (now called Copy) in async enviroment, also add tests

Author: Simone Mosciatti

redisql_lib/src/redis.rs

@@ -435,9 +435,9 @@ pub unsafe fn string_ptr_len(
     str: *mut rm::ffi::RedisModuleString,
 ) -> &'static str {
     let mut len = 0;
-    let base = rm::ffi::RedisModule_StringPtrLen.unwrap()(
-        str, &mut len,
-    ) as *mut u8;
+    let base =
+        rm::ffi::RedisModule_StringPtrLen.unwrap()(str, &mut len)
+            as *mut u8;
     let slice = slice::from_raw_parts(base, len);
     str::from_utf8_unchecked(slice)
 }
@@ -489,9 +489,9 @@ pub enum Command {
         arguments: Vec<&'static str>,
         client: BlockedClient,
     },
-    Copy {
-        source: Arc<Mutex<sql::RawConnection>>,
-        destination: Arc<Mutex<sql::RawConnection>>,
+    MakeCopy {
+        source: DBKey,
+        destination: DBKey,
         client: BlockedClient,
     },
 }
@@ -545,6 +545,32 @@ pub fn do_query(
     }
 }
 
+/// implements the copy of the source database into the destination one
+/// it also leak the two DBKeys
+pub fn do_copy(
+    source: DBKey,
+    destination: DBKey,
+) -> Result<QueryResult, err::RediSQLError> {
+    let source_connection = source.loop_data.get_db();
+    let destination_connection = destination.loop_data.get_db();
+
+    let backup_result = {
+        let source_db = source_connection.lock().unwrap();
+        let destination_db = destination_connection.lock().unwrap();
+        match make_backup(&source_db, &destination_db) {
+            Err(e) => Err(RediSQLError::from(e)),
+            Ok(_) => Ok(QueryResult::OK {}),
+        }
+    };
+
+    restore_previous_statements(&destination.loop_data);
+
+    std::mem::forget(source);
+    std::mem::forget(destination);
+
+    backup_result
+}
+
 fn bind_statement<'a>(
     stmt: &'a MultiStatement,
     arguments: &[&str],
@@ -577,6 +603,13 @@ impl error::Error for RedisError {
     }
 }
 
+/// restore_previous_statements wread the statements written in the database and add them to the
+/// loopdata datastructure.
+/// At the moment this function returns `()` no matter if there are errors or not.
+/// Errors are pretty unlikely, especially if nobody touched manually the metadata database, but
+/// still they may happens.
+/// I am not sure if it is a good idea or if I should upgrade the code to return an error, and
+/// maybe just ignore the error to keep the whole flow as it is now.
 fn restore_previous_statements<'a, L: 'a + LoopData>(
     loopdata: &L,
 ) -> () {
@@ -726,18 +759,13 @@ pub fn listen_and_execute<'a, L: 'a + LoopData>(
                     },
                 );
             }
-            Ok(Command::Copy {
+            Ok(Command::MakeCopy {
                 source,
                 destination,
                 client,
             }) => {
-                let source = source.lock().unwrap();
-                let destination = destination.lock().unwrap();
-                let result = match make_backup(&source, &destination)
-                {
-                    Err(e) => Err(RediSQLError::from(e)),
-                    Ok(_) => Ok(QueryResult::OK {}),
-                };
+                // The DBKeys MUST be leaked, they are leaked by do_copy
+                let result = do_copy(source, destination);
                 return_value(&client, result);
             }
             Ok(Command::Stop) => {
@@ -819,7 +847,7 @@ impl Drop for DBKey {
 pub fn create_metadata_table(
     db: Arc<Mutex<sql::RawConnection>>,
 ) -> Result<(), sql::SQLite3Error> {
-    let statement = "CREATE TABLE RediSQLMetadata(data_type TEXT, key TEXT, value TEXT);";
+    let statement = "CREATE TABLE IF NOT EXISTS RediSQLMetadata(data_type TEXT, key TEXT, value TEXT);";
 
     let stmt = MultiStatement::new(db, statement)?;
     stmt.execute()?;
@@ -1018,10 +1046,8 @@ pub fn get_dbkeyptr_from_name(
         rm::ffi::RedisModule_KeyType.unwrap()(safe_key.key)
     };
     if unsafe {
-        rm::ffi::DBType
-            == rm::ffi::RedisModule_ModuleTypeGetType.unwrap()(
-                safe_key.key,
-            )
+        rm::ffi::DBType == rm::ffi::RedisModule_ModuleTypeGetType
+            .unwrap()(safe_key.key)
     } {
         let db_ptr = unsafe {
             rm::ffi::RedisModule_ModuleTypeGetValue.unwrap()(

src/lib.rs

@@ -801,12 +801,12 @@ unsafe extern "C" fn rdb_load(
 }
 
 #[allow(non_snake_case)]
-extern "C" fn Backup(
+extern "C" fn MakeCopy(
     ctx: *mut r::rm::ffi::RedisModuleCtx,
     argv: *mut *mut r::rm::ffi::RedisModuleString,
     argc: ::std::os::raw::c_int,
 ) -> i32 {
-    debug!("Backup | Start");
+    debug!("MakeCopy | Start");
     let (context, argvector) = r::create_argument(ctx, argv, argc);
 
     if argvector.len() != 3 {
@@ -862,23 +862,27 @@ extern "C" fn Backup(
         },
     };
 
+    /*
     let source_connection = source_db.loop_data.get_db();
     let dest_connection = dest_db.loop_data.get_db();
+    */
+    let ch = &source_db.tx.clone();
 
-    let cmd = r::Command::Copy {
-        source: source_connection,
-        destination: dest_connection,
+    let cmd = r::Command::MakeCopy {
+        source: source_db,
+        destination: dest_db,
         client: blocked_client,
     };
 
-    let ch = &source_db.tx.clone();
+    /*
     std::mem::forget(source_db);
     std::mem::forget(dest_db);
+    */
 
-    debug!("Backup | End");
+    debug!("MakeCopy | End");
     match ch.send(cmd) {
         Ok(()) => {
-            debug!("Backup | Successfully send command");
+            debug!("MakeCopy | Successfully send command");
             r::rm::ffi::REDISMODULE_OK
         }
         Err(_) => r::rm::ffi::REDISMODULE_OK,
@@ -1031,9 +1035,9 @@ pub extern "C" fn RedisModule_OnLoad(
 
     match register_function(
         &ctx,
-        String::from("REDISQL.BACKUP"),
+        String::from("REDISQL.COPY"),
         String::from("readonly"),
-        Backup,
+        MakeCopy,
     ) {
         Ok(()) => (),
         Err(e) => return e,

test/correctness/test.py

@@ -563,6 +563,35 @@ def test_statement_after_rdb_load(self):
       self.assertTrue([4L, "cat:4", "4"] in result)
 
 
+class TestCopy(TestRediSQLWithExec):
+    def test_copy_mem_from_mem(self):
+        done = self.exec_naked("REDISQL.CREATE_DB", "DB1")
+        self.assertEquals(done, "OK")
+        done = self.exec_naked("REDISQL.CREATE_DB", "DB2")
+        self.assertEquals(done, "OK")
+        done = self.exec_naked("REDISQL.EXEC", "DB1", "CREATE TABLE foo(a INT);")
+        self.assertEquals(done, ["DONE", 0L])
+
+        for i in xrange(10):
+            done = self.exec_naked("REDISQL.EXEC", "DB1", "INSERT INTO foo VALUES({})".format(i))
+            self.assertEquals(done, ["DONE", 1L])
+
+        done = self.exec_naked("REDISQL.COPY", "DB1", "DB2")
+        result = self.exec_naked("REDISQL.QUERY", "DB2", "SELECT a FROM foo ORDER BY a")
+        self.assertEquals(result, [[0L], [1L], [2L], [3L], [4L], [5L], [6L], [7L], [8L], [9L]])
+
+    def test_statements_copy_mem_from_mem(self):
+        done = self.exec_naked("REDISQL.CREATE_DB", "DB1")
+        self.assertEquals(done, "OK")
+        done = self.exec_naked("REDISQL.CREATE_DB", "DB2")
+        self.assertEquals(done, "OK")
+        done = self.exec_naked("REDISQL.CREATE_STATEMENT", "DB1", "select1", "SELECT 1;")
+        self.assertEquals(done, "OK")
+
+        done = self.exec_naked("REDISQL.COPY", "DB1", "DB2")
+        result = self.exec_naked("REDISQL.QUERY_STATEMENT", "DB2", "select1")
+        self.assertEquals(result, [[1L]])
+
 
 if __name__ == '__main__':
    unittest.main()