3448: code review comments.

Extend GetBlock() to optionally update the file modification time, so
PUT operations can update the timestamp of an existing block.

UnixVolume.Delete() returns nil if the file is too new to delete (the
reasoning here is that this is the correct thing for the server to do,
even if the result technically does not fulfill the user's request, so
the server should return success).

Refs #3448.

diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go
index 1c55bb3d84245f04fa14cb56f2e3f808fe9e3916..026464c733300b08ef1be68150326f956a3bcb7a 100644 (file)
--- a/services/keepstore/handlers.go
+++ b/services/keepstore/handlers.go
@@ -148,7 +148,7 @@ func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
                }
        }
 
-       block, err := GetBlock(hash)
+       block, err := GetBlock(hash, false)
 
        // Garbage collect after each GET. Fixes #2865.
        // TODO(twp): review Keep memory usage and see if there's
@@ -397,10 +397,11 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
        }
 }
 
-// GetBlock, PutBlock and TouchBlock implement lower-level code for
-// handling blocks by rooting through volumes connected to the local
-// machine.  Once the handler has determined that system policy permits
-// the request, it calls these methods to perform the actual operation.
+// ==============================
+// GetBlock and PutBlock implement lower-level code for handling
+// blocks by rooting through volumes connected to the local machine.
+// Once the handler has determined that system policy permits the
+// request, it calls these methods to perform the actual operation.
 //
 // TODO(twp): this code would probably be better located in the
 // VolumeManager interface. As an abstraction, the VolumeManager
@@ -408,7 +409,22 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) {
 // block is stored on, so it should be responsible for figuring out
 // which volume to check for fetching blocks, storing blocks, etc.
 
-func GetBlock(hash string) ([]byte, error) {
+// ==============================
+// GetBlock fetches and returns the block identified by "hash".  If
+// the update_timestamp argument is true, GetBlock also updates the
+// block's file modification time (for the sake of PutBlock, which
+// must update the file's timestamp when the block already exists).
+//
+// On success, GetBlock returns a byte slice with the block data, and
+// a nil error.
+//
+// If the block cannot be found on any volume, returns NotFoundError.
+//
+// If the block found does not have the correct MD5 hash, returns
+// DiskHashError.
+//
+
+func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
        // Attempt to read the requested hash from a keep volume.
        error_to_caller := NotFoundError
 
@@ -443,6 +459,10 @@ func GetBlock(hash string) ([]byte, error) {
                                        log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned\n",
                                                vol, hash)
                                }
+                               // Update the timestamp if the caller requested.
+                               if update_timestamp {
+                                       vol.Touch(hash)
+                               }
                                return buf, nil
                        }
                }
@@ -489,22 +509,16 @@ func PutBlock(block []byte, hash string) error {
        }
 
        // If we already have a block on disk under this identifier, return
-       // success (but check for MD5 collisions).
+       // success (but check for MD5 collisions).  While fetching the block,
+       // update its timestamp.
        // The only errors that GetBlock can return are DiskHashError and NotFoundError.
        // In either case, we want to write our new (good) block to disk,
        // so there is nothing special to do if err != nil.
-       if oldblock, err := GetBlock(hash); err == nil {
+       //
+       if oldblock, err := GetBlock(hash, true); err == nil {
                if bytes.Compare(block, oldblock) == 0 {
-                       // The block already exists; update the timestamp and
-                       // return.
-                       // Note that TouchBlock will fail (and therefore
-                       // so will PutBlock) if the block exists but is found on a
-                       // read-only volume. That is intentional: if the user has
-                       // requested N replicas of a block, we want to be sure that
-                       // there are at least N *writable* replicas, so a block
-                       // that cannot be written to should not count toward the
-                       // replication total.
-                       return TouchBlock(hash)
+                       // The block already exists; return success.
+                       return nil
                } else {
                        return CollisionError
                }
@@ -540,22 +554,6 @@ func PutBlock(block []byte, hash string) error {
        }
 }
 
-// TouchBlock finds the block identified by hash and updates its
-// filesystem modification time with the current time.
-func TouchBlock(hash string) error {
-       for _, vol := range KeepVM.Volumes() {
-               err := vol.Touch(hash)
-               if os.IsNotExist(err) {
-                       continue
-               }
-               // either err is nil (meaning success) or some error other
-               // than "file does not exist" (which we should return upward).
-               return err
-       }
-       // If we got here, the block was not found on any volume.
-       return os.ErrNotExist
-}
-
 // IsValidLocator
 //     Return true if the specified string is a valid Keep locator.
 //     When Keep is extended to support hash types other than MD5,

diff --git a/services/keepstore/keepstore_test.go b/services/keepstore/keepstore_test.go
index b153d6dce27309f3cb692ec0d6a84b724386891a..d43a00e153ee52a724f38c911bea9f591ca9eb75 100644 (file)
--- a/services/keepstore/keepstore_test.go
+++ b/services/keepstore/keepstore_test.go
@@ -60,7 +60,7 @@ func TestGetBlock(t *testing.T) {
        }
 
        // Check that GetBlock returns success.
-       result, err := GetBlock(TEST_HASH)
+       result, err := GetBlock(TEST_HASH, false)
        if err != nil {
                t.Errorf("GetBlock error: %s", err)
        }
@@ -80,7 +80,7 @@ func TestGetBlockMissing(t *testing.T) {
        defer func() { KeepVM.Quit() }()
 
        // Check that GetBlock returns failure.
-       result, err := GetBlock(TEST_HASH)
+       result, err := GetBlock(TEST_HASH, false)
        if err != NotFoundError {
                t.Errorf("Expected NotFoundError, got %v", result)
        }
@@ -101,7 +101,7 @@ func TestGetBlockCorrupt(t *testing.T) {
        vols[0].Put(TEST_HASH, BAD_BLOCK)
 
        // Check that GetBlock returns failure.
-       result, err := GetBlock(TEST_HASH)
+       result, err := GetBlock(TEST_HASH, false)
        if err != DiskHashError {
                t.Errorf("Expected DiskHashError, got %v (buf: %v)", err, result)
        }
@@ -156,7 +156,7 @@ func TestPutBlockOneVol(t *testing.T) {
                t.Fatalf("PutBlock: %v", err)
        }
 
-       result, err := GetBlock(TEST_HASH)
+       result, err := GetBlock(TEST_HASH, false)
        if err != nil {
                t.Fatalf("GetBlock: %v", err)
        }
@@ -185,7 +185,7 @@ func TestPutBlockMD5Fail(t *testing.T) {
        }
 
        // Confirm that GetBlock fails to return anything.
-       if result, err := GetBlock(TEST_HASH); err != NotFoundError {
+       if result, err := GetBlock(TEST_HASH, false); err != NotFoundError {
                t.Errorf("GetBlock succeeded after a corrupt block store (result = %s, err = %v)",
                        string(result), err)
        }
@@ -210,7 +210,7 @@ func TestPutBlockCorrupt(t *testing.T) {
        }
 
        // The block on disk should now match TEST_BLOCK.
-       if block, err := GetBlock(TEST_HASH); err != nil {
+       if block, err := GetBlock(TEST_HASH, false); err != nil {
                t.Errorf("GetBlock: %v", err)
        } else if bytes.Compare(block, TEST_BLOCK) != 0 {
                t.Errorf("GetBlock returned: '%s'", string(block))

diff --git a/services/keepstore/volume_unix.go b/services/keepstore/volume_unix.go
index 6960098c5a44166f2b1173d679625b3993d0567c..0cebe4369524ede3ca6fd785d9cf5b3d849acc6c 100644 (file)
--- a/services/keepstore/volume_unix.go
+++ b/services/keepstore/volume_unix.go
@@ -251,16 +251,16 @@ func (v *UnixVolume) Delete(loc string) error {
        lockfile(f)
        defer unlockfile(f)
 
-       // Return PermissionError if the block has been PUT more recently
-       // than -permission_ttl.  This guards against a race condition
-       // where a block is old enough that Data Manager has added it to
-       // the trash list, but the user submitted a PUT for the block
-       // since then.
+       // If the block has been PUT more recently than -permission_ttl,
+       // return success without removing the block.  This guards against
+       // a race condition where a block is old enough that Data Manager
+       // has added it to the trash list, but the user submitted a PUT
+       // for the block since then.
        if fi, err := os.Stat(p); err != nil {
                return err
        } else {
                if time.Since(fi.ModTime()) < permission_ttl {
-                       return PermissionError
+                       return nil
                }
        }
        return os.Remove(p)