Fix goroutine leak in SubOutlet

[adriansr]

This fixes a goroutine leaking every time that a harvester is closed.

[ph]

I am a bit confused , we have a call to close(o.done) this will trigger the select below and call o.closeEventChannel

[ph]

The channel can still be close.

[ph]

OK, I got it,

OnEvent -> Acquire mutex
Out of bound close -> Block on mutex if on event is currently running.

[ph]

@adriansr I am a bit worried that we add one more synchronization mechanism to this object, we have the isOpen (atomic bool), the chClosed has some execution state, the mutex and the channels.

I think I am missing some context from the environment and I am just trying to understand what we want to prevent, it is multiple calls to `close(o.ch)?

[ph]

// This mutex prevents the event channel to be closed if OnEvent is
// still running.

From your comment this is what we want to stop? So we could drop the atomic bool and use isOpen instead of chClosed

[ph]

by "drop" I mean using a single mutex.

Changed after review

[adriansr]

I've submitted a simpler version addressing @ph comments.

To my understanding, the atomic bool is still necessary to prevent closing done twice.

[andrewkroh]

the atomic bool is still necessary to prevent closing done twice.

I didn't examine the code but based on the sounds of it this could be a good use case for sync.Once to prevent channel closing more than once.

[ph]

as @andrewkroh noted sync.Once could replace the semaphore isOpen.

[urso]

The open flag is also used in OnEvent. Once the outlet is closed, we MUST NOT run into the select block in OnEvent. A select is non-deterministic. Even if o.done is closed, we might end up sending something to o.ch. Depending on current state this can generate either a panic or some inconsistent state .

[ph]

@adriansr I've look at the failure on the CI, I think its one of our flaky test.

======================================================================
ERROR: Checks that states are properly removed after clean_inactive
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/travis/gopath/src/github.com/elastic/beats/filebeat/tests/system/test_registrar.py", line 805, in test_clean_inactive
    max_timeout=10)
  File "/Users/travis/gopath/src/github.com/elastic/beats/filebeat/tests/system/../../../libbeat/tests/system/beat/beat.py", line 325, in wait_until
    "Waited {} seconds.".format(max_timeout))
TimeoutError: Timeout waiting for 'cond' to be true. Waited 10 seconds.

@urso we can merge as is? WDYT?

[adriansr]

@ph just in case, I relaunched the failing job. Now it passed.

[ph]

@adriansr I've merged it, I will take a look as soon as possible to find why this test is not reliable.