Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

rafthttp: report unreachable when dropping messages

Browse Source
This commit is contained in:
Yicheng Qin 2015-03-10 22:47:59 -07:00
parent b1ff6ddd88
commit a230003255
2 changed files with 5 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
rafthttp/peer.go
View File

@ -91,6 +91,7 @@ type Peer interface {
type peer struct { type peer struct {
// id of the remote raft peer node // id of the remote raft peer node
id types.ID id types.ID
r Raft
msgAppWriter *streamWriter msgAppWriter *streamWriter
writer *streamWriter writer *streamWriter
@ -113,6 +114,7 @@ func startPeer(tr http.RoundTripper, urls types.URLs, local, to, cid types.ID, r
picker := newURLPicker(urls) picker := newURLPicker(urls)
p := &peer{ p := &peer{
id: to, id: to,
r: r,
msgAppWriter: startStreamWriter(fs, r), msgAppWriter: startStreamWriter(fs, r),
writer: startStreamWriter(fs, r), writer: startStreamWriter(fs, r),
pipeline: newPipeline(tr, picker, to, cid, fs, r, errorc), pipeline: newPipeline(tr, picker, to, cid, fs, r, errorc),
@ -156,6 +158,7 @@ func startPeer(tr http.RoundTripper, urls types.URLs, local, to, cid types.ID, r
select { select {
case writec <- m: case writec <- m:
default: default:
p.r.ReportUnreachable(m.To)
log.Printf("peer: dropping %s to %s since %s with %d-size buffer is blocked", log.Printf("peer: dropping %s to %s since %s with %d-size buffer is blocked",
m.Type, p.id, name, bufSizeMap[name]) m.Type, p.id, name, bufSizeMap[name])
} }

2
rafthttp/stream.go
View File

@ -116,6 +116,8 @@ func (cw *streamWriter) run() {
if m.Term > msgAppTerm { if m.Term > msgAppTerm {
cw.resetCloser() cw.resetCloser()
heartbeatc, msgc = nil, nil heartbeatc, msgc = nil, nil
// TODO: report to raft at peer level
cw.r.ReportUnreachable(m.To)
} }
continue continue
} }