Allow to configure maxConcurrentRequests

Change-Id: Ief4563f31e64bac598e572d3da34e8cf3a174bfb

client.go

diff --git a/client.go b/client.go
index d0ccf13..2b883ae 100644
--- a/client.go
+++ b/client.go
@@ -45,6 +45,19 @@
 	}
 }
 
+// MaxConcurrentRequestsPerFile sets the maximum concurrent requests allowed for a single file.
+//
+// The default maximum concurrent requests is 64.
+func MaxConcurrentRequestsPerFile(n int) ClientOption {
+	return func(c *Client) error {
+		if n < 1 {
+			return errors.Errorf("n must be greater or equal to 1")
+		}
+		c.maxConcurrentRequests = n
+		return nil
+	}
+}
+
 // NewClient creates a new SFTP client on conn, using zero or more option
 // functions.
 func NewClient(conn *ssh.Client, opts ...ClientOption) (*Client, error) {
@@ -79,7 +92,8 @@
 			},
 			inflight: make(map[uint32]chan<- result),
 		},
-		maxPacket: 1 << 15,
+		maxPacket:             1 << 15,
+		maxConcurrentRequests: 64,
 	}
 	if err := sftp.applyOptions(opts...); err != nil {
 		wr.Close()
@@ -106,8 +120,9 @@
 type Client struct {
 	clientConn
 
-	maxPacket int // max packet size read or written.
-	nextid    uint32
+	maxPacket             int // max packet size read or written.
+	nextid                uint32
+	maxConcurrentRequests int
 }
 
 // Create creates the named file mode 0666 (before umask), truncating it if
@@ -659,8 +674,6 @@
 	return f.path
 }
 
-const maxConcurrentRequests = 64
-
 // Read reads up to len(b) bytes from the File. It returns the number of bytes
 // read and an error, if any. Read follows io.Reader semantics, so when Read
 // encounters an error or EOF condition after successfully reading n > 0 bytes,
@@ -675,7 +688,7 @@
 	offset := f.offset
 	// maxConcurrentRequests buffer to deal with broadcastErr() floods
 	// also must have a buffer of max value of (desiredInFlight - inFlight)
-	ch := make(chan result, maxConcurrentRequests+1)
+	ch := make(chan result, f.c.maxConcurrentRequests+1)
 	type inflightRead struct {
 		b      []byte
 		offset uint64
@@ -740,7 +753,7 @@
 			if n < len(req.b) {
 				sendReq(req.b[l:], req.offset+uint64(l))
 			}
-			if desiredInFlight < maxConcurrentRequests {
+			if desiredInFlight < f.c.maxConcurrentRequests {
 				desiredInFlight++
 			}
 		default:
@@ -771,7 +784,7 @@
 	writeOffset := offset
 	fileSize := uint64(fi.Size())
 	// see comment on same line in Read() above
-	ch := make(chan result, maxConcurrentRequests+1)
+	ch := make(chan result, f.c.maxConcurrentRequests+1)
 	type inflightRead struct {
 		b      []byte
 		offset uint64
@@ -851,7 +864,7 @@
 				switch {
 				case offset > fileSize:
 					desiredInFlight = 1
-				case desiredInFlight < maxConcurrentRequests:
+				case desiredInFlight < f.c.maxConcurrentRequests:
 					desiredInFlight++
 				}
 				writeOffset += uint64(nbytes)
@@ -914,7 +927,7 @@
 	desiredInFlight := 1
 	offset := f.offset
 	// see comment on same line in Read() above
-	ch := make(chan result, maxConcurrentRequests+1)
+	ch := make(chan result, f.c.maxConcurrentRequests+1)
 	var firstErr error
 	written := len(b)
 	for len(b) > 0 || inFlight > 0 {
@@ -950,7 +963,7 @@
 				firstErr = err
 				break
 			}
-			if desiredInFlight < maxConcurrentRequests {
+			if desiredInFlight < f.c.maxConcurrentRequests {
 				desiredInFlight++
 			}
 		default:
@@ -975,7 +988,7 @@
 	desiredInFlight := 1
 	offset := f.offset
 	// see comment on same line in Read() above
-	ch := make(chan result, maxConcurrentRequests+1)
+	ch := make(chan result, f.c.maxConcurrentRequests+1)
 	var firstErr error
 	read := int64(0)
 	b := make([]byte, f.c.maxPacket)
@@ -1014,7 +1027,7 @@
 				firstErr = err
 				break
 			}
-			if desiredInFlight < maxConcurrentRequests {
+			if desiredInFlight < f.c.maxConcurrentRequests {
 				desiredInFlight++
 			}
 		default: