Resolving Concurrency Issues (#275)

* Resolving concurrency issues

* Fixing concurrency test

* Minor changes

azblob/chunkwriting.go

@@ -56,6 +56,7 @@ func copyFromReader(ctx context.Context, from io.Reader, to blockWriter, o Uploa
 	}
 	// If the error is not EOF, then we have a problem.
 	if err != nil && !errors.Is(err, io.EOF) {
+		cp.wg.Wait()
 		return nil, err
 	}
 

azblob/zt_highlevel_test.go

@@ -432,3 +432,31 @@ func (s *aztestsSuite) TestDoBatchTransferWithError(c *chk.C) {
 	mmf.isClosed = true
 	time.Sleep(time.Second * 5)
 }
+
+func (s *aztestsSuite) Test_CopyFromReader(c *chk.C) {
+	ctx := context.Background()
+	p, err := createSrcFile(_1MiB * 12)
+	if err != nil {
+		c.Assert(err, chk.IsNil)
+	}
+
+	defer os.Remove(p)
+
+	from, err := os.Open(p)
+	if err != nil {
+		c.Assert(err, chk.IsNil)
+	}
+
+	br := newFakeBlockWriter()
+	defer br.cleanup()
+
+	br.errOnBlock = 1
+	transferManager, err := NewStaticBuffer(_1MiB, 1)
+	if err != nil {
+		panic(err)
+	}
+	defer transferManager.Close()
+	_, err = copyFromReader(ctx, from, br, UploadStreamToBlockBlobOptions{TransferManager: transferManager})
+	c.Assert(err, chk.NotNil)
+	c.Assert(err.Error(), chk.Equals, "write error: multiple Read calls return no data or error")
+}