(entriesChannel chan<- *common.EntryContext)
| 945 | } |
| 946 | |
| 947 | func (b *BinlogReader) DataStreamEvents(entriesChannel chan<- *common.EntryContext) error { |
| 948 | bigTxThrottlingCount := 0 |
| 949 | for { |
| 950 | if b.shutdown { |
| 951 | break |
| 952 | } |
| 953 | |
| 954 | b.logger.Trace("b.HasBigTx.Wait. before") |
| 955 | // Throttle if this job has un-acked big tx, or |
| 956 | // there are too much global jobs with big tx. |
| 957 | for !b.shutdown { |
| 958 | localCount := atomic.LoadInt32(&b.BigTxCount) |
| 959 | globalLimit := g.BigTxReachMax() |
| 960 | if localCount <= b.mysqlContext.BigTxSrcQueue && !globalLimit { |
| 961 | bigTxThrottlingCount = 0 |
| 962 | break |
| 963 | } |
| 964 | |
| 965 | bigTxThrottlingCount += 1 |
| 966 | sleepMs := 10 |
| 967 | if bigTxThrottlingCount%(1000/sleepMs) == 0 { |
| 968 | // Force to read an event every 1000ms. |
| 969 | break |
| 970 | } |
| 971 | time.Sleep(time.Duration(sleepMs) * time.Millisecond) |
| 972 | if bigTxThrottlingCount * sleepMs >= 15 * 1000 { |
| 973 | b.logger.Warn("reader big tx throttling for 15s", "local", localCount, "global", globalLimit) |
| 974 | bigTxThrottlingCount = 0 |
| 975 | } |
| 976 | } |
| 977 | b.logger.Trace("b.HasBigTx.Wait. after") |
| 978 | |
| 979 | ev, err := b.binlogStreamer.GetEvent(b.ctx) |
| 980 | if err != nil { |
| 981 | b.logger.Error("error GetEvent.", "err", err) |
| 982 | return err |
| 983 | } |
| 984 | for g.IsLowMemory() { |
| 985 | time.Sleep(900 * time.Millisecond) |
| 986 | } |
| 987 | if b.shutdown { |
| 988 | return nil |
| 989 | } |
| 990 | |
| 991 | if ev.Header.EventType == replication.HEARTBEAT_EVENT { |
| 992 | continue |
| 993 | } |
| 994 | |
| 995 | b.currentCoordMutex.Lock() |
| 996 | b.currentCoord.LogPos = int64(ev.Header.LogPos) |
| 997 | b.currentCoordMutex.Unlock() |
| 998 | |
| 999 | if ev.Header.EventType == replication.ROTATE_EVENT { |
| 1000 | serverUUID, err := sql.GetServerUUID(b.db) |
| 1001 | if err != nil { |
| 1002 | return errors.Wrap(err, "on rotate_event. GetServerUUID") |
| 1003 | } |
| 1004 | if serverUUID != b.serverUUID { |
no test coverage detected