()
| 60 | } |
| 61 | |
| 62 | func (i *waitItem) run() { |
| 63 | // startFetch and apply and trigger pending |
| 64 | i.r.peersLock.RLock() |
| 65 | defer i.r.peersLock.RUnlock() |
| 66 | |
| 67 | // check log existence |
| 68 | if l, err := i.r.wal.Get(i.index); err == nil { |
| 69 | i.set(l) |
| 70 | return |
| 71 | } |
| 72 | |
| 73 | var ( |
| 74 | req = &kt.FetchRequest{ |
| 75 | Instance: i.r.instanceID, |
| 76 | Index: i.index, |
| 77 | } |
| 78 | resp *kt.FetchResponse |
| 79 | err error |
| 80 | ) |
| 81 | |
| 82 | // fetch log |
| 83 | caller := i.r.WaiterNewCallerFunc(i.r.peers.Leader) |
| 84 | if pcaller, ok := caller.(*rpc.PersistentCaller); ok && pcaller != nil { |
| 85 | defer pcaller.Close() |
| 86 | } |
| 87 | for { |
| 88 | select { |
| 89 | case <-i.stopCh: |
| 90 | return |
| 91 | case <-time.After(i.r.logWaitTimeout): |
| 92 | } |
| 93 | |
| 94 | resp = new(kt.FetchResponse) |
| 95 | |
| 96 | if err = caller.Call(i.r.fetchRPCMethod, req, resp); err != nil { |
| 97 | log.WithFields(log.Fields{ |
| 98 | "index": i.index, |
| 99 | "instance": i.r.instanceID, |
| 100 | }).WithError(err).Debug("send fetch rpc failed") |
| 101 | continue |
| 102 | } else if resp.Log == nil { |
| 103 | log.WithFields(log.Fields{ |
| 104 | "index": i.index, |
| 105 | "instance": i.r.instanceID, |
| 106 | }).Debug("could not fetch log") |
| 107 | continue |
| 108 | } |
| 109 | |
| 110 | if err = i.r.followerApply(resp.Log, false); err != nil { |
| 111 | // apply log |
| 112 | log.WithFields(log.Fields{ |
| 113 | "index": i.index, |
| 114 | "instance": i.r.instanceID, |
| 115 | }).WithError(err).Debug("apply log failed") |
| 116 | continue |
| 117 | } |
| 118 | |
| 119 | return |
no test coverage detected