sf_load_read_rule_config support default value

src/sf_binlog_writer.c

@@ -351,10 +351,11 @@ static int deal_record_by_version(SFBinlogWriterBuffer *wb)
     writer = wb->writer;
     distance = wb->version - writer->version_ctx.next;
     if (distance >= (writer->version_ctx.ring.size - 1)) {
-        logWarning("file: "__FILE__", line: %d, "
+        logWarning("file: "__FILE__", line: %d, subdir_name: %s, "
                 "current version: %"PRId64" is too large, "
                 "exceeds %"PRId64" + %d", __LINE__,
-                wb->version, writer->version_ctx.next,
+                writer->cfg.subdir_name, wb->version,
+                writer->version_ctx.next,
                 writer->version_ctx.ring.size - 1);
         repush_to_queue(writer->thread, wb);
         fc_sleep_ms(10);

src/sf_configs.c

@@ -94,13 +94,14 @@ void sf_net_retry_config_to_string(SFNetRetryConfig *net_retry_cfg,
             net_retry_cfg->network.interval_ms);
 }
 
-void sf_load_read_rule_config(SFDataReadRule *rule, IniFullContext *ini_ctx)
+void sf_load_read_rule_config_ex(SFDataReadRule *rule,
+        IniFullContext *ini_ctx, const SFDataReadRule def_rule)
 {
     char *read_rule;
     read_rule = iniGetStrValueEx(ini_ctx->section_name,
             "read_rule", ini_ctx->context, true);
     if (read_rule == NULL || *read_rule == '\0') {
-        *rule = sf_data_read_rule_any_available;
+        *rule = def_rule;
     } else if (strncasecmp(read_rule, "any", 3) == 0) {
         *rule = sf_data_read_rule_any_available;
     } else if (strncasecmp(read_rule, "slave", 5) == 0) {

src/sf_configs.h

@@ -94,7 +94,8 @@ static inline int sf_calc_next_retry_interval(SFNetRetryIntervalContext *ctx)
     return ctx->interval_ms;
 }
 
-void sf_load_read_rule_config(SFDataReadRule *rule, IniFullContext *ini_ctx);
+void sf_load_read_rule_config_ex(SFDataReadRule *rule,
+        IniFullContext *ini_ctx, const SFDataReadRule def_rule);
 
 static inline const char *sf_get_read_rule_caption(
         const SFDataReadRule read_rule)
@@ -111,6 +112,8 @@ static inline const char *sf_get_read_rule_caption(
     }
 }
 
+#define sf_load_read_rule_config(rule, ini_ctx) \
+    sf_load_read_rule_config_ex(rule, ini_ctx, sf_data_read_rule_master_only)
 
 #define SF_NET_RETRY_FINISHED(retry_times, counter, result)  \
         !((SF_IS_RETRIABLE_ERROR(result) && ((retry_times > 0 &&  \

src/sf_nio.c

@@ -308,6 +308,10 @@ int sf_nio_notify(struct fast_task_info *task, const int stage)
     if (!__sync_bool_compare_and_swap(&task->nio_stages.notify,
                 SF_NIO_STAGE_NONE, stage))
     {
+        logDebug("file: "__FILE__", line: %d, "
+                "current stage: %d != %d, skip set stage to %d",
+                __LINE__, __sync_fetch_and_sub(&task->nio_stages.notify, 0),
+                SF_NIO_STAGE_NONE, stage);
         return EAGAIN;
     }
 

src/sf_proto.c

@@ -75,8 +75,10 @@ int sf_check_response(ConnectionInfo *conn, SFResponseInfo *response,
                     response->error.length, result, STRERROR(result));
         }
     } else {
-        response->error.length = 0;
-        response->error.message[0] = '\0';
+        response->error.length = snprintf(response->error.message,
+                sizeof(response->error.message),
+                "response status %d, error info: %s",
+                response->header.status, STRERROR(response->header.status));
     }
 
     return response->header.status;

src/sf_proto.h

@@ -63,6 +63,12 @@
         int2buff(_body_len, (header)->body_len); \
     } while (0)
 
+#define SF_PROTO_SET_HEADER_EX(header, _cmd, _flags, _body_len) \
+    do {  \
+        SF_PROTO_SET_HEADER(header, _cmd, _body_len); \
+        short2buff(_flags, (header)->flags); \
+    } while (0)
+
 #define SF_PROTO_SET_RESPONSE_HEADER(proto_header, resp_header) \
     do {  \
         (proto_header)->cmd = (resp_header).cmd;       \