Repository: benhoyt/goawk
Pull request #206 (2023-08-30 13:08:15)

Abstract redirects and pipes into inputStream and outputStream interface
This creates the `inputStream` and `outputStream` interfaces which are satisfied by structs for all combinations of file/pipe and input/output:

1. Input File - `struct inFileStream`
2. Input Command - `struct inCmdStream`
3. Output File - `struct outFileStream`
4. Output Command - `struct outCmdStream`

The streams satisfy the `io.ReadCloser` and `io.WriteCloser` interfaces while the `outputStream` also has a `Flush() error` method. Both types of streams also have an `ExitCode() int` method which makes it trivial to determine the result of a `close()` on those streams.

This PR homogenizes behavior and fixes a number of minor issues related to command pipes. Gawk behavior is mimicked for the following:

1. The exit status of the child process is returned from `close()` when closing a pipe. Exit due to signal is `256 + signal` and if both a signal and coredump occurred it is `512 + signal`. (see #203)
2. The same rules apply to the result of the `system()` builtin function. (see #205)

A separate bug was fixed so that the child process information is cleaned up when `close()` is called on an input pipe (i.e. via `getline`). (see #204)

benhoyt: Thanks for this -- this is excellent work! I hope to take a closer look at the code changes this weekend, but at first glance is looks like a good approach.

FYI, I've just merged https://github.com/benhoyt/goawk/pull/207 to update to Go 1.16 so we can use `io.Discard` etc now. So if you merge the latest master into this branch, that should fix the test failures.
juster: That's great thanks! Do you mind if I rebase instead of merge? I rebase a lot it makes the commit history cleaner but spams up the issues & PRs... 😞 

I noticed that the Linux test had a failure. It seemed that the signal may have been handled and the sh process itself did not exit due to signal. I'll have to dig in more there and research which sh implementation is used and how it handles signals, etc. Ofc that is just a guess for now.

benhoyt: @juster Yep, rebasing is fine. My development style is usually "push any old commit to the PR branch, then squash and merge when merging the PR", but if you keep a clean commit history in the PR, I'm happy to apply directly to master instead.
juster: I found the problem with the `ExitSigPipe` test on Linux runners:

```
--- FAIL: TestStreamInCmd (0.00s)
    --- FAIL: TestStreamInCmd/ExitSigPipe (0.52s)
        iostream_test.go:168: expected Close() result: 269, got: 141
```

This test is running the `yes` utility because it prints non-stop. When I close the stream it will cause a `SIGPIPE` (13) signal to be sent to `yes` because it tries to print to a closed pipe.  The difference here is that `sh` is linked to `bash` and `bash` has a special exit code if one of its commands dies due to signal:

> When a command terminates on a fatal signal whose number is N, Bash uses the value 128+N as the exit status.

[Bash Reference Manual](https://www.gnu.org/software/bash/manual/bash.html#Exit-Status)

On my old Macbook `sh` is linked to `dash` which does not trap the signal and provide its own exit status. Instead, it also dies with `SIGPIPE`.

This is an unfortunate design choice! Because now there is no way to be sure what the exit value of `system` for example will be if there is a signal. It varies depending on what is used for `sh`. GNU cleverness strikes again 😉 .

Oh well. As we've seen in #205 signal exits are kind of an overlooked gray area. I think I will try to avoid spawning a shell at all for this test and just run `yes` directly with an `os/exec.Cmd`.

In my testing I also found some signals are caught by default with shells. For example, I will avoid `SIGQUIT` (3) which is caught and just quits the shell. I used this command to try all the signals:

```
kill -l | ./goawk 'BEGIN{j=1}{for(i=1;i<=NF;i++)printf "%2s %6s %s\n", j++, $i, system("bash -c \047kill -"i" $$\047")}'
```

juster: > I haven't reviewed `iostream.go` or `iostream_test.go` yet (the bulk of the PR!), but here's a few early comments I had time for yesterday. I'll try to get to the rest over the weekend.
> 
> Thanks again for this. Note that I'm fairly thorough in my reviews as I like to get quality code merged and code that I can "own" from here on out. Hopefully we'll end up with something we're both proud of!

Not a problem and no rush. I think I want to merge the "cleanup" commit into the first commit and pretend it was always cleaned up.

BTW, I found a simple way around the bash problem. I just need to use `exec`. What a lovely, strange command 😄.
benhoyt: One other thing: it looks like there's also merge/rebase conflicts from the `ioutil` -> `io` change that I merged -- those will need to be resolved.

Patch:
diff --git a/interp/interp.go b/interp/interp.go
index 6895e59..7f5c8c3 100644
--- a/interp/interp.go
+++ b/interp/interp.go
@@ -20,7 +20,6 @@ import (
 	"math"
 	"math/rand"
 	"os"
-	"os/exec"
 	"regexp"
 	"runtime"
 	"strconv"
@@ -80,9 +79,8 @@ type interp struct {
 	hadFiles      bool
 	input         io.Reader
 	inputBuffer   []byte
-	inputStreams  map[string]io.ReadCloser
-	outputStreams map[string]io.WriteCloser
-	commands      map[string]*exec.Cmd
+	inputStreams  map[string]inputStream
+	outputStreams map[string]outputStream
 	noExec        bool
 	noFileWrites  bool
 	noFileReads   bool
@@ -392,9 +390,8 @@ func newInterp(program *parser.Program) *interp {
 	p.outputRecordSep = "\n"
 	p.subscriptSep = "\x1c"
 
-	p.inputStreams = make(map[string]io.ReadCloser)
-	p.outputStreams = make(map[string]io.WriteCloser)
-	p.commands = make(map[string]*exec.Cmd)
+	p.inputStreams = make(map[string]inputStream)
+	p.outputStreams = make(map[string]outputStream)
 	p.scanners = make(map[string]*bufio.Scanner)
 
 	return p
diff --git a/interp/io.go b/interp/io.go
index 897f366..33659c9 100644
--- a/interp/io.go
+++ b/interp/io.go
@@ -96,51 +96,6 @@ func (p *interp) writeCSV(output io.Writer, fields []string) error {
 	return nil
 }
 
-// Buffered version of WriteCloser so output is buffered when redirecting to a
-// file (eg: print >"out")
-type bufferedWriteCloser struct {
-	*bufio.Writer
-	io.Closer
-}
-
-func newBufferedWriteCloser(w io.WriteCloser) *bufferedWriteCloser {
-	writer := bufio.NewWriterSize(w, outputBufSize)
-	return &bufferedWriteCloser{writer, w}
-}
-
-func (wc *bufferedWriteCloser) Close() error {
-	err := wc.Writer.Flush()
-	if err != nil {
-		return err
-	}
-	return wc.Closer.Close()
-}
-
-// Buffered version of WriteCloser for pipes so output is buffered when
-// redirecting to a command; waits for the command to finish on close.
-type commandWriteCloser struct {
-	*bufio.Writer
-	io.Closer
-	command *exec.Cmd
-}
-
-func newCommandWriteCloser(w io.WriteCloser, command *exec.Cmd) *commandWriteCloser {
-	writer := bufio.NewWriterSize(w, outputBufSize)
-	return &commandWriteCloser{writer, w, command}
-}
-
-func (wc *commandWriteCloser) Close() error {
-	err := wc.Writer.Flush()
-	if err != nil {
-		return err
-	}
-	err = wc.Closer.Close()
-	if err != nil {
-		return err
-	}
-	return wc.command.Wait()
-}
-
 // Determine the output stream for given redirect token and
 // destination (file or pipe name)
 func (p *interp) getOutputStream(redirect Token, destValue value) (io.Writer, error) {
@@ -169,13 +124,13 @@ func (p *interp) getOutputStream(redirect Token, destValue value) (io.Writer, er
 		} else {
 			flags |= os.O_APPEND
 		}
-		w, err := os.OpenFile(name, flags, 0644)
+		f, err := os.OpenFile(name, flags, 0644)
 		if err != nil {
 			return nil, newError("output redirection error: %s", err)
 		}
-		buffered := newBufferedWriteCloser(w)
-		p.outputStreams[name] = buffered
-		return buffered, nil
+		out := newOutFileStream(f, outputBufSize)
+		p.outputStreams[name] = out
+		return out, nil
 
 	case PIPE:
 		// Pipe to command
@@ -183,22 +138,16 @@ func (p *interp) getOutputStream(redirect Token, destValue value) (io.Writer, er
 			return nil, newError("can't write to pipe due to NoExec")
 		}
 		cmd := p.execShell(name)
-		w, err := cmd.StdinPipe()
-		if err != nil {
-			return nil, newError("error connecting to stdin pipe: %v", err)
-		}
 		cmd.Stdout = p.output
 		cmd.Stderr = p.errorOutput
 		p.flushOutputAndError() // ensure synchronization
-		err = cmd.Start()
+		out, err := newOutCmdStream(cmd)
 		if err != nil {
 			p.printErrorf("%s\n", err)
-			return io.Discard, nil
+			out = newOutNullStream()
 		}
-		p.commands[name] = cmd
-		buffered := newCommandWriteCloser(w, cmd)
-		p.outputStreams[name] = buffered
-		return buffered, nil
+		p.outputStreams[name] = out
+		return out, nil
 
 	default:
 		// Should never happen
@@ -238,13 +187,14 @@ func (p *interp) getInputScannerFile(name string) (*bufio.Scanner, error) {
 	if p.noFileReads {
 		return nil, newError("can't read from file due to NoFileReads")
 	}
-	r, err := os.Open(name)
+	f, err := os.Open(name)
 	if err != nil {
 		return nil, err // *os.PathError is handled by caller (getline returns -1)
 	}
-	scanner := p.newScanner(r, make([]byte, inputBufSize))
+	in := newInFileStream(f)
+	scanner := p.newScanner(in, make([]byte, inputBufSize))
 	p.scanners[name] = scanner
-	p.inputStreams[name] = r
+	p.inputStreams[name] = in
 	return scanner, nil
 }
 
@@ -262,19 +212,15 @@ func (p *interp) getInputScannerPipe(name string) (*bufio.Scanner, error) {
 	cmd := p.execShell(name)
 	cmd.Stdin = p.stdin
 	cmd.Stderr = p.errorOutput
-	r, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, newError("error connecting to stdout pipe: %v", err)
-	}
 	p.flushOutputAndError() // ensure synchronization
-	err = cmd.Start()
+	in, err := newInCmdStream(cmd)
 	if err != nil {
 		p.printErrorf("%s\n", err)
 		return bufio.NewScanner(strings.NewReader("")), nil
 	}
-	scanner := p.newScanner(r, make([]byte, inputBufSize))
-	p.commands[name] = cmd
-	p.inputStreams[name] = r
+
+	scanner := p.newScanner(in, make([]byte, inputBufSize))
+	p.inputStreams[name] = in
 	p.scanners[name] = scanner
 	return scanner, nil
 }
@@ -838,7 +784,7 @@ func writeOutput(w io.Writer, s string) error {
 	return err
 }
 
-// Close all streams, commands, and so on (after program execution).
+// Close all streams and so on (after program execution).
 func (p *interp) closeAll() {
 	if prevInput, ok := p.input.(io.Closer); ok {
 		_ = prevInput.Close()
@@ -849,9 +795,6 @@ func (p *interp) closeAll() {
 	for _, w := range p.outputStreams {
 		_ = w.Close()
 	}
-	for _, cmd := range p.commands {
-		_ = cmd.Wait()
-	}
 	if f, ok := p.output.(flusher); ok {
 		_ = f.Flush()
 	}
diff --git a/interp/iostream.go b/interp/iostream.go
new file mode 100644
index 0000000..572cdf6
--- /dev/null
+++ b/interp/iostream.go
@@ -0,0 +1,224 @@
+package interp
+
+// I/O streams are interfaces which allow file redirects and command pipelines to be treated
+// equivalently.
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"os/exec"
+	"syscall"
+)
+
+const (
+	notClosedExitCode = -127
+)
+
+var (
+	doubleCloseError = errors.New("close: stream already closed")
+)
+
+// firstError returns the first non-nil error or nil if all errors are nil.
+func firstError(errs ...error) error {
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Close the cmd and convert the error result into the result returned from goawk builtin functions.
+// A nil error is returned if that error describes a non-zero exit status or an unhandled signal.
+// Any other type of error returns -1 and err.
+//
+// The result mimicks gawk for expected child process errors:
+// 1. Returns the exit status of the child process and nil error on normal process exit.
+// 2. Returns 256 + signal on unhandled signal exit.
+// 3. Returns 512 + signal on unhandled signal exit which caused a core dump.
+func waitExitCode(cmd *exec.Cmd) (int, error) {
+	err := cmd.Wait()
+	if err == nil {
+		return 0, nil
+	}
+	ee, ok := err.(*exec.ExitError)
+	if !ok {
+		// Wait() returned an io error.
+		return -1, err
+	}
+	status, ok := ee.ProcessState.Sys().(syscall.WaitStatus)
+	if !ok {
+		// Maybe not all platforms support WaitStatus?
+		return -1, err
+	}
+	switch {
+	case status.CoreDump():
+		return 512 + int(status.Signal()), nil
+	case status.Signaled():
+		return 256 + int(status.Signal()), nil
+	case status.Exited():
+		return status.ExitStatus(), nil
+	default:
+		return -1, err
+	}
+}
+
+type inputStream interface {
+	io.ReadCloser
+	ExitCode() int
+}
+
+type outputStream interface {
+	io.WriteCloser
+	Flush() error
+	ExitCode() int
+}
+
+type outFileStream struct {
+	*bufio.Writer
+	closer   io.Closer
+	exitCode int
+	closed   bool
+}
+
+func newOutFileStream(wc io.WriteCloser, size int) outputStream {
+	b := bufio.NewWriterSize(wc, size)
+	return &outFileStream{b, wc, notClosedExitCode, false}
+}
+
+func (s *outFileStream) Close() error {
+	if s.closed {
+		return doubleCloseError
+	}
+	s.closed = true
+	flushErr := s.Writer.Flush()
+	closeErr := s.closer.Close()
+	if err := firstError(flushErr, closeErr); err != nil {
+		s.exitCode = -1
+		return err
+	}
+	s.exitCode = 0
+	return nil
+}
+
+func (s *outFileStream) ExitCode() int {
+	return s.exitCode
+}
+
+type outCmdStream struct {
+	*bufio.Writer
+	closer   io.Closer
+	cmd      *exec.Cmd
+	exitCode int
+	closed   bool
+}
+
+func newOutCmdStream(cmd *exec.Cmd) (outputStream, error) {
+	w, err := cmd.StdinPipe()
+	if err != nil {
+		return nil, newError("error connecting to stdin pipe: %v", err)
+	}
+	err = cmd.Start()
+	if err != nil {
+		w.Close()
+		return nil, err
+	}
+	out := &outCmdStream{bufio.NewWriterSize(w, outputBufSize), w, cmd, notClosedExitCode, false}
+	return out, nil
+}
+
+func (s *outCmdStream) Close() error {
+	if s.closed {
+		return doubleCloseError
+	}
+	s.closed = true
+	flushErr := s.Writer.Flush()
+	closeErr := s.closer.Close()
+	var waitErr error
+	s.exitCode, waitErr = waitExitCode(s.cmd)
+	return firstError(waitErr, flushErr, closeErr)
+}
+
+func (s *outCmdStream) ExitCode() int {
+	return s.exitCode
+}
+
+// An outNullStream allows writes to not do anything while fulfilling the outputStream interface.
+type outNullStream struct {
+	io.Writer
+	closed bool
+}
+
+func newOutNullStream() outputStream { return &outNullStream{io.Discard, false} }
+func (s outNullStream) Flush() error { return nil }
+func (s *outNullStream) Close() error {
+	if s.closed {
+		return doubleCloseError
+	}
+	s.closed = true
+	return nil
+}
+func (s outNullStream) ExitCode() int { return -1 }
+
+type inFileStream struct {
+	io.ReadCloser
+	exitCode int
+	closed   bool
+}
+
+func newInFileStream(rc io.ReadCloser) inputStream {
+	return &inFileStream{rc, notClosedExitCode, false}
+}
+
+func (s *inFileStream) Close() error {
+	if s.closed {
+		return doubleCloseError
+	}
+	s.closed = true
+	if err := s.ReadCloser.Close(); err != nil {
+		s.exitCode = -1
+		return err
+	}
+	s.exitCode = 0
+	return nil
+}
+
+func (s *inFileStream) ExitCode() int {
+	return s.exitCode
+}
+
+type inCmdStream struct {
+	io.ReadCloser
+	cmd      *exec.Cmd
+	exitCode int
+	closed   bool
+}
+
+func newInCmdStream(cmd *exec.Cmd) (inputStream, error) {
+	r, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, newError("error connecting to stdout pipe: %v", err)
+	}
+	err = cmd.Start()
+	if err != nil {
+		r.Close()
+		return nil, err
+	}
+	return &inCmdStream{r, cmd, notClosedExitCode, false}, nil
+}
+
+func (s *inCmdStream) Close() error {
+	if s.closed {
+		return doubleCloseError
+	}
+	s.closed = true
+	closeErr := s.ReadCloser.Close()
+	var waitErr error
+	s.exitCode, waitErr = waitExitCode(s.cmd)
+	return firstError(waitErr, closeErr)
+}
+
+func (s *inCmdStream) ExitCode() int {
+	return s.exitCode
+}
diff --git a/interp/newexecute.go b/interp/newexecute.go
index 83287fd..f1b9ed6 100644
--- a/interp/newexecute.go
+++ b/interp/newexecute.go
@@ -96,9 +96,6 @@ func (p *interp) resetCore() {
 	for k := range p.outputStreams {
 		delete(p.outputStreams, k)
 	}
-	for k := range p.commands {
-		delete(p.commands, k)
-	}
 
 	p.sp = 0
 	p.localArrays = p.localArrays[:0]
diff --git a/interp/vm.go b/interp/vm.go
index 45cc952..0ba778a 100644
--- a/interp/vm.go
+++ b/interp/vm.go
@@ -6,7 +6,6 @@ import (
 	"io"
 	"math"
 	"os"
-	"os/exec"
 	"strings"
 	"time"
 
@@ -911,33 +910,25 @@ func (p *interp) callBuiltin(builtinOp compiler.BuiltinOp) error {
 		p.replaceTop(num(math.Atan2(y.num(), x.num())))
 
 	case compiler.BuiltinClose:
+		var err error
+		code := -1
 		name := p.toString(p.peekTop())
-		var c io.Closer = p.inputStreams[name]
-		if c != nil {
+		if stream := p.inputStreams[name]; stream != nil {
 			// Close input stream
 			delete(p.inputStreams, name)
-			err := c.Close()
-			if err != nil {
-				p.replaceTop(num(-1))
-			} else {
-				p.replaceTop(num(0))
-			}
-		} else {
-			c = p.outputStreams[name]
-			if c != nil {
-				// Close output stream
-				delete(p.outputStreams, name)
-				err := c.Close()
-				if err != nil {
-					p.replaceTop(num(-1))
-				} else {
-					p.replaceTop(num(0))
-				}
-			} else {
-				// Nothing to close
-				p.replaceTop(num(-1))
-			}
+			delete(p.scanners, name)
+			err = stream.Close()
+			code = stream.ExitCode()
+		} else if stream := p.outputStreams[name]; stream != nil {
+			// Close output stream
+			delete(p.outputStreams, name)
+			err = stream.Close()
+			code = stream.ExitCode()
+		}
+		if err != nil {
+			p.printErrorf("error closing %q: %v\n", name, err)
 		}
+		p.replaceTop(num(float64(code)))
 
 	case compiler.BuiltinCos:
 		p.replaceTop(num(math.Cos(p.peekTop().num())))
@@ -1085,20 +1076,21 @@ func (p *interp) callBuiltin(builtinOp compiler.BuiltinOp) error {
 		cmd.Stdout = p.output
 		cmd.Stderr = p.errorOutput
 		_ = p.flushAll() // ensure synchronization
-		err := cmd.Run()
-		ret := 0.0
+		err := cmd.Start()
+		if err != nil {
+			// Could not start the shell so skip waiting on it.
+			p.printErrorf("%v\n", err)
+			p.replaceTop(num(-1.0))
+			return nil
+		}
+		exitCode, err := waitExitCode(cmd)
 		if err != nil {
 			if p.checkCtx && p.ctx.Err() != nil {
 				return p.ctx.Err()
 			}
-			if exitErr, ok := err.(*exec.ExitError); ok {
-				ret = float64(exitErr.ProcessState.ExitCode())
-			} else {
-				p.printErrorf("%v\n", err)
-				ret = -1
-			}
+			p.printErrorf("%v\n", err)
 		}
-		p.replaceTop(num(ret))
+		p.replaceTop(num(float64(exitCode)))
 
 	case compiler.BuiltinTolower:
 		p.replaceTop(str(strings.ToLower(p.toString(p.peekTop()))))


Test patch:
diff --git a/goawk_test.go b/goawk_test.go
index 8bf581f..e864056 100644
--- a/goawk_test.go
+++ b/goawk_test.go
@@ -234,9 +234,6 @@ func TestGAWK(t *testing.T) {
 
 		"gsubtst7":     true, // something wrong with gsub or field split/join
 		"splitwht":     true, // other awks handle split(s, a, " ") differently from split(s, a, / /)
-		"status-close": true, // hmmm, not sure what's up here
-		"sigpipe1":     true, // probable race condition: sometimes fails, sometimes passes
-
 		"parse1": true, // incorrect parsing of $$a++++ (see TODOs in interp_test.go too)
 
 		"rscompat": true, // GoAWK allows multi-char RS by default
@@ -251,6 +248,7 @@ func TestGAWK(t *testing.T) {
 		"eofsplit": true, // reads from /etc/passwd
 		"getline5": true, // removes a file while it's open
 		"iobug1":   true, // reads from /dev/null
+		"sigpipe1": true, // requires POSIX signals
 	}
 
 	sortLines := map[string]bool{
diff --git a/interp/interp_test.go b/interp/interp_test.go
index b64252d..2d8276f 100644
--- a/interp/interp_test.go
+++ b/interp/interp_test.go
@@ -587,13 +587,14 @@ BEGIN {
 		"", "error\n0\n", "", ""},
 	{`BEGIN { print system("exit 42") }  # !fuzz !posix`, "", "42\n", "", ""},
 	{`BEGIN { system("cat") }`, "foo\nbar", "foo\nbar", "", ""},
+	{`BEGIN { print system("exec /bin/kill -9 $$") } # !awk !posix !windows`, "", "265\n", "", ""},
 
 	// Test bytes/unicode handling (GoAWK currently has char==byte, unlike Gawk).
-	{`BEGIN { print match("food", "foo"), RSTART, RLENGTH }  !gawk`, "", "1 1 3\n", "", ""},
-	{`BEGIN { print match("x food y", "fo"), RSTART, RLENGTH }  !gawk`, "", "3 3 2\n", "", ""},
-	{`BEGIN { print match("x food y", "fox"), RSTART, RLENGTH }  !gawk`, "", "0 0 -1\n", "", ""},
-	{`BEGIN { print match("x food y", /[fod]+/), RSTART, RLENGTH }  !gawk`, "", "3 3 4\n", "", ""},
-	{`BEGIN { print match("絵 fööd y", /[föd]+/), RSTART, RLENGTH }  !gawk`, "", "5 5 6\n", "", ""},
+	{`BEGIN { print match("food", "foo"), RSTART, RLENGTH } # !gawk`, "", "1 1 3\n", "", ""},
+	{`BEGIN { print match("x food y", "fo"), RSTART, RLENGTH } # !gawk`, "", "3 3 2\n", "", ""},
+	{`BEGIN { print match("x food y", "fox"), RSTART, RLENGTH } # !gawk`, "", "0 0 -1\n", "", ""},
+	{`BEGIN { print match("x food y", /[fod]+/), RSTART, RLENGTH } # !gawk`, "", "3 3 4\n", "", ""},
+	{`BEGIN { print match("絵 fööd y", /[föd]+/), RSTART, RLENGTH } # !gawk`, "", "5 5 6\n", "", ""},
 	{`{ print length, length(), length("buzz"), length("") }  # !gawk`, "foo bar", "7 7 4 0\n", "", ""},
 	{`BEGIN { print length("a"), length("絵") }  # !gawk`, "", "1 3\n", "", ""},
 	{`BEGIN { print index("foo", "f"), index("foo0", 0), index("foo", "o"), index("foo", "x") }  # !gawk`, "", "1 4 2 0\n", "", ""},
@@ -840,6 +841,13 @@ BEGIN { x[1]=3; f5(x); print x[1] }
 	{`BEGIN { print >"out"; close("out"); getline <"out"; print >"out" }  # !awk !gawk`, "", "", "can't write to reader stream", ""},
 	{`BEGIN { print >"out"; close("out"); getline <"out"; print |"out" }  # !awk !gawk`, "", "", "can't write to reader stream", ""},
 
+	// The value of close() on a pipe emulates gawk behavior. Results are identical for both
+	// input and output. Windows does not do POSIX signals.
+	{`BEGIN { cmd="exit 9"; print "" |cmd; print close(cmd) } # !awk !posix`, "", "9\n", "", ""},
+	{`BEGIN { cmd="exec /bin/kill -9 $$"; print "" |cmd; print close(cmd) } # !awk !posix !windows`, "", "265\n", "", ""},
+	{`BEGIN { cmd="exit 9"; cmd |getline; print close(cmd) } # !awk !posix`, "", "9\n", "", ""},
+	{`BEGIN { cmd="exec /bin/kill -9 $$"; cmd |getline; print close(cmd) } # !awk !posix !windows`, "", "265\n", "", ""},
+
 	// Redirecting to or from a filename of "-" means write to stdout or read from stdin
 	{`BEGIN { print getline x < "-"; print x }`, "a\nb\n", "1\na\n", "", ""},
 	{`{ print $0; print getline x <"-"; print x }`, "one\ntwo\n", "one\n0\n\ntwo\n0\n\n", "", ""},
@@ -983,6 +991,28 @@ BEGIN { x[1]=3; f5(x); print x[1] }
 	{`{ print $1<$2 }`, "1_0 2", "1\n", "", ""},
 }
 
+// Flags parses the "skip flags" in it.src's first comment and returns them as
+// a map. An enable flag ("# +flag") appears as {"flag": true} in the map, and
+// a disable flag ("# !flag") appears as {"flag": false} in the map.
+func (it interpTest) Flags() map[string]bool {
+	flags := make(map[string]bool)
+	i := strings.Index(it.src, "#")
+	if i < 0 {
+		return flags
+	}
+	for _, field := range strings.Fields(it.src[i+1:]) {
+		switch {
+		case len(field) <= 1:
+			continue
+		case field[0] == '!':
+			flags[field[1:]] = false
+		case field[0] == '+':
+			flags[field[1:]] = true
+		}
+	}
+	return flags
+}
+
 func TestInterp(t *testing.T) {
 	// Ensure very long lines work (> 64KB)
 	longLine := strings.Repeat("x", 70000)
@@ -995,20 +1025,30 @@ func TestInterp(t *testing.T) {
 		if len(testName) > 70 {
 			testName = testName[:70]
 		}
+		flags := test.Flags()
+		var skipWindows bool
+		if f, ok := flags["windows"]; !f && ok && runtime.GOOS == "windows" {
+			skipWindows = true
+		}
 
 		// Run it through external awk program first
 		if awkExe != "" {
 			runAWK := func(t *testing.T, posix bool) {
-				if strings.Contains(test.src, "!"+awkExe) {
+				if skipWindows {
+					t.Skip("skipping windows")
+				}
+
+				var f, ok bool
+				if f, ok = flags[awkExe]; !f && ok {
 					t.Skipf("skipping under %s", awkExe)
 				}
-				if strings.Contains(test.src, "!"+runtime.GOOS+"-"+awkExe) {
+				if f, ok = flags[runtime.GOOS+"-"+awkExe]; !f && ok {
 					t.Skipf("skipping on %s under %s", runtime.GOOS, awkExe)
 				}
-				if posix && strings.Contains(test.src, "!posix") {
+				if f, ok = flags["posix"]; posix && !f && ok {
 					t.Skipf("skipping in --posix mode")
 				}
-				if !posix && strings.Contains(test.src, "+posix") {
+				if !posix && f {
 					t.Skip("skipping in non-posix mode")
 				}
 
@@ -1055,6 +1095,9 @@ func TestInterp(t *testing.T) {
 
 		// Then test it in GoAWK
 		t.Run(testName, func(t *testing.T) {
+			if skipWindows {
+				t.Skip("skipping windows")
+			}
 			testGoAWK(t, test.src, test.in, test.out, test.err, nil, nil)
 		})
 	}
diff --git a/interp/iostream_test.go b/interp/iostream_test.go
new file mode 100644
index 0000000..8955f36
--- /dev/null
+++ b/interp/iostream_test.go
@@ -0,0 +1,73 @@
+package interp
+
+import (
+	"errors"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"testing"
+)
+
+const (
+	outputStreamBufferSize = 1024
+)
+
+func TestStreamDoubleClose(t *testing.T) {
+	dir := t.TempDir()
+	t.Run("InFile", func(t *testing.T) {
+		f, err := os.Create(filepath.Join(dir, "infile"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		in := newInFileStream(f)
+		checkDoubleClose(t, in)
+	})
+	t.Run("OutFile", func(t *testing.T) {
+		f, err := os.Create(filepath.Join(dir, "outfile"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		out := newOutFileStream(f, outputStreamBufferSize)
+		checkDoubleClose(t, out)
+	})
+	t.Run("InCmd", func(t *testing.T) {
+		cmd := execDefaultShell("echo close me")
+		in, err := newInCmdStream(cmd)
+		if err != nil {
+			t.Fatal(err)
+		}
+		checkDoubleClose(t, in)
+	})
+	t.Run("OutCmd", func(t *testing.T) {
+		cmd := execDefaultShell("echo close me")
+		out, err := newOutCmdStream(cmd)
+		if err != nil {
+			t.Fatal(err)
+		}
+		checkDoubleClose(t, out)
+	})
+}
+
+func execDefaultShell(scriptlet string) *exec.Cmd {
+	cmdline := append(defaultShellCommand, scriptlet)
+	return exec.Command(cmdline[0], cmdline[1:]...)
+}
+
+type streamCloser interface {
+	ExitCode() int
+	Close() error
+}
+
+func checkDoubleClose(t *testing.T, sc streamCloser) {
+	t.Helper()
+	if err := sc.Close(); err != nil {
+		t.Fatal(err)
+	}
+	exitCode := sc.ExitCode()
+	if err := sc.Close(); !errors.Is(err, doubleCloseError) {
+		t.Error("expected stream.Close() to return error on double close")
+	}
+	if sc.ExitCode() != exitCode {
+		t.Error("expected stream.ExitCode() to stay the same")
+	}
+}