syncthing/protocol/protocol.go

445 lines
8.8 KiB
Go
Raw Normal View History

2013-12-15 10:43:31 +00:00
package protocol
import (
"compress/flate"
"errors"
"fmt"
2013-12-15 10:43:31 +00:00
"io"
2014-01-23 12:12:45 +00:00
"log"
2013-12-15 10:43:31 +00:00
"sync"
"time"
2013-12-15 10:43:31 +00:00
"github.com/calmh/syncthing/buffers"
2014-02-15 11:08:55 +00:00
"github.com/calmh/syncthing/xdr"
2013-12-15 10:43:31 +00:00
)
2014-02-20 16:40:15 +00:00
const BlockSize = 128 * 1024
2013-12-15 10:43:31 +00:00
const (
2013-12-28 13:10:36 +00:00
messageTypeIndex = 1
messageTypeRequest = 2
messageTypeResponse = 3
messageTypePing = 4
messageTypePong = 5
messageTypeIndexUpdate = 6
2014-01-23 12:12:45 +00:00
messageTypeOptions = 7
2013-12-15 10:43:31 +00:00
)
const (
2014-03-16 07:14:55 +00:00
FlagDeleted uint32 = 1 << 12
FlagInvalid = 1 << 13
)
var (
2014-02-24 12:29:30 +00:00
ErrClusterHash = fmt.Errorf("configuration error: mismatched cluster hash")
ErrClosed = errors.New("connection closed")
)
2013-12-15 10:43:31 +00:00
type Model interface {
// An index was received from the peer node
Index(nodeID string, files []FileInfo)
2013-12-28 13:10:36 +00:00
// An index update was received from the peer node
IndexUpdate(nodeID string, files []FileInfo)
2013-12-15 10:43:31 +00:00
// A request was made by the peer node
2014-02-20 16:40:15 +00:00
Request(nodeID, repo string, name string, offset int64, size int) ([]byte, error)
2013-12-15 10:43:31 +00:00
// The peer node closed the connection
Close(nodeID string, err error)
2013-12-15 10:43:31 +00:00
}
type Connection struct {
2013-12-24 16:45:16 +00:00
sync.RWMutex
2014-01-23 12:12:45 +00:00
id string
receiver Model
reader io.Reader
2014-02-20 16:40:15 +00:00
xr *xdr.Reader
2014-01-23 12:12:45 +00:00
writer io.Writer
2014-02-20 16:40:15 +00:00
xw *xdr.Writer
2014-01-23 12:12:45 +00:00
closed bool
awaiting map[int]chan asyncResult
2014-02-24 12:29:30 +00:00
nextID int
indexSent map[string]map[string][2]int64
peerOptions map[string]string
myOptions map[string]string
2014-01-23 12:12:45 +00:00
optionsLock sync.Mutex
2013-12-30 14:53:54 +00:00
hasSentIndex bool
hasRecvdIndex bool
2013-12-30 14:53:54 +00:00
statisticsLock sync.Mutex
2013-12-15 10:43:31 +00:00
}
2013-12-15 14:58:27 +00:00
type asyncResult struct {
val []byte
err error
}
2013-12-31 03:10:54 +00:00
const (
pingTimeout = 2 * time.Minute
pingIdleTime = 5 * time.Minute
)
2014-01-23 12:12:45 +00:00
func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver Model, options map[string]string) *Connection {
2013-12-15 10:43:31 +00:00
flrd := flate.NewReader(reader)
flwr, err := flate.NewWriter(writer, flate.BestSpeed)
if err != nil {
panic(err)
}
c := Connection{
id: nodeID,
receiver: receiver,
reader: flrd,
2014-02-20 16:40:15 +00:00
xr: xdr.NewReader(flrd),
writer: flwr,
2014-02-20 16:40:15 +00:00
xw: xdr.NewWriter(flwr),
awaiting: make(map[int]chan asyncResult),
indexSent: make(map[string]map[string][2]int64),
2013-12-15 10:43:31 +00:00
}
go c.readerLoop()
go c.pingerLoop()
2013-12-15 10:43:31 +00:00
2014-01-23 12:12:45 +00:00
if options != nil {
c.myOptions = options
2014-01-23 12:12:45 +00:00
go func() {
c.Lock()
2014-02-24 12:29:30 +00:00
header{0, c.nextID, messageTypeOptions}.encodeXDR(c.xw)
2014-02-20 16:40:15 +00:00
var om OptionsMessage
for k, v := range options {
om.Options = append(om.Options, Option{k, v})
}
om.encodeXDR(c.xw)
err := c.xw.Error()
if err == nil {
err = c.flush()
}
2014-01-23 12:12:45 +00:00
if err != nil {
log.Println("Warning: Write error during initial handshake:", err)
2014-01-23 12:12:45 +00:00
}
2014-02-24 12:29:30 +00:00
c.nextID++
2014-01-23 12:12:45 +00:00
c.Unlock()
}()
}
2013-12-15 10:43:31 +00:00
return &c
}
2014-01-09 12:58:35 +00:00
func (c *Connection) ID() string {
return c.id
}
2013-12-15 10:43:31 +00:00
// Index writes the list of file information to the connected peer node
func (c *Connection) Index(repo string, idx []FileInfo) {
2013-12-24 16:45:16 +00:00
c.Lock()
2013-12-28 13:10:36 +00:00
var msgType int
if c.indexSent[repo] == nil {
2013-12-28 13:10:36 +00:00
// This is the first time we send an index.
msgType = messageTypeIndex
c.indexSent[repo] = make(map[string][2]int64)
2013-12-28 13:10:36 +00:00
for _, f := range idx {
c.indexSent[repo][f.Name] = [2]int64{f.Modified, int64(f.Version)}
2013-12-28 13:10:36 +00:00
}
} else {
// We have sent one full index. Only send updates now.
msgType = messageTypeIndexUpdate
var diff []FileInfo
for _, f := range idx {
if vs, ok := c.indexSent[repo][f.Name]; !ok || f.Modified != vs[0] || int64(f.Version) != vs[1] {
2013-12-28 13:10:36 +00:00
diff = append(diff, f)
c.indexSent[repo][f.Name] = [2]int64{f.Modified, int64(f.Version)}
2013-12-28 13:10:36 +00:00
}
}
idx = diff
}
2014-02-24 12:29:30 +00:00
header{0, c.nextID, msgType}.encodeXDR(c.xw)
2014-02-20 16:40:15 +00:00
_, err := IndexMessage{repo, idx}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
2014-02-24 12:29:30 +00:00
c.nextID = (c.nextID + 1) & 0xfff
c.hasSentIndex = true
2013-12-24 16:45:16 +00:00
c.Unlock()
if err != nil {
2014-01-09 12:58:35 +00:00
c.close(err)
return
2013-12-19 23:01:34 +00:00
}
2013-12-15 10:43:31 +00:00
}
// Request returns the bytes for the specified block after fetching them from the connected peer.
2014-02-20 16:40:15 +00:00
func (c *Connection) Request(repo string, name string, offset int64, size int) ([]byte, error) {
2013-12-30 20:27:20 +00:00
c.Lock()
if c.closed {
c.Unlock()
2013-12-28 15:30:02 +00:00
return nil, ErrClosed
}
2013-12-15 14:58:27 +00:00
rc := make(chan asyncResult)
2014-02-24 12:29:30 +00:00
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypeRequest}.encodeXDR(c.xw)
2014-02-20 16:40:15 +00:00
_, err := RequestMessage{repo, name, uint64(offset), uint32(size)}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
2013-12-19 23:01:34 +00:00
}
if err != nil {
2013-12-24 16:45:16 +00:00
c.Unlock()
2014-01-09 12:58:35 +00:00
c.close(err)
2013-12-19 23:01:34 +00:00
return nil, err
}
2014-02-24 12:29:30 +00:00
c.nextID = (c.nextID + 1) & 0xfff
2013-12-24 16:45:16 +00:00
c.Unlock()
2013-12-15 10:43:31 +00:00
2013-12-15 14:58:27 +00:00
res, ok := <-rc
if !ok {
return nil, ErrClosed
2013-12-15 10:43:31 +00:00
}
2013-12-15 14:58:27 +00:00
return res.val, res.err
2013-12-15 10:43:31 +00:00
}
2014-01-09 12:58:35 +00:00
func (c *Connection) ping() bool {
2013-12-30 20:27:20 +00:00
c.Lock()
if c.closed {
c.Unlock()
2013-12-31 01:52:36 +00:00
return false
2013-12-28 15:30:02 +00:00
}
2013-12-30 20:27:20 +00:00
rc := make(chan asyncResult, 1)
2014-02-24 12:29:30 +00:00
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypePing}.encodeXDR(c.xw)
2013-12-19 23:01:34 +00:00
err := c.flush()
if err != nil {
c.Unlock()
2014-01-09 12:58:35 +00:00
c.close(err)
return false
2014-02-20 16:40:15 +00:00
} else if c.xw.Error() != nil {
2013-12-24 16:45:16 +00:00
c.Unlock()
2014-02-20 16:40:15 +00:00
c.close(c.xw.Error())
2013-12-31 01:52:36 +00:00
return false
2013-12-19 23:01:34 +00:00
}
2014-02-24 12:29:30 +00:00
c.nextID = (c.nextID + 1) & 0xfff
2013-12-24 16:45:16 +00:00
c.Unlock()
2013-12-15 10:43:31 +00:00
2013-12-31 01:52:36 +00:00
res, ok := <-rc
return ok && res.err == nil
2013-12-15 10:43:31 +00:00
}
type flusher interface {
Flush() error
}
2013-12-19 23:01:34 +00:00
func (c *Connection) flush() error {
2013-12-15 10:43:31 +00:00
if f, ok := c.writer.(flusher); ok {
2013-12-19 23:01:34 +00:00
return f.Flush()
2013-12-15 10:43:31 +00:00
}
2013-12-19 23:01:34 +00:00
return nil
2013-12-15 10:43:31 +00:00
}
2014-01-09 12:58:35 +00:00
func (c *Connection) close(err error) {
2013-12-24 16:45:16 +00:00
c.Lock()
2013-12-19 23:01:34 +00:00
if c.closed {
2013-12-24 16:45:16 +00:00
c.Unlock()
2013-12-19 23:01:34 +00:00
return
}
c.closed = true
2013-12-15 10:43:31 +00:00
for _, ch := range c.awaiting {
close(ch)
}
c.awaiting = nil
2013-12-24 16:45:16 +00:00
c.Unlock()
2013-12-15 14:58:27 +00:00
2014-01-09 12:58:35 +00:00
c.receiver.Close(c.id, err)
2013-12-15 10:43:31 +00:00
}
func (c *Connection) isClosed() bool {
2013-12-24 16:45:16 +00:00
c.RLock()
defer c.RUnlock()
2013-12-15 10:43:31 +00:00
return c.closed
}
func (c *Connection) readerLoop() {
2013-12-31 03:10:54 +00:00
loop:
for {
2014-02-20 16:40:15 +00:00
var hdr header
hdr.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
2013-12-31 03:10:54 +00:00
break loop
2013-12-15 10:43:31 +00:00
}
2013-12-21 07:06:54 +00:00
if hdr.version != 0 {
2014-02-24 12:29:30 +00:00
c.close(fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version))
2013-12-31 03:10:54 +00:00
break loop
2013-12-21 07:06:54 +00:00
}
2013-12-15 10:43:31 +00:00
switch hdr.msgType {
case messageTypeIndex:
2014-02-20 16:40:15 +00:00
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
2013-12-31 03:10:54 +00:00
break loop
2013-12-15 10:43:31 +00:00
} else {
2014-02-20 16:40:15 +00:00
c.receiver.Index(c.id, im.Files)
2013-12-15 10:43:31 +00:00
}
c.Lock()
c.hasRecvdIndex = true
c.Unlock()
2013-12-15 10:43:31 +00:00
2013-12-28 13:10:36 +00:00
case messageTypeIndexUpdate:
2014-02-20 16:40:15 +00:00
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
2013-12-31 03:10:54 +00:00
break loop
2013-12-28 13:10:36 +00:00
} else {
2014-02-20 16:40:15 +00:00
c.receiver.IndexUpdate(c.id, im.Files)
2013-12-28 13:10:36 +00:00
}
2013-12-15 10:43:31 +00:00
case messageTypeRequest:
2014-02-20 16:40:15 +00:00
var req RequestMessage
req.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
2013-12-31 03:10:54 +00:00
break loop
}
go c.processRequest(hdr.msgID, req)
2013-12-15 10:43:31 +00:00
case messageTypeResponse:
2014-02-23 12:58:10 +00:00
data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size
2013-12-15 10:43:31 +00:00
2014-02-20 16:40:15 +00:00
if c.xr.Error() != nil {
c.close(c.xr.Error())
2013-12-31 03:10:54 +00:00
break loop
2014-02-20 16:40:15 +00:00
}
2013-12-15 10:43:31 +00:00
2014-02-20 16:40:15 +00:00
c.Lock()
rc, ok := c.awaiting[hdr.msgID]
delete(c.awaiting, hdr.msgID)
c.Unlock()
if ok {
rc <- asyncResult{data, c.xr.Error()}
close(rc)
2013-12-15 10:43:31 +00:00
}
case messageTypePing:
2013-12-24 16:45:16 +00:00
c.Lock()
2014-02-20 16:40:15 +00:00
header{0, hdr.msgID, messageTypePong}.encodeXDR(c.xw)
2013-12-19 23:01:34 +00:00
err := c.flush()
2013-12-24 16:45:16 +00:00
c.Unlock()
if err != nil {
2014-01-09 12:58:35 +00:00
c.close(err)
2013-12-31 03:10:54 +00:00
break loop
2014-02-20 16:40:15 +00:00
} else if c.xw.Error() != nil {
c.close(c.xw.Error())
2013-12-31 03:10:54 +00:00
break loop
2013-12-18 17:29:15 +00:00
}
2013-12-15 10:43:31 +00:00
case messageTypePong:
2013-12-24 16:45:16 +00:00
c.RLock()
2013-12-15 14:58:27 +00:00
rc, ok := c.awaiting[hdr.msgID]
2013-12-24 16:45:16 +00:00
c.RUnlock()
2013-12-15 14:58:27 +00:00
if ok {
rc <- asyncResult{}
2013-12-15 10:43:31 +00:00
close(rc)
2013-12-15 14:58:27 +00:00
2013-12-24 16:45:16 +00:00
c.Lock()
2013-12-15 10:43:31 +00:00
delete(c.awaiting, hdr.msgID)
2013-12-24 16:45:16 +00:00
c.Unlock()
2013-12-15 10:43:31 +00:00
}
2013-12-21 07:15:19 +00:00
2014-01-23 12:12:45 +00:00
case messageTypeOptions:
2014-02-20 16:40:15 +00:00
var om OptionsMessage
om.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
}
2014-01-23 12:12:45 +00:00
c.optionsLock.Lock()
2014-02-20 16:40:15 +00:00
c.peerOptions = make(map[string]string, len(om.Options))
for _, opt := range om.Options {
c.peerOptions[opt.Key] = opt.Value
}
2014-01-23 12:12:45 +00:00
c.optionsLock.Unlock()
if mh, rh := c.myOptions["clusterHash"], c.peerOptions["clusterHash"]; len(mh) > 0 && len(rh) > 0 && mh != rh {
c.close(ErrClusterHash)
break loop
}
2013-12-21 07:15:19 +00:00
default:
2014-02-24 12:29:30 +00:00
c.close(fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType))
2013-12-31 03:10:54 +00:00
break loop
2013-12-15 10:43:31 +00:00
}
}
}
2014-02-20 16:40:15 +00:00
func (c *Connection) processRequest(msgID int, req RequestMessage) {
data, _ := c.receiver.Request(c.id, req.Repository, req.Name, int64(req.Offset), int(req.Size))
2013-12-31 03:10:54 +00:00
c.Lock()
2014-02-20 16:40:15 +00:00
header{0, msgID, messageTypeResponse}.encodeXDR(c.xw)
_, err := c.xw.WriteBytes(data)
2014-01-18 03:06:44 +00:00
if err == nil {
err = c.flush()
}
2013-12-31 03:10:54 +00:00
c.Unlock()
buffers.Put(data)
if err != nil {
2014-01-09 12:58:35 +00:00
c.close(err)
2013-12-15 10:43:31 +00:00
}
}
func (c *Connection) pingerLoop() {
2013-12-31 01:52:36 +00:00
var rc = make(chan bool, 1)
2013-12-31 03:10:54 +00:00
for {
2013-12-31 01:52:36 +00:00
time.Sleep(pingIdleTime / 2)
c.RLock()
ready := c.hasRecvdIndex && c.hasSentIndex
c.RUnlock()
if ready {
go func() {
2014-01-09 12:58:35 +00:00
rc <- c.ping()
}()
select {
case ok := <-rc:
if !ok {
2014-02-24 12:29:30 +00:00
c.close(fmt.Errorf("ping failure"))
}
case <-time.After(pingTimeout):
2014-02-24 12:29:30 +00:00
c.close(fmt.Errorf("ping timeout"))
}
}
}
}
2013-12-23 17:12:44 +00:00
type Statistics struct {
2014-01-05 15:16:37 +00:00
At time.Time
InBytesTotal int
OutBytesTotal int
2013-12-23 17:12:44 +00:00
}
func (c *Connection) Statistics() Statistics {
2013-12-30 14:53:54 +00:00
c.statisticsLock.Lock()
defer c.statisticsLock.Unlock()
2013-12-23 17:12:44 +00:00
stats := Statistics{
2014-01-05 15:16:37 +00:00
At: time.Now(),
2014-02-20 16:40:15 +00:00
InBytesTotal: int(c.xr.Tot()),
OutBytesTotal: int(c.xw.Tot()),
2013-12-23 17:12:44 +00:00
}
2014-01-05 15:16:37 +00:00
2013-12-23 17:12:44 +00:00
return stats
}
2014-01-23 12:12:45 +00:00
func (c *Connection) Option(key string) string {
c.optionsLock.Lock()
defer c.optionsLock.Unlock()
return c.peerOptions[key]
2014-01-23 12:12:45 +00:00
}