igor-dv opened a new issue, #40260:
URL: https://github.com/apache/beam/issues/40260
### What happened?
We run Prism in a long-lived worker process where each job starts its own go
harness in that process.
The problem is that the harness caches Data and State gRPC streams on
`DataChannelManager` / `StateChannelManager` and never closes them so the
streams and their read goroutines stay. Over many jobs the process accumulates
connections and goroutines until it is unusable.
In addition to that, there is another problem that shows up on a long-lived
process when the harness sends output to the runner on a stream that another
goroutine is already reading. If the runner has closed that stream, the send
fails with EOF and then tries to read too, and those two reads deadlock.
Test to reproduce the leak:
```go
package harness
import (
"context"
"fmt"
"io"
"runtime"
"testing"
"time"
fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
)
type blockRecvClient struct{ block chan struct{} }
func (c *blockRecvClient) Recv() (*fnpb.Elements, error) { <-c.block; return
nil, io.EOF }
func (c *blockRecvClient) Send(*fnpb.Elements) error { return nil }
func TestDataChannelManagerLeak(t *testing.T) {
const jobs = 50
block := make(chan struct{})
defer close(block)
runtime.GC()
time.Sleep(50 * time.Millisecond)
before := runtime.NumGoroutine()
for i := 0; i < jobs; i++ {
// One harness per job. Control EOFs; MainWithOptions returns
without closing the data manager.
m := &DataChannelManager{}
ch := makeDataChannel(context.Background(), "data",
&blockRecvClient{block: block}, func() {})
m.ports = map[string]*DataChannel{"data": ch}
_ = m
}
time.Sleep(50 * time.Millisecond)
after := runtime.NumGoroutine()
t.Logf("jobs=%d goroutines before=%d after=%d delta=%d", jobs, before,
after, after-before)
if after-before < jobs/2 {
t.Fatalf("did not reproduce leak: before=%d after=%d", before, after)
}
}
```
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [x] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
