4523: Change description column from text to string on jobs, groups, pipeline_instanc...
[arvados.git] / services / keepstore / 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         "os"
11         "strings"
12         "time"
13 )
14
15 type Volume interface {
16         Get(loc string) ([]byte, error)
17         Put(loc string, block []byte) error
18         Touch(loc string) error
19         Mtime(loc string) (time.Time, error)
20         Index(prefix string) string
21         Delete(loc string) error
22         Status() *VolumeStatus
23         String() string
24 }
25
26 // MockVolumes are Volumes used to test the Keep front end.
27 //
28 // If the Bad field is true, this volume should return an error
29 // on all writes and puts.
30 //
31 // The Touchable field signifies whether the Touch method will
32 // succeed.  Defaults to true.  Note that Bad and Touchable are
33 // independent: a MockVolume may be set up so that Put fails but Touch
34 // works or vice versa.
35 //
36 // TODO(twp): rename Bad to something more descriptive, e.g. Writable,
37 // and make sure that the tests that rely on it are testing the right
38 // thing.  We may need to simulate Writable, Touchable and Corrupt
39 // volumes in different ways.
40 //
41 type MockVolume struct {
42         Store      map[string][]byte
43         Timestamps map[string]time.Time
44         Bad        bool
45         Touchable  bool
46 }
47
48 func CreateMockVolume() *MockVolume {
49         return &MockVolume{
50                 Store:      make(map[string][]byte),
51                 Timestamps: make(map[string]time.Time),
52                 Bad:        false,
53                 Touchable:  true,
54         }
55 }
56
57 func (v *MockVolume) Get(loc string) ([]byte, error) {
58         if v.Bad {
59                 return nil, errors.New("Bad volume")
60         } else if block, ok := v.Store[loc]; ok {
61                 return block, nil
62         }
63         return nil, os.ErrNotExist
64 }
65
66 func (v *MockVolume) Put(loc string, block []byte) error {
67         if v.Bad {
68                 return errors.New("Bad volume")
69         }
70         v.Store[loc] = block
71         return v.Touch(loc)
72 }
73
74 func (v *MockVolume) Touch(loc string) error {
75         if v.Touchable {
76                 v.Timestamps[loc] = time.Now()
77                 return nil
78         }
79         return errors.New("Touch failed")
80 }
81
82 func (v *MockVolume) Mtime(loc string) (time.Time, error) {
83         var mtime time.Time
84         var err error
85         if v.Bad {
86                 err = errors.New("Bad volume")
87         } else if t, ok := v.Timestamps[loc]; ok {
88                 mtime = t
89         } else {
90                 err = os.ErrNotExist
91         }
92         return mtime, err
93 }
94
95 func (v *MockVolume) Index(prefix string) string {
96         var result string
97         for loc, block := range v.Store {
98                 if IsValidLocator(loc) && strings.HasPrefix(loc, prefix) {
99                         result = result + fmt.Sprintf("%s+%d %d\n",
100                                 loc, len(block), 123456789)
101                 }
102         }
103         return result
104 }
105
106 func (v *MockVolume) Delete(loc string) error {
107         if _, ok := v.Store[loc]; ok {
108                 if time.Since(v.Timestamps[loc]) < permission_ttl {
109                         return nil
110                 }
111                 delete(v.Store, loc)
112                 return nil
113         }
114         return os.ErrNotExist
115 }
116
117 func (v *MockVolume) Status() *VolumeStatus {
118         var used uint64
119         for _, block := range v.Store {
120                 used = used + uint64(len(block))
121         }
122         return &VolumeStatus{"/bogo", 123, 1000000 - used, used}
123 }
124
125 func (v *MockVolume) String() string {
126         return "[MockVolume]"
127 }
128
129 // A VolumeManager manages a collection of volumes.
130 //
131 // - Volumes is a slice of available Volumes.
132 // - Choose() returns a Volume suitable for writing to.
133 // - Quit() instructs the VolumeManager to shut down gracefully.
134 //
135 type VolumeManager interface {
136         Volumes() []Volume
137         Choose() Volume
138         Quit()
139 }
140
141 type RRVolumeManager struct {
142         volumes   []Volume
143         nextwrite chan Volume
144         quit      chan int
145 }
146
147 func MakeRRVolumeManager(vols []Volume) *RRVolumeManager {
148         // Create a new VolumeManager struct with the specified volumes,
149         // and with new Nextwrite and Quit channels.
150         // The Quit channel is buffered with a capacity of 1 so that
151         // another routine may write to it without blocking.
152         vm := &RRVolumeManager{vols, make(chan Volume), make(chan int, 1)}
153
154         // This goroutine implements round-robin volume selection.
155         // It sends each available Volume in turn to the Nextwrite
156         // channel, until receiving a notification on the Quit channel
157         // that it should terminate.
158         go func() {
159                 var i int = 0
160                 for {
161                         select {
162                         case <-vm.quit:
163                                 return
164                         case vm.nextwrite <- vm.volumes[i]:
165                                 i = (i + 1) % len(vm.volumes)
166                         }
167                 }
168         }()
169
170         return vm
171 }
172
173 func (vm *RRVolumeManager) Volumes() []Volume {
174         return vm.volumes
175 }
176
177 func (vm *RRVolumeManager) Choose() Volume {
178         return <-vm.nextwrite
179 }
180
181 func (vm *RRVolumeManager) Quit() {
182         vm.quit <- 1
183 }