backend: handle LOADING the same way as MASTERDOWN
@@ -22,6 +22,7 @@ import (
const (
stateConnected = 1
stateDataStale = 2
+ stateLoading = 3
)
type BackendConn struct {
@@ -137,6 +138,55 @@ func (bc *BackendConn) KeepAlive() bool {
bc, bc.addr, bc.database)
}
}
+
+ case stateLoading:
+ m := &Request{}
+ m.Multi = []*redis.Resp{
+ redis.NewBulkBytes([]byte("INFO")),
+ }
+ m.Batch = &sync.WaitGroup{}
+ bc.PushBack(m)
+
+ keepAliveCallback <- func() {
+ m.Batch.Wait()
+ var err = func() error {
+ if err := m.Err; err != nil {
+ return err
+ }
+ switch resp := m.Resp; {
+ case resp == nil:
+ return ErrRespIsRequired
+ case resp.IsError():
+ return fmt.Errorf("bad info resp: %s", resp.Value)
+ case resp.IsBulkBytes():
+ var info = make(map[string]string)
+ for _, line := range strings.Split(string(resp.Value), "\n") {
+ kv := strings.SplitN(line, ":", 2)
+ if len(kv) != 2 {
+ continue
+ }
+ if key := strings.TrimSpace(kv[0]); key != "" {
+ info[key] = strings.TrimSpace(kv[1])
+ }
+ }
+ if info["loading"] == "1" {
+ return nil
+ }
+ if bc.state.CompareAndSwap(stateLoading, stateConnected) {
+ log.Warnf("backend conn [%p] to %s, db-%d state = Connected (keepalive)",
+ bc, bc.addr, bc.database)
+ }
+ return nil
+ default:
+ return fmt.Errorf("bad info resp: should be string, but got %s", resp.Type)
+ }
+ }()
+ if err != nil && bc.closed.IsFalse() {
+ log.WarnErrorf(err, "backend conn [%p] to %s, db-%d recover from Loading failed",
+ bc, bc.addr, bc.database)
+ }
+ }
+
}
return true
}
@@ -266,6 +316,7 @@ func (bc *BackendConn) run() {
}
var errMasterDown = []byte("MASTERDOWN")
+var errLoading = []byte("LOADING")
func (bc *BackendConn) loopReader(tasks <-chan *Request, c *redis.Conn, round int) (err error) {
defer func() {
@@ -288,6 +339,11 @@ func (bc *BackendConn) loopReader(tasks <-chan *Request, c *redis.Conn, round in
log.Warnf("backend conn [%p] to %s, db-%d state = DataStale, caused by 'MASTERDOWN'",
bc, bc.addr, bc.database)
}
+ case bytes.HasPrefix(resp.Value, errLoading):
+ if bc.state.CompareAndSwap(stateConnected, stateLoading) {
+ log.Warnf("backend conn [%p] to %s, db-%d state = Loading",
+ bc, bc.addr, bc.database)
+ }
}
}
bc.setResponse(r, resp, nil)
