garnet/go/src/thinfs/thinio/conductor.go - fuchsia - Git at Google

 // Copyright 2016 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 // Package thinio provides functionality for orchestrating I/O operations on a block.Device.
 package thinio

 import (
 	"errors"
 	"fmt"
 	"sync"
 	"time"

 	"thinfs/block"
 	"thinfs/cache"

 	"github.com/golang/glog"
 )

 // ErrClosed is returned if a caller attempts to perform any operations on a Conductor after
 // it has been closed.
 var ErrClosed = errors.New("conductor has been closed")

 type errKind int

 const (
 	readErr errKind = iota
 	writeErr
 	flushErr
 	closeErr
 )

 type devError struct {
 	kind errKind
 	off  int64
 	data []byte
 	err  error
 }

 func (de devError) Error() string {
 	switch de.kind {
 	case readErr:
 		return fmt.Sprintf("failed to read block at offset %#x: %v", de.off, de.err)
 	case writeErr:
 		return fmt.Sprintf("failed to write block at offset %#x: %v", de.off, de.err)
 	case flushErr:
 		return fmt.Sprintf("failed to flush device: %v", de.err)
 	case closeErr:
 		return fmt.Sprintf("failed to close device: %v", de.err)
 	default:
 		return "unknown error"
 	}
 }

 // device implements cache.BackingStore, converting cache.Keys to the corresponding
 // block device offsets and performing any necessary I/O.
 type device struct {
 	dev block.Device

 	// Tells the conductor that an error has occurred.
 	recordError func(error)
 }

 // Get implements cache.BackingStore.Get for device.
 func (d *device) Get(k cache.Key) cache.Value {
 	off := k.(int64)
 	if off%d.dev.BlockSize() != 0 {
 		// panic because this indicates an internal error.
 		panic(fmt.Sprintf("off (%#x) is not block size aligned\n", off))
 	}
 	if glog.V(2) {
 		glog.Infof("Fetching block at offset %#x from the device\n", off)
 	}

 	var err error
 	buf := make([]byte, d.dev.BlockSize())
 	for dur := time.Millisecond; dur < 2*time.Second; dur *= 2 {
 		_, err = d.dev.ReadAt(buf, off)
 		if err == nil {
 			return buf
 		}

 		glog.Warningf("Failed to read block at offset %#x from device: %v.  Retrying after %v\n", off, err, dur)
 		time.Sleep(dur)
 	}

 	d.recordError(devError{
 		kind: readErr,
 		off:  off,
 		err:  err,
 	})
 	return nil
 }

 // Put implements cache.BackingStore.Put for device.
 func (d *device) Put(k cache.Key, v cache.Value) {
 	off := k.(int64)
 	if off%d.dev.BlockSize() != 0 {
 		// panic because this indicates an internal error.
 		panic(fmt.Sprintf("off (%#x) is not block size aligned\n", off))
 	}
 	if glog.V(2) {
 		glog.Infof("Writing block at offset %#x to the device\n", off)
 	}

 	var err error
 	buf := v.([]byte)
 	for dur := time.Millisecond; dur < 2*time.Second; dur *= 2 {
 		_, err = d.dev.WriteAt(buf, off)
 		if err == nil {
 			return
 		}

 		glog.Warningf("Failed to write block at offset %#x to device: %v.  Retrying after %v\n", off, err, dur)
 		time.Sleep(dur)
 	}

 	d.recordError(devError{
 		kind: writeErr,
 		off:  off,
 		data: buf,
 		err:  err,
 	})
 }

 // BlockSize is a convenience function for getting the block size of the underlying block.Device.
 func (d *device) BlockSize() int64 {
 	return d.dev.BlockSize()
 }

 // Discard is a convenience function for calling Discard on the underlying block.Device.
 func (d *device) Discard(off, len int64) error {
 	return d.dev.Discard(off, len)
 }

 // Flush is a convenience function for calling Flush on the underlying block.Device.
 func (d *device) Flush() {
 	// Yes the error *could* be recorded here, but I don't want it popping up on another random
 	// function.
 	// TODO(smklein): Redo the error handling in this file; it doesn't make much sense to defer
 	// errors to unrelated functions.
 	d.dev.Flush()
 }

 // Close is a convenience function for calling Close on the underlying block.Device.
 func (d *device) Close() {
 	if err := d.dev.Close(); err != nil {
 		d.recordError(devError{
 			kind: closeErr,
 			err:  err,
 		})
 	}
 }

 // Conductor orchestrates all I/O operations on a block.Device, maintaining an internal cache
 // of disk blocks for better performance
 type Conductor struct {
 	// TODO(smklein): This lock is too coarse-grained and prevents parallel access to the device.
 	mu    sync.Mutex
 	dev   *device
 	cache *cache.C
 	errs  []error // TODO(chirantan): recover from error?
 }

 // NewConductor returns a new, initialized Conductor for orchestrating I/O on dev.  cacheSize
 // should be the size in bytes of the cache that the Conductor should use for caching disk blocks.
 func NewConductor(dev block.Device, cacheSize int) *Conductor {
 	numEntries := cacheSize / int(dev.BlockSize())
 	store := &device{dev: dev}

 	if glog.V(1) {
 		glog.Infof("Creating Conductor with a cache of %v entries holding %v bytes\n", numEntries, cacheSize)
 	}

 	conductor := &Conductor{
 		dev:   store,
 		cache: cache.New(numEntries, store),
 	}

 	store.recordError = func(err error) {
 		conductor.errs = append(conductor.errs, err)
 	}

 	return conductor
 }

 // DeviceSize returns the size of the underlying device in bytes.
 func (c *Conductor) DeviceSize() int64 {
 	return c.dev.dev.Size()
 }

 // ReadAt implements io.ReaderAt for Conductor.
 func (c *Conductor) ReadAt(p []byte, off int64) (int, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.errs) > 0 {
 		return 0, c.errs[0]
 	}

 	if glog.V(2) {
 		glog.Infof("Reading %v bytes from offset %v\n", len(p), off)
 	}

 	var n int
 	if rem := off % c.dev.BlockSize(); rem != 0 {
 		addr := off - rem

 		entry := c.cache.Get(addr)
 		if len(c.errs) > 0 {
 			return n, c.errs[0]
 		}

 		buf := entry.Value.([]byte)
 		copied := copy(p, buf[rem:])
 		n += copied
 		p = p[copied:]
 		off += int64(copied)
 	}

 	// Now we know that off is block size aligned.
 	for len(p) > 0 {
 		entry := c.cache.Get(off)
 		if len(c.errs) > 0 {
 			return n, c.errs[0]
 		}

 		buf := entry.Value.([]byte)
 		copied := copy(p, buf)
 		n += copied
 		p = p[copied:]
 		off += int64(copied)
 	}

 	return n, nil
 }

 // WriteAt implements io.WriterAt for Conductor.
 func (c *Conductor) WriteAt(p []byte, off int64) (int, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.errs) > 0 {
 		return 0, c.errs[0]
 	}

 	if glog.V(2) {
 		glog.Infof("Writing %v bytes to offset %v\n", len(p), off)
 	}

 	var n int
 	if rem := off % c.dev.BlockSize(); rem != 0 {
 		addr := off - rem

 		entry := c.cache.Get(addr)
 		if len(c.errs) > 0 {
 			return n, c.errs[0]
 		}

 		buf := entry.Value.([]byte)
 		copied := copy(buf[rem:], p)
 		entry.IsDirty = true
 		n += copied
 		p = p[copied:]
 		off += int64(copied)
 	}

 	// Now we know that off is block size aligned.
 	for int64(len(p)) >= c.dev.BlockSize() {
 		buf := make([]byte, c.dev.BlockSize())
 		copy(buf, p)

 		c.cache.Put(off, buf)
 		if len(c.errs) > 0 {
 			return n, c.errs[0]
 		}

 		copied := len(buf)
 		n += copied
 		p = p[copied:]
 		off += int64(copied)
 	}

 	if len(p) > 0 {
 		entry := c.cache.Get(off)
 		if len(c.errs) > 0 {
 			return n, c.errs[0]
 		}

 		buf := entry.Value.([]byte)
 		copied := copy(buf, p)
 		entry.IsDirty = true
 		n += copied
 		p = p[copied:]
 		off += int64(copied)
 	}

 	return n, nil
 }

 // Discard discards the data in the address range [off, off+size).  Returns an error, if any.
 func (c *Conductor) Discard(off, size int64) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	if glog.V(2) {
 		glog.Infof("Discarding data in range [%v, %v)\n", off, off+size)
 	}

 	// A failure to discard is not considered a fatal error so it is not persisted in c.errs.
 	return c.dev.Discard(off, size)
 }

 // Flush flushes the Conductor's in-memory copy of recently written data to the underlying
 // block.Device.  Flush does not return until all previous writes are committed to stable
 // storage. Returns an error, if any.
 func (c *Conductor) Flush() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	if glog.V(2) {
 		glog.Info("Committing all data in Conductor to stable storage")
 	}

 	c.cache.Flush()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	c.dev.Flush()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	return nil
 }

 // Close closes the Conductor as well as the underlying block.Device, rendering them unusable
 // for I/O.  Returns an error, if any. Future operations on the Conductor will return ErrClosed.
 // Callers must call Flush before calling Close to ensure that all pending changes have been
 // flushed to stable storage.
 func (c *Conductor) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	c.dev.Close()
 	if len(c.errs) > 0 {
 		return c.errs[0]
 	}

 	c.errs = []error{ErrClosed}
 	c.dev = nil

 	return nil
 }

 // Errors returns a slice of all the errors that have occurred during the Conductor's
 // operation, if any.  Callers must not modify the returned slice, which is invalidated
 // if any other method is invoked on c.
 func (c *Conductor) Errors() []error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	return c.errs
 }
	// Copyright 2016 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	// Package thinio provides functionality for orchestrating I/O operations on a block.Device.
	package thinio

	import (
	"errors"
	"fmt"
	"sync"
	"time"

	"thinfs/block"
	"thinfs/cache"

	"github.com/golang/glog"
	)

	// ErrClosed is returned if a caller attempts to perform any operations on a Conductor after
	// it has been closed.
	var ErrClosed = errors.New("conductor has been closed")

	type errKind int

	const (
	readErr errKind = iota
	writeErr
	flushErr
	closeErr
	)

	type devError struct {
	kind errKind
	off int64
	data []byte
	err error
	}

	func (de devError) Error() string {
	switch de.kind {
	case readErr:
	return fmt.Sprintf("failed to read block at offset %#x: %v", de.off, de.err)
	case writeErr:
	return fmt.Sprintf("failed to write block at offset %#x: %v", de.off, de.err)
	case flushErr:
	return fmt.Sprintf("failed to flush device: %v", de.err)
	case closeErr:
	return fmt.Sprintf("failed to close device: %v", de.err)
	default:
	return "unknown error"
	}
	}

	// device implements cache.BackingStore, converting cache.Keys to the corresponding
	// block device offsets and performing any necessary I/O.
	type device struct {
	dev block.Device

	// Tells the conductor that an error has occurred.
	recordError func(error)
	}

	// Get implements cache.BackingStore.Get for device.
	func (d *device) Get(k cache.Key) cache.Value {
	off := k.(int64)
	if off%d.dev.BlockSize() != 0 {
	// panic because this indicates an internal error.
	panic(fmt.Sprintf("off (%#x) is not block size aligned\n", off))
	}
	if glog.V(2) {
	glog.Infof("Fetching block at offset %#x from the device\n", off)
	}

	var err error
	buf := make([]byte, d.dev.BlockSize())
	for dur := time.Millisecond; dur < 2time.Second; dur = 2 {
	_, err = d.dev.ReadAt(buf, off)
	if err == nil {
	return buf
	}

	glog.Warningf("Failed to read block at offset %#x from device: %v. Retrying after %v\n", off, err, dur)
	time.Sleep(dur)
	}

	d.recordError(devError{
	kind: readErr,
	off: off,
	err: err,
	})
	return nil
	}

	// Put implements cache.BackingStore.Put for device.
	func (d *device) Put(k cache.Key, v cache.Value) {
	off := k.(int64)
	if off%d.dev.BlockSize() != 0 {
	// panic because this indicates an internal error.
	panic(fmt.Sprintf("off (%#x) is not block size aligned\n", off))
	}
	if glog.V(2) {
	glog.Infof("Writing block at offset %#x to the device\n", off)
	}

	var err error
	buf := v.([]byte)
	for dur := time.Millisecond; dur < 2time.Second; dur = 2 {
	_, err = d.dev.WriteAt(buf, off)
	if err == nil {
	return
	}

	glog.Warningf("Failed to write block at offset %#x to device: %v. Retrying after %v\n", off, err, dur)
	time.Sleep(dur)
	}

	d.recordError(devError{
	kind: writeErr,
	off: off,
	data: buf,
	err: err,
	})
	}

	// BlockSize is a convenience function for getting the block size of the underlying block.Device.
	func (d *device) BlockSize() int64 {
	return d.dev.BlockSize()
	}

	// Discard is a convenience function for calling Discard on the underlying block.Device.
	func (d *device) Discard(off, len int64) error {
	return d.dev.Discard(off, len)
	}

	// Flush is a convenience function for calling Flush on the underlying block.Device.
	func (d *device) Flush() {
	// Yes the error could be recorded here, but I don't want it popping up on another random
	// function.
	// TODO(smklein): Redo the error handling in this file; it doesn't make much sense to defer
	// errors to unrelated functions.
	d.dev.Flush()
	}

	// Close is a convenience function for calling Close on the underlying block.Device.
	func (d *device) Close() {
	if err := d.dev.Close(); err != nil {
	d.recordError(devError{
	kind: closeErr,
	err: err,
	})
	}
	}

	// Conductor orchestrates all I/O operations on a block.Device, maintaining an internal cache
	// of disk blocks for better performance
	type Conductor struct {
	// TODO(smklein): This lock is too coarse-grained and prevents parallel access to the device.
	mu sync.Mutex
	dev *device
	cache *cache.C
	errs []error // TODO(chirantan): recover from error?
	}

	// NewConductor returns a new, initialized Conductor for orchestrating I/O on dev. cacheSize
	// should be the size in bytes of the cache that the Conductor should use for caching disk blocks.
	func NewConductor(dev block.Device, cacheSize int) *Conductor {
	numEntries := cacheSize / int(dev.BlockSize())
	store := &device{dev: dev}

	if glog.V(1) {
	glog.Infof("Creating Conductor with a cache of %v entries holding %v bytes\n", numEntries, cacheSize)
	}

	conductor := &Conductor{
	dev: store,
	cache: cache.New(numEntries, store),
	}

	store.recordError = func(err error) {
	conductor.errs = append(conductor.errs, err)
	}

	return conductor
	}

	// DeviceSize returns the size of the underlying device in bytes.
	func (c *Conductor) DeviceSize() int64 {
	return c.dev.dev.Size()
	}

	// ReadAt implements io.ReaderAt for Conductor.
	func (c *Conductor) ReadAt(p []byte, off int64) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.errs) > 0 {
	return 0, c.errs[0]
	}

	if glog.V(2) {
	glog.Infof("Reading %v bytes from offset %v\n", len(p), off)
	}

	var n int
	if rem := off % c.dev.BlockSize(); rem != 0 {
	addr := off - rem

	entry := c.cache.Get(addr)
	if len(c.errs) > 0 {
	return n, c.errs[0]
	}

	buf := entry.Value.([]byte)
	copied := copy(p, buf[rem:])
	n += copied
	p = p[copied:]
	off += int64(copied)
	}

	// Now we know that off is block size aligned.
	for len(p) > 0 {
	entry := c.cache.Get(off)
	if len(c.errs) > 0 {
	return n, c.errs[0]
	}

	buf := entry.Value.([]byte)
	copied := copy(p, buf)
	n += copied
	p = p[copied:]
	off += int64(copied)
	}

	return n, nil
	}

	// WriteAt implements io.WriterAt for Conductor.
	func (c *Conductor) WriteAt(p []byte, off int64) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.errs) > 0 {
	return 0, c.errs[0]
	}

	if glog.V(2) {
	glog.Infof("Writing %v bytes to offset %v\n", len(p), off)
	}

	var n int
	if rem := off % c.dev.BlockSize(); rem != 0 {
	addr := off - rem

	entry := c.cache.Get(addr)
	if len(c.errs) > 0 {
	return n, c.errs[0]
	}

	buf := entry.Value.([]byte)
	copied := copy(buf[rem:], p)
	entry.IsDirty = true
	n += copied
	p = p[copied:]
	off += int64(copied)
	}

	// Now we know that off is block size aligned.
	for int64(len(p)) >= c.dev.BlockSize() {
	buf := make([]byte, c.dev.BlockSize())
	copy(buf, p)

	c.cache.Put(off, buf)
	if len(c.errs) > 0 {
	return n, c.errs[0]
	}

	copied := len(buf)
	n += copied
	p = p[copied:]
	off += int64(copied)
	}

	if len(p) > 0 {
	entry := c.cache.Get(off)
	if len(c.errs) > 0 {
	return n, c.errs[0]
	}

	buf := entry.Value.([]byte)
	copied := copy(buf, p)
	entry.IsDirty = true
	n += copied
	p = p[copied:]
	off += int64(copied)
	}

	return n, nil
	}

	// Discard discards the data in the address range [off, off+size). Returns an error, if any.
	func (c *Conductor) Discard(off, size int64) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	if glog.V(2) {
	glog.Infof("Discarding data in range [%v, %v)\n", off, off+size)
	}

	// A failure to discard is not considered a fatal error so it is not persisted in c.errs.
	return c.dev.Discard(off, size)
	}

	// Flush flushes the Conductor's in-memory copy of recently written data to the underlying
	// block.Device. Flush does not return until all previous writes are committed to stable
	// storage. Returns an error, if any.
	func (c *Conductor) Flush() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	if glog.V(2) {
	glog.Info("Committing all data in Conductor to stable storage")
	}

	c.cache.Flush()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	c.dev.Flush()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	return nil
	}

	// Close closes the Conductor as well as the underlying block.Device, rendering them unusable
	// for I/O. Returns an error, if any. Future operations on the Conductor will return ErrClosed.
	// Callers must call Flush before calling Close to ensure that all pending changes have been
	// flushed to stable storage.
	func (c *Conductor) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	c.dev.Close()
	if len(c.errs) > 0 {
	return c.errs[0]
	}

	c.errs = []error{ErrClosed}
	c.dev = nil

	return nil
	}

	// Errors returns a slice of all the errors that have occurred during the Conductor's
	// operation, if any. Callers must not modify the returned slice, which is invalidated
	// if any other method is invoked on c.
	func (c *Conductor) Errors() []error {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.errs
	}