BinlogSyncer.Close() will be blocked if BinlogStreamer exists

[lxohi]

BinlogSyncer.wg will be Add(1) when creating BinlogStreamer. And BinlogSyncer.wg only will be decrease when an error comes to BinlogSyncer.onStream.
BinlogSyncer.Close() was waiting on wg.
So if I call BinlogSyncer.Close() when there is no error in BinlogSyncer.onStream, process will be blocked.

[siddontang]

The code in close sets a 100ms timeout here

if b.c != nil {
        b.c.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
    }

how about using b.c.Close() directly?

[lxohi]

        if b.c != nil {
                b.c.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
        }
        b.c.Close()

I tried to add b.c.Close() here in replication/binlogsyncer.go.
After adding this line, the process exit successfully.

[siddontang]

Em, SetReadDeadline will have no effect if reading some data in 100ms and then be cleared. Maybe Close is right here.
can you send me a PR to fix this?

[lxohi]

Sure, but I found another problem when verifying the fix code.
Even if I called b.c.Close() here, there is still a case that blocking on b.wg.Wait().
Because BinlogSyncer.parseEvent is sending BinlogEvent to a buffered channel(s.ch), so onStream method may be blocking on it.

[siddontang]

How about SetReadDeadline(0)?

[lxohi]

The point is BinlogSyncer.parseEvent thread may be already blocked on s.ch before BinlogSyncer.Close() was called.
As long as BinlogSyncer.Close() waiting on b.wg.Wait(), no matter how to close the TCP connection, onStream method on the other thread will not decrease the WaitGroup.

[siddontang]

@andwxh , can above PR fix this problem?

[lxohi]

Yeah! This time its completely fixed. Thank you.

[lxohi]

I wondered is it a good practice to use two variables (wg & stopCh) in Close(), in order to ensure onStream thread stopped. But it works now.