2015-01-25 06:19:16 +03:00
|
|
|
// Copyright 2015 CoreOS, Inc.
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
2014-11-16 21:21:05 +03:00
|
|
|
|
|
|
|
package rafthttp
|
|
|
|
|
|
|
|
import (
|
2015-02-06 20:12:32 +03:00
|
|
|
"fmt"
|
|
|
|
"log"
|
2014-11-16 21:21:05 +03:00
|
|
|
"net/http"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
|
|
|
"github.com/coreos/etcd/pkg/types"
|
2014-11-18 03:40:52 +03:00
|
|
|
"github.com/coreos/etcd/raft/raftpb"
|
2014-11-16 21:21:05 +03:00
|
|
|
)
|
|
|
|
|
|
|
|
const (
|
2014-11-22 03:42:12 +03:00
|
|
|
appRespBatchMs = 50
|
2014-12-10 09:52:00 +03:00
|
|
|
propBatchMs = 10
|
2014-11-24 10:37:54 +03:00
|
|
|
|
2015-02-13 09:23:10 +03:00
|
|
|
DialTimeout = time.Second
|
2014-11-24 10:37:54 +03:00
|
|
|
ConnReadTimeout = 5 * time.Second
|
|
|
|
ConnWriteTimeout = 5 * time.Second
|
2014-11-16 21:21:05 +03:00
|
|
|
)
|
|
|
|
|
2014-12-29 04:44:26 +03:00
|
|
|
type peer struct {
|
2015-01-04 07:02:43 +03:00
|
|
|
sync.Mutex
|
|
|
|
|
2014-12-14 20:50:59 +03:00
|
|
|
id types.ID
|
|
|
|
cid types.ID
|
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
tr http.RoundTripper
|
|
|
|
// the url this sender post to
|
|
|
|
u string
|
|
|
|
r Raft
|
|
|
|
fs *stats.FollowerStats
|
2014-11-19 00:53:14 +03:00
|
|
|
|
2014-12-10 09:52:00 +03:00
|
|
|
batcher *Batcher
|
|
|
|
propBatcher *ProposalBatcher
|
2014-11-19 00:53:14 +03:00
|
|
|
|
2015-02-05 20:40:46 +03:00
|
|
|
pipeline *pipeline
|
|
|
|
stream *stream
|
2014-12-15 07:00:32 +03:00
|
|
|
|
2015-02-06 20:12:32 +03:00
|
|
|
sendc chan raftpb.Message
|
|
|
|
updatec chan string
|
|
|
|
attachc chan *streamWriter
|
|
|
|
pausec chan struct{}
|
|
|
|
resumec chan struct{}
|
|
|
|
stopc chan struct{}
|
|
|
|
done chan struct{}
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
|
|
|
|
2015-01-03 07:00:29 +03:00
|
|
|
func NewPeer(tr http.RoundTripper, u string, id types.ID, cid types.ID, r Raft, fs *stats.FollowerStats, errorc chan error) *peer {
|
2015-02-06 20:12:32 +03:00
|
|
|
p := &peer{
|
2014-12-31 00:45:11 +03:00
|
|
|
id: id,
|
2015-02-05 20:40:46 +03:00
|
|
|
cid: cid,
|
2014-12-31 00:45:11 +03:00
|
|
|
tr: tr,
|
|
|
|
u: u,
|
|
|
|
r: r,
|
|
|
|
fs: fs,
|
2015-02-05 20:40:46 +03:00
|
|
|
pipeline: newPipeline(tr, u, id, cid, fs, errorc),
|
2014-12-31 00:45:11 +03:00
|
|
|
stream: &stream{},
|
|
|
|
batcher: NewBatcher(100, appRespBatchMs*time.Millisecond),
|
|
|
|
propBatcher: NewProposalBatcher(100, propBatchMs*time.Millisecond),
|
2015-02-06 20:12:32 +03:00
|
|
|
|
|
|
|
sendc: make(chan raftpb.Message),
|
|
|
|
updatec: make(chan string),
|
|
|
|
attachc: make(chan *streamWriter),
|
|
|
|
pausec: make(chan struct{}),
|
|
|
|
resumec: make(chan struct{}),
|
|
|
|
stopc: make(chan struct{}),
|
|
|
|
done: make(chan struct{}),
|
2014-12-31 00:45:11 +03:00
|
|
|
}
|
2015-02-06 20:12:32 +03:00
|
|
|
go p.run()
|
|
|
|
return p
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
2015-02-06 20:12:32 +03:00
|
|
|
func (p *peer) run() {
|
|
|
|
var paused bool
|
|
|
|
// non-blocking main loop
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case m := <-p.sendc:
|
|
|
|
if paused {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
p.send(m)
|
|
|
|
case u := <-p.updatec:
|
|
|
|
p.u = u
|
|
|
|
p.pipeline.update(u)
|
|
|
|
case sw := <-p.attachc:
|
|
|
|
sw.fs = p.fs
|
|
|
|
if err := p.stream.attach(sw); err != nil {
|
|
|
|
sw.stop()
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
go sw.handle()
|
|
|
|
case <-p.pausec:
|
|
|
|
paused = true
|
|
|
|
case <-p.resumec:
|
|
|
|
paused = false
|
|
|
|
case <-p.stopc:
|
|
|
|
p.pipeline.stop()
|
|
|
|
p.stream.stop()
|
|
|
|
close(p.done)
|
|
|
|
return
|
|
|
|
}
|
2015-01-04 07:02:43 +03:00
|
|
|
}
|
2014-11-16 21:21:05 +03:00
|
|
|
}
|
|
|
|
|
2015-02-06 20:12:32 +03:00
|
|
|
func (p *peer) Send(m raftpb.Message) {
|
|
|
|
select {
|
|
|
|
case p.sendc <- m:
|
|
|
|
case <-p.done:
|
|
|
|
log.Panicf("peer: unexpected stopped")
|
2015-01-04 07:02:43 +03:00
|
|
|
}
|
2015-02-06 20:12:32 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (p *peer) Update(u string) {
|
|
|
|
select {
|
|
|
|
case p.updatec <- u:
|
|
|
|
case <-p.done:
|
|
|
|
log.Panicf("peer: unexpected stopped")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// attachStream attaches a streamWriter to the peer.
|
|
|
|
// If attach succeeds, peer will take charge of the given streamWriter.
|
|
|
|
func (p *peer) attachStream(sw *streamWriter) error {
|
|
|
|
select {
|
|
|
|
case p.attachc <- sw:
|
2014-12-05 01:20:58 +03:00
|
|
|
return nil
|
2015-02-06 20:12:32 +03:00
|
|
|
case <-p.done:
|
|
|
|
return fmt.Errorf("peer: stopped")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Pause pauses the peer. The peer will simply drops all incoming
|
|
|
|
// messages without retruning an error.
|
|
|
|
func (p *peer) Pause() {
|
|
|
|
select {
|
|
|
|
case p.pausec <- struct{}{}:
|
|
|
|
case <-p.done:
|
2014-12-05 01:20:58 +03:00
|
|
|
}
|
2015-02-06 20:12:32 +03:00
|
|
|
}
|
2014-12-05 01:20:58 +03:00
|
|
|
|
2015-02-06 20:12:32 +03:00
|
|
|
// Resume resumes a paused peer.
|
|
|
|
func (p *peer) Resume() {
|
|
|
|
select {
|
|
|
|
case p.resumec <- struct{}{}:
|
|
|
|
case <-p.done:
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Stop performs any necessary finalization and terminates the peer
|
|
|
|
// elegantly.
|
|
|
|
func (p *peer) Stop() {
|
|
|
|
select {
|
|
|
|
case p.stopc <- struct{}{}:
|
|
|
|
case <-p.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
<-p.done
|
|
|
|
}
|
|
|
|
|
|
|
|
// send sends the data to the remote node. It is always non-blocking.
|
|
|
|
// It may be fail to send data if it returns nil error.
|
|
|
|
// TODO (xiangli): reasonable retry logic
|
|
|
|
func (p *peer) send(m raftpb.Message) error {
|
2014-12-31 00:45:11 +03:00
|
|
|
// move all the stream related stuff into stream
|
|
|
|
p.stream.invalidate(m.Term)
|
|
|
|
if shouldInitStream(m) && !p.stream.isOpen() {
|
|
|
|
u := p.u
|
2015-01-04 07:02:43 +03:00
|
|
|
// todo: steam open should not block.
|
2014-12-31 21:02:45 +03:00
|
|
|
p.stream.open(types.ID(m.From), p.id, p.cid, m.Term, p.tr, u, p.r)
|
2014-12-29 04:44:26 +03:00
|
|
|
p.batcher.Reset(time.Now())
|
2014-11-22 03:42:12 +03:00
|
|
|
}
|
2014-12-10 09:52:00 +03:00
|
|
|
|
|
|
|
var err error
|
|
|
|
switch {
|
|
|
|
case isProposal(m):
|
2014-12-29 04:44:26 +03:00
|
|
|
p.propBatcher.Batch(m)
|
2014-12-31 00:45:11 +03:00
|
|
|
case canBatch(m) && p.stream.isOpen():
|
2014-12-29 04:44:26 +03:00
|
|
|
if !p.batcher.ShouldBatch(time.Now()) {
|
2015-02-05 20:40:46 +03:00
|
|
|
err = p.pipeline.send(m)
|
2014-12-10 09:52:00 +03:00
|
|
|
}
|
|
|
|
case canUseStream(m):
|
2014-12-31 00:45:11 +03:00
|
|
|
if ok := p.stream.write(m); !ok {
|
2015-02-05 20:40:46 +03:00
|
|
|
err = p.pipeline.send(m)
|
2014-11-22 03:42:12 +03:00
|
|
|
}
|
2014-12-10 09:52:00 +03:00
|
|
|
default:
|
2015-02-05 20:40:46 +03:00
|
|
|
err = p.pipeline.send(m)
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
2014-12-10 09:52:00 +03:00
|
|
|
// send out batched MsgProp if needed
|
|
|
|
// TODO: it is triggered by all outcoming send now, and it needs
|
|
|
|
// more clear solution. Either use separate goroutine to trigger it
|
|
|
|
// or use streaming.
|
2014-12-29 04:44:26 +03:00
|
|
|
if !p.propBatcher.IsEmpty() {
|
2014-12-10 09:52:00 +03:00
|
|
|
t := time.Now()
|
2014-12-29 04:44:26 +03:00
|
|
|
if !p.propBatcher.ShouldBatch(t) {
|
2015-02-05 20:40:46 +03:00
|
|
|
p.pipeline.send(p.propBatcher.Message)
|
2014-12-29 04:44:26 +03:00
|
|
|
p.propBatcher.Reset(t)
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
|
|
|
}
|
2014-12-10 09:52:00 +03:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
func isProposal(m raftpb.Message) bool { return m.Type == raftpb.MsgProp }
|