From 024a6b0e8ae8b07a7f78ad45b2f7c1443f3c9aec Mon Sep 17 00:00:00 2001
From: YuQing <384681@qq.com>
Date: Thu, 27 Jul 2023 10:49:25 +0800
Subject: [PATCH] improve robustness of binlog writer flow control

---
 src/sf_binlog_writer.c | 6 ++++--
 src/sf_binlog_writer.h | 2 +-
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/sf_binlog_writer.c b/src/sf_binlog_writer.c
index 3de5770..f809070 100644
--- a/src/sf_binlog_writer.c
+++ b/src/sf_binlog_writer.c
@@ -632,8 +632,10 @@ void sf_push_to_binlog_write_queue(SFBinlogWriterInfo *writer,
     int64_t last_timestamp;
 
     last_timestamp = FC_ATOMIC_GET(writer->thread->flow_ctrol.last_timestamp);
-    if (last_timestamp > 0 && g_current_time - last_timestamp >
-            writer->thread->flow_ctrol.max_delay)
+    if ((last_timestamp > 0 && g_current_time - last_timestamp > writer->
+                thread->flow_ctrol.max_delay) && !(writer->order_by ==
+                    SF_BINLOG_WRITER_TYPE_ORDER_BY_VERSION && buffer->
+                    version.first - writer->version_ctx.next < 128))
     {
         time_t start_time;
         time_t last_log_timestamp;
diff --git a/src/sf_binlog_writer.h b/src/sf_binlog_writer.h
index 9d7c5b6..c01f13c 100644
--- a/src/sf_binlog_writer.h
+++ b/src/sf_binlog_writer.h
@@ -89,7 +89,7 @@ typedef struct sf_binlog_writer_info {
 
     struct {
         SFBinlogWriterBufferRing ring;
-        int64_t next;
+        volatile int64_t next;
         int64_t change_count;  //version change count
     } version_ctx;
     SFBinlogWriterThread *thread;