Fix a race condition in the TCP input

[ph]

Pass the net.Conn object when creating the client instead of passing it to the
Handle() method that keep a reference to it. By doing this we do not
have to worry about read or write race over the internal field. The
client still need a reference to the connection when the out of bound
call to Close() is executed to make sure we are getting out of a
Read() call early.

Tested with :

while true; do go test -v -race; done

Found in #13021

[ph]

cc @vjsamuel for awareness, I break the new interface :(

[andrewkroh]

Does this fix #12982?

[vjsamuel]

thanks for the heads up.

[urso]

The default split handler stores the connection object only for shutdown. This is because the server expects the handler to close the connection as well. But what if we change the server instead.

type client struct {
  conn net.Conn
  handler ConnectionHandler
}

func (c *client) Stop() {
  c.handler.Stop()
  c.conn.Close()
}

...

  conn, err := s.Listener.Accept()
  ...
  handler := s.factory(*s.config)
  client := &client{ conn: conn, handler: handler}
  s.registerClient(client)

  ...
  go func() {
    ...

    client.handler.Handle(conn)

    ...
  }()

In order to reduce the house-keeping + reduce the interface of ConnectionHandler even further we can pass some shutdown context to the handler as well (no need for client type anymore).
For example:

type ConnectionHandler struct {
  Handle(CloseRef, net.Conn) error
}

type CloseRef struct {  // subset of context.Context btw.
  Done() <- chan struct
  Err() error
}

type server struct {
  closer closeRef
  wg sync.WaitGroup
  ...
}

type (s *server) Stop() {
  closer.Close()
}

   ...

   conn, err := s.Listener.Accept()
   ...
   handler := s.factory(*s.config)
   closeRef := withCloser(s.closer, conn) // shutdown on s.closer would close 'conn' and propagate shutdown.

  ...
  go func() {
    ...

    defer closeRef.Close() // close connection and unlink from signal propagation

    ...

    handler.Handle(closeRef, conn)

    ...
  }()

We have this use-case quite often. And we fail to implement it correctly every now and so often. Close a connection and tell some loop to quit. Would be nice if it's just part of setting up the actual go-routine, instead of forcing us to do extra house-keeping just to implement shutdown.

[ph]

@andrewkroh Yes this is exactly that.

@urso I see where you are going with that, I will add a commit to this PR with the suggestion.

[ph]

@urso I have added a closeRef/Closer in the code to make sure to propagate the close, seems to work fine.

[urso]

All in all I like how the closer bundles all complexity, while setup+shutdown handling become much more symmetric in code -> less bookeeping and biolerplate.

Next step: generalize closer into a generic common reusable concept.
But let's keep the PR as is for now and add an issue to clean this up later.

[urso]

if the splitter returns an error due to us closing the connection, shall we supress the error then and just return nil?

[ph]

This is currently what I do I suppress errors in that scenario, in the beats input, I've decided otherwise and it created a lot of confusion for the users.

[ph]

@urso I have made the interface public and cleaned up a bit so we can extract it without too much trouble.

[urso]

I don't see any code that can set c.err besides close. That is, one could replace the mutex with a simple atomic.Bool and implement the body like: if !c.active.Load() { return ErrClosed }; return nil.

[ph]

We still have to lock the children map though? I can revisit if when we move it to external repo.