Commit8e61cac

committed

fix: Race when writing to a closed pipe

This is such an intermittent race it's difficult to track,but regardless this is an improvement to the code.

1 parent9b19dc9 commit8e61cacCopy full SHA for 8e61cac

File tree

+23

-8

lines changed

+23

-8

lines changed

Lines changed: 6 additions & 3 deletions

Original file line number	Diff line number	Diff line change
`@@ -31,9 +31,9 @@ func Serve(ctx context.Context, server proto.DRPCProvisionerServer, options *Ser`
`31`	`31`	`ifoptions.Listener==nil {`
`32`	`32`	`config:=yamux.DefaultConfig()`
`33`	`33`	`config.LogOutput=io.Discard`
`34`		`-stdio,err:=yamux.Server(readWriteCloser{`
`35`		`-ReadCloser:os.Stdin,`
`36`		`-Writer:os.Stdout,`
	`34`	`+stdio,err:=yamux.Server(&readWriteCloser{`
	`35`	`+ReadCloser:os.Stdin,`
	`36`	`+WriteCloser:os.Stdout,`
`37`	`37`	`},config)`
`38`	`38`	`iferr!=nil {`
`39`	`39`	`returnxerrors.Errorf("create yamux: %w",err)`
`@@ -54,6 +54,9 @@ func Serve(ctx context.Context, server proto.DRPCProvisionerServer, options *Ser`
`54`	`54`	`// short-lived processes that can be executed concurrently.`
`55`	`55`	`err=srv.Serve(ctx,options.Listener)`
`56`	`56`	`iferr!=nil {`
	`57`	`+iferrors.Is(err,io.EOF) {`
	`58`	`+returnnil`
	`59`	`+}`
`57`	`60`	`iferrors.Is(err,context.Canceled) {`
`58`	`61`	`returnnil`
`59`	`62`	`}`

Lines changed: 17 additions & 5 deletions

Original file line number	Diff line number	Diff line change
`@@ -3,6 +3,7 @@ package provisionersdk`
`3`	`3`	`import (`
`4`	`4`	`"context"`
`5`	`5`	`"io"`
	`6`	`+"sync"`
`6`	`7`
`7`	`8`	`"github.com/hashicorp/yamux"`
`8`	`9`	`"storj.io/drpc"`
`@@ -22,16 +23,16 @@ func TransportPipe() (yamux.Session, yamux.Session) {`
`22`	`23`	`yamuxConfig:=yamux.DefaultConfig()`
`23`	`24`	`yamuxConfig.LogOutput=io.Discard`
`24`	`25`	`client,err:=yamux.Client(&readWriteCloser{`
`25`		`-ReadCloser:clientReader,`
`26`		`-Writer:serverWriter,`
	`26`	`+ReadCloser:clientReader,`
	`27`	`+WriteCloser:serverWriter,`
`27`	`28`	`},yamuxConfig)`
`28`	`29`	`iferr!=nil {`
`29`	`30`	`panic(err)`
`30`	`31`	`}`
`31`	`32`
`32`	`33`	`server,err:=yamux.Server(&readWriteCloser{`
`33`		`-ReadCloser:serverReader,`
`34`		`-Writer:clientWriter,`
	`34`	`+ReadCloser:serverReader,`
	`35`	`+WriteCloser:clientWriter,`
`35`	`36`	`},yamuxConfig)`
`36`	`37`	`iferr!=nil {`
`37`	`38`	`panic(err)`
`@@ -45,8 +46,19 @@ func Conn(session *yamux.Session) drpc.Conn {`
`45`	`46`	`}`
`46`	`47`
`47`	`48`	`typereadWriteCloserstruct {`
	`49`	`+closeMutex sync.Mutex`
`48`	`50`	`io.ReadCloser`
`49`		`-io.Writer`
	`51`	`+io.WriteCloser`
	`52`	`+}`
	`53`	`+`
	`54`	`+func (c*readWriteCloser)Close()error {`
	`55`	`+c.closeMutex.Lock()`
	`56`	`+deferc.closeMutex.Unlock()`
	`57`	`+err:=c.ReadCloser.Close()`
	`58`	`+iferr!=nil {`
	`59`	`+returnerr`
	`60`	`+}`
	`61`	`+returnc.WriteCloser.Close()`
`50`	`62`	`}`
`51`	`63`
`52`	`64`	`// Allows concurrent requests on a single dRPC connection.`

Comments

(0)