syncthing/cmd/syncthing/filemonitor.go

package main

import (
	"bytes"
	"errors"
	"fmt"
	"os"
	"path"
	"sync"
	"time"

	"github.com/calmh/syncthing/buffers"
	"github.com/calmh/syncthing/scanner"
)

type fileMonitor struct {
	name        string // in-repo name
	path        string // full path
	writeDone   sync.WaitGroup
	model       *Model
	global      scanner.File
	localBlocks []scanner.Block
	copyError   error
	writeError  error
}

func (m *fileMonitor) FileBegins(cc <-chan content) error {
	if debugPull {
		dlog.Println("file begins:", m.name)
	}

	tmp := defTempNamer.TempName(m.path)

	dir := path.Dir(tmp)
	_, err := os.Stat(dir)
	if err != nil && os.IsNotExist(err) {
		err = os.MkdirAll(dir, 0777)
		if err != nil {
			return err
		}
	}

	outFile, err := os.Create(tmp)
	if err != nil {
		return err
	}

	m.writeDone.Add(1)

	var writeWg sync.WaitGroup
	if len(m.localBlocks) > 0 {
		writeWg.Add(1)
		inFile, err := os.Open(m.path)
		if err != nil {
			return err
		}

		// Copy local blocks, close infile when done
		go m.copyLocalBlocks(inFile, outFile, &writeWg)
	}

	// Write remote blocks,
	writeWg.Add(1)
	go m.copyRemoteBlocks(cc, outFile, &writeWg)

	// Wait for both writing routines, then close the outfile
	go func() {
		writeWg.Wait()
		outFile.Close()
		m.writeDone.Done()
	}()

	return nil
}

func (m *fileMonitor) copyLocalBlocks(inFile, outFile *os.File, writeWg *sync.WaitGroup) {
	defer inFile.Close()
	defer writeWg.Done()

	var buf = buffers.Get(BlockSize)
	defer buffers.Put(buf)

	for _, lb := range m.localBlocks {
		buf = buf[:lb.Size]
		_, err := inFile.ReadAt(buf, lb.Offset)
		if err != nil {
			m.copyError = err
			return
		}
		_, err = outFile.WriteAt(buf, lb.Offset)
		if err != nil {
			m.copyError = err
			return
		}
	}
}

func (m *fileMonitor) copyRemoteBlocks(cc <-chan content, outFile *os.File, writeWg *sync.WaitGroup) {
	defer writeWg.Done()

	for content := range cc {
		_, err := outFile.WriteAt(content.data, content.offset)
		buffers.Put(content.data)
		if err != nil {
			m.writeError = err
			return
		}
	}
}

func (m *fileMonitor) FileDone() error {
	if debugPull {
		dlog.Println("file done:", m.name)
	}

	m.writeDone.Wait()

	tmp := defTempNamer.TempName(m.path)
	defer os.Remove(tmp)

	if m.copyError != nil {
		return m.copyError
	}
	if m.writeError != nil {
		return m.writeError
	}

	err := hashCheck(tmp, m.global.Blocks)
	if err != nil {
		return err
	}

	err = os.Chtimes(tmp, time.Unix(m.global.Modified, 0), time.Unix(m.global.Modified, 0))
	if err != nil {
		return err
	}

	err = os.Chmod(tmp, os.FileMode(m.global.Flags&0777))
	if err != nil {
		return err
	}

	err = os.Rename(tmp, m.path)
	if err != nil {
		return err
	}

	m.model.updateLocal(m.global)
	return nil
}

func hashCheck(name string, correct []scanner.Block) error {
	rf, err := os.Open(name)
	if err != nil {
		return err
	}
	defer rf.Close()

	current, err := scanner.Blocks(rf, BlockSize)
	if err != nil {
		return err
	}
	if len(current) != len(correct) {
		return errors.New("incorrect number of blocks")
	}
	for i := range current {
		if bytes.Compare(current[i].Hash, correct[i].Hash) != 0 {
			return fmt.Errorf("hash mismatch: %x != %x", current[i], correct[i])
		}
	}

	return nil
}