Properly passing data on stdin to a command and receiving data from stdout of that command in golang

I have the following program:

package main

import "bytes"
import "io"
import "log"
import "os"
import "os/exec"
import "time"

func main() {
	runCatFromStdinWorks(populateStdin("aaa\n"))
	runCatFromStdinWorks(populateStdin("bbb\n"))
}

func populateStdin(str string) func(io.WriteCloser) {
	return func(stdin io.WriteCloser) {
		defer stdin.Close()
		io.Copy(stdin, bytes.NewBufferString(str))
	}
}

func runCatFromStdinWorks(populate_stdin_func func(io.WriteCloser)) {
	cmd := exec.Command("cat")
	stdin, err := cmd.StdinPipe()
	if err != nil {
		log.Panic(err)
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		log.Panic(err)
	}
	err = cmd.Start()
	if err != nil {
		log.Panic(err)
	}
	go populate_stdin_func(stdin)
	go func() {
            // Removing the following lines allow some output
            // to be fetched from cat's stdout sometimes
            time.Sleep(5 * time.Second)
            io.Copy(os.Stdout, stdout)
    }()
	err = cmd.Wait()
	if err != nil {
		log.Panic(err)
	}
}

When running in a loop, I get no results, like so:

$ while true; do go run cat_thingy.go; echo ; done



^C

This result comes after installing golang-go on an Ubuntu 12.04 from apt in a virtual machine (go version go1). I have not been able to replicate on a go installation on a Macbook Air (go version go1.0.3). It seems to be some kind of race condition. In fact, if I put a sleep(1*time.Second), I never see the issue at the expense of a random sleep in my code.

Is there something I am doing wrong in the code, or is this a bug? If it is a bug, has it been fixed?

UPDATE: Possible Clue

I found that the Command.Wait will close the pipes for communicating to/from the cat subprocess even if they still have unread data. I am not really sure the about the proper way to handle that. I guess I could create a channel to notify when the writing to stdin is done, but I would still need to know if the cat process has ended to make sure that nothing else was going to be written to its stdout pipe. I know that I can use cmd.Process.Wait to determine when the process ends, but is it safe to then call cmd.Wait?

UPDATE: Getting Closer

Here's a new cut at the code. I believe that this works as far as writing to stdin and reading from the stdout. I think I can make it properly stream the data (instead of buffering it all) if I replace the io.Copy from the stdout handling goroutine without something that streams.

package main

import "bytes"
import "fmt"
import "io"
import "log"
import "os/exec"
import "runtime"

const inputBufferBlockLength = 3*64*(2<<10) // enough to be bigger than 2x the pipe buffer of 64KiB
const numInputBlocks = 6

func main() {
    runtime.GOMAXPROCS(5)
    runCatFromStdin(populateStdin(numInputBlocks))
}

func populateStdin(numInputBlocks int) func(io.WriteCloser, chan bool) {
    return func(stdin io.WriteCloser) {
        defer stdin.Close()
        repeatedByteBases := []string{"a", "b", "c", "d", "e", "f"}
        for i := 0; i < numInputBlocks; i++ {
          repeatedBytes := bytes.NewBufferString(repeatedByteBases[i]).Bytes()
          fmt.Printf("%s\n", repeatedBytes)
          io.Copy(stdin, bytes.NewReader(bytes.Repeat(repeatedBytes, inputBufferBlockLength)))
        }
    }
}

func runCatFromStdin(populate_stdin_func func(io.WriteCloser)) {
    cmd := exec.Command("cat")
    stdin, err := cmd.StdinPipe()
    if err != nil {
        log.Panic(err)
    }
    stdout, err := cmd.StdoutPipe()
    if err != nil {
        log.Panic(err)
    }
    err = cmd.Start()
    if err != nil {
        log.Panic(err)
    }
    go populate_stdin_func(stdin)
    output_done_channel := make(chan bool)
    go func() {
        out_bytes := new(bytes.Buffer)
        io.Copy(out_bytes, stdout)
        fmt.Printf("%s\n", out_bytes)
        fmt.Println(out_bytes.Len())
        fmt.Println(inputBufferBlockLength*numInputBlocks)
        output_done_channel <- true
    }()
    <-output_done_channel
    err = cmd.Wait()
    if err != nil {
        log.Panic(err)
    }
}

Here is a version of your first code which works. Note the addition of the sync.WaitGroup to make sure you finish with the sending and receiving go routines before closing the command.

package main

import (
	"bytes"
	"io"
	"log"
	"os"
	"os/exec"
	"sync"
	"time"
)

func main() {
	runCatFromStdinWorks(populateStdin("aaa\n"))
	runCatFromStdinWorks(populateStdin("bbb\n"))
}

func populateStdin(str string) func(io.WriteCloser) {
	return func(stdin io.WriteCloser) {
		defer stdin.Close()
		io.Copy(stdin, bytes.NewBufferString(str))
	}
}

func runCatFromStdinWorks(populate_stdin_func func(io.WriteCloser)) {
	cmd := exec.Command("cat")
	stdin, err := cmd.StdinPipe()
	if err != nil {
		log.Panic(err)
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		log.Panic(err)
	}
	err = cmd.Start()
	if err != nil {
		log.Panic(err)
	}
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		populate_stdin_func(stdin)
	}()
	go func() {
		defer wg.Done()
		time.Sleep(5 * time.Second)
		io.Copy(os.Stdout, stdout)
	}()
	wg.Wait()
	err = cmd.Wait()
	if err != nil {
		log.Panic(err)
	}
}

(This is just another way of saying what @peterSO said though

> Go statements
>
> A "go" statement starts the execution of a function or method call as
> an independent concurrent thread of control, or goroutine, within the
> same address space.
>
> GoStmt = "go" Expression .
>
> The expression must be a call. The function value and parameters are
> evaluated as usual in the calling goroutine, but unlike with a regular
> call, program execution does not wait for the invoked function to
> complete. Instead, the function begins executing independently in a
> new goroutine. When the function terminates, its goroutine also
> terminates. If the function has any return values, they are discarded
> when the function completes.

Convert the gratuitous goroutines to function calls.

package main

import (
	"bytes"
	"io"
	"log"
	"os"
	"os/exec"
)

func main() {
	runCatFromStdinWorks(populateStdin("aaa\n"))
	runCatFromStdinWorks(populateStdin("bbb\n"))
}

func populateStdin(str string) func(io.WriteCloser) {
	return func(stdin io.WriteCloser) {
		defer stdin.Close()
		io.Copy(stdin, bytes.NewBufferString(str))
	}
}

func runCatFromStdinWorks(populate_stdin_func func(io.WriteCloser)) {
	cmd := exec.Command("cat")
	stdin, err := cmd.StdinPipe()
	if err != nil {
		log.Panic(err)
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		log.Panic(err)
	}
	err = cmd.Start()
	if err != nil {
		log.Panic(err)
	}
	populate_stdin_func(stdin)
	io.Copy(os.Stdout, stdout)
	err = cmd.Wait()
	if err != nil {
		log.Panic(err)
	}
}