[arvados] created: 2.6.0-262-gea34f3601

[git repository hosting]

        at  ea34f3601ac909daea5295a2f4e1236c5f75f3b9 (commit)


commit ea34f3601ac909daea5295a2f4e1236c5f75f3b9
Author: Tom Clegg <tom at curii.com>
Date:   Tue Jun 13 17:47:17 2023 -0400

    20559: Serialize writes for a given collection ID.
    
    Arvados-DCO-1.1-Signed-off-by: Tom Clegg <tom at curii.com>

diff --git a/services/keep-web/handler.go b/services/keep-web/handler.go
index 27981c487..573c60d0a 100644
--- a/services/keep-web/handler.go
+++ b/services/keep-web/handler.go
@@ -35,6 +35,9 @@ type handler struct {
 	Cache     cache
 	Cluster   *arvados.Cluster
 	setupOnce sync.Once
+
+	lockMtx sync.Mutex
+	lock    map[string]*sync.RWMutex
 }
 
 var urlPDHDecoder = strings.NewReplacer(" ", "+", "-", "+")
@@ -538,7 +541,17 @@ 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.collectionLocker(collectionID)
+	if writing {
+		locker.Lock()
+		defer locker.Unlock()
+	} else {
+		locker.RLock()
+		defer locker.RUnlock()
+	}
+
+	if writing {
 		// Save the collection only if/when all
 		// webdav->filesystem operations succeed --
 		// and send a 500 error if the modified
@@ -949,3 +962,17 @@ func (h *handler) determineCollection(fs arvados.CustomFileSystem, path string)
 	}
 	return nil, ""
 }
+
+func (h *handler) collectionLocker(collectionID string) *sync.RWMutex {
+	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
+	}
+	return locker
+}

commit 6cf4c8d0116e4ea6cd1f0c03c19aa52db06e6543
Author: Tom Clegg <tom at curii.com>
Date:   Tue Jun 13 17:47:14 2023 -0400

    20559: Test concurrent writes using same token.
    
    Arvados-DCO-1.1-Signed-off-by: Tom Clegg <tom at curii.com>

diff --git a/services/keep-web/handler_test.go b/services/keep-web/handler_test.go
index c9b48f99a..60fe4acc4 100644
--- a/services/keep-web/handler_test.go
+++ b/services/keep-web/handler_test.go
@@ -18,6 +18,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"git.arvados.org/arvados.git/lib/config"
@@ -1624,3 +1625,59 @@ func (s *IntegrationSuite) TestUploadLoggingPermission(c *check.C) {
 		}
 	}
 }
+
+func (s *IntegrationSuite) TestConcurrentWrites(c *check.C) {
+	client := arvados.NewClientFromEnv()
+	client.AuthToken = arvadostest.ActiveTokenV2
+	// Start small, and increase concurrency (2^2, 4^2, 8^2, 16^2)
+	// only until hitting failure. Avoids unnecessarily long
+	// failure reports.
+	for n := 2; n < 32 && !c.Failed(); n++ {
+		var coll arvados.Collection
+		err := client.RequestAndDecode(&coll, "POST", "arvados/v1/collections", nil, nil)
+		c.Assert(err, check.IsNil)
+		defer client.RequestAndDecode(&coll, "DELETE", "arvados/v1/collections/"+coll.UUID, nil, nil)
+
+		var wg sync.WaitGroup
+		for i := 0; i < n && !c.Failed(); i++ {
+			i := i
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				u := mustParseURL(fmt.Sprintf("http://%s.collections.example.com/i=%d", coll.UUID, i))
+				resp := httptest.NewRecorder()
+				req, err := http.NewRequest("MKCOL", u.String(), nil)
+				c.Assert(err, check.IsNil)
+				req.Header.Set("Authorization", "Bearer "+client.AuthToken)
+				s.handler.ServeHTTP(resp, req)
+				c.Assert(resp.Code, check.Equals, http.StatusCreated)
+				for j := 0; j < n && !c.Failed(); j++ {
+					j := j
+					wg.Add(1)
+					go func() {
+						defer wg.Done()
+						content := fmt.Sprintf("i=%d/j=%d", i, j)
+						u := mustParseURL("http://" + coll.UUID + ".collections.example.com/" + content)
+
+						resp := httptest.NewRecorder()
+						req, err := http.NewRequest("PUT", u.String(), strings.NewReader(content))
+						c.Assert(err, check.IsNil)
+						req.Header.Set("Authorization", "Bearer "+client.AuthToken)
+						s.handler.ServeHTTP(resp, req)
+						c.Check(resp.Code, check.Equals, http.StatusCreated)
+
+						time.Sleep(time.Second)
+						resp = httptest.NewRecorder()
+						req, err = http.NewRequest("GET", u.String(), nil)
+						c.Assert(err, check.IsNil)
+						req.Header.Set("Authorization", "Bearer "+client.AuthToken)
+						s.handler.ServeHTTP(resp, req)
+						c.Check(resp.Code, check.Equals, http.StatusOK)
+						c.Check(resp.Body.String(), check.Equals, content)
+					}()
+				}
+			}()
+		}
+		wg.Wait()
+	}
+}

-----------------------------------------------------------------------


hooks/post-receive
--