[ARVADOS] updated: 782bcb07365ce13e76640dff9d08266a81489d1f
git at public.curoverse.com
git at public.curoverse.com
Thu Oct 8 13:25:29 EDT 2015
Summary of changes:
services/keepstore/azure_blob_volume_test.go | 1 -
1 file changed, 1 deletion(-)
discards 5bce36a3b98012f002a084a78f65b2af8922adfd (commit)
discards 5928ca8a32c2f91c247f47d8ff05e018ce810f53 (commit)
via 782bcb07365ce13e76640dff9d08266a81489d1f (commit)
via c1c6fd9beb9612c56efd73bb80883cea39f04b28 (commit)
This update added new revisions after undoing existing revisions. That is
to say, the old revision is not a strict subset of the new revision. This
situation occurs when you --force push a change and generate a repository
containing something like this:
* -- * -- B -- O -- O -- O (5bce36a3b98012f002a084a78f65b2af8922adfd)
\
N -- N -- N (782bcb07365ce13e76640dff9d08266a81489d1f)
When this happens we assume that you've already had alert emails for all
of the O revisions, and so we here report only the revisions in the N
branch from the common base, B.
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 782bcb07365ce13e76640dff9d08266a81489d1f
Author: Tom Clegg <tom at curoverse.com>
Date: Thu Oct 8 13:30:18 2015 -0400
7159: Test race deadline
diff --git a/services/keepstore/azure_blob_volume_test.go b/services/keepstore/azure_blob_volume_test.go
index 889a026..74c94ec 100644
--- a/services/keepstore/azure_blob_volume_test.go
+++ b/services/keepstore/azure_blob_volume_test.go
@@ -408,6 +408,43 @@ func TestAzureBlobVolumeCreateBlobRace(t *testing.T) {
<-allDone
}
+func TestAzureBlobVolumeCreateBlobRaceDeadline(t *testing.T) {
+ defer func(t http.RoundTripper) {
+ http.DefaultTransport = t
+ }(http.DefaultTransport)
+ http.DefaultTransport = &http.Transport{
+ Dial: (&azStubDialer{}).Dial,
+ }
+
+ v := NewTestableAzureBlobVolume(t, false, 3)
+ defer v.Teardown()
+
+ azureWriteRaceInterval = 2 * time.Second
+ azureWriteRacePollTime = 5 * time.Millisecond
+
+ v.PutRaw(TestHash, []byte{})
+ v.TouchWithDate(TestHash, time.Now().Add(-1982 * time.Millisecond))
+
+ allDone := make(chan struct{})
+ go func() {
+ defer close(allDone)
+ buf, err := v.Get(TestHash)
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ if len(buf) != 0 {
+ t.Errorf("Got %+q, expected empty buf", buf)
+ }
+ bufs.Put(buf)
+ }()
+ select {
+ case <-allDone:
+ case <-time.After(time.Second):
+ t.Error("Get should have stopped waiting for race when block was 2s old")
+ }
+}
+
func (v *TestableAzureBlobVolume) PutRaw(locator string, data []byte) {
v.azHandler.PutRaw(v.containerName, locator, data)
}
commit c1c6fd9beb9612c56efd73bb80883cea39f04b28
Author: Tom Clegg <tom at curoverse.com>
Date: Thu Oct 8 12:52:17 2015 -0400
7159: Work around CreateBlob race by polling for updates when a brand new blob is found empty.
diff --git a/services/keepstore/azure_blob_volume.go b/services/keepstore/azure_blob_volume.go
index 79123a9..836b8f1 100644
--- a/services/keepstore/azure_blob_volume.go
+++ b/services/keepstore/azure_blob_volume.go
@@ -19,6 +19,8 @@ var (
azureStorageAccountName string
azureStorageAccountKeyFile string
azureStorageReplication int
+ azureWriteRaceInterval time.Duration = 15 * time.Second
+ azureWriteRacePollTime time.Duration = time.Second
)
func readKeyFromFile(file string) (string, error) {
@@ -117,6 +119,34 @@ func (v *AzureBlobVolume) Check() error {
}
func (v *AzureBlobVolume) Get(loc string) ([]byte, error) {
+ var deadline time.Time
+ haveDeadline := false
+ buf, err := v.get(loc)
+ for err == nil && len(buf) == 0 && loc[:32] != "d41d8cd98f00b204e9800998ecf8427e" {
+ // Seeing a brand new empty block probably means we're
+ // in a race with CreateBlob, which under the hood
+ // (apparently) does "CreateEmpty" and "CommitData"
+ // with no additional transaction locking.
+ if !haveDeadline {
+ t, err := v.Mtime(loc)
+ if err != nil {
+ log.Print("Got empty block (possible race) but Mtime failed: ", err)
+ break
+ }
+ deadline = t.Add(azureWriteRaceInterval)
+ haveDeadline = true
+ }
+ if time.Now().After(deadline) {
+ break
+ }
+ bufs.Put(buf)
+ time.Sleep(azureWriteRacePollTime)
+ buf, err = v.get(loc)
+ }
+ return buf, err
+}
+
+func (v *AzureBlobVolume) get(loc string) ([]byte, error) {
rdr, err := v.bsClient.GetBlob(v.containerName, loc)
if err != nil {
if strings.Contains(err.Error(), "404 Not Found") {
diff --git a/services/keepstore/azure_blob_volume_test.go b/services/keepstore/azure_blob_volume_test.go
index 66b0ea0..889a026 100644
--- a/services/keepstore/azure_blob_volume_test.go
+++ b/services/keepstore/azure_blob_volume_test.go
@@ -49,6 +49,7 @@ type azBlob struct {
type azStubHandler struct {
sync.Mutex
blobs map[string]*azBlob
+ race chan chan struct{}
}
func newAzStubHandler() *azStubHandler {
@@ -75,6 +76,21 @@ func (h *azStubHandler) PutRaw(container, hash string, data []byte) {
}
}
+func (h *azStubHandler) unlockAndRace() {
+ if h.race == nil {
+ return
+ }
+ h.Unlock()
+ // Signal caller that race is starting by reading from
+ // h.race. If we get a channel, block until that channel is
+ // ready to receive. If we get nil (or h.race is closed) just
+ // proceed.
+ if c := <-h.race; c != nil {
+ c <- struct{}{}
+ }
+ h.Lock()
+}
+
func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
h.Lock()
defer h.Unlock()
@@ -108,6 +124,18 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
switch {
case r.Method == "PUT" && r.Form.Get("comp") == "":
// "Put Blob" API
+ if _, ok := h.blobs[container+"|"+hash]; !ok {
+ // Like the real Azure service, we offer a
+ // race window during which other clients can
+ // list/get the new blob before any data is
+ // committed.
+ h.blobs[container+"|"+hash] = &azBlob{
+ Mtime: time.Now(),
+ Uncommitted: make(map[string][]byte),
+ Etag: makeEtag(),
+ }
+ h.unlockAndRace()
+ }
h.blobs[container+"|"+hash] = &azBlob{
Data: body,
Mtime: time.Now(),
@@ -182,6 +210,7 @@ func (h *azStubHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
log.Printf("write %+q: %s", blob.Data, err)
}
}
+ h.unlockAndRace()
case r.Method == "DELETE" && hash != "":
// "Delete Blob" API
if !blobExists {
@@ -265,7 +294,7 @@ type TestableAzureBlobVolume struct {
t *testing.T
}
-func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) TestableVolume {
+func NewTestableAzureBlobVolume(t *testing.T, readonly bool, replication int) *TestableAzureBlobVolume {
azHandler := newAzStubHandler()
azStub := httptest.NewServer(azHandler)
@@ -336,6 +365,49 @@ func TestAzureBlobVolumeReplication(t *testing.T) {
}
}
+func TestAzureBlobVolumeCreateBlobRace(t *testing.T) {
+ defer func(t http.RoundTripper) {
+ http.DefaultTransport = t
+ }(http.DefaultTransport)
+ http.DefaultTransport = &http.Transport{
+ Dial: (&azStubDialer{}).Dial,
+ }
+
+ v := NewTestableAzureBlobVolume(t, false, 3)
+ defer v.Teardown()
+
+ azureWriteRaceInterval = time.Second
+ azureWriteRacePollTime = time.Millisecond
+
+ allDone := make(chan struct{})
+ v.azHandler.race = make(chan chan struct{})
+ go func() {
+ err := v.Put(TestHash, TestBlock)
+ if err != nil {
+ t.Error(err)
+ }
+ }()
+ continuePut := make(chan struct{})
+ // Wait for the stub's Put to create the empty blob
+ v.azHandler.race <- continuePut
+ go func() {
+ buf, err := v.Get(TestHash)
+ if err != nil {
+ t.Error(err)
+ } else {
+ bufs.Put(buf)
+ }
+ close(allDone)
+ }()
+ // Wait for the stub's Get to get the empty blob
+ close(v.azHandler.race)
+ // Allow stub's Put to continue, so the real data is ready
+ // when the volume's Get retries
+ <-continuePut
+ // Wait for volume's Get to return the real data
+ <-allDone
+}
+
func (v *TestableAzureBlobVolume) PutRaw(locator string, data []byte) {
v.azHandler.PutRaw(v.containerName, locator, data)
}
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list