Bugfix: fixed the following bugs (#221)

- fixed the bug that the service may fail to respond to the active packet on the server side under fiber
- fixed the bug that data loss in LZ4 compression
- fixed the bug that LockFreeQueue queue size may overflow
- fixed the bug that gdb_fiber_plugin.py script execution failures when encounter illegal characters

Author: chhy2009

trpc/compressor/lz4/lz4_util.cc

@@ -147,7 +147,7 @@ bool DoCompress(LZ4F_compressionContext_t& ctx, const NoncontiguousBuffer& in, N
         TRPC_FMT_ERROR("CompressedToOutputStream error, compressed_size={}", compressed_size);
         return false;
       }
-      left_to_copy = -current_size;
+      left_to_copy -= current_size;
       current_pos += current_size;
     }
   }

trpc/runtime/iomodel/reactor/fiber/BUILD

@@ -150,6 +150,7 @@ cc_test(
         "//trpc/runtime:fiber_runtime",
         "//trpc/runtime/iomodel/reactor/common:default_io_handler",
         "//trpc/util:latch",
+	"//trpc/util:time",
         "@com_google_googletest//:gtest",
         "@com_google_googletest//:gtest_main",
     ],

trpc/runtime/iomodel/reactor/fiber/fiber_tcp_connection.cc

@@ -36,9 +36,13 @@ FiberTcpConnection::~FiberTcpConnection() {
 }
 
 void FiberTcpConnection::Established() {
-  SetEstablishTimestamp(trpc::time::GetMilliSeconds());
-  SetConnectionState(ConnectionState::kConnected);
-  SetConnActiveTime(trpc::time::GetMilliSeconds());
+  if (IsClient()) {
+    // The server's connection status are set separately before calling Established
+    auto now_ms = trpc::time::GetMilliSeconds();
+    SetEstablishTimestamp(now_ms);
+    SetConnectionState(ConnectionState::kConnected);
+    SetConnActiveTime(now_ms);
+  }
 
   TRPC_ASSERT(GetIoHandler());
   TRPC_ASSERT(GetConnectionHandler());

trpc/runtime/iomodel/reactor/fiber/fiber_tcp_connection_test.cc

@@ -28,6 +28,7 @@
 #include "trpc/runtime/iomodel/reactor/fiber/fiber_reactor.h"
 #include "trpc/util/latch.h"
 #include "trpc/util/net_util.h"
+#include "trpc/util/time.h"
 
 namespace trpc {
 
@@ -105,6 +106,10 @@ class FiberTcpConnectionTestImpl {
       auto io_handle = std::make_unique<DefaultIoHandler>(server_conn_.Get());
       server_conn_->SetIoHandler(std::move(io_handle));
 
+      auto now_ms = trpc::time::GetMilliSeconds();
+      server_conn_->SetEstablishTimestamp(now_ms);
+      server_conn_->SetConnectionState(ConnectionState::kConnected);
+      server_conn_->SetConnActiveTime(now_ms);
       server_conn_->Established();
       server_conn_->StartHandshaking();
 

trpc/tools/gdb_plugin/gdb_fiber_plugin.py

@@ -287,8 +287,13 @@ def _get_memory_maps_in_core_dump():
             break  # TODO(luobogao): Parse executable sections.
 
         splited = seg.split()
-        start = int(splited[0], 16)
-        end = int(splited[2], 16)
+        try:
+            start = int(splited[0], 16)
+            end = int(splited[2], 16)
+        except ValueError:
+            # May encounter data like: While running this, GDB does not access memory
+            #skip them
+            break
         objfile = splited[-1] if not splited[-1].startswith('load') else ''
         segs.append(MemorySegment(start, end, objfile))
     return segs

trpc/transport/server/fiber/fiber_bind_adapter.cc

@@ -304,8 +304,8 @@ uint64_t FiberBindAdapter::GenConnectionId() {
   return (static_cast<uint64_t>(scheduling_group_index_) << 32) | ++conn_id;
 }
 
-void FiberBindAdapter::AddConnection(RefPtr<FiberTcpConnection>&& conn) {
-  connection_manager_.Add(conn->GetConnId(), std::move(conn));
+void FiberBindAdapter::AddConnection(const RefPtr<FiberTcpConnection>& conn) {
+  connection_manager_.Add(conn->GetConnId(), conn);
 }
 
 RefPtr<FiberTcpConnection> FiberBindAdapter::GetConnection(uint64_t conn_id) {
@@ -410,7 +410,7 @@ void FiberBindAdapter::DoClose(const CloseConnectionInfo& close_connection_info)
                    ", conn.ip:" << fiber_conn->GetPeerIp() << ", info.ip:" << close_connection_info.client_ip <<
                    ", conn.port:" << fiber_conn->GetPeerPort() << ", info.port:" << close_connection_info.client_port <<
                    ", conn.fd:" << fiber_conn->GetFd() << ", info.fd:" << close_connection_info.fd);
-    connection_manager_.Add(fiber_conn->GetConnId(), std::move(fiber_conn));
+    connection_manager_.Add(fiber_conn->GetConnId(), fiber_conn);
   }
 }
 

trpc/transport/server/fiber/fiber_bind_adapter.h

@@ -50,7 +50,7 @@ class FiberBindAdapter : public RefCounted<FiberBindAdapter> {
 
   uint64_t GenConnectionId();
 
-  void AddConnection(RefPtr<FiberTcpConnection>&& conn);
+  void AddConnection(const RefPtr<FiberTcpConnection>& conn);
 
   void UpdateConnection(Connection* conn) {}
 

trpc/transport/server/fiber/fiber_connection_manager.cc

@@ -37,11 +37,11 @@ FiberConnectionManager::~FiberConnectionManager() {
   }
 }
 
-void FiberConnectionManager::Add(uint64_t conn_id, RefPtr<FiberTcpConnection>&& conn) {
+void FiberConnectionManager::Add(uint64_t conn_id, const RefPtr<FiberTcpConnection>& conn) {
   auto&& shard = conn_shards_[GetHashIndex(conn_id, kShards)];
 
   std::scoped_lock _(shard.lock);
-  auto&& [it, inserted] = shard.map.emplace(conn_id, std::move(conn));
+  auto&& [it, inserted] = shard.map.insert(std::make_pair(conn_id, conn));
   (void)it;  // Suppresses compilation warnings.
 
   TRPC_ASSERT(inserted && "insert FiberConnectionManager with Duplicate conn_id");

trpc/transport/server/fiber/fiber_connection_manager.h

@@ -30,7 +30,7 @@ class FiberConnectionManager {
 
   ~FiberConnectionManager();
 
-  void Add(uint64_t conn_id, RefPtr<FiberTcpConnection>&& conn);
+  void Add(uint64_t conn_id, const RefPtr<FiberTcpConnection>& conn);
 
   RefPtr<FiberTcpConnection> Del(uint64_t conn_id);
 

trpc/transport/server/fiber/fiber_server_transport_impl.cc

@@ -27,6 +27,7 @@
 #include "trpc/transport/server/fiber/fiber_server_connection_handler_factory.h"
 #include "trpc/util/log/logging.h"
 #include "trpc/util/random.h"
+#include "trpc/util/time.h"
 
 namespace trpc {
 
@@ -159,12 +160,18 @@ bool FiberServerTransportImpl::AcceptConnection(AcceptConnectionInfo& connection
   conn_handler->Init();
 
   conn->SetConnectionHandler(std::move(conn_handler));
+  // The connection active time and status must be updated before executing AddConnection to avoid 
+  // the newly created connection being cleared as an idle connection, causing the assertion to fail.
+  auto now_ms = trpc::time::GetMilliSeconds();
+  conn->SetEstablishTimestamp(now_ms);
+  conn->SetConnectionState(ConnectionState::kConnected);
+  conn->SetConnActiveTime(now_ms);
+
+  bind_adapters_[scheduling_group_index]->AddConnection(conn);
 
   conn->Established();
   conn->StartHandshaking();
 
-  bind_adapters_[scheduling_group_index]->AddConnection(std::move(conn));
-
   FrameStats::GetInstance()->GetServerStats().AddConnCount(1);
 
   return true;