use reference to contenxt instead of whole context

Author: Simone Mosciatti

engine_pro

@@ -15,7 +15,7 @@ fn main() {
     cc::Build::new()
         .file("src/CDeps/Redis/redismodule.c")
         .include("src/CDeps/Redis/include")
-        .debug(true)
+        //.debug(true)
         .compile("libredismodule.a");
 
     cc::Build::new()

redisql_lib/build.rs

@@ -203,8 +203,8 @@ impl<'a> StatementCache<'a> for ReplicationBook {
 type ProtectedRedisContext =
     Arc<Mutex<Arc<Mutex<RefCell<Option<Context>>>>>>;
 
-pub struct RedisContextSet<'a>(
-    MutexGuard<'a, Arc<Mutex<RefCell<Option<Context>>>>>,
+pub struct RedisContextSet<'external>(
+    MutexGuard<'external, Arc<Mutex<RefCell<Option<Context>>>>>,
 );
 
 impl<'a> RedisContextSet<'a> {
@@ -219,6 +219,12 @@ impl<'a> RedisContextSet<'a> {
         }
         RedisContextSet(wrap)
     }
+    pub fn release(&self) -> Context {
+        let locked = self.0.lock().unwrap();
+        let none_to_ctx = RefCell::new(None);
+        locked.swap(&none_to_ctx);
+        none_to_ctx.into_inner().unwrap()
+    }
 }
 
 impl<'a> Drop for RedisContextSet<'a> {
@@ -291,11 +297,11 @@ impl Loop {
 }
 
 pub trait RedisReply {
-    fn reply(&self, ctx: rm::Context) -> i32;
+    fn reply(&self, ctx: &rm::Context) -> i32;
 }
 
 impl RedisReply for sql::Entity {
-    fn reply(&self, ctx: rm::Context) -> i32 {
+    fn reply(&self, ctx: &rm::Context) -> i32 {
         match *self {
             sql::Entity::Integer { int } => {
                 rm::ReplyWithLongLong(ctx, i64::from(int))
@@ -357,7 +363,7 @@ fn reply_with_done(
     rm::ffi::REDISMODULE_OK
 }
 
-fn reply_with_array(ctx: rm::Context, array: Vec<sql::Row>) -> i32 {
+fn reply_with_array(ctx: &rm::Context, array: Vec<sql::Row>) -> i32 {
     let len = array.len() as c_long;
     unsafe {
         rm::ffi::RedisModule_ReplyWithArray.unwrap()(
@@ -373,21 +379,21 @@ fn reply_with_array(ctx: rm::Context, array: Vec<sql::Row>) -> i32 {
             );
         }
         for entity in row {
-            entity.reply(ctx);
+            entity.reply(&ctx);
         }
     }
     rm::ffi::REDISMODULE_OK
 }
 
 impl RedisReply for sql::SQLite3Error {
-    fn reply(&self, ctx: Context) -> i32 {
+    fn reply(&self, ctx: &Context) -> i32 {
         let error = format!("{}", self);
         reply_with_error(ctx.as_ptr(), error)
     }
 }
 
 impl RedisReply for RediSQLError {
-    fn reply(&self, ctx: Context) -> i32 {
+    fn reply(&self, ctx: &Context) -> i32 {
         let error = format!("{}", self);
         reply_with_error(ctx.as_ptr(), error)
     }
@@ -502,7 +508,8 @@ pub enum QueryResult {
 }
 
 impl QueryResult {
-    pub fn reply(self, ctx: rm::Context) -> i32 {
+    pub fn reply(self, ctx: &rm::Context) -> i32 {
+        debug!("Start replying!");
         match self {
             QueryResult::OK { .. } => reply_with_ok(ctx.as_ptr()),
             QueryResult::DONE { modified_rows, .. } => {
@@ -629,16 +636,12 @@ pub fn listen_and_execute<'a, L: 'a + LoopData>(
         match rx.recv() {
             Ok(Command::Exec { query, client }) => {
                 debug!("Exec | Query = {:?}", query);
-
                 loopdata.with_contex_set(
                     Context::thread_safe(&client),
                     |_| {
-                        debug!("A");
                         let result =
                             do_execute(&loopdata.get_db(), query);
-                        debug!("B");
                         return_value(&client, result);
-                        debug!("C");
                     },
                 );
                 debug!("Exec | DONE, returning result");
@@ -997,8 +1000,9 @@ pub fn get_dbkeyptr_from_name(
     name: &str,
 ) -> Result<*mut DBKey, i32> {
     let context = Context::new(ctx);
-    let key_name = rm::RMString::new(context, name);
-    let key = OpenKey(context, &key_name, rm::ffi::REDISMODULE_WRITE);
+    let key_name = rm::RMString::new(&context, name);
+    let key =
+        OpenKey(&context, &key_name, rm::ffi::REDISMODULE_WRITE);
     let safe_key = RedisKey { key };
     let key_type = unsafe {
         rm::ffi::RedisModule_KeyType.unwrap()(safe_key.key)
@@ -1067,7 +1071,7 @@ pub fn reply_with_error_from_key_type(
     ctx: *mut rm::ffi::RedisModuleCtx,
     key_type: i32,
 ) -> i32 {
-    let context = Context::new(ctx);
+    let context = &Context::new(ctx);
     match key_type {
         rm::ffi::REDISMODULE_KEYTYPE_EMPTY => {
             ReplyWithError(context, "ERR - Error the key is empty\0")
@@ -1111,15 +1115,15 @@ fn remove_statement(
 
 #[allow(non_snake_case)]
 pub unsafe fn Replicate(
-    _ctx: rm::Context,
+    _ctx: &rm::Context,
     _command: &str,
     _argv: *mut *mut rm::ffi::RedisModuleString,
     _argc: std::os::raw::c_int,
 ) {
 }
 
 pub fn register_function(
-    context: rm::Context,
+    context: &rm::Context,
     name: String,
     flags: String,
     f: extern "C" fn(
@@ -1139,7 +1143,7 @@ pub fn register_function(
 }
 
 pub fn register_write_function(
-    ctx: rm::Context,
+    ctx: &rm::Context,
     name: String,
     f: extern "C" fn(
         *mut rm::ffi::RedisModuleCtx,

redisql_lib/src/redis.rs

@@ -13,17 +13,20 @@ pub mod ffi {
     include!(concat!(env!("OUT_DIR"), "/bindings_redis.rs"));
 }
 
-#[allow(dead_code)]
-#[derive(Copy, Clone, Debug)]
+#[derive(Debug)]
 pub struct Context {
     ctx: *mut ffi::RedisModuleCtx,
+    thread_safe: bool,
 }
 
 impl Context {
     pub fn new(ctx: *mut ffi::RedisModuleCtx) -> Context {
-        Context { ctx }
+        Context {
+            ctx,
+            thread_safe: false,
+        }
     }
-    pub fn as_ptr(self) -> *mut ffi::RedisModuleCtx {
+    pub fn as_ptr(&self) -> *mut ffi::RedisModuleCtx {
         self.ctx
     }
     pub fn thread_safe(blocked_client: &BlockedClient) -> Context {
@@ -32,19 +35,43 @@ impl Context {
                 blocked_client.as_ptr(),
             )
         };
-        Context { ctx }
+        Context {
+            ctx,
+            thread_safe: true,
+        }
+    }
+    pub fn lock(&self) {
+        unsafe {
+            ffi::RedisModule_ThreadSafeContextLock.unwrap()(
+                self.as_ptr(),
+            );
+        }
+    }
+    pub fn release(&self) {
+        unsafe {
+            ffi::RedisModule_ThreadSafeContextUnlock.unwrap()(
+                self.as_ptr(),
+            );
+        }
     }
 }
 
-unsafe impl Send for Context {}
-unsafe impl Sync for Context {}
-
-impl From<Context> for *mut ffi::RedisModuleCtx {
-    fn from(c: Context) -> *mut ffi::RedisModuleCtx {
-        c.ctx
+impl Drop for Context {
+    fn drop(&mut self) {
+        if self.thread_safe {
+            debug!("Free thread safe context");
+            unsafe {
+                ffi::RedisModule_FreeThreadSafeContext.unwrap()(
+                    self.as_ptr(),
+                );
+            }
+        }
     }
 }
 
+unsafe impl Send for Context {}
+unsafe impl Sync for Context {}
+
 pub struct BlockedClient {
     pub client: *mut ffi::RedisModuleBlockedClient,
 }
@@ -58,13 +85,13 @@ impl BlockedClient {
 }
 
 #[derive(Debug)]
-pub struct RMString {
+pub struct RMString<'a> {
     ptr: *mut ffi::RedisModuleString,
-    ctx: Context,
+    ctx: &'a Context,
 }
 
-impl RMString {
-    pub fn new(ctx: Context, s: &str) -> RMString {
+impl<'a> RMString<'a> {
+    pub fn new(ctx: &'a Context, s: &str) -> RMString<'a> {
         let ptr = unsafe {
             ffi::RedisModule_CreateString.unwrap()(
                 ctx.as_ptr(),
@@ -79,7 +106,7 @@ impl RMString {
     }
 }
 
-impl Drop for RMString {
+impl<'a> Drop for RMString<'a> {
     fn drop(&mut self) {
         unsafe {
             ffi::RedisModule_FreeString.unwrap()(
@@ -92,7 +119,7 @@ impl Drop for RMString {
 
 #[allow(non_snake_case)]
 pub fn CreateCommand(
-    ctx: Context,
+    ctx: &Context,
     name: String,
     f: ffi::RedisModuleCmdFunc,
     flags: String,
@@ -117,15 +144,15 @@ pub fn CreateCommand(
 }
 
 #[allow(non_snake_case)]
-pub fn ReplicateVerbatim(ctx: Context) -> i32 {
+pub fn ReplicateVerbatim(ctx: &Context) -> i32 {
     unsafe {
         ffi::RedisModule_ReplicateVerbatim.unwrap()(ctx.as_ptr())
     }
 }
 
 #[allow(non_snake_case)]
 pub unsafe fn Replicate(
-    ctx: Context,
+    ctx: &Context,
     command: &str,
     argv: *mut *mut ffi::RedisModuleString,
     argc: c_int,
@@ -142,7 +169,7 @@ pub unsafe fn Replicate(
 }
 
 #[allow(non_snake_case)]
-pub fn ReplyWithError(ctx: Context, error: &str) -> i32 {
+pub fn ReplyWithError(ctx: &Context, error: &str) -> i32 {
     unsafe {
         ffi::RedisModule_ReplyWithError.unwrap()(
             ctx.as_ptr(),
@@ -153,7 +180,7 @@ pub fn ReplyWithError(ctx: Context, error: &str) -> i32 {
 
 #[allow(non_snake_case)]
 pub fn OpenKey(
-    ctx: Context,
+    ctx: &Context,
     name: &RMString,
     mode: i32,
 ) -> *mut ffi::RedisModuleKey {
@@ -192,26 +219,26 @@ pub unsafe fn SaveStringBuffer(
 }
 
 #[allow(non_snake_case)]
-pub fn ReplyWithNull(ctx: Context) -> i32 {
+pub fn ReplyWithNull(ctx: &Context) -> i32 {
     unsafe { ffi::RedisModule_ReplyWithNull.unwrap()(ctx.as_ptr()) }
 }
 
 #[allow(non_snake_case)]
-pub fn ReplyWithLongLong(ctx: Context, ll: i64) -> i32 {
+pub fn ReplyWithLongLong(ctx: &Context, ll: i64) -> i32 {
     unsafe {
         ffi::RedisModule_ReplyWithLongLong.unwrap()(ctx.as_ptr(), ll)
     }
 }
 
 #[allow(non_snake_case)]
-pub fn ReplyWithDouble(ctx: Context, dd: f64) -> i32 {
+pub fn ReplyWithDouble(ctx: &Context, dd: f64) -> i32 {
     unsafe {
         ffi::RedisModule_ReplyWithDouble.unwrap()(ctx.as_ptr(), dd)
     }
 }
 
 #[allow(non_snake_case)]
-pub fn ReplyWithStringBuffer(ctx: Context, buffer: &[u8]) -> i32 {
+pub fn ReplyWithStringBuffer(ctx: &Context, buffer: &[u8]) -> i32 {
     let ptr = buffer.as_ptr() as *const c_char;
     let len = buffer.len();
     unsafe {