blobdiffer.go

/*
 * Copyright 2016 ClusterHQ
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package blobdiffer

import (
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"
	"syscall"

	"github.com/ClusterHQ/fli/dl/datalayer"
	"github.com/ClusterHQ/fli/dl/delta"
	"github.com/ClusterHQ/fli/dl/filediffer/attrcmp"
	dlhash "github.com/ClusterHQ/fli/dl/hash"
	"github.com/ClusterHQ/fli/dl/record"
	"github.com/ClusterHQ/fli/errors"
)

type (
	// FileDiffer defines the file differ factory interface
	FileDiffer interface {
		DiffContents(f1, f2 FileInfo, target string, records chan<- record.Record, hf dlhash.Factory) error
	}

	// Factory is a helper for creating a new blob differ factory
	Factory struct {
		FileDiffer FileDiffer
		// Limit is the max number of file differs can run at the same time
		Limit int
		hf    dlhash.Factory
	}

	// SafeFactory is a helper for creating a new xattrs-based blob differ factory
	SafeFactory struct {
		FileDiffer FileDiffer
		// Limit is the max number of file differs can run at the same time
		Limit int
		hf    dlhash.Factory
	}

	differ struct {
		// Origin and target are the full mount point of a particular storage type, for example, on ZFS,
		// it is like: /chq/populated/.zfs/snapshot/populated
		origin, target string

		// Channel is where the syscall records are sent to
		records chan record.Record

		// File differ type
		fileDiffer FileDiffer

		// wgInt is the wait group used internally by all file differ go routines
		wgInt *sync.WaitGroup

		// wgExt is the wait group wait on by caller, differ needs to exit it
		wgExt *sync.WaitGroup

		// fileDifferQueue is to prevent too many file differs running at the same time, servers as an upper
		// limit of how many file differs can run simultaneously.
		fileDifferQueue chan struct{}

		// hf is the hash factory used to do record checksum
		hf dlhash.Factory

		// errc is the channel for file differ to report back errors to differ
		errc chan error

		// exErrCh is the channel for differ to report back errors to its caller
		exErrCh chan<- error

		// cancelCh is the channel for cancellation
		cancelCh <-chan bool
	}

	xattrDiffer struct {
		differ
	}

	// FileInfo ...
	FileInfo interface {
		io.ReadCloser
		Size() int64
		Path() string
	}

	fileInfoImpl struct {
		io.ReadCloser
		size  int64
		fpath string
	}

	attrDifferFn func(f1 string, f2 string, target string, records chan<- record.Record, hf dlhash.Factory) error
)

var (
	_ datalayer.BlobDifferFactory = Factory{}
	_ delta.DifferenceOperations  = &differ{}
)

// New creates a new BlobDiffer
func (f Factory) New(origin, target string, hf dlhash.Factory, exErrCh chan<- error,
	cancelCh <-chan bool, wg *sync.WaitGroup) <-chan record.Record {
	// TODO: To avoid changing too many places(cli and fli) right now.
	if f.Limit == 0 {
		// TODO: Chose a default or make all caller configurable
		f.Limit = 128
	}
	return newDiffer(origin, target, f.FileDiffer, f.Limit, hf, exErrCh, cancelCh, wg)
}

// New creates a new BlobDiffer
func (f SafeFactory) New(origin, target string, hf dlhash.Factory, exErrCh chan<- error,
	cancelCh <-chan bool, wg *sync.WaitGroup) <-chan record.Record {
	// TODO: To avoid changing too many places(cli and fli) right now.
	if f.Limit == 0 {
		// TODO: Chose a default or make all caller configurable
		f.Limit = 128
	}
	return newSafeDiffer(origin, target, f.FileDiffer, f.Limit, hf, exErrCh, cancelCh, wg)
}

// newDiffer starts the delta algorithm in a go routine.
// records generated by differ are sent to the channel for others to consume
func newDiffer(origin, target string, fileDiffer FileDiffer, fileDifferLimit int,
	hf dlhash.Factory, exErrCh chan<- error, cancelCh <-chan bool, wg *sync.WaitGroup) <-chan record.Record {
	d := &differ{
		origin:          origin,
		target:          target,
		fileDiffer:      fileDiffer,
		wgInt:           &sync.WaitGroup{},
		fileDifferQueue: make(chan struct{}, fileDifferLimit),
		hf:              hf,
		records:         make(chan record.Record, datalayer.DifferChannelSize),
		// Enough room for all file differs and the one sync executed differ request
		errc:     make(chan error, fileDifferLimit+1),
		exErrCh:  exErrCh,
		cancelCh: cancelCh,
		wgExt:    wg,
	}
	go d.run(origin, target)
	return d.records
}

// newDiffer starts the delta algorithm in a go routine.
// records generated by differ are sent to the channel for others to consume
func newSafeDiffer(origin, target string, fileDiffer FileDiffer, fileDifferLimit int,
	hf dlhash.Factory, exErrCh chan<- error, cancelCh <-chan bool, wg *sync.WaitGroup) <-chan record.Record {
	d := &xattrDiffer{differ{
		origin:          origin,
		target:          target,
		fileDiffer:      fileDiffer,
		wgInt:           &sync.WaitGroup{},
		fileDifferQueue: make(chan struct{}, fileDifferLimit),
		hf:              hf,
		records:         make(chan record.Record, datalayer.DifferChannelSize),
		errc:            make(chan error),
		exErrCh:         exErrCh,
		cancelCh:        cancelCh,
		wgExt:           wg,
	}}
	go d.run(origin, target)
	return d.records
}

// start starts the differ process.
func (d *differ) run(origin, target string) {
	defer d.wgExt.Done()
	err := delta.Alloc(origin, target, d, false).DoParsing()

	// Wait for all file differs to finish
	d.wgInt.Wait()

	// Send end of transfer
	record.Send(record.NewEOT(), d.records, d.hf)

	// All done, closing
	close(d.records)

	select {
	case errDiffer := <-d.errc:
		d.exErrCh <- errDiffer
	default:
		if err != nil {
			d.exErrCh <- errors.Errorf("Differ error %v", err)
		}
	}
}

//GetOriginPath gets the path of the origin file
func (d differ) GetOriginPath() string {
	return d.origin
}

//GetTargetPath gets the path of the target file
func (d differ) GetTargetPath() string {
	return d.target
}

// diffFile compares two files, gets all the pwrite syscall records that correspond to that diff, and sends those
// records to the channel provided by the differ.
// setmtime record is generated at the end of all pwrites to reflect the correct file mtime on the receiving side.
// origin and target are relative paths to differ's origin and target respectively
// If origin is "", it implies that it is diffing against a blank file(generate pwrites for a new file)
func (d differ) diffFile(origin, target string) {
	// If there are too many differs running already, wait; otherwise, start the back ground job
	d.fileDifferQueue <- struct{}{}

	p1 := ""
	if origin != "" {
		p1 = filepath.Join(d.origin, origin)
	}
	p2 := filepath.Join(d.target, target)

	d.wgInt.Add(1)
	go diffFile(p1, p2, target, d.records, d.wgInt, d.fileDiffer, diffAttrs, d.fileDifferQueue, d.errc, d.hf)
	return
}

func diffAttrs(f1 string, f2 string, target string, reords chan<- record.Record, hf dlhash.Factory) error {
	err := attrcmp.DiffXattrs(f1, f2, target, reords, hf)
	if err != nil {
		return err
	}

	// DiffAttrs should be called last because it deals with setting mtime.
	return attrcmp.DiffAttrs(f1, f2, target, reords, hf)
}

// diffFile opens the base and target, generate diff between them, send over to remote.
func diffFile(srcBaseFullPath string, srcTargetFullPath string, remoteTarget string, records chan<- record.Record,
	wg *sync.WaitGroup, fileDiffer FileDiffer, attrDiffer attrDifferFn, fileDifferQueue chan struct{},
	errc chan<- error, hf dlhash.Factory) error {
	defer wg.Done()
	defer func() { <-fileDifferQueue }()

	var (
		err error
		f1  FileInfo
		f2  FileInfo
	)

	if len(srcBaseFullPath) != 0 {
		f1, err = open(srcBaseFullPath)
		if err != nil {
			errc <- err
			return err
		}
		defer f1.Close()
	} else {
		f1 = fileInfoImpl{nil, 0, ""}
	}

	f2, err = open(srcTargetFullPath)
	if err != nil {
		errc <- err
		return err
	}
	defer f2.Close()

	// Truncate if the new file is smaller the the old one
	if f2.Size() < f1.Size() {
		err = record.Send(record.NewTruncate(srcTargetFullPath, f2.Size()), records, hf)
		if err != nil {
			errc <- err
			return err
		}
	}

	err = fileDiffer.DiffContents(f1, f2, remoteTarget, records, hf)
	if err != nil {
		errc <- err
		return err
	}

	err = attrDiffer(f1.Path(), f2.Path(), remoteTarget, records, hf)
	if err != nil {
		errc <- err
	}

	return nil
}

func (d differ) diffNewFile(target string) {
	d.diffFile("", target)
}

// ResolveDiff implements DifferenceOperations
func (d *differ) ResolveDiff(origin string, target string) error {
	// Note: origin and target are always the same as of the current delta code
	f1 := filepath.Join(d.origin, origin)
	info, err := os.Lstat(f1)
	if err != nil {
		return err
	}

	if info.IsDir() {
		// TODO: handle it, ignoring for now
		return nil
	}

	d.diffFile(origin, target)
	return nil
}

// Create implements DifferenceOperations
func (d *differ) Create(path string) error {
	err := record.Send(record.NewCreate(path, record.DefaultCreateMode), d.records, d.hf)
	if err != nil {
		return err
	}

	d.diffNewFile(path)
	return nil
}

// Link implements DifferenceOperations
func (d *differ) Link(path string, target string) error {
	return record.Send(record.NewHardlink(target, path), d.records, d.hf)
}

// Symlink implements DifferenceOperations
func (d *differ) Symlink(path string) error {
	oldname, err := os.Readlink(filepath.Join(d.GetTargetPath(), path))
	if err != nil {
		return err
	}

	return record.Send(record.NewSymlink(oldname, path), d.records, d.hf)
}

// Unlink implements DifferenceOperations
func (d *differ) Unlink(path string) error {
	return record.Send(record.NewRemove(path), d.records, d.hf)
}

// Rmdir implements DifferenceOperations
func (d *differ) Rmdir(path string) error {
	return record.Send(record.NewRemove(path), d.records, d.hf)
}

// Mkdir implements DifferenceOperations
func (d *differ) Mkdir(path string) error {
	return record.Send(record.NewMkdir(path, record.DefaultCreateMode), d.records, d.hf)
}

// Mknod implements DifferenceOperations
func (d *differ) Mknod(path string) error {
	fullpath := filepath.Join(d.GetTargetPath(), path)
	s := syscall.Stat_t{}

	if err := syscall.Stat(fullpath, &s); err != nil {
		return err
	}

	// Note: This is non-portable except for FIFOs.
	// See http://pubs.opengroup.org/onlinepubs/009695399/functions/mknod.html
	// Also, using Stat_t.Mode rather than FileInfo.Mode is not portable.
	return record.Send(record.NewMknod(path, s.Mode, int(s.Rdev)), d.records, d.hf)
}

// Stop implements DifferenceOperations
func (d *differ) Stop() bool {
	return len(d.errc) != 0 || len(d.cancelCh) != 0
}

// open opens the file and returns the file handle and its size in bytes
func open(path string) (FileInfo, error) {
	info, err := os.Lstat(path)
	if err != nil {
		return fileInfoImpl{}, fmt.Errorf("Failed to stat %s", path)
	}
	fd, err := os.Open(path)
	if err != nil {
		return fileInfoImpl{}, fmt.Errorf("Failed to open %s", path)
	}
	return fileInfoImpl{ReadCloser: fd, size: info.Size(), fpath: path}, nil
}

func (fi fileInfoImpl) Size() int64 {
	return fi.size
}

func (fi fileInfoImpl) Path() string {
	return fi.fpath
}

// Mknod implements DifferenceOperations
func (d *xattrDiffer) Mknod(path string) error {
	fullpath := filepath.Join(d.GetTargetPath(), path)
	devmode, err := attrcmp.GetUintXattr(fullpath, "devmode")
	if err != nil {
		return err
	}
	dev, err := attrcmp.GetUintXattr(fullpath, "dev")
	if err != nil {
		return err
	}

	return record.Send(record.NewMknod(path, uint32(devmode), int(dev)), d.records, d.hf)
}

// ResolveDiff implements DifferenceOperations
func (d *xattrDiffer) ResolveDiff(origin string, target string) error {
	// Note: origin and target are always the same as of the current delta code
	f1 := filepath.Join(d.origin, origin)
	info, err := os.Lstat(f1)
	if err != nil {
		return err
	}

	if info.IsDir() {
		// TODO: handle it, ignoring for now
		return nil
	}

	return d.diffFile(origin, target)
}

// Create implements DifferenceOperations
func (d *xattrDiffer) Create(path string) error {
	err := record.Send(record.NewCreate(path, record.DefaultCreateMode), d.records, d.hf)
	if err != nil {
		return err
	}

	return d.diffFile("", path)
}

func (d xattrDiffer) diffFile(origin, target string) error {
	// If there are too many differs running already, wait; otherwise, start the back ground job
	d.fileDifferQueue <- struct{}{}

	p1 := ""
	if origin != "" {
		p1 = filepath.Join(d.origin, origin)
	}
	p2 := filepath.Join(d.target, target)

	d.wgInt.Add(1)
	go diffFile(p1, p2, target, d.records, d.wgInt, d.fileDiffer, diffXattrAttrs, d.fileDifferQueue, d.errc, d.hf)
	return nil
}

func diffXattrAttrs(f1 string, f2 string, target string, records chan<- record.Record, hf dlhash.Factory) error {
	err := attrcmp.DiffXattrsInXattrs(f1, f2, target, records, hf)
	if err != nil {
		return err
	}

	// DiffAttrs should be called last because it deals with setting mtime.
	return attrcmp.DiffAttrsInXattrs(f1, f2, target, records, hf)
}