grpc / grpc/grpc-rust

dynamic-load-balance channel can not work

Open
#2,257 5 comments 1 reaction 0 assignees View on GitHub
Dominant language
Rust
Stars
12.5k
Forks
1.3k
Avg merge
4d 7h
Merged PRs (30d)
24

Description

## Bug Report

### Version
v0.12.3

### Platform
x86_64-unknown-linux

### Crates

### Description

I use tonic as a grpc client with `Channel::balance_channel(10)`, when client call server(multi node) at the first time, it works, and then wait a moment (1 min) the tcp connection status is turned to `CLOSE_WAIT` (i.e. the server close the connection as thers is no other streams), and from now on, the later call to server failed to response like this:

```shell
2025-04-24 17:55:57.317958 ERROR grpc{trace_id="2125102617454885569858015d101d"}: src/grpc/handler.rs:47: err=Status { code: Cancelled, message: "operation was canceled", source: Some(tonic::transport::Error(Transport, hyper::Error(Canceled, "connection closed"))
) }
2025-04-24 17:55:57.318031 ERROR grpc{trace_id="2125102617454885569858015d101d"}: src/grpc/handler.rs:63: Status {
code: Cancelled,
message: "operation was canceled",
source: Some(
tonic::transport::Error(
Transport,
hyper::Error(
Canceled,
"connection closed",
),
),
),
}
```

TCP Status
![Image](https://github.com/user-attachments/assets/09436a2e-0024-42f5-9210-855a9319f5e8)

I don't know if there is something wrong with my usage. Please help confirm. The following is the creation of the channel, thank you.
```rust

fn create_channel(vipkey: K) -> ChannelType
where
K: AsRef + Send + 'static,
{
let (channel, tx) = Channel::balance_channel(10);
let channel = ServiceBuilder::new()
.timeout(Duration::from_secs(3))
.service(channel);

tokio::spawn(async move {
let mut prev_hosts: Vec> = vec![];
let vipkey = vipkey.as_ref();
loop {
let hosts = vs::srv_hosts(vipkey).await.unwrap_or_default();
let hosts: Vec<_> = hosts.into_iter().filter(|h| h.valid).collect();
for host in hosts.iter() {
if prev_hosts
.iter()
.any(|h| h.ip == host.ip && h.port == host.port)
{
continue;
}
let key = format!("http://{}:{}", host.ip.as_str(), host.port);
let ep = Endpoint::from_shared(key.clone()).unwrap();
let change = Change::Insert(key, ep);
if let Err(err) = tx.send(change).await {
error!("send change error: {:?}", err);
}
}

for host in prev_hosts.iter() {
if hosts.iter().any(|h| h.ip == host.ip && h.port == host.port) {
continue;
}
let key = format!("http://{}:{}", host.ip.as_str(), host.port);
let change = Change::Remove(key);
if let Err(err) = tx.send(change).await {
error!("send change error: {:?}", err);
}
}
prev_hosts = hosts;
tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
}
});

channel
}

```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.