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