123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481 |
- package protocol
- import (
- "bufio"
- "compress/flate"
- "errors"
- "fmt"
- "io"
- "log"
- "sync"
- "time"
- "github.com/calmh/syncthing/buffers"
- "github.com/calmh/syncthing/xdr"
- )
- const BlockSize = 128 * 1024
- const (
- messageTypeIndex = 1
- messageTypeRequest = 2
- messageTypeResponse = 3
- messageTypePing = 4
- messageTypePong = 5
- messageTypeIndexUpdate = 6
- messageTypeOptions = 7
- )
- const (
- FlagDeleted uint32 = 1 << 12
- FlagInvalid = 1 << 13
- FlagDirectory = 1 << 14
- )
- 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)
- // 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)
- Statistics() Statistics
- Option(key string) string
- }
- 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
- peerOptions map[string]string
- myOptions map[string]string
- optionsLock sync.Mutex
- 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, options map[string]string) 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()
- if options != nil {
- c.myOptions = options
- go func() {
- c.Lock()
- header{0, c.nextID, messageTypeOptions}.encodeXDR(c.xw)
- 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()
- }
- if err != nil {
- log.Println("Warning: Write error during initial handshake:", err)
- }
- c.nextID++
- c.Unlock()
- }()
- }
- 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
- }
- 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 messageTypeOptions:
- var om OptionsMessage
- om.decodeXDR(c.xr)
- if c.xr.Error() != nil {
- c.close(c.xr.Error())
- break loop
- }
- c.optionsLock.Lock()
- c.peerOptions = make(map[string]string, len(om.Options))
- for _, opt := range om.Options {
- c.peerOptions[opt.Key] = opt.Value
- }
- 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
- }
- 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()),
- }
- }
- func (c *rawConnection) Option(key string) string {
- c.optionsLock.Lock()
- defer c.optionsLock.Unlock()
- return c.peerOptions[key]
- }
|