Repository: tetratelabs/wazero (go)

Issue:
Non-blocking stdio isn't working correctly
**Describe the bug**

Non-blocking stdio isn't working correctly.

**To Reproduce**

Use [`gotip`](https://pkg.go.dev/golang.org/dl/gotip) that includes at least https://github.com/golang/go/commit/13529cc5f443ef4e242da3716daa0032aa8d34f2 (note that this will be part of the Go 1.21 release in ~August).

If I compile the following module:

```go
package main

import (
	"io"
	"os"
	"time"
)

func main() {
	go func() {
		time.Sleep(1 * time.Second)
		os.Stdout.WriteString("waiting for stdin...\n")
	}()

	b, err := io.ReadAll(os.Stdin)
	if err != nil {
		panic(err)
	}
	os.Stdout.Write(b)
	return
}
```

using:

```
$ GOOS=wasip1 GOARCH=wasm gotip build -o stdio.wasm stdio.go
```

and then run the script with wazero, it returns errno=28 (EINVAL) from `poll_oneoff` which causes Go to crash:

```
$ (sleep 3; echo foo) | wazero run stdio.wasm 
waiting for stdin...
errno= 28  len(pollsubs)= 1
fatal error: poll_oneoff failed

runtime stack:
runtime.throw({0x17c29, 0x12})
	/Users/chris/sdk/gotip/src/runtime/panic.go:1077 +0x3 fp=0xeb778 sp=0xeb750 pc=0x12250003
runtime.netpoll(0xffffffffffffffff)
	/Users/chris/sdk/gotip/src/runtime/netpoll_wasip1.go:218 +0x45 fp=0xeb7e8 sp=0xeb778 pc=0x11f30045
```

**Expected behavior**

[wasi-go](https://github.com/stealthrocket/wasi-go) does the right thing here:

```
$ (sleep 3; echo foo) | wasirun stdio.wasm                  
waiting for stdin...
foo
```

**Environment (please complete the relevant information):**

```
$ gotip version
go version devel go1.21-13529cc Tue Jun 27 01:09:18 2023 +0000 darwin/arm64
$ uname -a
Darwin MBP.local 22.5.0 Darwin Kernel Version 22.5.0: Thu Jun  8 22:22:20 PDT 2023; root:xnu-8796.121.3~7/RELEASE_ARM64_T6000 arm64
$ wazero version
v1.2.1
```


Pull request:
fs: nonblocking writes should use syscall.Write
resolves #1538 on non-Windows platforms. Windows will be addressed in #1579.

Using blocking `file.Write` APIs together with nonblocking I/O, causes stdout corruption under certain conditions. 

In particular, a reliable way to reproduce the issue is to capture the stdout of a sub-process running wazero. This PR fixes the writing issue by introducing `writeFd` (similar to `readFd` introduced in #1517), a function that just invokes the syscall directly when I/O is non-blocking. 

It also introduces the constants `nonBlockingFileReadSupported`, `nonBlockingFileWriteSupported`, superseding the generic `nonBlockingFileIoSupported`: the reason is still related to #1579, where Windows-related concerns will be addressed; for now we ignore the issue on that platform. However, because the test is _not skipped_ on Windows, it looks like that platform is not affected (also verified on my Windows machine).


Patch:
diff --git a/internal/sysfs/file_unix.go b/internal/sysfs/file_unix.go
index 4516cc5f..4ed51a9f 100644
--- a/internal/sysfs/file_unix.go
+++ b/internal/sysfs/file_unix.go
@@ -8,7 +8,10 @@ import (
 	"github.com/tetratelabs/wazero/experimental/sys"
 )
 
-const nonBlockingFileIoSupported = true
+const (
+	nonBlockingFileReadSupported  = true
+	nonBlockingFileWriteSupported = true
+)
 
 // readFd exposes syscall.Read.
 func readFd(fd uintptr, buf []byte) (int, sys.Errno) {
@@ -19,3 +22,13 @@ func readFd(fd uintptr, buf []byte) (int, sys.Errno) {
 	errno := sys.UnwrapOSError(err)
 	return n, errno
 }
+
+// writeFd exposes syscall.Write.
+func writeFd(fd uintptr, buf []byte) (int, sys.Errno) {
+	if len(buf) == 0 {
+		return 0, 0 // Short-circuit 0-len writes.
+	}
+	n, err := syscall.Write(int(fd), buf)
+	errno := sys.UnwrapOSError(err)
+	return n, errno
+}
diff --git a/internal/sysfs/file_unsupported.go b/internal/sysfs/file_unsupported.go
index c10d39aa..eb8b5537 100644
--- a/internal/sysfs/file_unsupported.go
+++ b/internal/sysfs/file_unsupported.go
@@ -2,11 +2,19 @@
 
 package sysfs
 
-import "syscall"
+import "github.com/tetratelabs/wazero/experimental/sys"
 
-const nonBlockingFileIoSupported = false
+const (
+	nonBlockingFileReadSupported  = false
+	nonBlockingFileWriteSupported = false
+)
 
 // readFd returns ENOSYS on unsupported platforms.
-func readFd(fd uintptr, buf []byte) (int, syscall.Errno) {
-	return -1, syscall.ENOSYS
+func readFd(fd uintptr, buf []byte) (int, sys.Errno) {
+	return -1, sys.ENOSYS
+}
+
+// writeFd returns ENOSYS on unsupported platforms.
+func writeFd(fd uintptr, buf []byte) (int, sys.Errno) {
+	return -1, sys.ENOSYS
 }
diff --git a/internal/sysfs/file_windows.go b/internal/sysfs/file_windows.go
index 29b8060b..e89db54d 100644
--- a/internal/sysfs/file_windows.go
+++ b/internal/sysfs/file_windows.go
@@ -7,7 +7,10 @@ import (
 	"github.com/tetratelabs/wazero/experimental/sys"
 )
 
-const nonBlockingFileIoSupported = true
+const (
+	nonBlockingFileReadSupported  = true
+	nonBlockingFileWriteSupported = false
+)
 
 var kernel32 = syscall.NewLazyDLL("kernel32.dll")
 
@@ -56,3 +59,7 @@ func peekNamedPipe(handle syscall.Handle) (uint32, error) {
 	}
 	return totalBytesAvail, err
 }
+
+func writeFd(fd uintptr, buf []byte) (int, sys.Errno) {
+	return -1, sys.ENOSYS
+}
diff --git a/internal/sysfs/osfile.go b/internal/sysfs/osfile.go
index de167de0..b64f030a 100644
--- a/internal/sysfs/osfile.go
+++ b/internal/sysfs/osfile.go
@@ -145,7 +145,7 @@ func (f *osFile) Read(buf []byte) (n int, errno experimentalsys.Errno) {
 	if len(buf) == 0 {
 		return 0, 0 // Short-circuit 0-len reads.
 	}
-	if nonBlockingFileIoSupported && f.IsNonblock() {
+	if nonBlockingFileReadSupported && f.IsNonblock() {
 		n, errno = readFd(f.fd, buf)
 	} else {
 		n, errno = read(f.file, buf)
@@ -214,7 +214,12 @@ func (f *osFile) Readdir(n int) (dirents []fsapi.Dirent, errno experimentalsys.E
 
 // Write implements the same method as documented on fsapi.File
 func (f *osFile) Write(buf []byte) (n int, errno experimentalsys.Errno) {
-	if n, errno = write(f.file, buf); errno != 0 {
+	if len(buf) == 0 {
+		return 0, 0 // Short-circuit 0-len writes.
+	}
+	if nonBlockingFileWriteSupported && f.IsNonblock() {
+		n, errno = writeFd(f.fd, buf)
+	} else if n, errno = write(f.file, buf); errno != 0 {
 		// Defer validation overhead until we've already had an error.
 		errno = fileError(f, f.closed, errno)
 	}


Test patch:
diff --git a/imports/wasi_snapshot_preview1/testdata/gotip/wasi.go b/imports/wasi_snapshot_preview1/testdata/gotip/wasi.go
index d71e15a0..78fd70ce 100644
--- a/imports/wasi_snapshot_preview1/testdata/gotip/wasi.go
+++ b/imports/wasi_snapshot_preview1/testdata/gotip/wasi.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io"
@@ -8,6 +9,8 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -53,6 +56,8 @@ func main() {
 		}
 	case "stdout":
 		mainStdout()
+	case "largestdout":
+		mainLargeStdout()
 	}
 }
 
@@ -259,3 +264,46 @@ func mainStdin() error {
 func mainStdout() {
 	os.Stdout.WriteString("test")
 }
+
+func mainLargeStdout() {
+	const ntest = 1024
+
+	var decls, calls bytes.Buffer
+
+	for i := 1; i <= ntest; i++ {
+		s := strconv.Itoa(i)
+		decls.WriteString(strings.Replace(decl, "$", s, -1))
+		calls.WriteString(strings.Replace("call(test$)\n\t", "$", s, -1))
+	}
+
+	program = strings.Replace(program, "$DECLS", decls.String(), 1)
+	program = strings.Replace(program, "$CALLS", calls.String(), 1)
+	fmt.Print(program)
+}
+
+var program = `package main
+
+var count int
+
+func call(f func() bool) {
+	if f() {
+		count++
+	}
+}
+
+$DECLS
+
+func main() {
+	$CALLS
+	if count != 0 {
+		println("failed", count, "case(s)")
+	}
+}
+`
+
+const decl = `
+type T$ [$]uint8
+func test$() bool {
+	v := T${1}
+	return v == [$]uint8{2} || v != [$]uint8{1}
+}`
diff --git a/imports/wasi_snapshot_preview1/wasi_stdlib_test.go b/imports/wasi_snapshot_preview1/wasi_stdlib_test.go
index 5537fb09..78e84300 100644
--- a/imports/wasi_snapshot_preview1/wasi_stdlib_test.go
+++ b/imports/wasi_snapshot_preview1/wasi_stdlib_test.go
@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"os/exec"
 	"path"
 	"runtime"
 	"sort"
@@ -269,6 +270,59 @@ func compileAndRunWithPreStart(t *testing.T, ctx context.Context, config wazero.
 	return
 }
 
+// compileAndRunForked executes the test case with the wazero runtime in a separate process, and waits for it to terminate.
+//
+// Stdout is captured to a buffer, and stderr is dumped to os.Stderr.
+// It returns the capture buffer and boolean; the boolean is true if we are running in the outer process.
+// If it is false, it means we are running in the subprocess; all the verification logic should be handled
+// when the boolean is true.
+//
+// A typical usage pattern will be:
+//
+//	if buf, hasRun := compileAndRunForked(...); hasRun {
+//		validateContents(buf)
+//	}
+func compileAndRunForked(t *testing.T, ctx context.Context, config wazero.ModuleConfig, tname string, bin []byte) ([]byte, bool) {
+	var buf bytes.Buffer
+	// We use the technique described in https://go.dev/talks/2014/testing.slide#23
+	// We check if we are running forked by ensuring that a "magic" environment variable is set.
+	if os.Getenv("_TEST_FORKED") != "1" {
+		// If said variable is not set, then we need to exec this same executable, specifying the name of test.
+		// We could use t.Name(), but because t may be arbitrarily nested, it is better to get the name of the test
+		// from a parameter.
+		cmd := exec.Command(os.Args[0], "-test.run", tname)
+		cmd.Stdout = &buf
+		cmd.Stderr = os.Stderr
+		cmd.Env = append(os.Environ(), "_TEST_FORKED=1")
+		err := cmd.Run()
+		if e, ok := err.(*exec.ExitError); ok && !e.Success() {
+			require.NoError(t, e, "The test quit with an error code: %v\n", e)
+		}
+		res := buf.Bytes()
+		// This is a test, so in case of success, it will include the "PASS\n" string:
+		// we remove that from the output, to return a clean stdout.
+		return res[0 : len(res)-len("PASS\n")], true
+	}
+
+	r := wazero.NewRuntime(ctx)
+	defer r.Close(ctx)
+
+	_, err := wasi_snapshot_preview1.Instantiate(ctx, r)
+	require.NoError(t, err)
+
+	mod, err := r.InstantiateWithConfig(ctx, bin, config.
+		WithStartFunctions()) // clear
+	require.NoError(t, err)
+
+	_, err = mod.ExportedFunction("_start").Call(ctx)
+	if exitErr, ok := err.(*sys.ExitError); ok {
+		require.Zero(t, exitErr.ExitCode())
+	} else {
+		require.NoError(t, err)
+	}
+	return nil, false
+}
+
 func Test_Poll(t *testing.T) {
 	// The following test cases replace Stdin with a custom reader.
 	// For more precise coverage, see poll_test.go.
@@ -537,3 +591,49 @@ func testStdin(t *testing.T, bin []byte) {
 	require.Equal(t, "foo", string(buf))
 	<-ch
 }
+
+func Test_LargeStdout(t *testing.T) {
+	toolchains := map[string][]byte{}
+	if wasmGotip != nil {
+		toolchains["gotip"] = wasmGotip
+	}
+
+	for toolchain, bin := range toolchains {
+		toolchain := toolchain
+		bin := bin
+		name := t.Name()
+		t.Run(toolchain, func(t *testing.T) {
+			testLargeStdout(t, name, bin)
+		})
+	}
+}
+
+func testLargeStdout(t *testing.T, tname string, bin []byte) {
+	// This test dumps a large Go source file to stdout. The generated result
+	// should be valid code, otherwise it means that stdout is corrupted.
+	//
+	// The error conditions are more easily reproduced by executing in a subprocess
+	// and capturing its stdout.
+	if buf, hasRun := compileAndRunForked(t, testCtx, wazero.NewModuleConfig().
+		WithArgs("wasi", "largestdout").
+		WithStdout(os.Stdout), tname, bin); hasRun {
+
+		tempDir := t.TempDir()
+		temp, err := os.Create(joinPath(tempDir, "out.go"))
+		require.NoError(t, err)
+		defer temp.Close()
+
+		require.NoError(t, err)
+		_, _ = temp.Write(buf)
+		_ = temp.Close()
+
+		gotipBin, err := findGotipBin()
+		require.NoError(t, err)
+
+		cmd := exec.CommandContext(testCtx, gotipBin, "build", "-o",
+			joinPath(tempDir, "outbin"), temp.Name())
+		require.NoError(t, err)
+		output, err := cmd.CombinedOutput()
+		require.NoError(t, err, string(output))
+	}
+}
diff --git a/internal/sysfs/file_test.go b/internal/sysfs/file_test.go
index 9d84cf1f..35de1672 100644
--- a/internal/sysfs/file_test.go
+++ b/internal/sysfs/file_test.go
@@ -92,6 +92,37 @@ func TestReadFdNonblock(t *testing.T) {
 	require.EqualErrno(t, experimentalsys.EAGAIN, e)
 }
 
+func TestWriteFdNonblock(t *testing.T) {
+	// Test using os.Pipe as it is known to support non-blocking reads.
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+	defer w.Close()
+
+	fd := w.Fd()
+	err = setNonblock(fd, true)
+
+	require.NoError(t, err)
+
+	// Create a buffer (the content is not relevant)
+	buf := make([]byte, 1024)
+	// Write to the file until the pipe buffer gets filled up.
+	numWrites := 100
+	for i := 0; i < numWrites; i++ {
+		_, e := writeFd(fd, buf)
+		if e != 0 {
+			if runtime.GOOS == "windows" {
+				// This is currently not supported on Windows.
+				require.EqualErrno(t, experimentalsys.ENOSYS, e)
+			} else {
+				require.EqualErrno(t, experimentalsys.EAGAIN, e)
+			}
+			return
+		}
+	}
+	t.Fatal("writeFd should return EAGAIN at some point")
+}
+
 func TestFileSetAppend(t *testing.T) {
 	tmpDir := t.TempDir()