syncthing/protocol/protocol.go

468 lines
9.6 KiB
Go

package protocol
import (
"bufio"
"compress/flate"
"errors"
"fmt"
"io"
"sync"
"time"
"github.com/calmh/syncthing/buffers"
"github.com/calmh/syncthing/xdr"
)
const BlockSize = 128 * 1024
const (
messageTypeClusterConfig = 0
messageTypeIndex = 1
messageTypeRequest = 2
messageTypeResponse = 3
messageTypePing = 4
messageTypePong = 5
messageTypeIndexUpdate = 6
)
const (
FlagDeleted uint32 = 1 << 12
FlagInvalid = 1 << 13
FlagDirectory = 1 << 14
)
const (
FlagShareTrusted uint32 = 1 << 0
FlagShareReadOnly = 1 << 1
FlagShareBits = 0x000000ff
)
var (
ErrClusterHash = fmt.Errorf("configuration error: mismatched cluster hash")
ErrClosed = errors.New("connection closed")
)
type Model interface {
// An index was received from the peer node
Index(nodeID string, repo string, files []FileInfo)
// An index update was received from the peer node
IndexUpdate(nodeID string, repo string, files []FileInfo)
// A request was made by the peer node
Request(nodeID string, repo string, name string, offset int64, size int) ([]byte, error)
// A cluster configuration message was received
ClusterConfig(nodeID string, config ClusterConfigMessage)
// The peer node closed the connection
Close(nodeID string, err error)
}
type Connection interface {
ID() string
Index(repo string, files []FileInfo)
Request(repo string, name string, offset int64, size int) ([]byte, error)
ClusterConfig(config ClusterConfigMessage)
Statistics() Statistics
}
type rawConnection struct {
sync.RWMutex
id string
receiver Model
reader io.ReadCloser
xr *xdr.Reader
writer io.WriteCloser
wb *bufio.Writer
xw *xdr.Writer
closed chan struct{}
awaiting map[int]chan asyncResult
nextID int
indexSent map[string]map[string][2]int64
hasSentIndex bool
hasRecvdIndex bool
}
type asyncResult struct {
val []byte
err error
}
const (
pingTimeout = 2 * time.Minute
pingIdleTime = 5 * time.Minute
)
func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver Model) Connection {
flrd := flate.NewReader(reader)
flwr, err := flate.NewWriter(writer, flate.BestSpeed)
if err != nil {
panic(err)
}
wb := bufio.NewWriter(flwr)
c := rawConnection{
id: nodeID,
receiver: nativeModel{receiver},
reader: flrd,
xr: xdr.NewReader(flrd),
writer: flwr,
wb: wb,
xw: xdr.NewWriter(wb),
closed: make(chan struct{}),
awaiting: make(map[int]chan asyncResult),
indexSent: make(map[string]map[string][2]int64),
}
go c.readerLoop()
go c.pingerLoop()
return wireFormatConnection{&c}
}
func (c *rawConnection) ID() string {
return c.id
}
// Index writes the list of file information to the connected peer node
func (c *rawConnection) Index(repo string, idx []FileInfo) {
c.Lock()
if c.isClosed() {
c.Unlock()
return
}
var msgType int
if c.indexSent[repo] == nil {
// This is the first time we send an index.
msgType = messageTypeIndex
c.indexSent[repo] = make(map[string][2]int64)
for _, f := range idx {
c.indexSent[repo][f.Name] = [2]int64{f.Modified, int64(f.Version)}
}
} 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] {
diff = append(diff, f)
c.indexSent[repo][f.Name] = [2]int64{f.Modified, int64(f.Version)}
}
}
idx = diff
}
header{0, c.nextID, msgType}.encodeXDR(c.xw)
_, err := IndexMessage{repo, idx}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
c.nextID = (c.nextID + 1) & 0xfff
c.hasSentIndex = true
c.Unlock()
if err != nil {
c.close(err)
return
}
}
// Request returns the bytes for the specified block after fetching them from the connected peer.
func (c *rawConnection) Request(repo string, name string, offset int64, size int) ([]byte, error) {
c.Lock()
if c.isClosed() {
c.Unlock()
return nil, ErrClosed
}
rc := make(chan asyncResult)
if _, ok := c.awaiting[c.nextID]; ok {
panic("id taken")
}
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypeRequest}.encodeXDR(c.xw)
_, err := RequestMessage{repo, name, uint64(offset), uint32(size)}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
if err != nil {
c.Unlock()
c.close(err)
return nil, err
}
c.nextID = (c.nextID + 1) & 0xfff
c.Unlock()
res, ok := <-rc
if !ok {
return nil, ErrClosed
}
return res.val, res.err
}
// ClusterConfig send the cluster configuration message to the peer and returns any error
func (c *rawConnection) ClusterConfig(config ClusterConfigMessage) {
c.Lock()
defer c.Unlock()
if c.isClosed() {
return
}
header{0, c.nextID, messageTypeClusterConfig}.encodeXDR(c.xw)
c.nextID = (c.nextID + 1) & 0xfff
_, err := config.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
if err != nil {
c.close(err)
}
}
func (c *rawConnection) ping() bool {
c.Lock()
if c.isClosed() {
c.Unlock()
return false
}
rc := make(chan asyncResult, 1)
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypePing}.encodeXDR(c.xw)
err := c.flush()
if err != nil {
c.Unlock()
c.close(err)
return false
} else if c.xw.Error() != nil {
c.Unlock()
c.close(c.xw.Error())
return false
}
c.nextID = (c.nextID + 1) & 0xfff
c.Unlock()
res, ok := <-rc
return ok && res.err == nil
}
type flusher interface {
Flush() error
}
func (c *rawConnection) flush() error {
c.wb.Flush()
if f, ok := c.writer.(flusher); ok {
return f.Flush()
}
return nil
}
func (c *rawConnection) close(err error) {
c.Lock()
select {
case <-c.closed:
c.Unlock()
return
default:
}
close(c.closed)
for _, ch := range c.awaiting {
close(ch)
}
c.awaiting = nil
c.writer.Close()
c.reader.Close()
c.Unlock()
c.receiver.Close(c.id, err)
}
func (c *rawConnection) isClosed() bool {
select {
case <-c.closed:
return true
default:
return false
}
}
func (c *rawConnection) readerLoop() {
loop:
for !c.isClosed() {
var hdr header
hdr.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
}
if hdr.version != 0 {
c.close(fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version))
break loop
}
switch hdr.msgType {
case messageTypeIndex:
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
} else {
// We run this (and the corresponding one for update, below)
// in a separate goroutine to avoid blocking the read loop.
// There is otherwise a potential deadlock where both sides
// has the model locked because it's sending a large index
// update and can't receive the large index update from the
// other side.
go c.receiver.Index(c.id, im.Repository, im.Files)
}
c.Lock()
c.hasRecvdIndex = true
c.Unlock()
case messageTypeIndexUpdate:
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
} else {
go c.receiver.IndexUpdate(c.id, im.Repository, im.Files)
}
case messageTypeRequest:
var req RequestMessage
req.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
}
go c.processRequest(hdr.msgID, req)
case messageTypeResponse:
data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
}
go func(hdr header, err error) {
c.Lock()
rc, ok := c.awaiting[hdr.msgID]
delete(c.awaiting, hdr.msgID)
c.Unlock()
if ok {
rc <- asyncResult{data, err}
close(rc)
}
}(hdr, c.xr.Error())
case messageTypePing:
c.Lock()
header{0, hdr.msgID, messageTypePong}.encodeXDR(c.xw)
err := c.flush()
c.Unlock()
if err != nil {
c.close(err)
break loop
} else if c.xw.Error() != nil {
c.close(c.xw.Error())
break loop
}
case messageTypePong:
c.RLock()
rc, ok := c.awaiting[hdr.msgID]
c.RUnlock()
if ok {
rc <- asyncResult{}
close(rc)
c.Lock()
delete(c.awaiting, hdr.msgID)
c.Unlock()
}
case messageTypeClusterConfig:
var cm ClusterConfigMessage
cm.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
break loop
} else {
go c.receiver.ClusterConfig(c.id, cm)
}
default:
c.close(fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType))
break loop
}
}
}
func (c *rawConnection) processRequest(msgID int, req RequestMessage) {
data, _ := c.receiver.Request(c.id, req.Repository, req.Name, int64(req.Offset), int(req.Size))
c.Lock()
header{0, msgID, messageTypeResponse}.encodeXDR(c.xw)
_, err := c.xw.WriteBytes(data)
if err == nil {
err = c.flush()
}
c.Unlock()
buffers.Put(data)
if err != nil {
c.close(err)
}
}
func (c *rawConnection) pingerLoop() {
var rc = make(chan bool, 1)
ticker := time.Tick(pingIdleTime / 2)
for {
select {
case <-ticker:
c.RLock()
ready := c.hasRecvdIndex && c.hasSentIndex
c.RUnlock()
if ready {
go func() {
rc <- c.ping()
}()
select {
case ok := <-rc:
if !ok {
c.close(fmt.Errorf("ping failure"))
}
case <-time.After(pingTimeout):
c.close(fmt.Errorf("ping timeout"))
}
}
case <-c.closed:
return
}
}
}
type Statistics struct {
At time.Time
InBytesTotal int
OutBytesTotal int
}
func (c *rawConnection) Statistics() Statistics {
return Statistics{
At: time.Now(),
InBytesTotal: int(c.xr.Tot()),
OutBytesTotal: int(c.xw.Tot()),
}
}