pkg/loader: fix loader run while quitting

[lichunzhu]

What problem does this PR solve?

TOOL-1579
When loader meets some errors in loader.Run it will quit immediately. But if there is some other functions trying to syncing txns to loader.input they may get blocked. This problem is similar to #714.

What is changed and how it works?

Add select in mysql dsyncer. When loader finished loader.run, mysql.Error() will be activated which can avoid block.

Check List

Tests

• Unit test
• Integration test

Code changes

Side effects

Related changes

[lichunzhu]

/run-all-tests

[suzaku]

When loader.Run stopped, MySQLSyncer would call setErr, and the errCh channel would be closed regardless of whether the passed error is nil or not.
So maybe we can simply wrap the blocking loader.Input() <- with a select like the following to solve the problem.

diff --git a/drainer/sync/mysql.go b/drainer/sync/mysql.go
index c9067b3e..10343e3e 100644
--- a/drainer/sync/mysql.go
+++ b/drainer/sync/mysql.go
@@ -83,9 +83,12 @@ func (m *MysqlSyncer) Sync(item *Item) error {
 
 	txn.Metadata = item
 
-	m.loader.Input() <- txn
-
-	return nil
+	select {
+	case <-m.errCh:
+		return m.err
+	case m.loader.Input() <- txn:
+		return nil
+	}
 }
 
 // Close implements Syncer interface

[lichunzhu]

@suzaku Nice suggestion! It works for MysqlSyncer. But the farther developers may make the same mistakes again. Actually I think it's very easy to make mistakes here if some other developers want to reuse the loader.Input code. So shall we leave some comments or other things?

[suzaku]

One possible way is to encapsulate the enqueueing of txns with a method that may return an error if the loader's not in a consistent state.
So that users don't have to remember to use select when sending to the input channel.

Maybe we can do that in a separate PR.

For this problem, can we add a test case to prove that we have solved the hanging problem?

[lichunzhu]

/run-integration-tests tidb=release-3.0 tikv=release-3.0 pd=release-3.0

[lichunzhu]

/run-all-tests tidb=release-3.0 tikv=release-3.0 pd=release-3.0

[suzaku]

LGTM