518 lines
10 KiB
Go
518 lines
10 KiB
Go
package protocol
|
|
|
|
import (
|
|
"bufio"
|
|
"compress/flate"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"sync"
|
|
"time"
|
|
|
|
"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 {
|
|
id string
|
|
receiver Model
|
|
|
|
reader io.ReadCloser
|
|
cr *countingReader
|
|
xr *xdr.Reader
|
|
writer io.WriteCloser
|
|
|
|
cw *countingWriter
|
|
wb *bufio.Writer
|
|
xw *xdr.Writer
|
|
wmut sync.Mutex
|
|
|
|
indexSent map[string]map[string][2]int64
|
|
awaiting []chan asyncResult
|
|
imut sync.Mutex
|
|
|
|
nextID chan int
|
|
outbox chan []encodable
|
|
closed chan struct{}
|
|
}
|
|
|
|
type asyncResult struct {
|
|
val []byte
|
|
err error
|
|
}
|
|
|
|
const (
|
|
pingTimeout = 4 * time.Minute
|
|
pingIdleTime = 5 * time.Minute
|
|
)
|
|
|
|
func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver Model) Connection {
|
|
cr := &countingReader{Reader: reader}
|
|
cw := &countingWriter{Writer: writer}
|
|
|
|
flrd := flate.NewReader(cr)
|
|
flwr, err := flate.NewWriter(cw, flate.BestSpeed)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
wb := bufio.NewWriter(flwr)
|
|
|
|
c := rawConnection{
|
|
id: nodeID,
|
|
receiver: nativeModel{receiver},
|
|
reader: flrd,
|
|
cr: cr,
|
|
xr: xdr.NewReader(flrd),
|
|
writer: flwr,
|
|
cw: cw,
|
|
wb: wb,
|
|
xw: xdr.NewWriter(wb),
|
|
awaiting: make([]chan asyncResult, 0x1000),
|
|
indexSent: make(map[string]map[string][2]int64),
|
|
outbox: make(chan []encodable),
|
|
nextID: make(chan int),
|
|
closed: make(chan struct{}),
|
|
}
|
|
|
|
go c.readerLoop()
|
|
go c.writerLoop()
|
|
go c.pingerLoop()
|
|
go c.idGenerator()
|
|
|
|
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.imut.Lock()
|
|
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
|
|
}
|
|
c.imut.Unlock()
|
|
|
|
c.send(header{0, -1, msgType}, IndexMessage{repo, idx})
|
|
}
|
|
|
|
// 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) {
|
|
var id int
|
|
select {
|
|
case id = <-c.nextID:
|
|
case <-c.closed:
|
|
return nil, ErrClosed
|
|
}
|
|
|
|
c.imut.Lock()
|
|
if ch := c.awaiting[id]; ch != nil {
|
|
panic("id taken")
|
|
}
|
|
rc := make(chan asyncResult)
|
|
c.awaiting[id] = rc
|
|
c.imut.Unlock()
|
|
|
|
ok := c.send(header{0, id, messageTypeRequest},
|
|
RequestMessage{repo, name, uint64(offset), uint32(size)})
|
|
if !ok {
|
|
return nil, ErrClosed
|
|
}
|
|
|
|
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.send(header{0, -1, messageTypeClusterConfig}, config)
|
|
}
|
|
|
|
func (c *rawConnection) ping() bool {
|
|
var id int
|
|
select {
|
|
case id = <-c.nextID:
|
|
case <-c.closed:
|
|
return false
|
|
}
|
|
|
|
rc := make(chan asyncResult, 1)
|
|
c.imut.Lock()
|
|
c.awaiting[id] = rc
|
|
c.imut.Unlock()
|
|
|
|
ok := c.send(header{0, id, messageTypePing})
|
|
if !ok {
|
|
return false
|
|
}
|
|
|
|
res, ok := <-rc
|
|
return ok && res.err == nil
|
|
}
|
|
|
|
func (c *rawConnection) readerLoop() (err error) {
|
|
defer func() {
|
|
c.close(err)
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case <-c.closed:
|
|
return ErrClosed
|
|
default:
|
|
}
|
|
|
|
var hdr header
|
|
hdr.decodeXDR(c.xr)
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
}
|
|
if hdr.version != 0 {
|
|
return fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version)
|
|
}
|
|
|
|
switch hdr.msgType {
|
|
case messageTypeIndex:
|
|
if err := c.handleIndex(); err != nil {
|
|
return err
|
|
}
|
|
|
|
case messageTypeIndexUpdate:
|
|
if err := c.handleIndexUpdate(); err != nil {
|
|
return err
|
|
}
|
|
|
|
case messageTypeRequest:
|
|
if err := c.handleRequest(hdr); err != nil {
|
|
return err
|
|
}
|
|
|
|
case messageTypeResponse:
|
|
if err := c.handleResponse(hdr); err != nil {
|
|
return err
|
|
}
|
|
|
|
case messageTypePing:
|
|
c.send(header{0, hdr.msgID, messageTypePong})
|
|
|
|
case messageTypePong:
|
|
c.handlePong(hdr)
|
|
|
|
case messageTypeClusterConfig:
|
|
if err := c.handleClusterConfig(); err != nil {
|
|
return err
|
|
}
|
|
|
|
default:
|
|
return fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *rawConnection) handleIndex() error {
|
|
var im IndexMessage
|
|
im.decodeXDR(c.xr)
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
} 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)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *rawConnection) handleIndexUpdate() error {
|
|
var im IndexMessage
|
|
im.decodeXDR(c.xr)
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
} else {
|
|
go c.receiver.IndexUpdate(c.id, im.Repository, im.Files)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *rawConnection) handleRequest(hdr header) error {
|
|
var req RequestMessage
|
|
req.decodeXDR(c.xr)
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
}
|
|
go c.processRequest(hdr.msgID, req)
|
|
return nil
|
|
}
|
|
|
|
func (c *rawConnection) handleResponse(hdr header) error {
|
|
data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size
|
|
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
}
|
|
|
|
go func(hdr header, err error) {
|
|
c.imut.Lock()
|
|
rc := c.awaiting[hdr.msgID]
|
|
c.awaiting[hdr.msgID] = nil
|
|
c.imut.Unlock()
|
|
|
|
if rc != nil {
|
|
rc <- asyncResult{data, err}
|
|
close(rc)
|
|
}
|
|
}(hdr, c.xr.Error())
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *rawConnection) handlePong(hdr header) {
|
|
c.imut.Lock()
|
|
if rc := c.awaiting[hdr.msgID]; rc != nil {
|
|
go func() {
|
|
rc <- asyncResult{}
|
|
close(rc)
|
|
}()
|
|
|
|
c.awaiting[hdr.msgID] = nil
|
|
}
|
|
c.imut.Unlock()
|
|
}
|
|
|
|
func (c *rawConnection) handleClusterConfig() error {
|
|
var cm ClusterConfigMessage
|
|
cm.decodeXDR(c.xr)
|
|
if err := c.xr.Error(); err != nil {
|
|
return err
|
|
} else {
|
|
go c.receiver.ClusterConfig(c.id, cm)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type encodable interface {
|
|
encodeXDR(*xdr.Writer) (int, error)
|
|
}
|
|
type encodableBytes []byte
|
|
|
|
func (e encodableBytes) encodeXDR(xw *xdr.Writer) (int, error) {
|
|
return xw.WriteBytes(e)
|
|
}
|
|
|
|
func (c *rawConnection) send(h header, es ...encodable) bool {
|
|
if h.msgID < 0 {
|
|
select {
|
|
case id := <-c.nextID:
|
|
h.msgID = id
|
|
case <-c.closed:
|
|
return false
|
|
}
|
|
}
|
|
msg := append([]encodable{h}, es...)
|
|
|
|
select {
|
|
case c.outbox <- msg:
|
|
return true
|
|
case <-c.closed:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (c *rawConnection) writerLoop() {
|
|
var err error
|
|
for es := range c.outbox {
|
|
c.wmut.Lock()
|
|
for _, e := range es {
|
|
e.encodeXDR(c.xw)
|
|
}
|
|
|
|
if err = c.flush(); err != nil {
|
|
c.wmut.Unlock()
|
|
c.close(err)
|
|
return
|
|
}
|
|
c.wmut.Unlock()
|
|
}
|
|
}
|
|
|
|
type flusher interface {
|
|
Flush() error
|
|
}
|
|
|
|
func (c *rawConnection) flush() error {
|
|
if err := c.xw.Error(); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := c.wb.Flush(); err != nil {
|
|
return err
|
|
}
|
|
|
|
if f, ok := c.writer.(flusher); ok {
|
|
return f.Flush()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *rawConnection) close(err error) {
|
|
c.imut.Lock()
|
|
c.wmut.Lock()
|
|
defer c.imut.Unlock()
|
|
defer c.wmut.Unlock()
|
|
|
|
select {
|
|
case <-c.closed:
|
|
return
|
|
default:
|
|
close(c.closed)
|
|
|
|
for i, ch := range c.awaiting {
|
|
if ch != nil {
|
|
close(ch)
|
|
c.awaiting[i] = nil
|
|
}
|
|
}
|
|
|
|
c.writer.Close()
|
|
c.reader.Close()
|
|
|
|
c.receiver.Close(c.id, err)
|
|
}
|
|
}
|
|
|
|
func (c *rawConnection) idGenerator() {
|
|
nextID := 0
|
|
for {
|
|
nextID = (nextID + 1) & 0xfff
|
|
select {
|
|
case c.nextID <- nextID:
|
|
case <-c.closed:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *rawConnection) pingerLoop() {
|
|
var rc = make(chan bool, 1)
|
|
ticker := time.Tick(pingIdleTime / 2)
|
|
for {
|
|
select {
|
|
case <-ticker:
|
|
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
|
|
}
|
|
|
|
case <-c.closed:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
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.send(header{0, msgID, messageTypeResponse},
|
|
encodableBytes(data))
|
|
}
|
|
|
|
type Statistics struct {
|
|
At time.Time
|
|
InBytesTotal int
|
|
OutBytesTotal int
|
|
}
|
|
|
|
func (c *rawConnection) Statistics() Statistics {
|
|
return Statistics{
|
|
At: time.Now(),
|
|
InBytesTotal: int(c.cr.Tot()),
|
|
OutBytesTotal: int(c.cw.Tot()),
|
|
}
|
|
}
|