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

etcdhttp: non-blocking sender

Browse Source
This commit is contained in:
Xiang Li 2014-09-05 13:53:49 -07:00
parent 769c043537
commit cbec48e8f6
1 changed files with 32 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
etcdserver/etcdhttp/http.go
View File

@ -73,37 +73,41 @@ const DefaultTimeout = 500 * time.Millisecond
func Sender(p Peers) func(msgs []raftpb.Message) {
return func(msgs []raftpb.Message) {
for _, m := range msgs {
// TODO: create workers that deal with message sending
// concurrently as to not block progress
for {
url := p.Pick(m.To)
if url == "" {
// TODO: unknown peer id.. what do we do? I
// don't think his should ever happen, need to
// look into this further.
log.Println("etcdhttp: no addr for %d", m.To)
break
}
url += "/raft"
// TODO: don't block. we should be able to have 1000s
// of messages out at a time.
data, err := m.Marshal()
if err != nil {
log.Println("etcdhttp: dropping message:", err)
break // drop bad message
}
if httpPost(url, data) {
break // success
}
// TODO: backoff
}
// TODO: reuse go routines
// limit the number of outgoing connections for the same receiver
go send(p, m)
}
}
}
func send(p Peers, m raftpb.Message) {
// TODO (xiangli): reasonable retry logic
for i := 0; i < 3; i++ {
url := p.Pick(m.To)
if url == "" {
// TODO: unknown peer id.. what do we do? I
// don't think his should ever happen, need to
// look into this further.
log.Println("etcdhttp: no addr for %d", m.To)
return
}
url += "/raft"
// TODO: don't block. we should be able to have 1000s
// of messages out at a time.
data, err := m.Marshal()
if err != nil {
log.Println("etcdhttp: dropping message:", err)
return // drop bad message
}
if httpPost(url, data) {
return // success
}
// TODO: backoff
}
}
func httpPost(url string, data []byte) bool {
// TODO: set timeouts
resp, err := http.Post(url, "application/protobuf", bytes.NewBuffer(data))
@ -111,6 +115,7 @@ func httpPost(url string, data []byte) bool {
elog.TODO()
return false
}
resp.Body.Close()
if resp.StatusCode != 200 {
elog.TODO()
return false