Fix output synchronization by flushing appropriately (#80)

Author: benhoyt

goawk_test.go

@@ -395,6 +395,10 @@ func TestCommandLine(t *testing.T) {
 		{[]string{"--", "{ print $1 }", "-file"}, "", "", `file "-file" not found`},
 		{[]string{"{ print $1 }", "-file"}, "", "", `file "-file" not found`},
 
+		// Output synchronization
+		{[]string{`BEGIN { print "1"; print "2"|"cat" }`}, "", "1\n2\n", ""},
+		{[]string{`BEGIN { print "1"; "echo 2" | getline x; print x }`}, "", "1\n2\n", ""},
+
 		// Parse error formatting
 		{[]string{"@"}, "", "", "<cmdline>:1:1: unexpected char\n@\n^"},
 		{[]string{"BEGIN {\n\tx*;\n}"}, "", "", "<cmdline>:2:4: expected expression instead of ;\n    x*;\n      ^"},
@@ -415,6 +419,13 @@ func TestCommandLine(t *testing.T) {
 	}
 }
 
+func TestDevStdout(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("/dev/stdout not presnt on Windows")
+	}
+	runAWKs(t, []string{`BEGIN { print "1"; print "2">"/dev/stdout" }`}, "", "1\n2\n", "")
+}
+
 func runGoAWK(args []string, stdin string) (stdout, stderr string, err error) {
 	cmd := exec.Command(goAWKExe, args...)
 	if stdin != "" {

interp/functions.go

@@ -195,9 +195,10 @@ func (p *interp) callBuiltin(op Token, argExprs []Expr) (value, error) {
 		cmd := p.execShell(cmdline)
 		cmd.Stdout = p.output
 		cmd.Stderr = p.errorOutput
+		_ = p.flushAll() // ensure synchronization
 		err := cmd.Start()
 		if err != nil {
-			fmt.Fprintln(p.errorOutput, err)
+			p.printErrorf("%s\n", err)
 			return num(-1), nil
 		}
 		err = cmd.Wait()
@@ -206,7 +207,7 @@ func (p *interp) callBuiltin(op Token, argExprs []Expr) (value, error) {
 				code := exitErr.ProcessState.ExitCode()
 				return num(float64(code)), nil
 			} else {
-				fmt.Fprintf(p.errorOutput, "unexpected error running command %q: %v\n", cmdline, err)
+				p.printErrorf("unexpected error running command %q: %v\n", cmdline, err)
 				return num(-1), nil
 			}
 		}

interp/interp.go

@@ -893,6 +893,7 @@ func (p *interp) eval(expr Expr) (value, error) {
 			}
 			line = scanner.Text()
 		default:
+			p.flushOutputAndError() // Flush output in case they've written a prompt
 			var err error
 			line, err = p.nextLine()
 			if err == io.EOF {

interp/interp_test.go

@@ -1099,6 +1099,38 @@ BEGIN {
 	if !reflect.DeepEqual(f.flushes, expected) {
 		t.Fatalf("expected flushes %q, got %q", expected, f.flushes)
 	}
+
+	// Ensure output is flushed before getline reads from stdin
+	src = `
+BEGIN {
+	printf "Prompt: "
+	getline x
+	print x
+}`
+	f = &mockFlusher{}
+	testGoAWK(t, src, "42\n", "", "", nil, func(config *interp.Config) {
+		config.Output = f
+	})
+	expected = []string{"Prompt: ", "Prompt: 42\n"}
+	if !reflect.DeepEqual(f.flushes, expected) {
+		t.Fatalf("expected flushes %q, got %q", expected, f.flushes)
+	}
+
+	// Ensure output is flushed before system()
+	src = `
+BEGIN {
+  print "one"
+  system("echo .")
+  print "two"
+}`
+	f = &mockFlusher{}
+	testGoAWK(t, src, "", "", "", nil, func(config *interp.Config) {
+		config.Output = f
+	})
+	expected = []string{"one\n", "one\n.\ntwo\n"}
+	if !reflect.DeepEqual(f.flushes, expected) {
+		t.Fatalf("expected flushes %q, got %q", expected, f.flushes)
+	}
 }
 
 type errorFlusher struct {

interp/io.go

@@ -33,7 +33,7 @@ type bufferedWriteCloser struct {
 	io.Closer
 }
 
-func newBufferedWriteClose(w io.WriteCloser) *bufferedWriteCloser {
+func newBufferedWriteCloser(w io.WriteCloser) *bufferedWriteCloser {
 	writer := bufio.NewWriterSize(w, outputBufSize)
 	return &bufferedWriteCloser{writer, w}
 }
@@ -76,6 +76,7 @@ func (p *interp) getOutputStream(redirect Token, dest Expr) (io.Writer, error) {
 		if p.noFileWrites {
 			return nil, newError("can't write to file due to NoFileWrites")
 		}
+		p.flushOutputAndError() // ensure synchronization
 		flags := os.O_CREATE | os.O_WRONLY
 		if redirect == GREATER {
 			flags |= os.O_TRUNC
@@ -86,7 +87,7 @@ func (p *interp) getOutputStream(redirect Token, dest Expr) (io.Writer, error) {
 		if err != nil {
 			return nil, newError("output redirection error: %s", err)
 		}
-		buffered := newBufferedWriteClose(w)
+		buffered := newBufferedWriteCloser(w)
 		p.outputStreams[name] = buffered
 		return buffered, nil
 
@@ -102,13 +103,14 @@ func (p *interp) getOutputStream(redirect Token, dest Expr) (io.Writer, error) {
 		}
 		cmd.Stdout = p.output
 		cmd.Stderr = p.errorOutput
+		p.flushOutputAndError() // ensure synchronization
 		err = cmd.Start()
 		if err != nil {
-			fmt.Fprintln(p.errorOutput, err)
+			p.printErrorf("%s\n", err)
 			return ioutil.Discard, nil
 		}
 		p.commands[name] = cmd
-		buffered := newBufferedWriteClose(w)
+		buffered := newBufferedWriteCloser(w)
 		p.outputStreams[name] = buffered
 		return buffered, nil
 
@@ -166,9 +168,10 @@ func (p *interp) getInputScannerPipe(name string) (*bufio.Scanner, error) {
 	if err != nil {
 		return nil, newError("error connecting to stdout pipe: %v", err)
 	}
+	p.flushOutputAndError() // ensure synchronization
 	err = cmd.Start()
 	if err != nil {
-		fmt.Fprintln(p.errorOutput, err)
+		p.printErrorf("%s\n", err)
 		return bufio.NewScanner(strings.NewReader("")), nil
 	}
 	scanner := p.newScanner(r)
@@ -468,7 +471,7 @@ func (p *interp) flushAll() bool {
 func (p *interp) flushStream(name string) bool {
 	writer := p.outputStreams[name]
 	if writer == nil {
-		fmt.Fprintf(p.errorOutput, "error flushing %q: not an output file or pipe\n", name)
+		p.printErrorf("error flushing %q: not an output file or pipe\n", name)
 		return false
 	}
 	return p.flushWriter(name, writer)
@@ -481,11 +484,35 @@ type flusher interface {
 // Flush given output writer, and report whether it was flushed successfully
 // (logging an error if not).
 func (p *interp) flushWriter(name string, writer io.Writer) bool {
-	flusher := writer.(flusher)
+	flusher, ok := writer.(flusher)
+	if !ok {
+		return true // not a flusher, don't error
+	}
 	err := flusher.Flush()
 	if err != nil {
-		fmt.Fprintf(p.errorOutput, "error flushing %q: %v\n", name, err)
+		p.printErrorf("error flushing %q: %v\n", name, err)
 		return false
 	}
 	return true
 }
+
+// Flush output and error streams.
+func (p *interp) flushOutputAndError() {
+	if flusher, ok := p.output.(flusher); ok {
+		_ = flusher.Flush()
+	}
+	if flusher, ok := p.errorOutput.(flusher); ok {
+		_ = flusher.Flush()
+	}
+}
+
+// Print a message to the error output stream, flushing as necessary.
+func (p *interp) printErrorf(format string, args ...interface{}) {
+	if flusher, ok := p.output.(flusher); ok {
+		_ = flusher.Flush() // ensure synchronization
+	}
+	fmt.Fprintf(p.errorOutput, format, args...)
+	if flusher, ok := p.errorOutput.(flusher); ok {
+		_ = flusher.Flush()
+	}
+}