Handle recvPacket in a single goroutine.

Previously recvPacket would be invoked in several goroutines. This meant
that when multiple concurrent requests were in flight there were N
goroutines each waiting on recvPacket. For optimal throughput the goal
is to send a new request as quickly as possible once a response is
received. The previous mechanism worked counter to this because the
goroutine sending new requests would be competing against N recvPacket
goroutines that may become runnable as data streams in. Having a single
goroutine responsible for recvPacket means that the recv and send
goroutines will ping-pong back and forth optimizing throughput.

This changes shows a ~10-25% increase in throughput in the the *Delay*
benchmark tests.

$ go test -bench=. -integration
PASS
BenchmarkRead1k	       2	 840068631 ns/op	  12.48 MB/s
BenchmarkRead16k	      20	  72968548 ns/op	 143.70 MB/s
BenchmarkRead32k	      30	  56871347 ns/op	 184.38 MB/s
BenchmarkRead128k	     100	  34150953 ns/op	 307.05 MB/s
BenchmarkRead512k	     100	  15730685 ns/op	 666.59 MB/s
BenchmarkRead1MiB	     200	  10462421 ns/op	1002.24 MB/s
BenchmarkRead4MiB	     200	   7325236 ns/op	1431.47 MB/s
BenchmarkRead4MiBDelay10Msec	      10	 186893765 ns/op	  56.11 MB/s
BenchmarkRead4MiBDelay50Msec	       2	 907127114 ns/op	  11.56 MB/s
BenchmarkRead4MiBDelay150Msec	       1	2708025060 ns/op	   3.87 MB/s
BenchmarkWrite1k	       1	1623940932 ns/op	   6.46 MB/s
BenchmarkWrite16k	      10	 174293843 ns/op	  60.16 MB/s
BenchmarkWrite32k	      10	 120377272 ns/op	  87.11 MB/s
BenchmarkWrite128k	      20	  54592205 ns/op	 192.08 MB/s
BenchmarkWrite512k	      50	  66449591 ns/op	 157.80 MB/s
BenchmarkWrite1MiB	      50	  70965660 ns/op	 147.76 MB/s
BenchmarkWrite4MiB	      50	  69234861 ns/op	 151.45 MB/s
BenchmarkWrite4MiBDelay10Msec	       5	 276624260 ns/op	  37.91 MB/s
BenchmarkWrite4MiBDelay50Msec	       1	1318396552 ns/op	   7.95 MB/s
BenchmarkWrite4MiBDelay150Msec	       1	3918416658 ns/op	   2.68 MB/s
BenchmarkCopyDown10MiBDelay10Msec	      10	 152240808 ns/op	  68.88 MB/s
BenchmarkCopyDown10MiBDelay50Msec	       2	 715003188 ns/op	  14.67 MB/s
BenchmarkCopyDown10MiBDelay150Msec	       1	2116878801 ns/op	   4.95 MB/s
BenchmarkCopyUp10MiBDelay10Msec	      10	 192748258 ns/op	  54.40 MB/s
BenchmarkCopyUp10MiBDelay50Msec	       2	 691486538 ns/op	  15.16 MB/s
BenchmarkCopyUp10MiBDelay150Msec	       1	1997162991 ns/op	   5.25 MB/s
BenchmarkMarshalInit	 2000000	       644 ns/op
BenchmarkMarshalOpen	 3000000	       562 ns/op
BenchmarkMarshalWriteWorstCase	   20000	     75166 ns/op
BenchmarkMarshalWrite1k	  500000	      3862 ns/op
ok  	github.com/pkg/sftp	71.174s

Author: ggriffiniii

client.go

@@ -68,7 +68,12 @@ func NewClientPipe(rd io.Reader, wr io.WriteCloser, opts ...func(*Client) error)
 		wr.Close()
 		return nil, err
 	}
-	return sftp, sftp.recvVersion()
+	if err := sftp.recvVersion(); err != nil {
+		wr.Close()
+		return nil, err
+	}
+	go sftp.recv()
+	return sftp, nil
 }
 
 // Client represents an SFTP session on a *ssh.ClientConn SSH connection.
@@ -127,6 +132,41 @@ func (c *Client) recvVersion() error {
 	return nil
 }
 
+func (c *Client) recv() {
+	sendAll := func(res result) {
+		c.mu.Lock()
+		listeners := make([]chan<- result, len(c.inflight))
+		for _, ch := range c.inflight {
+			listeners = append(listeners, ch)
+		}
+		c.mu.Unlock()
+		for _, ch := range listeners {
+			ch <- res
+		}
+	}
+	for {
+		typ, data, err := recvPacket(c.r)
+		if err != nil {
+			// Return the error to all listeners.
+			sendAll(result{err: err})
+			return
+		}
+		sid, _ := unmarshalUint32(data)
+		c.mu.Lock()
+		ch, ok := c.inflight[sid]
+		delete(c.inflight, sid)
+		c.mu.Unlock()
+		if !ok {
+			// This is an unexpected occurrence. Send the error
+			// back to all listeners so that they terminate
+			// gracefully.
+			sendAll(result{err: fmt.Errorf("sid: %v not fond", sid)})
+			return
+		}
+		ch <- result{typ: typ, data: data}
+	}
+}
+
 // Walk returns a new Walker rooted at root.
 func (c *Client) Walk(root string) *fs.Walker {
 	return fs.WalkFS(root, c)
@@ -526,33 +566,9 @@ func (c *Client) dispatchRequest(ch chan<- result, p idmarshaler) {
 	err := sendPacket(c.w, p)
 	if err != nil {
 		go func() { ch <- result{err: err} }()
+		delete(c.inflight, p.id())
 		return
 	}
-	go func() {
-		c.mu.Lock()
-		typ, data, err := recvPacket(c.r)
-		if err != nil {
-			c.mu.Unlock()
-			ch <- result{err: err}
-			return
-		}
-
-		// the packet we received may not be the one we sent
-		// look up the channel of the owner and dispatch it to
-		// the sender.
-		sid, _ := unmarshalUint32(data)
-		ch1, ok := c.inflight[sid]
-		delete(c.inflight, sid)
-		c.mu.Unlock()
-		if !ok {
-			// send error back to the caller which started this goroutine
-			// this may not be the caller who issued the request that we
-			// have a response for.
-			ch <- result{err: fmt.Errorf("sid: %v not found", sid)}
-			return
-		}
-		ch1 <- result{typ: typ, data: data}
-	}()
 }
 
 // Creates the specified directory. An error will be returned if a file or

client_integration_test.go

@@ -112,14 +112,6 @@ func testClient(t testing.TB, readonly bool, delay time.Duration) (*Client, *exe
 		t.Fatal(err)
 	}
 
-	if err := sftp.sendInit(); err != nil {
-		defer cmd.Wait()
-		t.Fatal(err)
-	}
-	if err := sftp.recvVersion(); err != nil {
-		defer cmd.Wait()
-		t.Fatal(err)
-	}
 	return sftp, cmd
 }