This change fixes the semantics of Close to actually wait for in-flight requests before closing the ReadWriteCloser. (Previously, the Close method closed the ReadWriteCloser immediately, which I suspect is what led to many of the failures observed in golang/go#49387 and golang/go#46520.) It achieves this by explicitly tracking the number of in-flight requests, including requests with pending async responses, and explicitly rejecting new Call requests (while keeping the read loop open!) once Close has begun. To make it easier for me to reason about the request lifetimes, I reduced the number of long-lived goroutines from three to just one (the Read loop), with an additional Handler goroutine that runs only while the Handler queue is non-empty. Now, it is clearer (I hope!) that the number of in-flight async requests strictly decreases after Close has begun, even though the Read goroutine continues to read requests (and, importantly, responses) and to forward Notifications to the preempter. For golang/go#49387 For golang/go#46520 Change-Id: Idf5960f848108a7ced78c5382099c8692e9b181e Reviewed-on: https://go-review.googlesource.com/c/tools/+/388134 gopls-CI: kokoro <noreply+kokoro@google.com> Run-TryBot: Bryan Mills <bcmills@google.com> TryBot-Result: Gopher Robot <gobot@golang.org> Reviewed-by: Alan Donovan <adonovan@google.com>
138 lines
3.6 KiB
Go
138 lines
3.6 KiB
Go
// Copyright 2018 The Go MCP SDK Authors. All rights reserved.
|
|
// Use of this source code is governed by an MIT-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package jsonrpc2
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"net"
|
|
"os"
|
|
)
|
|
|
|
// This file contains implementations of the transport primitives that use the standard network
|
|
// package.
|
|
|
|
// NetListenOptions is the optional arguments to the NetListen function.
|
|
type NetListenOptions struct {
|
|
NetListenConfig net.ListenConfig
|
|
NetDialer net.Dialer
|
|
}
|
|
|
|
// NetListener returns a new Listener that listens on a socket using the net package.
|
|
func NetListener(ctx context.Context, network, address string, options NetListenOptions) (Listener, error) {
|
|
ln, err := options.NetListenConfig.Listen(ctx, network, address)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &netListener{net: ln}, nil
|
|
}
|
|
|
|
// netListener is the implementation of Listener for connections made using the net package.
|
|
type netListener struct {
|
|
net net.Listener
|
|
}
|
|
|
|
// Accept blocks waiting for an incoming connection to the listener.
|
|
func (l *netListener) Accept(context.Context) (io.ReadWriteCloser, error) {
|
|
return l.net.Accept()
|
|
}
|
|
|
|
// Close will cause the listener to stop listening. It will not close any connections that have
|
|
// already been accepted.
|
|
func (l *netListener) Close() error {
|
|
addr := l.net.Addr()
|
|
err := l.net.Close()
|
|
if addr.Network() == "unix" {
|
|
rerr := os.Remove(addr.String())
|
|
if rerr != nil && err == nil {
|
|
err = rerr
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
|
|
// Dialer returns a dialer that can be used to connect to the listener.
|
|
func (l *netListener) Dialer() Dialer {
|
|
return NetDialer(l.net.Addr().Network(), l.net.Addr().String(), net.Dialer{})
|
|
}
|
|
|
|
// NetDialer returns a Dialer using the supplied standard network dialer.
|
|
func NetDialer(network, address string, nd net.Dialer) Dialer {
|
|
return &netDialer{
|
|
network: network,
|
|
address: address,
|
|
dialer: nd,
|
|
}
|
|
}
|
|
|
|
type netDialer struct {
|
|
network string
|
|
address string
|
|
dialer net.Dialer
|
|
}
|
|
|
|
func (n *netDialer) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
|
|
return n.dialer.DialContext(ctx, n.network, n.address)
|
|
}
|
|
|
|
// NetPipeListener returns a new Listener that listens using net.Pipe.
|
|
// It is only possibly to connect to it using the Dialer returned by the
|
|
// Dialer method, each call to that method will generate a new pipe the other
|
|
// side of which will be returned from the Accept call.
|
|
func NetPipeListener(ctx context.Context) (Listener, error) {
|
|
return &netPiper{
|
|
done: make(chan struct{}),
|
|
dialed: make(chan io.ReadWriteCloser),
|
|
}, nil
|
|
}
|
|
|
|
// netPiper is the implementation of Listener build on top of net.Pipes.
|
|
type netPiper struct {
|
|
done chan struct{}
|
|
dialed chan io.ReadWriteCloser
|
|
}
|
|
|
|
// Accept blocks waiting for an incoming connection to the listener.
|
|
func (l *netPiper) Accept(context.Context) (io.ReadWriteCloser, error) {
|
|
// Block until the pipe is dialed or the listener is closed,
|
|
// preferring the latter if already closed at the start of Accept.
|
|
select {
|
|
case <-l.done:
|
|
return nil, errClosed
|
|
default:
|
|
}
|
|
select {
|
|
case rwc := <-l.dialed:
|
|
return rwc, nil
|
|
case <-l.done:
|
|
return nil, errClosed
|
|
}
|
|
}
|
|
|
|
// Close will cause the listener to stop listening. It will not close any connections that have
|
|
// already been accepted.
|
|
func (l *netPiper) Close() error {
|
|
// unblock any accept calls that are pending
|
|
close(l.done)
|
|
return nil
|
|
}
|
|
|
|
func (l *netPiper) Dialer() Dialer {
|
|
return l
|
|
}
|
|
|
|
func (l *netPiper) Dial(ctx context.Context) (io.ReadWriteCloser, error) {
|
|
client, server := net.Pipe()
|
|
|
|
select {
|
|
case l.dialed <- server:
|
|
return client, nil
|
|
|
|
case <-l.done:
|
|
client.Close()
|
|
server.Close()
|
|
return nil, errClosed
|
|
}
|
|
}
|