20559: Serialize writes for a given collection ID.

[arvados.git] / services / keep-web / handler.go

diff --git a/services/keep-web/handler.go b/services/keep-web/handler.go
index 3cdaf5d2b51c5e2d663fd8e53dc29739ffe76551..e9b497a0b1165f5327bc74e7970f69d34318cf6e 100644 (file)
--- a/services/keep-web/handler.go
+++ b/services/keep-web/handler.go
@@ -18,6 +18,7 @@ import (
        "strconv"
        "strings"
        "sync"
+       "time"
 
        "git.arvados.org/arvados.git/lib/cmd"
        "git.arvados.org/arvados.git/lib/webdavfs"
@@ -35,6 +36,9 @@ type handler struct {
        Cache     cache
        Cluster   *arvados.Cluster
        setupOnce sync.Once
+
+       lockMtx sync.Mutex
+       lock    map[string]*sync.RWMutex
 }
 
 var urlPDHDecoder = strings.NewReplacer(" ", "+", "-", "+")
@@ -530,7 +534,11 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
        }
        h.logUploadOrDownload(r, session.arvadosclient, sessionFS, fsprefix+strings.Join(targetPath, "/"), nil, tokenUser)
 
-       if writeMethod[r.Method] {
+       writing := writeMethod[r.Method]
+       locker := h.collectionLock(collectionID, writing)
+       defer locker.Unlock()
+
+       if writing {
                // Save the collection only if/when all
                // webdav->filesystem operations succeed --
                // and send a 500 error if the modified
@@ -942,6 +950,31 @@ func (h *handler) determineCollection(fs arvados.CustomFileSystem, path string)
        return nil, ""
 }
 
+var lockTidyInterval = time.Minute * 10
+
+// Lock the specified collection for reading or writing. Caller must
+// call Unlock() on the returned Locker when the operation is
+// finished.
+func (h *handler) collectionLock(collectionID string, writing bool) sync.Locker {
+       h.lockMtx.Lock()
+       defer h.lockMtx.Unlock()
+       locker := h.lock[collectionID]
+       if locker == nil {
+               locker = new(sync.RWMutex)
+               if h.lock == nil {
+                       h.lock = map[string]*sync.RWMutex{}
+               }
+               h.lock[collectionID] = locker
+       }
+       if writing {
+               locker.Lock()
+               return locker
+       } else {
+               locker.RLock()
+               return locker.RLocker()
+       }
+}
+
 func ServeCORSPreflight(w http.ResponseWriter, header http.Header) bool {
        method := header.Get("Access-Control-Request-Method")
        if method == "" {