[ARVADOS] updated: 6cca3e129b6de3fc4b34a1664f494fcbbbd940d1

[git at public.curoverse.com]

Summary of changes:
 sdk/go/keepclient/keepclient.go | 19 +++++++++++--------
 sdk/go/keepclient/support.go    | 25 +++++++++++++++----------
 2 files changed, 26 insertions(+), 18 deletions(-)

       via  6cca3e129b6de3fc4b34a1664f494fcbbbd940d1 (commit)
      from  bb663993cc851971d9444d834acfc416363931b2 (commit)

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 6cca3e129b6de3fc4b34a1664f494fcbbbd940d1
Author: radhika <radhika at curoverse.com>
Date:   Mon Sep 21 17:36:03 2015 -0400

    7162: Use replicasPerService to avoid sending redundant replication requests.

diff --git a/sdk/go/keepclient/keepclient.go b/sdk/go/keepclient/keepclient.go
index 479669a..2d18ec9 100644
--- a/sdk/go/keepclient/keepclient.go
+++ b/sdk/go/keepclient/keepclient.go
@@ -34,14 +34,17 @@ const X_Keep_Replicas_Stored = "X-Keep-Replicas-Stored"
 
 // Information about Arvados and Keep servers.
 type KeepClient struct {
-	Arvados       *arvadosclient.ArvadosClient
-	Want_replicas int
-	Using_proxy   bool
-	localRoots    *map[string]string
+	Arvados            *arvadosclient.ArvadosClient
+	Want_replicas      int
+	Using_proxy        bool
+	localRoots         *map[string]string
 	writableLocalRoots *map[string]string
-	gatewayRoots  *map[string]string
-	lock          sync.RWMutex
-	Client        *http.Client
+	gatewayRoots       *map[string]string
+	lock               sync.RWMutex
+	Client             *http.Client
+
+	// set to 1 if all writable services are of disk type, otherwise 0
+	replicasPerService int
 }
 
 // Create a new KeepClient.  This will contact the API server to discover Keep
@@ -247,7 +250,7 @@ func (kc *KeepClient) getSortedRoots(locator string) []string {
 			// +K at abcde means fetch from proxy at
 			// keep.abcde.arvadosapi.com
 			found = append(found, "https://keep."+hint[2:]+".arvadosapi.com")
-		} else if len(hint) == 29 && arvadosclient.UUIDMatch(hint[2:]){
+		} else if len(hint) == 29 && arvadosclient.UUIDMatch(hint[2:]) {
 			// +K at abcde-abcde-abcdeabcdeabcde means fetch
 			// from gateway with given uuid
 			if gwURI, ok := kc.GatewayRoots()[hint[2:]]; ok {
diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go
index 592d77e..4cc6447 100644
--- a/sdk/go/keepclient/support.go
+++ b/sdk/go/keepclient/support.go
@@ -98,6 +98,8 @@ func (this *KeepClient) DiscoverKeepServers() error {
 	gatewayRoots := make(map[string]string)
 	writableLocalRoots := make(map[string]string)
 
+	this.replicasPerService = 1 // set to 1 until writable non-disk services are found
+
 	for _, service := range m.Items {
 		scheme := "http"
 		if service.SSL {
@@ -111,16 +113,16 @@ func (this *KeepClient) DiscoverKeepServers() error {
 		}
 		listed[url] = true
 
-		switch service.SvcType {
-		case "disk":
-			localRoots[service.Uuid] = url
-		default:
-			localRoots[service.Uuid] = url
+		localRoots[service.Uuid] = url
+		if service.SvcType == "proxy" {
 			this.Using_proxy = true
 		}
 
 		if service.ReadOnly == false {
 			writableLocalRoots[service.Uuid] = url
+			if service.SvcType != "disk" {
+				this.replicasPerService = 0
+			}
 		}
 
 		// Gateway services are only used when specified by
@@ -177,10 +179,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read
 
 	req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.Arvados.ApiToken))
 	req.Header.Add("Content-Type", "application/octet-stream")
-
-	if this.Using_proxy {
-		req.Header.Add(X_Keep_Desired_Replicas, fmt.Sprint(this.Want_replicas))
-	}
+	req.Header.Add(X_Keep_Desired_Replicas, fmt.Sprint(this.Want_replicas))
 
 	var resp *http.Response
 	if resp, err = this.Client.Do(req); err != nil {
@@ -237,7 +236,13 @@ func (this KeepClient) putReplicas(
 	remaining_replicas := this.Want_replicas
 
 	for remaining_replicas > 0 {
-		for active < remaining_replicas {
+		replicasPerThread := this.replicasPerService
+		if replicasPerThread < 1 {
+			// unlimited or unknown
+			replicasPerThread = remaining_replicas
+		}
+
+		for active*replicasPerThread < remaining_replicas {
 			// Start some upload requests
 			if next_server < len(sv) {
 				log.Printf("[%v] Begin upload %s to %s", requestId, hash, sv[next_server])

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


hooks/post-receive
--