syncthing/internal/protocol/protocol.go

640 lines
14 KiB
Go
Raw Normal View History

2014-11-16 20:13:20 +00:00
// Copyright (C) 2014 The Syncthing Authors.
2014-09-29 19:43:32 +00:00
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program. If not, see <http://www.gnu.org/licenses/>.
2014-06-01 20:50:14 +00:00
2013-12-15 10:43:31 +00:00
package protocol
import (
"encoding/binary"
"encoding/hex"
2013-12-15 10:43:31 +00:00
"errors"
"fmt"
2013-12-15 10:43:31 +00:00
"io"
"sync"
"time"
lz4 "github.com/bkaradzic/go-lz4"
2013-12-15 10:43:31 +00:00
)
const (
BlockSize = 128 * 1024
)
2014-02-20 16:40:15 +00:00
2013-12-15 10:43:31 +00:00
const (
messageTypeClusterConfig = 0
messageTypeIndex = 1
messageTypeRequest = 2
messageTypeResponse = 3
messageTypePing = 4
messageTypePong = 5
messageTypeIndexUpdate = 6
2014-07-26 19:27:55 +00:00
messageTypeClose = 7
2013-12-15 10:43:31 +00:00
)
const (
stateInitial = iota
stateCCRcvd
stateIdxRcvd
)
const (
2014-05-23 10:53:11 +00:00
FlagDeleted uint32 = 1 << 12
FlagInvalid = 1 << 13
FlagDirectory = 1 << 14
FlagNoPermBits = 1 << 15
)
const (
FlagShareTrusted uint32 = 1 << 0
FlagShareReadOnly = 1 << 1
FlagIntroducer = 1 << 2
FlagShareBits = 0x000000ff
)
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 device
Index(deviceID DeviceID, folder string, files []FileInfo)
// An index update was received from the peer device
IndexUpdate(deviceID DeviceID, folder string, files []FileInfo)
// A request was made by the peer device
Request(deviceID DeviceID, folder string, name string, offset int64, size int) ([]byte, error)
// A cluster configuration message was received
ClusterConfig(deviceID DeviceID, config ClusterConfigMessage)
// The peer device closed the connection
Close(deviceID DeviceID, err error)
2013-12-15 10:43:31 +00:00
}
type Connection interface {
ID() DeviceID
Name() string
Index(folder string, files []FileInfo) error
IndexUpdate(folder string, files []FileInfo) error
Request(folder string, name string, offset int64, size int) ([]byte, error)
ClusterConfig(config ClusterConfigMessage)
Statistics() Statistics
}
type rawConnection struct {
id DeviceID
name string
receiver Model
state int
2014-07-14 21:52:11 +00:00
cr *countingReader
cw *countingWriter
awaiting [4096]chan asyncResult
2014-07-03 11:37:20 +00:00
awaitingMut sync.Mutex
idxMut sync.Mutex // ensures serialization of Index calls
nextID chan int
outbox chan hdrMsg
closed chan struct{}
2014-07-03 11:37:20 +00:00
once sync.Once
compressionThreshold int // compress messages larger than this many bytes
rdbuf0 []byte // used & reused by readMessage
rdbuf1 []byte // used & reused by readMessage
2013-12-15 10:43:31 +00:00
}
2013-12-15 14:58:27 +00:00
type asyncResult struct {
val []byte
err error
}
type hdrMsg struct {
hdr header
msg encodable
}
type encodable interface {
2014-10-21 06:40:05 +00:00
AppendXDR([]byte) ([]byte, error)
}
2013-12-31 03:10:54 +00:00
const (
2014-06-14 09:07:34 +00:00
pingTimeout = 30 * time.Second
pingIdleTime = 60 * time.Second
2013-12-31 03:10:54 +00:00
)
func NewConnection(deviceID DeviceID, reader io.Reader, writer io.Writer, receiver Model, name string, compress bool) Connection {
cr := &countingReader{Reader: reader}
cw := &countingWriter{Writer: writer}
2014-07-28 13:25:34 +00:00
compThres := 1<<31 - 1 // compression disabled
if compress {
compThres = 128 // compress messages that are 128 bytes long or larger
}
c := rawConnection{
id: deviceID,
name: name,
receiver: nativeModel{receiver},
state: stateInitial,
cr: cr,
cw: cw,
outbox: make(chan hdrMsg),
nextID: make(chan int),
closed: make(chan struct{}),
compressionThreshold: compThres,
2013-12-15 10:43:31 +00:00
}
go c.readerLoop()
go c.writerLoop()
go c.pingerLoop()
go c.idGenerator()
2013-12-15 10:43:31 +00:00
return wireFormatConnection{&c}
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) ID() DeviceID {
2014-01-09 12:58:35 +00:00
return c.id
}
func (c *rawConnection) Name() string {
return c.name
}
// Index writes the list of file information to the connected peer device
func (c *rawConnection) Index(folder string, idx []FileInfo) error {
select {
case <-c.closed:
return ErrClosed
default:
2013-12-28 13:10:36 +00:00
}
c.idxMut.Lock()
c.send(-1, messageTypeIndex, IndexMessage{folder, idx})
c.idxMut.Unlock()
return nil
}
// IndexUpdate writes the list of file information to the connected peer device as an update
func (c *rawConnection) IndexUpdate(folder string, idx []FileInfo) error {
select {
case <-c.closed:
return ErrClosed
default:
}
c.idxMut.Lock()
c.send(-1, messageTypeIndexUpdate, IndexMessage{folder, idx})
c.idxMut.Unlock()
return nil
2013-12-15 10:43:31 +00:00
}
// Request returns the bytes for the specified block after fetching them from the connected peer.
func (c *rawConnection) Request(folder string, name string, offset int64, size int) ([]byte, error) {
var id int
select {
case id = <-c.nextID:
case <-c.closed:
2013-12-28 15:30:02 +00:00
return nil, ErrClosed
}
2014-07-03 11:37:20 +00:00
c.awaitingMut.Lock()
if ch := c.awaiting[id]; ch != nil {
panic("id taken")
}
2014-07-03 11:37:20 +00:00
rc := make(chan asyncResult, 1)
c.awaiting[id] = rc
2014-07-03 11:37:20 +00:00
c.awaitingMut.Unlock()
ok := c.send(id, messageTypeRequest, RequestMessage{folder, name, uint64(offset), uint32(size)})
if !ok {
return nil, ErrClosed
2013-12-19 23:01:34 +00:00
}
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
}
// ClusterConfig send the cluster configuration message to the peer and returns any error
func (c *rawConnection) ClusterConfig(config ClusterConfigMessage) {
c.send(-1, messageTypeClusterConfig, config)
}
func (c *rawConnection) ping() bool {
var id int
select {
case id = <-c.nextID:
case <-c.closed:
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-07-03 11:37:20 +00:00
c.awaitingMut.Lock()
c.awaiting[id] = rc
2014-07-03 11:37:20 +00:00
c.awaitingMut.Unlock()
ok := c.send(id, messageTypePing, nil)
if !ok {
return false
2013-12-19 23:01:34 +00:00
}
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
}
func (c *rawConnection) readerLoop() (err error) {
defer func() {
c.close(err)
}()
for {
select {
case <-c.closed:
return ErrClosed
default:
}
hdr, msg, err := c.readMessage()
if err != nil {
return err
}
switch hdr.msgType {
case messageTypeIndex:
if c.state < stateCCRcvd {
return fmt.Errorf("protocol error: index message in state %d", c.state)
}
c.handleIndex(msg.(IndexMessage))
c.state = stateIdxRcvd
case messageTypeIndexUpdate:
if c.state < stateIdxRcvd {
return fmt.Errorf("protocol error: index update message in state %d", c.state)
}
c.handleIndexUpdate(msg.(IndexMessage))
case messageTypeRequest:
if c.state < stateIdxRcvd {
return fmt.Errorf("protocol error: request message in state %d", c.state)
}
// Requests are handled asynchronously
go c.handleRequest(hdr.msgID, msg.(RequestMessage))
case messageTypeResponse:
if c.state < stateIdxRcvd {
return fmt.Errorf("protocol error: response message in state %d", c.state)
}
c.handleResponse(hdr.msgID, msg.(ResponseMessage))
case messageTypePing:
c.send(hdr.msgID, messageTypePong, EmptyMessage{})
case messageTypePong:
c.handlePong(hdr.msgID)
case messageTypeClusterConfig:
if c.state != stateInitial {
return fmt.Errorf("protocol error: cluster config message in state %d", c.state)
}
go c.receiver.ClusterConfig(c.id, msg.(ClusterConfigMessage))
c.state = stateCCRcvd
2014-07-26 19:27:55 +00:00
case messageTypeClose:
return errors.New(msg.(CloseMessage).Reason)
2014-07-26 19:27:55 +00:00
default:
return fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
}
}
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) readMessage() (hdr header, msg encodable, err error) {
if cap(c.rdbuf0) < 8 {
c.rdbuf0 = make([]byte, 8)
} else {
c.rdbuf0 = c.rdbuf0[:8]
}
_, err = io.ReadFull(c.cr, c.rdbuf0)
if err != nil {
return
}
hdr = decodeHeader(binary.BigEndian.Uint32(c.rdbuf0[0:4]))
msglen := int(binary.BigEndian.Uint32(c.rdbuf0[4:8]))
if debug {
l.Debugf("read header %v (msglen=%d)", hdr, msglen)
}
if cap(c.rdbuf0) < msglen {
c.rdbuf0 = make([]byte, msglen)
} else {
c.rdbuf0 = c.rdbuf0[:msglen]
}
_, err = io.ReadFull(c.cr, c.rdbuf0)
if err != nil {
return
}
if debug {
l.Debugf("read %d bytes", len(c.rdbuf0))
}
msgBuf := c.rdbuf0
if hdr.compression {
c.rdbuf1 = c.rdbuf1[:cap(c.rdbuf1)]
c.rdbuf1, err = lz4.Decode(c.rdbuf1, c.rdbuf0)
if err != nil {
return
}
msgBuf = c.rdbuf1
2014-07-23 09:54:15 +00:00
if debug {
l.Debugf("decompressed to %d bytes", len(msgBuf))
2014-07-23 09:54:15 +00:00
}
}
if debug {
if len(msgBuf) > 1024 {
l.Debugf("message data:\n%s", hex.Dump(msgBuf[:1024]))
} else {
l.Debugf("message data:\n%s", hex.Dump(msgBuf))
}
}
switch hdr.msgType {
case messageTypeIndex, messageTypeIndexUpdate:
var idx IndexMessage
err = idx.UnmarshalXDR(msgBuf)
msg = idx
case messageTypeRequest:
var req RequestMessage
err = req.UnmarshalXDR(msgBuf)
msg = req
case messageTypeResponse:
var resp ResponseMessage
err = resp.UnmarshalXDR(msgBuf)
msg = resp
case messageTypePing, messageTypePong:
msg = EmptyMessage{}
case messageTypeClusterConfig:
var cc ClusterConfigMessage
err = cc.UnmarshalXDR(msgBuf)
msg = cc
case messageTypeClose:
var cm CloseMessage
err = cm.UnmarshalXDR(msgBuf)
msg = cm
default:
err = fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
2014-07-03 11:37:20 +00:00
}
return
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) handleIndex(im IndexMessage) {
if debug {
l.Debugf("Index(%v, %v, %d files)", c.id, im.Folder, len(im.Files))
}
c.receiver.Index(c.id, im.Folder, im.Files)
}
func (c *rawConnection) handleIndexUpdate(im IndexMessage) {
if debug {
l.Debugf("queueing IndexUpdate(%v, %v, %d files)", c.id, im.Folder, len(im.Files))
2013-12-15 10:43:31 +00:00
}
c.receiver.IndexUpdate(c.id, im.Folder, im.Files)
}
2013-12-15 14:58:27 +00:00
func (c *rawConnection) handleRequest(msgID int, req RequestMessage) {
data, _ := c.receiver.Request(c.id, req.Folder, req.Name, int64(req.Offset), int(req.Size))
c.send(msgID, messageTypeResponse, ResponseMessage{data})
2014-07-26 19:27:55 +00:00
}
func (c *rawConnection) handleResponse(msgID int, resp ResponseMessage) {
c.awaitingMut.Lock()
if rc := c.awaiting[msgID]; rc != nil {
c.awaiting[msgID] = nil
rc <- asyncResult{resp.Data, nil}
close(rc)
}
c.awaitingMut.Unlock()
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) handlePong(msgID int) {
c.awaitingMut.Lock()
if rc := c.awaiting[msgID]; rc != nil {
c.awaiting[msgID] = nil
rc <- asyncResult{}
close(rc)
}
c.awaitingMut.Unlock()
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) send(msgID int, msgType int, msg encodable) bool {
if msgID < 0 {
select {
case id := <-c.nextID:
msgID = id
case <-c.closed:
return false
2013-12-21 07:06:54 +00:00
}
}
hdr := header{
version: 0,
msgID: msgID,
msgType: msgType,
}
2013-12-15 10:43:31 +00:00
select {
case c.outbox <- hdrMsg{hdr, msg}:
return true
case <-c.closed:
return false
}
}
2013-12-15 10:43:31 +00:00
func (c *rawConnection) writerLoop() {
var msgBuf = make([]byte, 8) // buffer for wire format message, kept and reused
var uncBuf []byte // buffer for uncompressed message, kept and reused
for {
var tempBuf []byte
var err error
select {
case hm := <-c.outbox:
if hm.msg != nil {
// Uncompressed message in uncBuf
2014-10-21 06:40:05 +00:00
uncBuf, err = hm.msg.AppendXDR(uncBuf[:0])
if err != nil {
c.close(err)
return
}
if len(uncBuf) >= c.compressionThreshold {
// Use compression for large messages
hm.hdr.compression = true
// Make sure we have enough space for the compressed message plus header in msgBug
msgBuf = msgBuf[:cap(msgBuf)]
if maxLen := lz4.CompressBound(len(uncBuf)) + 8; maxLen > len(msgBuf) {
msgBuf = make([]byte, maxLen)
}
// Compressed is written to msgBuf, we keep tb for the length only
tempBuf, err = lz4.Encode(msgBuf[8:], uncBuf)
binary.BigEndian.PutUint32(msgBuf[4:8], uint32(len(tempBuf)))
msgBuf = msgBuf[0 : len(tempBuf)+8]
if debug {
l.Debugf("write compressed message; %v (len=%d)", hm.hdr, len(tempBuf))
}
} else {
// No point in compressing very short messages
hm.hdr.compression = false
msgBuf = msgBuf[:cap(msgBuf)]
if l := len(uncBuf) + 8; l > len(msgBuf) {
msgBuf = make([]byte, l)
}
binary.BigEndian.PutUint32(msgBuf[4:8], uint32(len(uncBuf)))
msgBuf = msgBuf[0 : len(uncBuf)+8]
copy(msgBuf[8:], uncBuf)
if debug {
l.Debugf("write uncompressed message; %v (len=%d)", hm.hdr, len(uncBuf))
}
}
} else {
if debug {
l.Debugf("write empty message; %v", hm.hdr)
}
binary.BigEndian.PutUint32(msgBuf[4:8], 0)
msgBuf = msgBuf[:8]
}
binary.BigEndian.PutUint32(msgBuf[0:4], encodeHeader(hm.hdr))
if err == nil {
var n int
n, err = c.cw.Write(msgBuf)
if debug {
l.Debugf("wrote %d bytes on the wire", n)
}
}
if err != nil {
c.close(err)
return
}
case <-c.closed:
return
}
}
}
2013-12-15 10:43:31 +00:00
func (c *rawConnection) close(err error) {
2014-07-03 11:37:20 +00:00
c.once.Do(func() {
close(c.closed)
2013-12-15 14:58:27 +00:00
2014-07-03 11:37:20 +00:00
c.awaitingMut.Lock()
for i, ch := range c.awaiting {
if ch != nil {
close(ch)
c.awaiting[i] = nil
2013-12-15 10:43:31 +00:00
}
}
2014-07-03 11:37:20 +00:00
c.awaitingMut.Unlock()
go c.receiver.Close(c.id, err)
2014-07-03 11:37:20 +00:00
})
2013-12-15 10:43:31 +00:00
}
func (c *rawConnection) idGenerator() {
nextID := 0
for {
nextID = (nextID + 1) & 0xfff
select {
case c.nextID <- nextID:
case <-c.closed:
return
}
2013-12-15 10:43:31 +00:00
}
}
func (c *rawConnection) pingerLoop() {
2013-12-31 01:52:36 +00:00
var rc = make(chan bool, 1)
ticker := time.Tick(pingIdleTime / 2)
2013-12-31 03:10:54 +00:00
for {
select {
case <-ticker:
if d := time.Since(c.cr.Last()); d < pingIdleTime {
if debug {
l.Debugln(c.id, "ping skipped after rd", d)
}
continue
}
if d := time.Since(c.cw.Last()); d < pingIdleTime {
if debug {
l.Debugln(c.id, "ping skipped after wr", d)
}
continue
}
go func() {
if debug {
l.Debugln(c.id, "ping ->")
}
rc <- c.ping()
}()
select {
case ok := <-rc:
2014-05-28 18:45:29 +00:00
if debug {
l.Debugln(c.id, "<- pong")
}
if !ok {
c.close(fmt.Errorf("ping failure"))
}
case <-time.After(pingTimeout):
c.close(fmt.Errorf("ping timeout"))
case <-c.closed:
return
}
case <-c.closed:
return
}
}
}
2013-12-23 17:12:44 +00:00
type Statistics struct {
2014-01-05 15:16:37 +00:00
At time.Time
InBytesTotal uint64
OutBytesTotal uint64
2013-12-23 17:12:44 +00:00
}
func (c *rawConnection) Statistics() Statistics {
return Statistics{
2014-01-05 15:16:37 +00:00
At: time.Now(),
InBytesTotal: c.cr.Tot(),
OutBytesTotal: c.cw.Tot(),
2013-12-23 17:12:44 +00:00
}
}