Revert "Fix protocol close test"

This reverts commit 92c1ce57a6.

build.sh

@@ -11,6 +11,8 @@ host=${host%%.*}
 ldflags="-w -X main.Version $version -X main.BuildStamp $date -X main.BuildUser $user -X main.BuildHost $host"
 
 build() {
+	go vet ./... || exit 1
+
 	if command -v godep >/dev/null ; then
 		godep=godep
 	else

cmd/stcli/logger.go

@@ -1,72 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"log"
-	"os"
-)
-
-var logger *log.Logger
-
-func init() {
-	log.SetOutput(os.Stderr)
-	logger = log.New(os.Stderr, "", log.Flags())
-}
-
-func debugln(vals ...interface{}) {
-	s := fmt.Sprintln(vals...)
-	logger.Output(2, "DEBUG: "+s)
-}
-
-func debugf(format string, vals ...interface{}) {
-	s := fmt.Sprintf(format, vals...)
-	logger.Output(2, "DEBUG: "+s)
-}
-
-func infoln(vals ...interface{}) {
-	s := fmt.Sprintln(vals...)
-	logger.Output(2, "INFO: "+s)
-}
-
-func infof(format string, vals ...interface{}) {
-	s := fmt.Sprintf(format, vals...)
-	logger.Output(2, "INFO: "+s)
-}
-
-func okln(vals ...interface{}) {
-	s := fmt.Sprintln(vals...)
-	logger.Output(2, "OK: "+s)
-}
-
-func okf(format string, vals ...interface{}) {
-	s := fmt.Sprintf(format, vals...)
-	logger.Output(2, "OK: "+s)
-}
-
-func warnln(vals ...interface{}) {
-	s := fmt.Sprintln(vals...)
-	logger.Output(2, "WARNING: "+s)
-}
-
-func warnf(format string, vals ...interface{}) {
-	s := fmt.Sprintf(format, vals...)
-	logger.Output(2, "WARNING: "+s)
-}
-
-func fatalln(vals ...interface{}) {
-	s := fmt.Sprintln(vals...)
-	logger.Output(2, "FATAL: "+s)
-	os.Exit(3)
-}
-
-func fatalf(format string, vals ...interface{}) {
-	s := fmt.Sprintf(format, vals...)
-	logger.Output(2, "FATAL: "+s)
-	os.Exit(3)
-}
-
-func fatalErr(err error) {
-	if err != nil {
-		fatalf(err.Error())
-	}
-}

cmd/stcli/tls.go

@@ -1,24 +1,11 @@
 package main
 
 import (
-	"crypto/rand"
-	"crypto/rsa"
 	"crypto/sha256"
 	"crypto/tls"
-	"crypto/x509"
-	"crypto/x509/pkix"
 	"encoding/base32"
-	"encoding/pem"
-	"math/big"
-	"os"
 	"path/filepath"
 	"strings"
-	"time"
-)
-
-const (
-	tlsRSABits = 3072
-	tlsName    = "syncthing"
 )
 
 func loadCert(dir string) (tls.Certificate, error) {
@@ -31,41 +18,3 @@ func certID(bs []byte) string {
 	id := hf.Sum(nil)
 	return strings.Trim(base32.StdEncoding.EncodeToString(id), "=")
 }
-
-func newCertificate(dir string) {
-	infoln("Generating RSA certificate and key...")
-
-	priv, err := rsa.GenerateKey(rand.Reader, tlsRSABits)
-	fatalErr(err)
-
-	notBefore := time.Now()
-	notAfter := time.Date(2049, 12, 31, 23, 59, 59, 0, time.UTC)
-
-	template := x509.Certificate{
-		SerialNumber: new(big.Int).SetInt64(0),
-		Subject: pkix.Name{
-			CommonName: tlsName,
-		},
-		NotBefore: notBefore,
-		NotAfter:  notAfter,
-
-		KeyUsage:              x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
-		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
-		BasicConstraintsValid: true,
-	}
-
-	derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv)
-	fatalErr(err)
-
-	certOut, err := os.Create(filepath.Join(dir, "cert.pem"))
-	fatalErr(err)
-	pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes})
-	certOut.Close()
-	okln("Created RSA certificate file")
-
-	keyOut, err := os.OpenFile(filepath.Join(dir, "key.pem"), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
-	fatalErr(err)
-	pem.Encode(keyOut, &pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(priv)})
-	keyOut.Close()
-	okln("Created RSA key file")
-}

cmd/syncthing/gui.go

@@ -14,7 +14,6 @@ import (
 	"time"
 
 	"code.google.com/p/go.crypto/bcrypt"
-	"github.com/calmh/syncthing/scanner"
 	"github.com/codegangsta/martini"
 )
 
@@ -166,23 +165,6 @@ func restPostReset(req *http.Request) {
 	go restart()
 }
 
-type guiFile scanner.File
-
-func (f guiFile) MarshalJSON() ([]byte, error) {
-	type t struct {
-		Name     string
-		Size     int64
-		Modified int64
-		Flags    uint32
-	}
-	return json.Marshal(t{
-		Name:     f.Name,
-		Size:     scanner.File(f).Size,
-		Modified: f.Modified,
-		Flags:    f.Flags,
-	})
-}
-
 var cpuUsagePercent [10]float64 // The last ten seconds
 var cpuUsageLock sync.RWMutex
 

cmd/syncthing/model.go

@@ -423,16 +423,14 @@ func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection)
 	cm := m.clusterConfig(nodeID)
 	protoConn.ClusterConfig(cm)
 
-	var idxToSend = make(map[string][]protocol.FileInfo)
-
-	m.rmut.RLock()
-	for _, repo := range m.nodeRepos[nodeID] {
-		idxToSend[repo] = m.protocolIndex(repo)
-	}
-	m.rmut.RUnlock()
-
 	go func() {
-		for repo, idx := range idxToSend {
+		m.rmut.RLock()
+		repos := m.nodeRepos[nodeID]
+		m.rmut.RUnlock()
+		for _, repo := range repos {
+			m.rmut.RLock()
+			idx := m.protocolIndex(repo)
+			m.rmut.RUnlock()
 			if debugNet {
 				dlog.Printf("IDX(out/initial): %s: %q: %d files", nodeID, repo, len(idx))
 			}

cmd/syncthing/util.go

@@ -10,32 +10,6 @@ import (
 	"github.com/calmh/syncthing/scanner"
 )
 
-func MetricPrefix(n int64) string {
-	if n > 1e9 {
-		return fmt.Sprintf("%.02f G", float64(n)/1e9)
-	}
-	if n > 1e6 {
-		return fmt.Sprintf("%.02f M", float64(n)/1e6)
-	}
-	if n > 1e3 {
-		return fmt.Sprintf("%.01f k", float64(n)/1e3)
-	}
-	return fmt.Sprintf("%d ", n)
-}
-
-func BinaryPrefix(n int64) string {
-	if n > 1<<30 {
-		return fmt.Sprintf("%.02f Gi", float64(n)/(1<<30))
-	}
-	if n > 1<<20 {
-		return fmt.Sprintf("%.02f Mi", float64(n)/(1<<20))
-	}
-	if n > 1<<10 {
-		return fmt.Sprintf("%.01f Ki", float64(n)/(1<<10))
-	}
-	return fmt.Sprintf("%d ", n)
-}
-
 func Rename(from, to string) error {
 	if runtime.GOOS == "windows" {
 		err := os.Remove(to)

protocol/protocol.go

@@ -64,26 +64,24 @@ type Connection interface {
 }
 
 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
-
-	close  chan error
-	closed chan struct{}
+	sync.RWMutex
 
+	id        string
+	receiver  Model
+	reader    io.ReadCloser
+	cr        *countingReader
+	xr        *xdr.Reader
+	writer    io.WriteCloser
+	cw        *countingWriter
+	wb        *bufio.Writer
+	xw        *xdr.Writer
+	closed    chan struct{}
 	awaiting  map[int]chan asyncResult
 	nextID    int
 	indexSent map[string]map[string][2]int64
-	imut      sync.Mutex
+
+	hasSentIndex  bool
+	hasRecvdIndex bool
 }
 
 type asyncResult struct {
@@ -117,13 +115,11 @@ func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver M
 		cw:        cw,
 		wb:        wb,
 		xw:        xdr.NewWriter(wb),
-		close:     make(chan error),
 		closed:    make(chan struct{}),
 		awaiting:  make(map[int]chan asyncResult),
 		indexSent: make(map[string]map[string][2]int64),
 	}
 
-	go c.closer()
 	go c.readerLoop()
 	go c.pingerLoop()
 
@@ -136,11 +132,11 @@ func (c *rawConnection) ID() string {
 
 // 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
 	}
-
-	c.imut.Lock()
 	var msgType int
 	if c.indexSent[repo] == nil {
 		// This is the first time we send an index.
@@ -163,48 +159,45 @@ func (c *rawConnection) Index(repo string, idx []FileInfo) {
 		idx = diff
 	}
 
-	id := c.nextID
+	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.imut.Unlock()
-
-	c.wmut.Lock()
-	header{0, id, msgType}.encodeXDR(c.xw)
-	IndexMessage{repo, idx}.encodeXDR(c.xw)
-	err := c.flush()
-	c.wmut.Unlock()
+	c.hasSentIndex = true
+	c.Unlock()
 
 	if err != nil {
-		c.close <- err
+		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
 	}
-
-	c.imut.Lock()
-	id := c.nextID
-	c.nextID = (c.nextID + 1) & 0xfff
 	rc := make(chan asyncResult)
-	if _, ok := c.awaiting[id]; ok {
+	if _, ok := c.awaiting[c.nextID]; ok {
 		panic("id taken")
 	}
-	c.awaiting[id] = rc
-	c.imut.Unlock()
-
-	c.wmut.Lock()
-	header{0, id, messageTypeRequest}.encodeXDR(c.xw)
-	RequestMessage{repo, name, uint64(offset), uint32(size)}.encodeXDR(c.xw)
-	err := c.flush()
-	c.wmut.Unlock()
-
+	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.close <- err
+		c.Unlock()
+		c.close(err)
 		return nil, err
 	}
+	c.nextID = (c.nextID + 1) & 0xfff
+	c.Unlock()
 
 	res, ok := <-rc
 	if !ok {
@@ -215,47 +208,46 @@ func (c *rawConnection) Request(repo string, name string, offset int64, size int
 
 // 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
 	}
 
-	c.imut.Lock()
-	id := c.nextID
+	header{0, c.nextID, messageTypeClusterConfig}.encodeXDR(c.xw)
 	c.nextID = (c.nextID + 1) & 0xfff
-	c.imut.Unlock()
-
-	c.wmut.Lock()
-	header{0, id, messageTypeClusterConfig}.encodeXDR(c.xw)
-	config.encodeXDR(c.xw)
-	err := c.flush()
-	c.wmut.Unlock()
 
+	_, err := config.encodeXDR(c.xw)
+	if err == nil {
+		err = c.flush()
+	}
 	if err != nil {
-		c.close <- err
+		c.close(err)
 	}
 }
 
 func (c *rawConnection) ping() bool {
+	c.Lock()
 	if c.isClosed() {
+		c.Unlock()
 		return false
 	}
-
-	c.imut.Lock()
-	id := c.nextID
-	c.nextID = (c.nextID + 1) & 0xfff
 	rc := make(chan asyncResult, 1)
-	c.awaiting[id] = rc
-	c.imut.Unlock()
-
-	c.wmut.Lock()
-	header{0, id, messageTypePing}.encodeXDR(c.xw)
+	c.awaiting[c.nextID] = rc
+	header{0, c.nextID, messageTypePing}.encodeXDR(c.xw)
 	err := c.flush()
-	c.wmut.Unlock()
-
 	if err != nil {
-		c.close <- err
+		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
@@ -266,24 +258,21 @@ type flusher interface {
 }
 
 func (c *rawConnection) flush() error {
-	if err := c.xw.Error(); err != nil {
-		return err
-	}
-
-	if err := c.wb.Flush(); err != nil {
-		return err
-	}
-
+	c.wb.Flush()
 	if f, ok := c.writer.(flusher); ok {
 		return f.Flush()
 	}
-
 	return nil
 }
 
-func (c *rawConnection) closer() {
-	err := <-c.close
-
+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)
@@ -291,6 +280,7 @@ func (c *rawConnection) closer() {
 	c.awaiting = nil
 	c.writer.Close()
 	c.reader.Close()
+	c.Unlock()
 
 	c.receiver.Close(c.id, err)
 }
@@ -309,12 +299,12 @@ loop:
 	for !c.isClosed() {
 		var hdr header
 		hdr.decodeXDR(c.xr)
-		if err := c.xr.Error(); err != nil {
-			c.close <- err
+		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)
+			c.close(fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version))
 			break loop
 		}
 
@@ -322,8 +312,8 @@ loop:
 		case messageTypeIndex:
 			var im IndexMessage
 			im.decodeXDR(c.xr)
-			if err := c.xr.Error(); err != nil {
-				c.close <- err
+			if c.xr.Error() != nil {
+				c.close(c.xr.Error())
 				break loop
 			} else {
 
@@ -336,12 +326,15 @@ loop:
 
 				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 err := c.xr.Error(); err != nil {
-				c.close <- err
+			if c.xr.Error() != nil {
+				c.close(c.xr.Error())
 				break loop
 			} else {
 				go c.receiver.IndexUpdate(c.id, im.Repository, im.Files)
@@ -350,8 +343,8 @@ loop:
 		case messageTypeRequest:
 			var req RequestMessage
 			req.decodeXDR(c.xr)
-			if err := c.xr.Error(); err != nil {
-				c.close <- err
+			if c.xr.Error() != nil {
+				c.close(c.xr.Error())
 				break loop
 			}
 			go c.processRequest(hdr.msgID, req)
@@ -359,16 +352,16 @@ loop:
 		case messageTypeResponse:
 			data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size
 
-			if err := c.xr.Error(); err != nil {
-				c.close <- err
+			if c.xr.Error() != nil {
+				c.close(c.xr.Error())
 				break loop
 			}
 
 			go func(hdr header, err error) {
-				c.imut.Lock()
+				c.Lock()
 				rc, ok := c.awaiting[hdr.msgID]
 				delete(c.awaiting, hdr.msgID)
-				c.imut.Unlock()
+				c.Unlock()
 
 				if ok {
 					rc <- asyncResult{data, err}
@@ -377,41 +370,44 @@ loop:
 			}(hdr, c.xr.Error())
 
 		case messageTypePing:
-			c.wmut.Lock()
+			c.Lock()
 			header{0, hdr.msgID, messageTypePong}.encodeXDR(c.xw)
 			err := c.flush()
-			c.wmut.Unlock()
+			c.Unlock()
 			if err != nil {
-				c.close <- err
+				c.close(err)
+				break loop
+			} else if c.xw.Error() != nil {
+				c.close(c.xw.Error())
 				break loop
 			}
 
 		case messageTypePong:
-			c.imut.Lock()
+			c.RLock()
 			rc, ok := c.awaiting[hdr.msgID]
+			c.RUnlock()
 
 			if ok {
-				go func() {
-					rc <- asyncResult{}
-					close(rc)
-				}()
+				rc <- asyncResult{}
+				close(rc)
 
+				c.Lock()
 				delete(c.awaiting, hdr.msgID)
+				c.Unlock()
 			}
-			c.imut.Unlock()
 
 		case messageTypeClusterConfig:
 			var cm ClusterConfigMessage
 			cm.decodeXDR(c.xr)
-			if err := c.xr.Error(); err != nil {
-				c.close <- err
+			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)
+			c.close(fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType))
 			break loop
 		}
 	}
@@ -420,16 +416,17 @@ 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.wmut.Lock()
+	c.Lock()
 	header{0, msgID, messageTypeResponse}.encodeXDR(c.xw)
-	c.xw.WriteBytes(data)
-	err := c.flush()
-	c.wmut.Unlock()
+	_, err := c.xw.WriteBytes(data)
+	if err == nil {
+		err = c.flush()
+	}
+	c.Unlock()
 
 	buffers.Put(data)
-
 	if err != nil {
-		c.close <- err
+		c.close(err)
 	}
 }
 
@@ -439,16 +436,22 @@ func (c *rawConnection) pingerLoop() {
 	for {
 		select {
 		case <-ticker:
-			go func() {
-				rc <- c.ping()
-			}()
-			select {
-			case ok := <-rc:
-				if !ok {
-					c.close <- fmt.Errorf("ping failure")
+			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 <-time.After(pingTimeout):
-				c.close <- fmt.Errorf("ping timeout")
 			}
 		case <-c.closed:
 			return

protocol/protocol_test.go

@@ -5,7 +5,6 @@ import (
 	"io"
 	"testing"
 	"testing/quick"
-	"time"
 )
 
 func TestHeaderFunctions(t *testing.T) {
@@ -173,13 +172,7 @@ func TestClose(t *testing.T) {
 	c0 := NewConnection("c0", ar, bw, m0).(wireFormatConnection).next.(*rawConnection)
 	NewConnection("c1", br, aw, m1)
 
-	c0.close <- nil
-
-	select {
-	case <-c0.closed:
-	case <-time.After(1 * time.Second):
-		t.Fatal("Did not close within a second")
-	}
+	c0.close(nil)
 
 	if !c0.isClosed() {
 		t.Fatal("Connection should be closed")