Test this on Linux

4 files changed, 37 insertions, 71 deletions

diff --git a/xdelta3/go/src/regtest.go b/xdelta3/go/src/regtest.go
index 4507b8c..85bd43a 100644
--- a/xdelta3/go/src/regtest.go
+++ b/xdelta3/go/src/regtest.go
@@ -57,14 +57,20 @@ func smokeTest(r *xdelta.Runner, p *xdelta.Program) {
 }
 
 func offsetTest(r *xdelta.Runner, p *xdelta.Program, bufsize, offset, length int64) {
+        // Note there is a strong potential to deadlock or fail due to
+        // a broken test in this test for several reasons:
+        // (a) decoder is not required to read the entire source file
+        // (b) decoder defers open to source file until first window received
+        // (c) open on a fifo blocks until a reader opens
+        // (d) sub-process Wait can invalidate busy file descriptors
         t, g := xdelta.NewTestGroup(r)
-        eargs := []string{"-e", "-0", fmt.Sprint("-B", bufsize), "-vv", fmt.Sprint("-W", winsize)}
+        eargs := []string{"-e", "-0", fmt.Sprint("-B", bufsize), "-vvvvvvv", fmt.Sprint("-W", winsize)}
         enc, err := t.Exec("encode", p, true, eargs)
         if err != nil {
                 g.Panic(err)
         }
         
-        dargs := []string{"-d", fmt.Sprint("-B", bufsize), "-vv", fmt.Sprint("-W", winsize)}
+        dargs := []string{"-d", fmt.Sprint("-B", bufsize), "-vvvvvvv", fmt.Sprint("-W", winsize)}
         dec, err := t.Exec("decode", p, true, dargs)
         if err != nil {
                 g.Panic(err)
@@ -79,10 +85,10 @@ func offsetTest(r *xdelta.Runner, p *xdelta.Program, bufsize, offset, length int
         t.CompareStreams(dec.Stdout, read, length)
 
         // The decoder output ("read", above) is compared with the
-        // test-provided output ("write", below).
-        xdelta.WriteRstreams(t, seed, offset, length,
-                t.NewDualWriter(enc.Srcin, dec.Srcin),
-                t.NewDualWriter(dec.Srcin, write))
+        // test-provided output ("write", below).  The following
+        // generates the input and output twice.
+        t.WriteRstreams("encode", seed, offset, length, enc.Srcin, dec.Srcin)
+        t.WriteRstreams("decode", seed, offset, length, dec.Srcin, write)
         t.Wait(g, enc, dec)
 }
 
diff --git a/xdelta3/go/src/xdelta/rstream.go b/xdelta3/go/src/xdelta/rstream.go
index eafcfeb..7d205c6 100644
--- a/xdelta3/go/src/xdelta/rstream.go
+++ b/xdelta3/go/src/xdelta/rstream.go
@@ -3,19 +3,20 @@ package xdelta
 
 import (
         "io"
+        "fmt"
         "math/rand"
 )
 
 const (
-        blocksize = 1<<20
+        blocksize = 1<<17
 )
 
-func WriteRstreams(t *TestGroup, seed, offset, len int64,
+func (t *TestGroup) WriteRstreams(desc string, seed, offset, len int64,
         src, tgt io.WriteCloser) {
-        t.Go("src-write", func (g Goroutine) {
+        t.Go("src-write:"+desc, func (g Goroutine) {
                 writeOne(g, seed, 0, len, src, false)
         })
-        t.Go("tgt-write", func (g Goroutine) {
+        t.Go("tgt-write:"+desc, func (g Goroutine) {
                 writeOne(g, seed, offset, len, tgt, true)
         })
 }
@@ -29,31 +30,37 @@ func writeOne(g Goroutine, seed, offset, len int64, stream io.WriteCloser, reada
         }
         if offset != 0 {
                 // Fill with other random data until the offset
-                if err := writeRand(rand.New(rand.NewSource(^seed)), offset, stream); err != nil {
+                fmt.Println(g, "pre-offset case", offset)
+                if err := writeRand(g, rand.New(rand.NewSource(^seed)), offset, stream); err != nil {
                         g.Panic(err)
                 }
         }
-        if err := writeRand(rand.New(rand.NewSource(seed)),
+        fmt.Println(g, "offset case", len - offset)
+        if err := writeRand(g, rand.New(rand.NewSource(seed)),
                 len - offset, stream); err != nil {
                 g.Panic(err)
         }
+        fmt.Println(g, "closing", len)
         if err := stream.Close(); err != nil {
                 g.Panic(err)
         }
         g.OK()
 }
 
-func writeRand(r *rand.Rand, len int64, s io.Writer) error {
+func writeRand(g Goroutine, r *rand.Rand, len int64, s io.Writer) error {
         blk := make([]byte, blocksize)
+        fmt.Println(g, "rstream", len)
         for len > 0 {
                 fillRand(r, blk)
                 c := blocksize
                 if len < blocksize {
                         c = int(len)
                 }
+                fmt.Println(g, "writing...", c, s)
                 if _, err := s.Write(blk[0:c]); err != nil {
                         return err
                 }
+                fmt.Println(g, "...written", c)
                 len -= int64(c)
         }
         return nil
diff --git a/xdelta3/go/src/xdelta/test.go b/xdelta3/go/src/xdelta/test.go
index 05de487..427e4c7 100644
--- a/xdelta3/go/src/xdelta/test.go
+++ b/xdelta3/go/src/xdelta/test.go
@@ -78,17 +78,14 @@ func (t *TestGroup) CopyStreams(r io.ReadCloser, w io.WriteCloser) Goroutine {
         return t.Go("copy", func(g Goroutine) {
                 _, err := io.Copy(w, r)
                 if err != nil {
-                        fmt.Println("CopyS", err)
                         g.Panic(err)
                 }
                 err = r.Close()
                 if err != nil {
-                        fmt.Println("CloseS1", err)
                         g.Panic(err)
                 }
                 err = w.Close()
                 if err != nil {
-                        fmt.Println("CloseS2", err)
                         g.Panic(err)
                 }
                 g.OK()
@@ -157,7 +154,7 @@ func (t *TestGroup) Exec(desc string, p *Program, srcfifo bool, flags []string)
         run.Cmd.Path = p.Path
         run.Cmd.Args = append(args, flags...)
         run.Cmd.Dir = t.Runner.Testdir
-
+        fmt.Println("Start command", run.Cmd.Args)
         if serr := run.Cmd.Start(); serr != nil {
                 return nil, serr
         }
@@ -169,7 +166,9 @@ func (r *Run) Wait() error {
 }
 
 func writeFifo(srcfile string, read io.Reader) error {
+        fmt.Println("About to open", srcfile)
         fifo, err := os.OpenFile(srcfile, os.O_WRONLY, 0600)
+        fmt.Println("Opened!!!", srcfile)
         if err != nil {
                 fifo.Close()
                 return err
diff --git a/xdelta3/go/src/xdelta/tgroup.go b/xdelta3/go/src/xdelta/tgroup.go
index 1852bac..f94c03e 100644
--- a/xdelta3/go/src/xdelta/tgroup.go
+++ b/xdelta3/go/src/xdelta/tgroup.go
@@ -2,7 +2,6 @@ package xdelta
 
 import (
         "fmt"
-        "io"
         "sync"
         "time"
 )
@@ -21,7 +20,7 @@ type Goroutine struct {
 }
 
 func NewTestGroup(r *Runner) (*TestGroup, Goroutine) {
-        g := Goroutine{"main", make(chan error)}
+        g := Goroutine{"main", make(chan error, 1)}
         wc := make(chan bool)
         tg := &TestGroup{Runner: r, running: []Goroutine{g}, waitChan: wc}
         go waitAll(tg, wc)
@@ -33,6 +32,7 @@ func (g *Goroutine) String() string {
 }
 
 func (g *Goroutine) OK() {
+        fmt.Println("OK", g)
         if g.errChan != nil {
                 g.errChan <- nil
                 _ = <- g.errChan
@@ -41,7 +41,7 @@ func (g *Goroutine) OK() {
 }
 
 func (g *Goroutine) Panic(err error) {
-        fmt.Println(g, err)
+        fmt.Println("PANIC", g, err)
         if g.errChan != nil {
                 g.errChan <- err
                 _ = <- g.errChan
@@ -50,7 +50,7 @@ func (g *Goroutine) Panic(err error) {
 }
 
 func (t *TestGroup) Go(name string, f func(Goroutine)) Goroutine {
-        g := Goroutine{name, make(chan error)}
+        g := Goroutine{name, make(chan error, 1)}
         t.Lock()
         t.running = append(t.running, g)
         t.Unlock()
@@ -88,6 +88,10 @@ func waitAll(t *TestGroup, wc chan bool) {
                         t.Unlock()
                         break
                 }
+                // fmt.Println("----------------------------------------------------------------------")
+                // for _, r := range t.running {
+                //      fmt.Println("Waiting for", r)
+                // }
                 runner := t.running[0]
                 t.running = t.running[1:]
                 t.Unlock()
@@ -110,53 +114,3 @@ func waitAll(t *TestGroup, wc chan bool) {
         }
         wc <- true
 }
-
-type dualWriter struct {
-        e, d chan []byte
-}
-
-func (d *dualWriter) Write(p []byte) (int, error) {
-        if len(p) != 0 {
-                d.e <- p
-                d.d <- p
-        }
-        return len(p), nil
-}
-
-func (d *dualWriter) Close() error {
-        d.e <- nil
-        d.d <- nil
-        _ = <- d.e
-        _ = <- d.d
-        return nil
-}
-
-func newWriter(c chan []byte, a io.WriteCloser) func (Goroutine) {
-        return func (g Goroutine) {
-                for {
-                        d := <- c
-                        if d == nil {
-                                if err := a.Close(); err != nil {
-                                        g.Panic(err)
-                                }
-                                c <- nil
-                                g.OK()
-                                return
-                        }
-                        if num, err := a.Write(d); err != nil {
-                                g.Panic(err)
-                        } else if num != len(d) {
-                                g.Panic(fmt.Errorf("Invalid write: %v != %v", num, len(d)))
-                        }
-                }
-        }
-}
-
-func (t *TestGroup) NewDualWriter(a1, a2 io.WriteCloser) io.WriteCloser {
-        c1 := make(chan []byte)
-        c2 := make(chan []byte)
-        r := &dualWriter{c1, c2}
-        t.Go("writer0", newWriter(c1, a1))
-        t.Go("writer1", newWriter(c2, a2))
-        return r
-}