Fix for concurrency bug

Author: RachelTucker

ds3/buildclient/buildClient.go

@@ -9,10 +9,10 @@ import (
     "os"
 )
 
-const endpointEnv = "DS3_ENDPOINT"
-const accessKeyEnv = "DS3_ACCESS_KEY"
-const secretKeyEnv = "DS3_SECRET_KEY"
-const proxyEnv = "DS3_PROXY"
+const EndpointEnv = "DS3_ENDPOINT"
+const AccessKeyEnv = "DS3_ACCESS_KEY"
+const SecretKeyEnv = "DS3_SECRET_KEY"
+const ProxyEnv = "DS3_PROXY"
 
 // Creates a client from cli arguments
 func FromArgs(args *commands.Arguments) (*ds3.Client, error) {
@@ -22,10 +22,10 @@ func FromArgs(args *commands.Arguments) (*ds3.Client, error) {
 // Creates a client from environment variables
 func FromEnv() (*ds3.Client, error) {
     //Retrieve the environment variables
-    endpoint := os.Getenv(endpointEnv)
-    accessKey := os.Getenv(accessKeyEnv)
-    secretKey := os.Getenv(secretKeyEnv)
-    proxy := os.Getenv(proxyEnv)
+    endpoint := os.Getenv(EndpointEnv)
+    accessKey := os.Getenv(AccessKeyEnv)
+    secretKey := os.Getenv(SecretKeyEnv)
+    proxy := os.Getenv(ProxyEnv)
 
     // Validate required arguments and create client if all required values are present
     switch {

ds3/networking/headers.go

@@ -53,7 +53,7 @@ type signatureFields struct {
     Path                    string
 }
 
-func buildAuthHeaderValue(creds *Credentials, fields *signatureFields) (string) {
+func (fields *signatureFields) BuildAuthHeaderValue(creds *Credentials) (string) {
     // Build the string that we need to compute the MAC on.
     stringToSign := fmt.Sprintf(
         "%s\n%s\n%s\n%s\n%s%s",

ds3/networking/httpRequestBuilder.go

@@ -108,14 +108,14 @@ func (builder *HttpRequestBuilder) Build(conn *ConnectionInfo) (*http.Request, e
 
     builder.signatureFields.Date = getCurrentTime()
 
-    authHeaderVal := buildAuthHeaderValue(conn.Credentials, &builder.signatureFields)
+    authHeaderVal := builder.signatureFields.BuildAuthHeaderValue(conn.Credentials)
 
     // Set the http request headers such as authorization and date.
     return builder.addHttpRequestHeaders(httpRequest, authHeaderVal)
 }
 
 func (builder *HttpRequestBuilder) buildUrl(conn *ConnectionInfo) string {
-    httpUrl := conn.Endpoint
+    var httpUrl url.URL = *conn.Endpoint
     httpUrl.Path = builder.signatureFields.Path
     httpUrl.RawQuery = encodeQueryParams(builder.queryParams)
     return httpUrl.String()

ds3_cli/commands/getObject.go

@@ -25,7 +25,7 @@ func getObject(client *ds3.Client, args *Arguments) error {
     // Build the request.
     request := models.NewGetObjectRequest(args.Bucket, args.Key)
     if args.End > 0 {
-        request.WithRange(args.Start, args.End)
+        request.WithRanges(models.Range{ int64(args.Start), int64(args.End) })
     }
 
     // Perform the request.

ds3_integration/concurrency_test.go

@@ -0,0 +1,146 @@
+package ds3_integration
+
+import (
+    "testing"
+    "sync"
+    "log"
+    "fmt"
+    "runtime/debug"
+    "spectra/ds3_go_sdk/ds3"
+    "spectra/ds3_go_sdk/ds3_utils/ds3Testing"
+    "spectra/ds3_go_sdk/ds3/models"
+    "spectra/ds3_go_sdk/ds3_integration/utils"
+    "spectra/ds3_go_sdk/ds3/networking"
+    "os"
+    "spectra/ds3_go_sdk/ds3/buildclient"
+    "net/url"
+)
+
+func putFileWithClient(name string, offset int64, jobId string, content *[]byte, group *sync.WaitGroup, t *testing.T) {
+    defer group.Done()
+    log.Printf("Putting file %s", name)
+
+    _, err := client.PutObject(models.NewPutObjectRequest(testBucket, name, ds3.BuildByteReaderWithSizeDecorator(*content)).
+        WithOffset(offset).
+        WithJob(jobId))
+
+    if err != nil {
+        t.Errorf("Unexpected error when putting file '%s': '%s'.", name, err.Error())
+        debug.PrintStack()
+        t.Fail()
+    }
+}
+
+func TestConcurrentClientUsage(t *testing.T) {
+    defer testutils.DeleteBucketContents(client, testBucket)
+
+    content, err := testutils.LoadBook(testutils.BookTitles[0])
+    ds3Testing.AssertNilError(t, err)
+
+    size := int64(len(content))
+
+    // create a job for putting n instances of file
+    n := 30
+
+    var ds3Objects []models.Ds3PutObject
+    for i := 0; i < n; i++ {
+        ds3Objects = append(ds3Objects, models.Ds3PutObject{ Name:fmt.Sprintf("file%d", i), Size:size })
+    }
+
+    bulkPut, err := client.PutBulkJobSpectraS3(models.NewPutBulkJobSpectraS3Request(testBucket, ds3Objects))
+    ds3Testing.AssertNilError(t, err)
+
+    // launch go routines to put files concurrently
+    var group sync.WaitGroup
+    group.Add(n)
+
+    for _, chunk := range bulkPut.MasterObjectList.Objects {
+        allocateChunk, allocateErr := client.AllocateJobChunkSpectraS3(models.NewAllocateJobChunkSpectraS3Request(chunk.ChunkId))
+        ds3Testing.AssertNilError(t, allocateErr)
+        for _, obj := range allocateChunk.Objects.Objects {
+            go putFileWithClient(*obj.Name, obj.Offset, bulkPut.MasterObjectList.JobId, &content, &group, t)
+        }
+    }
+
+    group.Wait()
+}
+
+func putFileWithSendNetwork(name string, offset int64, jobId string, content *[]byte, bucketName string, group *sync.WaitGroup, t *testing.T, network networking.Network, info *networking.ConnectionInfo) {
+    defer group.Done()
+
+    // manually create http request
+    httpRequest, err := networking.NewHttpRequestBuilder().
+        WithHttpVerb(ds3.HTTP_VERB_PUT).
+        WithPath("/" + bucketName + "/" + name).
+        WithOptionalQueryParam("job", &jobId).
+        WithOptionalQueryParam("offset", networking.Int64PtrToStrPtr(&offset)).
+        WithReader(ds3.BuildByteReaderWithSizeDecorator(*content)).
+        WithChecksum(models.NewNoneChecksum()).
+        WithHeaders(make(map[string]string)).
+        Build(info)
+
+    //networkRetryDecorator := networking.NewNetworkRetryDecorator(network, 5)
+    t.Logf("Created request to for file '%s' to url '%s'", name, httpRequest.URL.String())
+    response, err := network.Invoke(httpRequest)
+    if err != nil {
+        t.Errorf("Unexpected error when putting file '%s': '%s'.", name, err.Error())
+        debug.PrintStack()
+        t.Fail()
+    }
+
+    _, err = models.NewPutObjectResponse(response)
+    if err != nil {
+        t.Errorf("Unexpected error when putting file '%s': '%s'.", name, err.Error())
+        debug.PrintStack()
+        t.Fail()
+    }
+}
+
+func TestConcurrentSendNetworkUsage(t *testing.T) {
+    defer testutils.DeleteBucketContents(client, testBucket)
+
+    content, err := testutils.LoadBook(testutils.BookTitles[0])
+    ds3Testing.AssertNilError(t, err)
+
+    size := int64(len(content))
+
+    // create a job for putting n instances of file
+    n := 30
+
+    var ds3Objects []models.Ds3PutObject
+    for i := 0; i < n; i++ {
+        ds3Objects = append(ds3Objects, models.Ds3PutObject{ Name:fmt.Sprintf("file%d", i), Size:size })
+    }
+
+    bulkPut, err := client.PutBulkJobSpectraS3(models.NewPutBulkJobSpectraS3Request(testBucket, ds3Objects))
+    ds3Testing.AssertNilError(t, err)
+
+    // launch go routines to put files concurrently
+    var group sync.WaitGroup
+    group.Add(n)
+
+    //Retrieve the environment variables
+    endpoint := os.Getenv(buildclient.EndpointEnv)
+    accessKey := os.Getenv(buildclient.AccessKeyEnv)
+    secretKey := os.Getenv(buildclient.SecretKeyEnv)
+
+    endpointUrl, err := url.Parse(endpoint)
+    ds3Testing.AssertNilError(t, err)
+    info := networking.ConnectionInfo{
+        Endpoint:    endpointUrl,
+        Credentials: &networking.Credentials{AccessId: accessKey, Key: secretKey},
+        Proxy:       nil}
+
+    network := networking.NewSendNetwork(&info)
+
+    for _, chunk := range bulkPut.MasterObjectList.Objects {
+        //TODO Try get available job chunks ready for client processing instead
+        allocateChunk, allocateErr := client.AllocateJobChunkSpectraS3(models.NewAllocateJobChunkSpectraS3Request(chunk.ChunkId))
+        ds3Testing.AssertNilError(t, allocateErr)
+        for _, obj := range allocateChunk.Objects.Objects {
+            go putFileWithSendNetwork(*obj.Name, obj.Offset, bulkPut.MasterObjectList.JobId, &content, testBucket, &group, t, network, &info)
+        }
+    }
+
+    group.Wait()
+}