2014-07-13 00:45:33 +02:00
|
|
|
// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
|
|
|
|
// All rights reserved. Use of this source code is governed by an MIT-style
|
|
|
|
// license that can be found in the LICENSE file.
|
2014-06-01 22:50:14 +02:00
|
|
|
|
2013-12-15 11:43:31 +01:00
|
|
|
package protocol
|
|
|
|
|
|
|
|
import (
|
2014-07-28 11:31:22 +02:00
|
|
|
"encoding/binary"
|
|
|
|
"encoding/hex"
|
2013-12-15 11:43:31 +01:00
|
|
|
"errors"
|
2013-12-31 03:21:57 +01:00
|
|
|
"fmt"
|
2013-12-15 11:43:31 +01:00
|
|
|
"io"
|
|
|
|
"sync"
|
2013-12-15 16:19:45 +01:00
|
|
|
"time"
|
2014-07-23 12:18:25 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
lz4 "github.com/bkaradzic/go-lz4"
|
2013-12-15 11:43:31 +01:00
|
|
|
)
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
const (
|
2014-07-28 12:44:46 +02:00
|
|
|
BlockSize = 128 * 1024
|
2014-07-28 11:31:22 +02:00
|
|
|
)
|
2014-02-20 17:40:15 +01:00
|
|
|
|
2013-12-15 11:43:31 +01:00
|
|
|
const (
|
2014-04-13 15:28:26 +02:00
|
|
|
messageTypeClusterConfig = 0
|
|
|
|
messageTypeIndex = 1
|
|
|
|
messageTypeRequest = 2
|
|
|
|
messageTypeResponse = 3
|
|
|
|
messageTypePing = 4
|
|
|
|
messageTypePong = 5
|
|
|
|
messageTypeIndexUpdate = 6
|
2014-07-26 21:27:55 +02:00
|
|
|
messageTypeClose = 7
|
2013-12-15 11:43:31 +01:00
|
|
|
)
|
|
|
|
|
2014-07-02 21:33:30 +02:00
|
|
|
const (
|
|
|
|
stateInitial = iota
|
|
|
|
stateCCRcvd
|
|
|
|
stateIdxRcvd
|
|
|
|
)
|
|
|
|
|
2014-01-07 22:44:21 +01:00
|
|
|
const (
|
2014-05-23 12:53:11 +02:00
|
|
|
FlagDeleted uint32 = 1 << 12
|
|
|
|
FlagInvalid = 1 << 13
|
|
|
|
FlagDirectory = 1 << 14
|
|
|
|
FlagNoPermBits = 1 << 15
|
2014-01-07 22:44:21 +01:00
|
|
|
)
|
|
|
|
|
2014-04-13 15:28:26 +02:00
|
|
|
const (
|
|
|
|
FlagShareTrusted uint32 = 1 << 0
|
|
|
|
FlagShareReadOnly = 1 << 1
|
2014-09-23 16:04:20 +02:00
|
|
|
FlagIntroducer = 1 << 2
|
2014-04-13 15:28:26 +02:00
|
|
|
FlagShareBits = 0x000000ff
|
|
|
|
)
|
|
|
|
|
2014-02-09 23:13:06 +01:00
|
|
|
var (
|
2014-02-24 13:29:30 +01:00
|
|
|
ErrClusterHash = fmt.Errorf("configuration error: mismatched cluster hash")
|
|
|
|
ErrClosed = errors.New("connection closed")
|
2014-02-09 23:13:06 +01:00
|
|
|
)
|
|
|
|
|
2013-12-15 11:43:31 +01:00
|
|
|
type Model interface {
|
|
|
|
// An index was received from the peer node
|
2014-06-30 01:42:03 +02:00
|
|
|
Index(nodeID NodeID, repo string, files []FileInfo)
|
2013-12-28 14:10:36 +01:00
|
|
|
// An index update was received from the peer node
|
2014-06-30 01:42:03 +02:00
|
|
|
IndexUpdate(nodeID NodeID, repo string, files []FileInfo)
|
2013-12-15 11:43:31 +01:00
|
|
|
// A request was made by the peer node
|
2014-06-30 01:42:03 +02:00
|
|
|
Request(nodeID NodeID, repo string, name string, offset int64, size int) ([]byte, error)
|
2014-04-13 15:28:26 +02:00
|
|
|
// A cluster configuration message was received
|
2014-06-30 01:42:03 +02:00
|
|
|
ClusterConfig(nodeID NodeID, config ClusterConfigMessage)
|
2013-12-15 11:43:31 +01:00
|
|
|
// The peer node closed the connection
|
2014-06-30 01:42:03 +02:00
|
|
|
Close(nodeID NodeID, err error)
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-03-28 14:36:57 +01:00
|
|
|
type Connection interface {
|
2014-06-30 01:42:03 +02:00
|
|
|
ID() NodeID
|
2014-07-15 13:04:37 +02:00
|
|
|
Name() string
|
|
|
|
Index(repo string, files []FileInfo) error
|
|
|
|
IndexUpdate(repo string, files []FileInfo) error
|
2014-03-29 18:53:48 +01:00
|
|
|
Request(repo string, name string, offset int64, size int) ([]byte, error)
|
2014-04-13 15:28:26 +02:00
|
|
|
ClusterConfig(config ClusterConfigMessage)
|
2014-03-28 14:36:57 +01:00
|
|
|
Statistics() Statistics
|
|
|
|
}
|
|
|
|
|
|
|
|
type rawConnection struct {
|
2014-06-30 01:42:03 +02:00
|
|
|
id NodeID
|
2014-07-15 13:04:37 +02:00
|
|
|
name string
|
2014-05-04 17:40:40 +02:00
|
|
|
receiver Model
|
2014-07-02 21:33:30 +02:00
|
|
|
state int
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-14 23:52:11 +02:00
|
|
|
cr *countingReader
|
|
|
|
cw *countingWriter
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
awaiting [4096]chan asyncResult
|
2014-07-03 13:37:20 +02:00
|
|
|
awaitingMut sync.Mutex
|
2014-05-11 19:30:29 +02:00
|
|
|
|
2014-07-15 13:04:37 +02:00
|
|
|
idxMut sync.Mutex // ensures serialization of Index calls
|
2014-06-22 08:17:58 +02:00
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
nextID chan int
|
2014-07-28 11:31:22 +02:00
|
|
|
outbox chan hdrMsg
|
2014-05-11 19:30:29 +02:00
|
|
|
closed chan struct{}
|
2014-07-03 13:37:20 +02:00
|
|
|
once sync.Once
|
2014-07-28 11:31:22 +02:00
|
|
|
|
2014-07-28 12:44:46 +02:00
|
|
|
compressionThreshold int // compress messages larger than this many bytes
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
rdbuf0 []byte // used & reused by readMessage
|
|
|
|
rdbuf1 []byte // used & reused by readMessage
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2013-12-15 15:58:27 +01:00
|
|
|
type asyncResult struct {
|
|
|
|
val []byte
|
|
|
|
err error
|
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
type hdrMsg struct {
|
|
|
|
hdr header
|
|
|
|
msg encodable
|
|
|
|
}
|
|
|
|
|
|
|
|
type encodable interface {
|
|
|
|
AppendXDR([]byte) []byte
|
|
|
|
}
|
|
|
|
|
2013-12-31 04:10:54 +01:00
|
|
|
const (
|
2014-06-14 11:07:34 +02:00
|
|
|
pingTimeout = 30 * time.Second
|
|
|
|
pingIdleTime = 60 * time.Second
|
2013-12-31 04:10:54 +01:00
|
|
|
)
|
2013-12-15 16:19:45 +01:00
|
|
|
|
2014-07-28 12:44:46 +02:00
|
|
|
func NewConnection(nodeID NodeID, reader io.Reader, writer io.Writer, receiver Model, name string, compress bool) Connection {
|
2014-04-21 12:49:47 +02:00
|
|
|
cr := &countingReader{Reader: reader}
|
|
|
|
cw := &countingWriter{Writer: writer}
|
|
|
|
|
2014-07-28 15:25:34 +02:00
|
|
|
compThres := 1<<31 - 1 // compression disabled
|
2014-07-28 12:44:46 +02:00
|
|
|
if compress {
|
|
|
|
compThres = 128 // compress messages that are 128 bytes long or larger
|
|
|
|
}
|
2014-03-28 14:36:57 +01:00
|
|
|
c := rawConnection{
|
2014-07-28 12:44:46 +02:00
|
|
|
id: nodeID,
|
|
|
|
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 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
go c.readerLoop()
|
2014-05-11 19:30:29 +02:00
|
|
|
go c.writerLoop()
|
2013-12-15 16:19:45 +01:00
|
|
|
go c.pingerLoop()
|
2014-05-11 19:30:29 +02:00
|
|
|
go c.idGenerator()
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2014-03-28 14:36:57 +01:00
|
|
|
return wireFormatConnection{&c}
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-06-30 01:42:03 +02:00
|
|
|
func (c *rawConnection) ID() NodeID {
|
2014-01-09 13:58:35 +01:00
|
|
|
return c.id
|
|
|
|
}
|
|
|
|
|
2014-07-15 13:04:37 +02:00
|
|
|
func (c *rawConnection) Name() string {
|
|
|
|
return c.name
|
|
|
|
}
|
2014-05-28 10:11:17 +02:00
|
|
|
|
2014-07-15 13:04:37 +02:00
|
|
|
// Index writes the list of file information to the connected peer node
|
|
|
|
func (c *rawConnection) Index(repo string, idx []FileInfo) error {
|
|
|
|
select {
|
|
|
|
case <-c.closed:
|
|
|
|
return ErrClosed
|
|
|
|
default:
|
2013-12-28 14:10:36 +01:00
|
|
|
}
|
2014-07-15 13:04:37 +02:00
|
|
|
c.idxMut.Lock()
|
2014-07-28 11:31:22 +02:00
|
|
|
c.send(-1, messageTypeIndex, IndexMessage{repo, idx})
|
2014-07-15 13:04:37 +02:00
|
|
|
c.idxMut.Unlock()
|
|
|
|
return nil
|
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-15 13:04:37 +02:00
|
|
|
// IndexUpdate writes the list of file information to the connected peer node as an update
|
|
|
|
func (c *rawConnection) IndexUpdate(repo string, idx []FileInfo) error {
|
|
|
|
select {
|
|
|
|
case <-c.closed:
|
|
|
|
return ErrClosed
|
|
|
|
default:
|
2014-06-12 18:06:46 +02:00
|
|
|
}
|
2014-07-15 13:04:37 +02:00
|
|
|
c.idxMut.Lock()
|
2014-07-28 11:31:22 +02:00
|
|
|
c.send(-1, messageTypeIndexUpdate, IndexMessage{repo, idx})
|
2014-07-15 13:04:37 +02:00
|
|
|
c.idxMut.Unlock()
|
|
|
|
return nil
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Request returns the bytes for the specified block after fetching them from the connected peer.
|
2014-03-28 14:36:57 +01:00
|
|
|
func (c *rawConnection) Request(repo string, name string, offset int64, size int) ([]byte, error) {
|
2014-05-11 19:30:29 +02:00
|
|
|
var id int
|
|
|
|
select {
|
|
|
|
case id = <-c.nextID:
|
|
|
|
case <-c.closed:
|
2013-12-28 16:30:02 +01:00
|
|
|
return nil, ErrClosed
|
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Lock()
|
2014-05-11 19:30:29 +02:00
|
|
|
if ch := c.awaiting[id]; ch != nil {
|
2014-03-23 08:44:27 +01:00
|
|
|
panic("id taken")
|
|
|
|
}
|
2014-07-03 13:37:20 +02:00
|
|
|
rc := make(chan asyncResult, 1)
|
2014-05-04 17:40:40 +02:00
|
|
|
c.awaiting[id] = rc
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Unlock()
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
ok := c.send(id, messageTypeRequest, RequestMessage{repo, name, uint64(offset), uint32(size)})
|
2014-05-11 19:30:29 +02:00
|
|
|
if !ok {
|
|
|
|
return nil, ErrClosed
|
2013-12-20 00:01:34 +01:00
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2013-12-15 15:58:27 +01:00
|
|
|
res, ok := <-rc
|
|
|
|
if !ok {
|
|
|
|
return nil, ErrClosed
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
2013-12-15 15:58:27 +01:00
|
|
|
return res.val, res.err
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-04-13 15:28:26 +02:00
|
|
|
// ClusterConfig send the cluster configuration message to the peer and returns any error
|
|
|
|
func (c *rawConnection) ClusterConfig(config ClusterConfigMessage) {
|
2014-07-28 11:31:22 +02:00
|
|
|
c.send(-1, messageTypeClusterConfig, config)
|
2014-04-13 15:28:26 +02:00
|
|
|
}
|
|
|
|
|
2014-03-28 14:36:57 +01:00
|
|
|
func (c *rawConnection) ping() bool {
|
2014-05-11 19:30:29 +02:00
|
|
|
var id int
|
|
|
|
select {
|
|
|
|
case id = <-c.nextID:
|
|
|
|
case <-c.closed:
|
2013-12-31 02:52:36 +01:00
|
|
|
return false
|
2013-12-28 16:30:02 +01:00
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2013-12-30 21:27:20 +01:00
|
|
|
rc := make(chan asyncResult, 1)
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Lock()
|
2014-05-04 17:40:40 +02:00
|
|
|
c.awaiting[id] = rc
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Unlock()
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
ok := c.send(id, messageTypePing, nil)
|
2014-05-11 19:30:29 +02:00
|
|
|
if !ok {
|
2014-05-04 08:11:06 +02:00
|
|
|
return false
|
2013-12-20 00:01:34 +01:00
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2013-12-31 02:52:36 +01:00
|
|
|
res, ok := <-rc
|
|
|
|
return ok && res.err == nil
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
func (c *rawConnection) readerLoop() (err error) {
|
|
|
|
defer func() {
|
|
|
|
c.close(err)
|
|
|
|
}()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-c.closed:
|
|
|
|
return ErrClosed
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
hdr, msg, err := c.readMessage()
|
|
|
|
if err != nil {
|
2014-05-11 19:30:29 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
switch hdr.msgType {
|
|
|
|
case messageTypeIndex:
|
2014-07-02 21:33:30 +02:00
|
|
|
if c.state < stateCCRcvd {
|
|
|
|
return fmt.Errorf("protocol error: index message in state %d", c.state)
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
c.handleIndex(msg.(IndexMessage))
|
2014-07-02 21:33:30 +02:00
|
|
|
c.state = stateIdxRcvd
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypeIndexUpdate:
|
2014-07-02 21:33:30 +02:00
|
|
|
if c.state < stateIdxRcvd {
|
|
|
|
return fmt.Errorf("protocol error: index update message in state %d", c.state)
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
c.handleIndexUpdate(msg.(IndexMessage))
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypeRequest:
|
2014-07-02 21:33:30 +02:00
|
|
|
if c.state < stateIdxRcvd {
|
|
|
|
return fmt.Errorf("protocol error: request message in state %d", c.state)
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
// Requests are handled asynchronously
|
|
|
|
go c.handleRequest(hdr.msgID, msg.(RequestMessage))
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypeResponse:
|
2014-07-02 21:33:30 +02:00
|
|
|
if c.state < stateIdxRcvd {
|
|
|
|
return fmt.Errorf("protocol error: response message in state %d", c.state)
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
c.handleResponse(hdr.msgID, msg.(ResponseMessage))
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypePing:
|
2014-07-28 11:31:22 +02:00
|
|
|
c.send(hdr.msgID, messageTypePong, EmptyMessage{})
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypePong:
|
2014-07-28 11:31:22 +02:00
|
|
|
c.handlePong(hdr.msgID)
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case messageTypeClusterConfig:
|
2014-07-02 21:33:30 +02:00
|
|
|
if c.state != stateInitial {
|
|
|
|
return fmt.Errorf("protocol error: cluster config message in state %d", c.state)
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
go c.receiver.ClusterConfig(c.id, msg.(ClusterConfigMessage))
|
2014-07-02 21:33:30 +02:00
|
|
|
c.state = stateCCRcvd
|
2014-05-11 19:30:29 +02:00
|
|
|
|
2014-07-26 21:27:55 +02:00
|
|
|
case messageTypeClose:
|
2014-07-28 11:31:22 +02:00
|
|
|
return errors.New(msg.(CloseMessage).Reason)
|
2014-07-26 21:27:55 +02:00
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
default:
|
|
|
|
return fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
|
|
|
|
}
|
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
func (c *rawConnection) readMessage() (hdr header, msg encodable, err error) {
|
|
|
|
if cap(c.rdbuf0) < 8 {
|
|
|
|
c.rdbuf0 = make([]byte, 8)
|
2014-05-11 19:30:29 +02:00
|
|
|
} else {
|
2014-07-28 11:31:22 +02:00
|
|
|
c.rdbuf0 = c.rdbuf0[:8]
|
|
|
|
}
|
|
|
|
_, err = io.ReadFull(c.cr, c.rdbuf0)
|
|
|
|
if err != nil {
|
|
|
|
return
|
2014-05-04 17:40:40 +02:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
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)
|
2014-05-11 19:30:29 +02:00
|
|
|
} else {
|
2014-07-28 11:31:22 +02:00
|
|
|
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 11:54:15 +02:00
|
|
|
if debug {
|
2014-07-28 11:31:22 +02:00
|
|
|
l.Debugf("decompressed to %d bytes", len(msgBuf))
|
2014-07-23 11:54:15 +02:00
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02: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))
|
|
|
|
}
|
2014-05-11 19:30:29 +02:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
switch hdr.msgType {
|
|
|
|
case messageTypeIndex, messageTypeIndexUpdate:
|
|
|
|
var idx IndexMessage
|
|
|
|
err = idx.UnmarshalXDR(msgBuf)
|
|
|
|
msg = idx
|
2014-05-11 19:30:29 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
case messageTypeRequest:
|
|
|
|
var req RequestMessage
|
|
|
|
err = req.UnmarshalXDR(msgBuf)
|
|
|
|
msg = req
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
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 13:37:20 +02:00
|
|
|
}
|
2014-05-11 19:30:29 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
return
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
func (c *rawConnection) handleIndex(im IndexMessage) {
|
|
|
|
if debug {
|
|
|
|
l.Debugf("Index(%v, %v, %d files)", c.id, im.Repository, len(im.Files))
|
2014-05-04 08:11:06 +02:00
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
c.receiver.Index(c.id, im.Repository, im.Files)
|
2014-05-11 19:30:29 +02:00
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
func (c *rawConnection) handleIndexUpdate(im IndexMessage) {
|
|
|
|
if debug {
|
|
|
|
l.Debugf("queueing IndexUpdate(%v, %v, %d files)", c.id, im.Repository, len(im.Files))
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
c.receiver.IndexUpdate(c.id, im.Repository, im.Files)
|
2014-05-11 19:30:29 +02:00
|
|
|
}
|
2013-12-15 15:58:27 +01:00
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
func (c *rawConnection) handleRequest(msgID int, req RequestMessage) {
|
|
|
|
data, _ := c.receiver.Request(c.id, req.Repository, req.Name, int64(req.Offset), int(req.Size))
|
|
|
|
|
|
|
|
c.send(msgID, messageTypeResponse, ResponseMessage{data})
|
2014-07-26 21:27:55 +02:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02: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 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02: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 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-07-28 11:31:22 +02:00
|
|
|
func (c *rawConnection) send(msgID int, msgType int, msg encodable) bool {
|
|
|
|
if msgID < 0 {
|
2014-05-11 19:30:29 +02:00
|
|
|
select {
|
|
|
|
case id := <-c.nextID:
|
2014-07-28 11:31:22 +02:00
|
|
|
msgID = id
|
2014-05-11 19:30:29 +02:00
|
|
|
case <-c.closed:
|
|
|
|
return false
|
2013-12-21 08:06:54 +01:00
|
|
|
}
|
2014-05-11 19:30:29 +02:00
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
|
|
|
|
hdr := header{
|
|
|
|
version: 0,
|
|
|
|
msgID: msgID,
|
|
|
|
msgType: msgType,
|
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
select {
|
2014-07-28 11:31:22 +02:00
|
|
|
case c.outbox <- hdrMsg{hdr, msg}:
|
2014-05-11 19:30:29 +02:00
|
|
|
return true
|
|
|
|
case <-c.closed:
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
func (c *rawConnection) writerLoop() {
|
2014-07-28 11:31:22 +02:00
|
|
|
var msgBuf = make([]byte, 8) // buffer for wire format message, kept and reused
|
|
|
|
var uncBuf []byte // buffer for uncompressed message, kept and reused
|
2014-07-04 15:16:33 +02:00
|
|
|
for {
|
2014-07-28 11:31:22 +02:00
|
|
|
var tempBuf []byte
|
|
|
|
var err error
|
|
|
|
|
2014-07-04 15:16:33 +02:00
|
|
|
select {
|
2014-07-28 11:31:22 +02:00
|
|
|
case hm := <-c.outbox:
|
|
|
|
if hm.msg != nil {
|
|
|
|
// Uncompressed message in uncBuf
|
|
|
|
uncBuf = hm.msg.AppendXDR(uncBuf[:0])
|
|
|
|
|
2014-07-28 12:44:46 +02:00
|
|
|
if len(uncBuf) >= c.compressionThreshold {
|
2014-07-28 11:31:22 +02:00
|
|
|
// 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]
|
2014-07-04 15:16:33 +02:00
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
|
|
|
|
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 {
|
2014-07-04 15:16:33 +02:00
|
|
|
c.close(err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
case <-c.closed:
|
2014-05-11 19:30:29 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
func (c *rawConnection) close(err error) {
|
2014-07-03 13:37:20 +02:00
|
|
|
c.once.Do(func() {
|
2014-05-11 19:30:29 +02:00
|
|
|
close(c.closed)
|
2013-12-15 15:58:27 +01:00
|
|
|
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Lock()
|
2014-05-11 19:30:29 +02:00
|
|
|
for i, ch := range c.awaiting {
|
|
|
|
if ch != nil {
|
|
|
|
close(ch)
|
|
|
|
c.awaiting[i] = nil
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
2014-05-11 19:30:29 +02:00
|
|
|
}
|
2014-07-03 13:37:20 +02:00
|
|
|
c.awaitingMut.Unlock()
|
2014-02-09 23:13:06 +01:00
|
|
|
|
2014-05-12 02:35:44 +02:00
|
|
|
go c.receiver.Close(c.id, err)
|
2014-07-03 13:37:20 +02:00
|
|
|
})
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
|
2014-05-11 19:30:29 +02:00
|
|
|
func (c *rawConnection) idGenerator() {
|
|
|
|
nextID := 0
|
|
|
|
for {
|
|
|
|
nextID = (nextID + 1) & 0xfff
|
|
|
|
select {
|
|
|
|
case c.nextID <- nextID:
|
|
|
|
case <-c.closed:
|
|
|
|
return
|
|
|
|
}
|
2013-12-15 11:43:31 +01:00
|
|
|
}
|
|
|
|
}
|
2013-12-15 16:19:45 +01:00
|
|
|
|
2014-03-28 14:36:57 +01:00
|
|
|
func (c *rawConnection) pingerLoop() {
|
2013-12-31 02:52:36 +01:00
|
|
|
var rc = make(chan bool, 1)
|
2014-03-28 14:36:57 +01:00
|
|
|
ticker := time.Tick(pingIdleTime / 2)
|
2013-12-31 04:10:54 +01:00
|
|
|
for {
|
2014-03-28 14:36:57 +01:00
|
|
|
select {
|
|
|
|
case <-ticker:
|
2014-07-28 11:31:22 +02:00
|
|
|
if d := time.Since(c.cr.Last()); d < pingIdleTime {
|
2014-05-28 12:30:47 +02:00
|
|
|
if debug {
|
|
|
|
l.Debugln(c.id, "ping skipped after rd", d)
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
2014-07-28 11:31:22 +02:00
|
|
|
if d := time.Since(c.cw.Last()); d < pingIdleTime {
|
2014-05-28 12:30:47 +02:00
|
|
|
if debug {
|
|
|
|
l.Debugln(c.id, "ping skipped after wr", d)
|
|
|
|
}
|
|
|
|
continue
|
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
go func() {
|
2014-05-28 12:30:47 +02:00
|
|
|
if debug {
|
|
|
|
l.Debugln(c.id, "ping ->")
|
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
rc <- c.ping()
|
|
|
|
}()
|
|
|
|
select {
|
|
|
|
case ok := <-rc:
|
2014-05-28 20:45:29 +02:00
|
|
|
if debug {
|
|
|
|
l.Debugln(c.id, "<- pong")
|
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
if !ok {
|
|
|
|
c.close(fmt.Errorf("ping failure"))
|
2014-01-01 03:22:49 +01:00
|
|
|
}
|
2014-05-04 17:40:40 +02:00
|
|
|
case <-time.After(pingTimeout):
|
|
|
|
c.close(fmt.Errorf("ping timeout"))
|
2014-05-11 19:30:29 +02:00
|
|
|
case <-c.closed:
|
|
|
|
return
|
2013-12-15 16:19:45 +01:00
|
|
|
}
|
2014-05-11 19:30:29 +02:00
|
|
|
|
|
|
|
case <-c.closed:
|
|
|
|
return
|
2013-12-15 16:19:45 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2013-12-23 18:12:44 +01:00
|
|
|
|
|
|
|
type Statistics struct {
|
2014-01-05 16:16:37 +01:00
|
|
|
At time.Time
|
2014-06-01 21:56:05 +02:00
|
|
|
InBytesTotal uint64
|
|
|
|
OutBytesTotal uint64
|
2013-12-23 18:12:44 +01:00
|
|
|
}
|
|
|
|
|
2014-03-28 14:36:57 +01:00
|
|
|
func (c *rawConnection) Statistics() Statistics {
|
2014-03-29 18:53:48 +01:00
|
|
|
return Statistics{
|
2014-01-05 16:16:37 +01:00
|
|
|
At: time.Now(),
|
2014-06-01 21:56:05 +02:00
|
|
|
InBytesTotal: c.cr.Tot(),
|
|
|
|
OutBytesTotal: c.cw.Tot(),
|
2013-12-23 18:12:44 +01:00
|
|
|
}
|
|
|
|
}
|
2014-05-23 12:53:26 +02:00
|
|
|
|
|
|
|
func IsDeleted(bits uint32) bool {
|
|
|
|
return bits&FlagDeleted != 0
|
|
|
|
}
|
|
|
|
|
|
|
|
func IsInvalid(bits uint32) bool {
|
|
|
|
return bits&FlagInvalid != 0
|
|
|
|
}
|
|
|
|
|
|
|
|
func IsDirectory(bits uint32) bool {
|
|
|
|
return bits&FlagDirectory != 0
|
|
|
|
}
|
|
|
|
|
|
|
|
func HasPermissionBits(bits uint32) bool {
|
|
|
|
return bits&FlagNoPermBits == 0
|
|
|
|
}
|