Add instance recv limiter in peer conn (#2027)

easytier/locales/app.yml

@@ -217,6 +217,9 @@ core_clap:
   foreign_relay_bps_limit:
     en: "the maximum bps limit for foreign network relay, default is no limit. unit: BPS (bytes per second)"
     zh-CN: "作为共享节点时，限制非本地网络的流量转发速率，默认无限制，单位 BPS （字节每秒）"
+  instance_recv_bps_limit:
+    en: "the maximum total receive bps limit for this instance, default is no limit. unit: BPS (bytes per second)"
+    zh-CN: "限制当前网络实例整体入站流量的总接收速率，默认无限制，单位 BPS （字节每秒）"
   tcp_whitelist:
     en: "tcp port whitelist. Supports single ports (80) and ranges (8000-9000)"
     zh-CN: "TCP 端口白名单。支持单个端口（80）和范围（8000-9000）"

easytier/src/common/config.rs

@@ -69,6 +69,7 @@ pub fn gen_default_flags() -> Flags {
 
         quic_listen_port: u32::MAX,
         need_p2p: false,
+        instance_recv_bps_limit: u64::MAX,
     }
 }
 

easytier/src/core.rs

@@ -560,6 +560,13 @@ struct NetworkOptions {
     )]
     foreign_relay_bps_limit: Option<u64>,
 
+    #[arg(
+        long,
+        env = "ET_INSTANCE_RECV_BPS_LIMIT",
+        help = t!("core_clap.instance_recv_bps_limit").to_string(),
+    )]
+    instance_recv_bps_limit: Option<u64>,
+
     #[arg(
         long,
         value_delimiter = ',',
@@ -1060,6 +1067,9 @@ impl NetworkOptions {
         f.foreign_relay_bps_limit = self
             .foreign_relay_bps_limit
             .unwrap_or(f.foreign_relay_bps_limit);
+        f.instance_recv_bps_limit = self
+            .instance_recv_bps_limit
+            .unwrap_or(f.instance_recv_bps_limit);
         f.multi_thread_count = self.multi_thread_count.unwrap_or(f.multi_thread_count);
         f.disable_relay_kcp = self.disable_relay_kcp.unwrap_or(f.disable_relay_kcp);
         f.disable_relay_quic = self.disable_relay_quic.unwrap_or(f.disable_relay_quic);

easytier/src/launcher.rs

@@ -826,6 +826,10 @@ impl NetworkConfig {
             flags.mtu = mtu as u32;
         }
 
+        if let Some(instance_recv_bps_limit) = self.instance_recv_bps_limit {
+            flags.instance_recv_bps_limit = instance_recv_bps_limit;
+        }
+
         if let Some(enable_private_mode) = self.enable_private_mode {
             flags.private_mode = enable_private_mode;
         }
@@ -978,6 +982,8 @@ impl NetworkConfig {
         result.disable_sym_hole_punching = Some(flags.disable_sym_hole_punching);
         result.enable_magic_dns = Some(flags.accept_dns);
         result.mtu = Some(flags.mtu as i32);
+        result.instance_recv_bps_limit =
+            (flags.instance_recv_bps_limit != u64::MAX).then_some(flags.instance_recv_bps_limit);
         result.enable_private_mode = Some(flags.private_mode);
 
         if flags.relay_network_whitelist == "*" {

easytier/src/peers/peer_conn.rs

@@ -1365,6 +1365,17 @@ impl PeerConn {
                 &format!("{}:recv", conn_info_for_instrument.network_name),
                 limiter_config.into(),
             ))
+        } else if self.global_ctx.get_flags().instance_recv_bps_limit != u64::MAX {
+            let limiter_config = LimiterConfig {
+                burst_rate: None,
+                bps: Some(self.global_ctx.get_flags().instance_recv_bps_limit),
+                fill_duration_ms: None,
+            };
+            Some(
+                self.global_ctx
+                    .token_bucket_manager()
+                    .get_or_create("instance:recv", limiter_config.into()),
+            )
         } else {
             None
         };

easytier/src/proto/api_manage.proto

@@ -87,6 +87,7 @@ message NetworkConfig {
   optional string credential_file = 57;
   optional bool lazy_p2p = 58;
   optional bool need_p2p = 59;
+  optional uint64 instance_recv_bps_limit = 60;
 }
 
 message PortForwardConfig {

easytier/src/proto/common.proto

@@ -73,6 +73,7 @@ message FlagsInConfig {
 
   bool lazy_p2p = 37;
   bool need_p2p = 38;
+  uint64 instance_recv_bps_limit = 39;
 }
 
 message RpcDescriptor {

easytier/src/tests/three_node.rs

@@ -1535,6 +1535,48 @@ pub async fn relay_bps_limit_test(#[values(100, 200, 400, 800)] bps_limit: u64)
     drop_insts(insts).await;
 }
 
+#[rstest::rstest]
+#[serial_test::serial]
+#[tokio::test]
+pub async fn instance_recv_bps_limit_test(#[values(100, 800)] bps_limit: u64) {
+    let insts = init_three_node_ex(
+        "tcp",
+        |cfg| {
+            if cfg.get_inst_name() == "inst2" {
+                let mut f = cfg.get_flags();
+                f.instance_recv_bps_limit = bps_limit * 1024;
+                cfg.set_flags(f);
+            }
+            cfg
+        },
+        false,
+    )
+    .await;
+
+    let tcp_listener = TcpTunnelListener::new("tcp://0.0.0.0:22223".parse().unwrap());
+    let tcp_connector = TcpTunnelConnector::new("tcp://10.144.144.3:22223".parse().unwrap());
+
+    let bps = _tunnel_bench_netns(
+        tcp_listener,
+        tcp_connector,
+        NetNS::new(Some("net_c".into())),
+        NetNS::new(Some("net_a".into())),
+    )
+    .await;
+
+    println!("bps: {}", bps);
+
+    let bps = bps as u64 / 1024;
+    assert!(
+        bps >= bps_limit - 50 && bps <= bps_limit + 50,
+        "bps: {}, bps_limit: {}",
+        bps,
+        bps_limit
+    );
+
+    drop_insts(insts).await;
+}
+
 async fn assert_try_direct_connect_err<C>(inst: &Instance, connector: C)
 where
     C: crate::tunnel::TunnelConnector + std::fmt::Debug,