// directories.
var listen, volumearg string
+ var serialize_io bool
flag.StringVar(&listen, "listen", DEFAULT_ADDR,
"interface on which to listen for requests, in the format ipaddr:port. e.g. -listen=10.0.1.24:8000. Use -listen=:port to listen on all network interfaces.")
flag.StringVar(&volumearg, "volumes", "",
"Comma-separated list of directories to use for Keep volumes, e.g. -volumes=/var/keep1,/var/keep2. If empty or not supplied, Keep will scan mounted filesystems for volumes with a /keep top-level directory.")
+ flag.BoolVar(&serialize_io, "serialize-io", false,
+ "If set, all read and write operations on local Keep volumes will be serialized.")
flag.Parse()
// Look for local keep volumes.
for _, v := range keepvols {
if _, err := os.Stat(v); err == nil {
log.Println("adding Keep volume:", v)
- KeepVolumes = append(KeepVolumes, &UnixVolume{v})
+ newvol := &UnixVolume{v, nil}
+ if serialize_io {
+ newvol.queue = make(chan *IORequest)
+ }
+ KeepVolumes = append(KeepVolumes, newvol)
} else {
log.Printf("bad Keep volume: %s\n", err)
}
func GetBlock(hash string) ([]byte, error) {
// Attempt to read the requested hash from a keep volume.
for _, vol := range KeepVolumes {
- if buf, err := vol.Read(hash); err != nil {
+ if buf, err := vol.Get(hash); err != nil {
// IsNotExist is an expected error and may be ignored.
// (If all volumes report IsNotExist, we return a NotFoundError)
// A CorruptError should be returned immediately.
// Store the block on the first available Keep volume.
allFull := true
for _, vol := range KeepVolumes {
- err := vol.Write(hash, block)
+ err := vol.Put(hash, block)
if err == nil {
return nil // success!
}
)
type Volume interface {
- Read(loc string) ([]byte, error)
- Write(loc string, block []byte) error
+ Get(loc string) ([]byte, error)
+ Put(loc string, block []byte) error
Index(prefix string) string
Status() *VolumeStatus
String() string
}
+// IORequests are encapsulated requests to perform I/O on a Keep volume.
+// When running in serialized mode, the Keep front end sends IORequests
+// on a channel to an IORunner, which handles them one at a time and
+// returns an IOResponse.
+//
+type IOMethod int
+
+const (
+ KeepGet IOMethod = iota
+ KeepPut
+)
+
+type IORequest struct {
+ method IOMethod
+ loc string
+ data []byte
+ reply chan *IOResponse
+}
+
+type IOResponse struct {
+ data []byte
+ err error
+}
+
+// A UnixVolume is configured with:
+//
+// * root: the path to the volume's root directory
+// * queue: if non-nil, all I/O requests for this volume should be queued
+// on this channel. The response will be delivered on the IOResponse
+// channel included in the request.
+//
type UnixVolume struct {
- root string // path to this volume
+ root string // path to this volume
+ queue chan *IORequest
}
-func (v *UnixVolume) String() string {
- return fmt.Sprintf("[UnixVolume %s]", v.root)
+func (v *UnixVolume) IOHandler() {
+ for req := range v.queue {
+ var result IOResponse
+ switch req.method {
+ case KeepGet:
+ result.data, result.err = v.Read(req.loc)
+ case KeepPut:
+ result.err = v.Write(req.loc, req.data)
+ }
+ req.reply <- &result
+ }
+}
+
+func MakeUnixVolume(root string, queue chan *IORequest) UnixVolume {
+ v := UnixVolume{root, queue}
+ if queue != nil {
+ go v.IOHandler()
+ }
+ return v
+}
+
+func (v *UnixVolume) Get(loc string) ([]byte, error) {
+ if v.queue == nil {
+ return v.Read(loc)
+ }
+ reply := make(chan *IOResponse)
+ v.queue <- &IORequest{KeepGet, loc, nil, reply}
+ response := <-reply
+ return response.data, response.err
+}
+
+func (v *UnixVolume) Put(loc string, block []byte) error {
+ if v.queue == nil {
+ return v.Write(loc, block)
+ }
+ reply := make(chan *IOResponse)
+ v.queue <- &IORequest{KeepPut, loc, block, reply}
+ response := <-reply
+ return response.err
}
// Read retrieves a block identified by the locator string "loc", and
}
return
}
+
+func (v *UnixVolume) String() string {
+ return fmt.Sprintf("[UnixVolume %s]", v.root)
+}
"time"
)
-func TempUnixVolume(t *testing.T) UnixVolume {
+func TempUnixVolume(t *testing.T, queue chan *IORequest) UnixVolume {
d, err := ioutil.TempDir("", "volume_test")
if err != nil {
t.Fatal(err)
}
- return UnixVolume{d}
+ return MakeUnixVolume(d, queue)
}
func _teardown(v UnixVolume) {
+ if v.queue != nil {
+ close(v.queue)
+ }
os.RemoveAll(v.root)
}
}
}
-func TestRead(t *testing.T) {
- v := TempUnixVolume(t)
+func TestGet(t *testing.T) {
+ v := TempUnixVolume(t, nil)
defer _teardown(v)
_store(t, v, TEST_HASH, TEST_BLOCK)
- buf, err := v.Read(TEST_HASH)
+ buf, err := v.Get(TEST_HASH)
if err != nil {
t.Error(err)
}
}
}
-func TestReadNotFound(t *testing.T) {
- v := TempUnixVolume(t)
+func TestGetNotFound(t *testing.T) {
+ v := TempUnixVolume(t, nil)
defer _teardown(v)
_store(t, v, TEST_HASH, TEST_BLOCK)
- buf, err := v.Read(TEST_HASH_2)
+ buf, err := v.Get(TEST_HASH_2)
switch {
case os.IsNotExist(err):
break
}
}
-func TestWrite(t *testing.T) {
- v := TempUnixVolume(t)
+func TestPut(t *testing.T) {
+ v := TempUnixVolume(t, nil)
defer _teardown(v)
- err := v.Write(TEST_HASH, TEST_BLOCK)
+ err := v.Put(TEST_HASH, TEST_BLOCK)
if err != nil {
t.Error(err)
}
}
}
-func TestWriteBadVolume(t *testing.T) {
- v := TempUnixVolume(t)
+func TestPutBadVolume(t *testing.T) {
+ v := TempUnixVolume(t, nil)
defer _teardown(v)
os.Chmod(v.root, 000)
- err := v.Write(TEST_HASH, TEST_BLOCK)
+ err := v.Put(TEST_HASH, TEST_BLOCK)
if err == nil {
t.Error("Write should have failed")
}
}
+// Serialization tests.
+//
+// TODO(twp): a proper test of I/O serialization requires that
+// a second request start while the first one is still executing.
+// Doing this correctly requires some tricky synchronization.
+// For now we'll just launch a bunch of requests in goroutines
+// and demonstrate that they return accurate results.
+//
+func TestGetSerialized(t *testing.T) {
+ v := TempUnixVolume(t, make(chan *IORequest))
+ defer _teardown(v)
+
+ _store(t, v, TEST_HASH, TEST_BLOCK)
+ _store(t, v, TEST_HASH_2, TEST_BLOCK_2)
+ _store(t, v, TEST_HASH_3, TEST_BLOCK_3)
+
+ sem := make(chan int)
+ go func(sem chan int) {
+ buf, err := v.Get(TEST_HASH)
+ if err != nil {
+ t.Errorf("err1: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK) != 0 {
+ t.Errorf("buf should be %s, is %s", string(TEST_BLOCK), string(buf))
+ }
+ sem <- 1
+ }(sem)
+
+ go func(sem chan int) {
+ buf, err := v.Get(TEST_HASH_2)
+ if err != nil {
+ t.Errorf("err2: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK_2) != 0 {
+ t.Errorf("buf should be %s, is %s", string(TEST_BLOCK_2), string(buf))
+ }
+ sem <- 1
+ }(sem)
+
+ go func(sem chan int) {
+ buf, err := v.Get(TEST_HASH_3)
+ if err != nil {
+ t.Errorf("err3: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK_3) != 0 {
+ t.Errorf("buf should be %s, is %s", string(TEST_BLOCK_3), string(buf))
+ }
+ sem <- 1
+ }(sem)
+
+ // Wait for all goroutines to finish
+ for done := 0; done < 2; {
+ done += <-sem
+ }
+}
+
+func TestPutSerialized(t *testing.T) {
+ v := TempUnixVolume(t, make(chan *IORequest))
+ defer _teardown(v)
+
+ sem := make(chan int)
+ go func(sem chan int) {
+ err := v.Put(TEST_HASH, TEST_BLOCK)
+ if err != nil {
+ t.Errorf("err1: %v", err)
+ }
+ sem <- 1
+ }(sem)
+
+ go func(sem chan int) {
+ err := v.Put(TEST_HASH_2, TEST_BLOCK_2)
+ if err != nil {
+ t.Errorf("err2: %v", err)
+ }
+ sem <- 1
+ }(sem)
+
+ go func(sem chan int) {
+ err := v.Put(TEST_HASH_3, TEST_BLOCK_3)
+ if err != nil {
+ t.Errorf("err3: %v", err)
+ }
+ sem <- 1
+ }(sem)
+
+ // Wait for all goroutines to finish
+ for done := 0; done < 2; {
+ done += <-sem
+ }
+
+ // Double check that we actually wrote the blocks we expected to write.
+ buf, err := v.Get(TEST_HASH)
+ if err != nil {
+ t.Errorf("Get #1: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK) != 0 {
+ t.Errorf("Get #1: expected %s, got %s", string(TEST_BLOCK), string(buf))
+ }
+
+ buf, err = v.Get(TEST_HASH_2)
+ if err != nil {
+ t.Errorf("Get #2: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK_2) != 0 {
+ t.Errorf("Get #2: expected %s, got %s", string(TEST_BLOCK_2), string(buf))
+ }
+
+ buf, err = v.Get(TEST_HASH_3)
+ if err != nil {
+ t.Errorf("Get #3: %v", err)
+ }
+ if bytes.Compare(buf, TEST_BLOCK_3) != 0 {
+ t.Errorf("Get #3: expected %s, got %s", string(TEST_BLOCK_3), string(buf))
+ }
+}
+
func TestIsFull(t *testing.T) {
- v := TempUnixVolume(t)
+ v := TempUnixVolume(t, nil)
defer _teardown(v)
full_path := v.root + "/full"
projects / arvados.git / commitdiff