pipeline can deadlock if component finishes early

[Roger Peppe]

The following program deadlocks, but it seems like a reasonable
thing to want to do.

Related to this, sometimes a "write on closed pipe"
error is actually expected, for example when running {foo | head -20}).
I wonder if it might be nice to provide some support for
this , though I'm not quite sure what form it would take.
The Errors value is currently hard to analyse, as there's no way
of correlating the individual errors back to the components
that caused them.

package main

import (
 "bufio"
 "bytes"
 "errors"
 "fmt"
 "io"
 "log"
 "strings"

 "labix.org/v2/pipe"
)

func main() {
 err := pipe.Run(
  pipe.Line(
   pipe.Read(largeFile()),
   pipe.Exec("cat"),
   Search(func(line []byte) bool {
    log.Printf("found")
    return strings.HasPrefix(string(line), "0 ")
   }),
  ),
 )
 if err != nil {
  log.Fatal(err)
 }
 log.Printf("done")
}

var ErrNotFound = errors.New("not found")

func Search(look func(line []byte) bool) pipe.Pipe {
 return pipe.TaskFunc(func(st *pipe.State) error {
  s := bufio.NewScanner(st.Stdin)
  for s.Scan() {
   if look(s.Bytes()) {
    return nil
   }
  }
  return ErrNotFound
 })
}

func largeFile() io.Reader {
 var buf bytes.Buffer
 for i := 0; i < 10000; i++ {
  fmt.Fprintln(&buf, i, " a line to read and more and some more again")
 }
 return &buf
}