services/keep/src/keep/volume.go

   1 // A Volume is an interface representing a Keep back-end storage unit:
   2 // for example, a single mounted disk, a RAID array, an Amazon S3 volume,
   3 // etc.
   4
   5 package main
   6
   7 import (
   8         "errors"
   9         "fmt"
  10         "strings"
  11         "os"
  12 )
  13
  14 type Volume interface {
  15         Get(loc string) ([]byte, error)
  16         Put(loc string, block []byte) error
  17         Index(prefix string) string
  18         Status() *VolumeStatus
  19         String() string
  20 }
  21
  22 // MockVolumes are Volumes used to test the Keep front end.
  23 //
  24 // If the Bad field is true, this volume should return an error
  25 // on all writes and puts.
  26 //
  27 type MockVolume struct {
  28         Store map[string][]byte
  29         Bad   bool
  30 }
  31
  32 func CreateMockVolume() *MockVolume {
  33         return &MockVolume{make(map[string][]byte), false}
  34 }
  35
  36 func (v *MockVolume) Get(loc string) ([]byte, error) {
  37         if v.Bad {
  38                 return nil, errors.New("Bad volume")
  39         } else if block, ok := v.Store[loc]; ok {
  40                 return block, nil
  41         }
  42         return nil, os.ErrNotExist
  43 }
  44
  45 func (v *MockVolume) Put(loc string, block []byte) error {
  46         if v.Bad {
  47                 return errors.New("Bad volume")
  48         }
  49         v.Store[loc] = block
  50         return nil
  51 }
  52
  53 func (v *MockVolume) Index(prefix string) string {
  54         var result string
  55         for loc, block := range v.Store {
  56                 if IsValidLocator(loc) && strings.HasPrefix(loc, prefix) {
  57                         result = result + fmt.Sprintf("%s+%d %d\n",
  58                                 loc, len(block), 123456789)
  59                 }
  60         }
  61         return result
  62 }
  63
  64 func (v *MockVolume) Status() *VolumeStatus {
  65         var used uint64
  66         for _, block := range v.Store {
  67                 used = used + uint64(len(block))
  68         }
  69         return &VolumeStatus{"/bogo", 123, 1000000 - used, used}
  70 }
  71
  72 func (v *MockVolume) String() string {
  73         return "[MockVolume]"
  74 }
  75
  76 // A VolumeManager manages a collection of volumes.
  77 //
  78 // - Volumes is a slice of available Volumes.
  79 // - Choose() returns a Volume suitable for writing to.
  80 // - Quit() instructs the VolumeManager to shut down gracefully.
  81 //
  82 type VolumeManager interface {
  83         Volumes() []Volume
  84         Choose() Volume
  85         Quit()
  86 }
  87
  88 type RRVolumeManager struct {
  89         volumes   []Volume
  90         nextwrite chan Volume
  91         quit      chan int
  92 }
  93
  94 func MakeRRVolumeManager(vols []Volume) *RRVolumeManager {
  95         // Create a new VolumeManager struct with the specified volumes,
  96         // and with new Nextwrite and Quit channels.
  97         // The Quit channel is buffered with a capacity of 1 so that
  98         // another routine may write to it without blocking.
  99         vm := &RRVolumeManager{vols, make(chan Volume), make(chan int, 1)}
 100
 101         // This goroutine implements round-robin volume selection.
 102         // It sends each available Volume in turn to the Nextwrite
 103         // channel, until receiving a notification on the Quit channel
 104         // that it should terminate.
 105         go func() {
 106                 var i int = 0
 107                 for {
 108                         select {
 109                         case <-vm.quit:
 110                                 return
 111                         case vm.nextwrite <- vm.volumes[i]:
 112                                 i = (i + 1) % len(vm.volumes)
 113                         }
 114                 }
 115         }()
 116
 117         return vm
 118 }
 119
 120 func (vm *RRVolumeManager) Volumes() []Volume {
 121         return vm.volumes
 122 }
 123
 124 func (vm *RRVolumeManager) Choose() Volume {
 125         return <-vm.nextwrite
 126 }
 127
 128 func (vm *RRVolumeManager) Quit() {
 129         vm.quit <- 1
 130 }