Add integration tests that measure how throughput varies with latency.

This clearly shows the impact high latency networks have on the
throughput of sftp transfers. The current benchmark results on my
machine are:

$ go test -bench=. -integration
PASS
BenchmarkRead1k	      20	  93028268 ns/op	 112.72 MB/s
BenchmarkRead16k	     100	  14912438 ns/op	 703.16 MB/s
BenchmarkRead32k	     100	  12282661 ns/op	 853.71 MB/s
BenchmarkRead128k	     100	  10550935 ns/op	 993.83 MB/s
BenchmarkRead512k	     100	  12970518 ns/op	 808.44 MB/s
BenchmarkRead1MiB	     100	  11526693 ns/op	 909.70 MB/s
BenchmarkRead4MiB	     100	  15761135 ns/op	 665.30 MB/s
BenchmarkRead4MiBDelay10Msec	       1	2662626928 ns/op	   3.94 MB/s
BenchmarkRead4MiBDelay50Msec	       1	12977334971 ns/op	   0.81 MB/s
BenchmarkRead4MiBDelay150Msec	       1	38777294968 ns/op	   0.27 MB/s
BenchmarkWrite1k	       5	 235318220 ns/op	  44.56 MB/s
BenchmarkWrite16k	      50	  57347520 ns/op	 182.85 MB/s
BenchmarkWrite32k	      30	  66151560 ns/op	 158.51 MB/s
BenchmarkWrite128k	      20	  86642733 ns/op	 121.02 MB/s
BenchmarkWrite512k	      20	  72648883 ns/op	 144.34 MB/s
BenchmarkWrite1MiB	      20	  87145317 ns/op	 120.33 MB/s
BenchmarkWrite4MiB	      50	  64098344 ns/op	 163.59 MB/s
BenchmarkWrite4MiBDelay10Msec	       1	3360676836 ns/op	   3.12 MB/s
BenchmarkWrite4MiBDelay50Msec	       1	16321294488 ns/op	   0.64 MB/s
BenchmarkWrite4MiBDelay150Msec	       1	48731747397 ns/op	   0.22 MB/s
BenchmarkCopyDown10MiBDelay10Msec	       1	3369680836 ns/op	   3.11 MB/s
BenchmarkCopyDown10MiBDelay50Msec	       1	16228310257 ns/op	   0.65 MB/s
BenchmarkCopyDown10MiBDelay150Msec	       1	48505306499 ns/op	   0.22 MB/s
BenchmarkCopyUp10MiBDelay10Msec	       1	3393474907 ns/op	   3.09 MB/s
BenchmarkCopyUp10MiBDelay50Msec	       1	16273951972 ns/op	   0.64 MB/s
BenchmarkCopyUp10MiBDelay150Msec	       1	48461724780 ns/op	   0.22 MB/s
BenchmarkMarshalInit	 2000000	       779 ns/op
BenchmarkMarshalOpen	 2000000	       645 ns/op
BenchmarkMarshalWriteWorstCase	   20000	     70954 ns/op
BenchmarkMarshalWrite1k	  300000	      5072 ns/op
ok  	github.com/pkg/sftp	296.179s

client_integration_test.go

@@ -17,13 +17,15 @@ import (
 	"syscall"
 	"testing"
 	"testing/quick"
+	"time"
 
 	"github.com/kr/fs"
 )
 
 const (
-	READONLY  = true
-	READWRITE = false
+	READONLY                = true
+	READWRITE               = false
+	NO_DELAY  time.Duration = 0
 
 	debuglevel = "ERROR" // set to "DEBUG" for debugging
 )
@@ -31,9 +33,57 @@ const (
 var testIntegration = flag.Bool("integration", false, "perform integration tests against sftp server process")
 var testSftp = flag.String("sftp", "/usr/lib/openssh/sftp-server", "location of the sftp server binary")
 
+type delayedWrite struct {
+	t time.Time
+	b []byte
+}
+
+// delayedWriter wraps a writer and artificially delays the write. This is
+// meant to mimic connections with various latencies. Error's returned from the
+// underlying writer will panic so this should only be used over reliable
+// connections.
+type delayedWriter struct {
+	w      io.WriteCloser
+	ch     chan delayedWrite
+	closed chan struct{}
+}
+
+func newDelayedWriter(w io.WriteCloser, delay time.Duration) io.WriteCloser {
+	ch := make(chan delayedWrite, 128)
+	closed := make(chan struct{})
+	go func() {
+		for writeMsg := range ch {
+			time.Sleep(writeMsg.t.Add(delay).Sub(time.Now()))
+			n, err := w.Write(writeMsg.b)
+			if err != nil {
+				panic("write error")
+			}
+			if n < len(writeMsg.b) {
+				panic("showrt write")
+			}
+		}
+		w.Close()
+		close(closed)
+	}()
+	return delayedWriter{w: w, ch: ch, closed: closed}
+}
+
+func (w delayedWriter) Write(b []byte) (int, error) {
+	bcopy := make([]byte, len(b))
+	copy(bcopy, b)
+	w.ch <- delayedWrite{t: time.Now(), b: bcopy}
+	return len(b), nil
+}
+
+func (w delayedWriter) Close() error {
+	close(w.ch)
+	<-w.closed
+	return nil
+}
+
 // testClient returns a *Client connected to a localy running sftp-server
 // the *exec.Cmd returned must be defer Wait'd.
-func testClient(t testing.TB, readonly bool) (*Client, *exec.Cmd) {
+func testClient(t testing.TB, readonly bool, delay time.Duration) (*Client, *exec.Cmd) {
 	if !*testIntegration {
 		t.Skip("skipping intergration test")
 	}
@@ -46,6 +96,9 @@ func testClient(t testing.TB, readonly bool) (*Client, *exec.Cmd) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	if delay > NO_DELAY {
+		pw = newDelayedWriter(pw, delay)
+	}
 	pr, err := cmd.StdoutPipe()
 	if err != nil {
 		t.Fatal(err)
@@ -71,7 +124,7 @@ func testClient(t testing.TB, readonly bool) (*Client, *exec.Cmd) {
 }
 
 func TestNewClient(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 
 	if err := sftp.Close(); err != nil {
@@ -80,7 +133,7 @@ func TestNewClient(t *testing.T) {
 }
 
 func TestClientLstat(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -106,7 +159,7 @@ func TestClientLstat(t *testing.T) {
 }
 
 func TestClientLstatMissing(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -123,7 +176,7 @@ func TestClientLstatMissing(t *testing.T) {
 }
 
 func TestClientMkdir(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -141,7 +194,7 @@ func TestClientMkdir(t *testing.T) {
 }
 
 func TestClientOpen(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -200,7 +253,7 @@ func (s seek) end(t *testing.T, r io.ReadSeeker) {
 }
 
 func TestClientSeek(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -244,7 +297,7 @@ func TestClientSeek(t *testing.T) {
 }
 
 func TestClientCreate(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -263,7 +316,7 @@ func TestClientCreate(t *testing.T) {
 }
 
 func TestClientAppend(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -282,7 +335,7 @@ func TestClientAppend(t *testing.T) {
 }
 
 func TestClientCreateFailed(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -303,7 +356,7 @@ func TestClientCreateFailed(t *testing.T) {
 }
 
 func TestClientFileStat(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -334,7 +387,7 @@ func TestClientFileStat(t *testing.T) {
 }
 
 func TestClientRemove(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -351,7 +404,7 @@ func TestClientRemove(t *testing.T) {
 }
 
 func TestClientRemoveDir(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -368,7 +421,7 @@ func TestClientRemoveDir(t *testing.T) {
 }
 
 func TestClientRemoveFailed(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -385,7 +438,7 @@ func TestClientRemoveFailed(t *testing.T) {
 }
 
 func TestClientRename(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -406,7 +459,7 @@ func TestClientRename(t *testing.T) {
 }
 
 func TestClientReadLine(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -450,7 +503,7 @@ var clientReadTests = []struct {
 }
 
 func TestClientRead(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -538,7 +591,7 @@ var clientWriteTests = []struct {
 }
 
 func TestClientWrite(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -665,7 +718,7 @@ func mark(path string, info os.FileInfo, err error, errors *[]error, clear bool)
 }
 
 func TestClientWalk(t *testing.T) {
-	sftp, cmd := testClient(t, READONLY)
+	sftp, cmd := testClient(t, READONLY, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -748,11 +801,11 @@ func TestClientWalk(t *testing.T) {
 	}
 }
 
-func benchmarkRead(b *testing.B, bufsize int) {
+func benchmarkRead(b *testing.B, bufsize int, delay time.Duration) {
 	size := 10*1024*1024 + 123 // ~10MiB
 
 	// open sftp client
-	sftp, cmd := testClient(b, READONLY)
+	sftp, cmd := testClient(b, READONLY, delay)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -787,38 +840,50 @@ func benchmarkRead(b *testing.B, bufsize int) {
 }
 
 func BenchmarkRead1k(b *testing.B) {
-	benchmarkRead(b, 1*1024)
+	benchmarkRead(b, 1*1024, NO_DELAY)
 }
 
 func BenchmarkRead16k(b *testing.B) {
-	benchmarkRead(b, 16*1024)
+	benchmarkRead(b, 16*1024, NO_DELAY)
 }
 
 func BenchmarkRead32k(b *testing.B) {
-	benchmarkRead(b, 32*1024)
+	benchmarkRead(b, 32*1024, NO_DELAY)
 }
 
 func BenchmarkRead128k(b *testing.B) {
-	benchmarkRead(b, 128*1024)
+	benchmarkRead(b, 128*1024, NO_DELAY)
 }
 
 func BenchmarkRead512k(b *testing.B) {
-	benchmarkRead(b, 512*1024)
+	benchmarkRead(b, 512*1024, NO_DELAY)
 }
 
 func BenchmarkRead1MiB(b *testing.B) {
-	benchmarkRead(b, 1024*1024)
+	benchmarkRead(b, 1024*1024, NO_DELAY)
 }
 
 func BenchmarkRead4MiB(b *testing.B) {
-	benchmarkRead(b, 4*1024*1024)
+	benchmarkRead(b, 4*1024*1024, NO_DELAY)
 }
 
-func benchmarkWrite(b *testing.B, bufsize int) {
+func BenchmarkRead4MiBDelay10Msec(b *testing.B) {
+	benchmarkRead(b, 4*1024*1024, 10*time.Millisecond)
+}
+
+func BenchmarkRead4MiBDelay50Msec(b *testing.B) {
+	benchmarkRead(b, 4*1024*1024, 50*time.Millisecond)
+}
+
+func BenchmarkRead4MiBDelay150Msec(b *testing.B) {
+	benchmarkRead(b, 4*1024*1024, 150*time.Millisecond)
+}
+
+func benchmarkWrite(b *testing.B, bufsize int, delay time.Duration) {
 	size := 10*1024*1024 + 123 // ~10MiB
 
 	// open sftp client
-	sftp, cmd := testClient(b, false)
+	sftp, cmd := testClient(b, false, delay)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -871,35 +936,47 @@ func benchmarkWrite(b *testing.B, bufsize int) {
 }
 
 func BenchmarkWrite1k(b *testing.B) {
-	benchmarkWrite(b, 1*1024)
+	benchmarkWrite(b, 1*1024, NO_DELAY)
 }
 
 func BenchmarkWrite16k(b *testing.B) {
-	benchmarkWrite(b, 16*1024)
+	benchmarkWrite(b, 16*1024, NO_DELAY)
 }
 
 func BenchmarkWrite32k(b *testing.B) {
-	benchmarkWrite(b, 32*1024)
+	benchmarkWrite(b, 32*1024, NO_DELAY)
 }
 
 func BenchmarkWrite128k(b *testing.B) {
-	benchmarkWrite(b, 128*1024)
+	benchmarkWrite(b, 128*1024, NO_DELAY)
 }
 
 func BenchmarkWrite512k(b *testing.B) {
-	benchmarkWrite(b, 512*1024)
+	benchmarkWrite(b, 512*1024, NO_DELAY)
 }
 
 func BenchmarkWrite1MiB(b *testing.B) {
-	benchmarkWrite(b, 1024*1024)
+	benchmarkWrite(b, 1024*1024, NO_DELAY)
 }
 
 func BenchmarkWrite4MiB(b *testing.B) {
-	benchmarkWrite(b, 4*1024*1024)
+	benchmarkWrite(b, 4*1024*1024, NO_DELAY)
+}
+
+func BenchmarkWrite4MiBDelay10Msec(b *testing.B) {
+	benchmarkWrite(b, 4*1024*1024, 10*time.Millisecond)
+}
+
+func BenchmarkWrite4MiBDelay50Msec(b *testing.B) {
+	benchmarkWrite(b, 4*1024*1024, 50*time.Millisecond)
+}
+
+func BenchmarkWrite4MiBDelay150Msec(b *testing.B) {
+	benchmarkWrite(b, 4*1024*1024, 150*time.Millisecond)
 }
 
 func TestClientStatVFS(t *testing.T) {
-	sftp, cmd := testClient(t, READWRITE)
+	sftp, cmd := testClient(t, READWRITE, NO_DELAY)
 	defer cmd.Wait()
 	defer sftp.Close()
 
@@ -933,3 +1010,156 @@ func TestClientStatVFS(t *testing.T) {
 	}
 
 }
+
+func benchmarkCopyDown(b *testing.B, fileSize int64, delay time.Duration) {
+	// Create a temp file and fill it with zero's.
+	src, err := ioutil.TempFile("", "sftptest")
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer src.Close()
+	srcFilename := src.Name()
+	defer os.Remove(srcFilename)
+	zero, err := os.Open("/dev/zero")
+	if err != nil {
+		b.Fatal(err)
+	}
+	n, err := io.Copy(src, io.LimitReader(zero, fileSize))
+	if err != nil {
+		b.Fatal(err)
+	}
+	if n < fileSize {
+		b.Fatal("short copy")
+	}
+	zero.Close()
+	src.Close()
+
+	sftp, cmd := testClient(b, READONLY, delay)
+	defer cmd.Wait()
+	defer sftp.Close()
+	b.ResetTimer()
+	b.SetBytes(fileSize)
+
+	for i := 0; i < b.N; i++ {
+		dst, err := ioutil.TempFile("", "sftptest")
+		if err != nil {
+			b.Fatal(err)
+		}
+		defer os.Remove(dst.Name())
+
+		src, err := sftp.Open(srcFilename)
+		if err != nil {
+			b.Fatal(err)
+		}
+		defer src.Close()
+		n, err := io.Copy(dst, src)
+		if err != nil {
+			b.Fatalf("copy error: %v", err)
+		}
+		if n < fileSize {
+			b.Fatal("unable to copy all bytes")
+		}
+		dst.Close()
+		fi, err := os.Stat(dst.Name())
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		if fi.Size() != fileSize {
+			b.Fatalf("wrong file size: want %d, got %d", fileSize, fi.Size())
+		}
+		os.Remove(dst.Name())
+	}
+}
+
+func BenchmarkCopyDown10MiBDelay10Msec(b *testing.B) {
+	benchmarkCopyDown(b, 10*1024*1024, 10*time.Millisecond)
+}
+
+func BenchmarkCopyDown10MiBDelay50Msec(b *testing.B) {
+	benchmarkCopyDown(b, 10*1024*1024, 50*time.Millisecond)
+}
+
+func BenchmarkCopyDown10MiBDelay150Msec(b *testing.B) {
+	benchmarkCopyDown(b, 10*1024*1024, 150*time.Millisecond)
+}
+
+func benchmarkCopyUp(b *testing.B, fileSize int64, delay time.Duration) {
+	// Create a temp file and fill it with zero's.
+	src, err := ioutil.TempFile("", "sftptest")
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer src.Close()
+	srcFilename := src.Name()
+	defer os.Remove(srcFilename)
+	zero, err := os.Open("/dev/zero")
+	if err != nil {
+		b.Fatal(err)
+	}
+	n, err := io.Copy(src, io.LimitReader(zero, fileSize))
+	if err != nil {
+		b.Fatal(err)
+	}
+	if n < fileSize {
+		b.Fatal("short copy")
+	}
+	zero.Close()
+	src.Close()
+
+	sftp, cmd := testClient(b, false, delay)
+	defer cmd.Wait()
+	defer sftp.Close()
+
+	b.ResetTimer()
+	b.SetBytes(fileSize)
+
+	for i := 0; i < b.N; i++ {
+		tmp, err := ioutil.TempFile("", "sftptest")
+		if err != nil {
+			b.Fatal(err)
+		}
+		tmp.Close()
+		defer os.Remove(tmp.Name())
+
+		dst, err := sftp.Create(tmp.Name())
+		if err != nil {
+			b.Fatal(err)
+		}
+		defer dst.Close()
+		src, err := os.Open(srcFilename)
+		if err != nil {
+			b.Fatal(err)
+		}
+		defer src.Close()
+		n, err := io.Copy(dst, src)
+		if err != nil {
+			b.Fatalf("copy error: %v", err)
+		}
+		if n < fileSize {
+			b.Fatal("unable to copy all bytes")
+		}
+
+		fi, err := os.Stat(tmp.Name())
+		if err != nil {
+			b.Fatal(err)
+		}
+
+		if fi.Size() != fileSize {
+			b.Fatalf("wrong file size: want %d, got %d", fileSize, fi.Size())
+		}
+		os.Remove(tmp.Name())
+	}
+}
+
+func BenchmarkCopyUp10MiBDelay10Msec(b *testing.B) {
+	benchmarkCopyUp(b, 10*1024*1024, 10*time.Millisecond)
+}
+
+func BenchmarkCopyUp10MiBDelay50Msec(b *testing.B) {
+	benchmarkCopyUp(b, 10*1024*1024, 50*time.Millisecond)
+}
+
+func BenchmarkCopyUp10MiBDelay150Msec(b *testing.B) {
+	benchmarkCopyUp(b, 10*1024*1024, 150*time.Millisecond)
+}