bug fix debug_level, and add free_handle to allow POOL_ALLOC free resource

Author: Terry-Mao

etc/slave.cf

@@ -17,7 +17,7 @@ log /tmp/rs_slave.log
 log.level 3
 
 # debug level
-debug.level BINLOG
+debug.level BINLOG|ALLOC
 
 [slave]
 # listen addr

src/core/rs_log.c

@@ -20,10 +20,8 @@ static rs_str_t rs_debug_levels[] = {
     rs_null_string
 };
 
-
 char        *rs_log_path = "./rs.log";
 int         rs_log_fd = STDOUT_FILENO;
-
 uint32_t    rs_log_level = RS_LOG_INFO;
 uint32_t    rs_debug_level = 0;
 
@@ -48,8 +46,11 @@ int rs_log_set_levels(char *debug_level)
 
     for(p = debug_level; p != NULL; p = rs_strchr(p, '|')) {
 
-        i = 0;
-        for(t = rs_debug_levels[i]; t.len > 0 && t.data != NULL; 
+        if(p != debug_level) {
+            p++;
+        }
+
+        for(i = 0, t = rs_debug_levels[i]; t.len > 0 && t.data != NULL; 
                 t = rs_debug_levels[++i])
         {
             if(rs_strncmp(t.data, p, t.len) == 0) {

src/slave/rs_binlog_row.c

@@ -385,6 +385,7 @@ int rs_dm_binlog_row(rs_slave_info_t *si, void *data, uint32_t len, char type,
         rs_binlog_dm_func before_update_handle,
         rs_binlog_dm_func update_handle,
         rs_binlog_dm_func delete_handle,
+        rs_binlog_dm_func free_handle,
         rs_dm_pos_alloc_t *pas_arr, void *obj)
 {
     char                        *p, *ubp, *dp;
@@ -463,6 +464,7 @@ int rs_dm_binlog_row(rs_slave_info_t *si, void *data, uint32_t len, char type,
             ubp = use_bits_after;
         }
 
+        /* before parse struct handle */
         if(before_parse_handle != NULL) {
             before_parse_handle(obj);
         }
@@ -546,11 +548,11 @@ int rs_dm_binlog_row(rs_slave_info_t *si, void *data, uint32_t len, char type,
             cmp += meta->meta_len;
         }
 
+        /* after parse struct handle */
         if(after_parse_handle != NULL) {
             after_parse_handle(obj);
         }
 
-        /* append cmd */
         if(type == RS_WRITE_ROWS_EVENT) {
             handle = write_handle;
         } else if(type == RS_UPDATE_ROWS_EVENT) {
@@ -566,11 +568,13 @@ int rs_dm_binlog_row(rs_slave_info_t *si, void *data, uint32_t len, char type,
             return RS_ERR;
         }
 
-        if(handle == NULL) {
-            continue;
+        /* dm handle */
+        if(handle != NULL && handle(si, obj) == RS_ERR) {
+            return RS_ERR;
         }
 
-        if(handle(si, obj) == RS_ERR) {
+        /* free resource handle */
+        if(free_handle != NULL && free_handle(si, obj) == RS_ERR) {
             return RS_ERR;
         }
     }

src/slave/rs_binlog_row.h

@@ -74,6 +74,7 @@ int rs_dm_binlog_row(rs_slave_info_t *si, void *data, uint32_t len, char type,
         rs_binlog_dm_func before_update_handle,
         rs_binlog_dm_func update_handle,
         rs_binlog_dm_func delete_handle,
+        rs_binlog_dm_func free_handle,
         rs_dm_pos_alloc_t *pas_arr, void *obj);
 
 #endif

src/slave/rs_io_thread.c

@@ -105,10 +105,10 @@ void *rs_start_io_thread(void *data)
             rs_log_error(RS_LOG_ERR, 0, "close failed()");
         }
         si->svr_fd = -1;
+        rs_log_error(RS_LOG_ERR, 0, "retry connect to master");
         sleep(RS_RETRY_CONNECT_SLEEP_SEC);
     }
 
-
 free:;
 
      pthread_cleanup_pop(1);

src/slave/rs_mysql_test_test.c

@@ -19,24 +19,25 @@ rs_dm_pos_alloc_t rs_mysql_test_pas[] = {
     rs_pstr_t_init(&(test->col))
 
 
-void rs_init_test_test(void *obj);
-void rs_print_test_test(void *obj);
+static void rs_init_test_test(void *obj);
+static void rs_print_test_test(void *obj);
 
 static int rs_insert_test_test(rs_slave_info_t *si, void *obj);
 static int rs_before_update_test_test(rs_slave_info_t *si, void *obj);
 static int rs_update_test_test(rs_slave_info_t *si, void *obj);
 static int rs_delete_test_test(rs_slave_info_t *si, void *obj);
+static int rs_free_test_test(rs_slave_info_t *si, void *obj);
 
 /* test */
-void rs_init_test_test(void *obj)
+static void rs_init_test_test(void *obj)
 {
     rs_mysql_test_t         *test;
     test = (rs_mysql_test_t *) obj;
 
     rs_mysql_test_t_init(test);
 }
 
-void rs_print_test_test(void *obj)
+static void rs_print_test_test(void *obj)
 {
     rs_mysql_test_t         *test;
 
@@ -55,7 +56,7 @@ void rs_print_test_test(void *obj)
             );
 }
 
-int rs_insert_test_test(rs_slave_info_t *si, void *obj)
+static int rs_insert_test_test(rs_slave_info_t *si, void *obj)
 {
     rs_mysql_test_t *test;
 
@@ -70,21 +71,33 @@ int rs_insert_test_test(rs_slave_info_t *si, void *obj)
     return RS_OK;
 }
 
-int rs_before_update_test_test(rs_slave_info_t *si, void *obj)
+static int rs_before_update_test_test(rs_slave_info_t *si, void *obj)
 {
     return RS_OK;
 }
 
-int rs_update_test_test(rs_slave_info_t *si, void *obj)
+static int rs_update_test_test(rs_slave_info_t *si, void *obj)
 {
     return rs_insert_test_test(si, obj);
 }
 
-int rs_delete_test_test(rs_slave_info_t *si, void *obj)
+static int rs_delete_test_test(rs_slave_info_t *si, void *obj)
 {
     return RS_OK;
 }
 
+static int rs_free_test_test(rs_slave_info_t *si, void *obj)
+{
+    rs_mysql_test_t *test;
+
+    test = (rs_mysql_test_t *) obj;
+
+    if(test->col.data != NULL) {
+        rs_pfree(si->dpool, test->col.data, test->col.id);
+    }
+
+    return RS_OK;
+}
 
 int rs_dm_test_test(rs_slave_info_t *si, char *r, uint32_t rl, char t) {
     /* test.test */
@@ -98,6 +111,7 @@ int rs_dm_test_test(rs_slave_info_t *si, char *r, uint32_t rl, char t) {
                 rs_before_update_test_test,
                 rs_update_test_test, 
                 rs_delete_test_test, 
+                rs_free_test_test,
                 rs_mysql_test_pas, &test);
 
     return err;