poll.go

Documentation: github.com/tetratelabs/wazero/imports/wasi_snapshot_preview1

     1  package wasi_snapshot_preview1
     2  
     3  import (
     4  	"context"
     5  	"time"
     6  
     7  	"github.com/tetratelabs/wazero/api"
     8  	"github.com/tetratelabs/wazero/experimental/sys"
     9  	"github.com/tetratelabs/wazero/internal/fsapi"
    10  	internalsys "github.com/tetratelabs/wazero/internal/sys"
    11  	"github.com/tetratelabs/wazero/internal/wasip1"
    12  	"github.com/tetratelabs/wazero/internal/wasm"
    13  )
    14  
    15  // pollOneoff is the WASI function named PollOneoffName that concurrently
    16  // polls for the occurrence of a set of events.
    17  //
    18  // # Parameters
    19  //
    20  //   - in: pointer to the subscriptions (48 bytes each)
    21  //   - out: pointer to the resulting events (32 bytes each)
    22  //   - nsubscriptions: count of subscriptions, zero returns sys.EINVAL.
    23  //   - resultNevents: count of events.
    24  //
    25  // Result (Errno)
    26  //
    27  // The return value is 0 except the following error conditions:
    28  //   - sys.EINVAL: the parameters are invalid
    29  //   - sys.ENOTSUP: a parameters is valid, but not yet supported.
    30  //   - sys.EFAULT: there is not enough memory to read the subscriptions or
    31  //     write results.
    32  //
    33  // # Notes
    34  //
    35  //   - Since the `out` pointer nests Errno, the result is always 0.
    36  //   - This is similar to `poll` in POSIX.
    37  //
    38  // See https://github.com/WebAssembly/WASI/blob/snapshot-01/phases/snapshot/docs.md#poll_oneoff
    39  // See https://linux.die.net/man/3/poll
    40  var pollOneoff = newHostFunc(
    41  	wasip1.PollOneoffName, pollOneoffFn,
    42  	[]api.ValueType{i32, i32, i32, i32},
    43  	"in", "out", "nsubscriptions", "result.nevents",
    44  )
    45  
    46  type event struct {
    47  	eventType byte
    48  	userData  []byte
    49  	errno     wasip1.Errno
    50  }
    51  
    52  func pollOneoffFn(_ context.Context, mod api.Module, params []uint64) sys.Errno {
    53  	in := uint32(params[0])
    54  	out := uint32(params[1])
    55  	nsubscriptions := uint32(params[2])
    56  	resultNevents := uint32(params[3])
    57  
    58  	if nsubscriptions == 0 {
    59  		return sys.EINVAL
    60  	}
    61  
    62  	mem := mod.Memory()
    63  
    64  	// Ensure capacity prior to the read loop to reduce error handling.
    65  	inBuf, ok := mem.Read(in, nsubscriptions*48)
    66  	if !ok {
    67  		return sys.EFAULT
    68  	}
    69  	outBuf, ok := mem.Read(out, nsubscriptions*32)
    70  	// zero-out all buffer before writing
    71  	for i := range outBuf {
    72  		outBuf[i] = 0
    73  	}
    74  
    75  	if !ok {
    76  		return sys.EFAULT
    77  	}
    78  
    79  	// Eagerly write the number of events which will equal subscriptions unless
    80  	// there's a fault in parsing (not processing).
    81  	if !mod.Memory().WriteUint32Le(resultNevents, nsubscriptions) {
    82  		return sys.EFAULT
    83  	}
    84  
    85  	// Loop through all subscriptions and write their output.
    86  
    87  	// Extract FS context, used in the body of the for loop for FS access.
    88  	fsc := mod.(*wasm.ModuleInstance).Sys.FS()
    89  	// Slice of events that are processed out of the loop (blocking stdin subscribers).
    90  	var blockingStdinSubs []*event
    91  	// The timeout is initialized at max Duration, the loop will find the minimum.
    92  	var timeout time.Duration = 1<<63 - 1
    93  	// Count of all the subscriptions that have been already written back to outBuf.
    94  	// nevents*32 returns at all times the offset where the next event should be written:
    95  	// this way we ensure that there are no gaps between records.
    96  	nevents := uint32(0)
    97  
    98  	// Layout is subscription_u: Union
    99  	// https://github.com/WebAssembly/WASI/blob/snapshot-01/phases/snapshot/docs.md#subscription_u
   100  	for i := uint32(0); i < nsubscriptions; i++ {
   101  		inOffset := i * 48
   102  		outOffset := nevents * 32
   103  
   104  		eventType := inBuf[inOffset+8] // +8 past userdata
   105  		// +8 past userdata +8 contents_offset
   106  		argBuf := inBuf[inOffset+8+8:]
   107  		userData := inBuf[inOffset : inOffset+8]
   108  
   109  		evt := &event{
   110  			eventType: eventType,
   111  			userData:  userData,
   112  			errno:     wasip1.ErrnoSuccess,
   113  		}
   114  
   115  		switch eventType {
   116  		case wasip1.EventTypeClock: // handle later
   117  			newTimeout, err := processClockEvent(argBuf)
   118  			if err != 0 {
   119  				return err
   120  			}
   121  			// Min timeout.
   122  			if newTimeout < timeout {
   123  				timeout = newTimeout
   124  			}
   125  			// Ack the clock event to the outBuf.
   126  			writeEvent(outBuf[outOffset:], evt)
   127  			nevents++
   128  		case wasip1.EventTypeFdRead:
   129  			fd := int32(le.Uint32(argBuf))
   130  			if fd < 0 {
   131  				return sys.EBADF
   132  			}
   133  			if file, ok := fsc.LookupFile(fd); !ok {
   134  				evt.errno = wasip1.ErrnoBadf
   135  				writeEvent(outBuf[outOffset:], evt)
   136  				nevents++
   137  			} else if fd != internalsys.FdStdin && file.File.IsNonblock() {
   138  				writeEvent(outBuf[outOffset:], evt)
   139  				nevents++
   140  			} else {
   141  				// if the fd is Stdin, and it is in blocking mode,
   142  				// do not ack yet, append to a slice for delayed evaluation.
   143  				blockingStdinSubs = append(blockingStdinSubs, evt)
   144  			}
   145  		case wasip1.EventTypeFdWrite:
   146  			fd := int32(le.Uint32(argBuf))
   147  			if fd < 0 {
   148  				return sys.EBADF
   149  			}
   150  			if _, ok := fsc.LookupFile(fd); ok {
   151  				evt.errno = wasip1.ErrnoNotsup
   152  			} else {
   153  				evt.errno = wasip1.ErrnoBadf
   154  			}
   155  			nevents++
   156  			writeEvent(outBuf[outOffset:], evt)
   157  		default:
   158  			return sys.EINVAL
   159  		}
   160  	}
   161  
   162  	sysCtx := mod.(*wasm.ModuleInstance).Sys
   163  	if nevents == nsubscriptions {
   164  		// We already wrote back all the results. We already wrote this number
   165  		// earlier to offset `resultNevents`.
   166  		// We only need to observe the timeout (nonzero if there are clock subscriptions)
   167  		// and return.
   168  		if timeout > 0 {
   169  			sysCtx.Nanosleep(int64(timeout))
   170  		}
   171  		return 0
   172  	}
   173  
   174  	// If there are blocking stdin subscribers, check for data with given timeout.
   175  	stdin, ok := fsc.LookupFile(internalsys.FdStdin)
   176  	if !ok {
   177  		return sys.EBADF
   178  	}
   179  	// Wait for the timeout to expire, or for some data to become available on Stdin.
   180  
   181  	if stdinReady, errno := stdin.File.Poll(fsapi.POLLIN, int32(timeout.Milliseconds())); errno != 0 {
   182  		return errno
   183  	} else if stdinReady {
   184  		// stdin has data ready to for reading, write back all the events
   185  		for i := range blockingStdinSubs {
   186  			evt := blockingStdinSubs[i]
   187  			evt.errno = 0
   188  			writeEvent(outBuf[nevents*32:], evt)
   189  			nevents++
   190  		}
   191  	}
   192  
   193  	if nevents != nsubscriptions {
   194  		if !mod.Memory().WriteUint32Le(resultNevents, nevents) {
   195  			return sys.EFAULT
   196  		}
   197  	}
   198  
   199  	return 0
   200  }
   201  
   202  // processClockEvent supports only relative name events, as that's what's used
   203  // to implement sleep in various compilers including Rust, Zig and TinyGo.
   204  func processClockEvent(inBuf []byte) (time.Duration, sys.Errno) {
   205  	_ /* ID */ = le.Uint32(inBuf[0:8])          // See below
   206  	timeout := le.Uint64(inBuf[8:16])           // nanos if relative
   207  	_ /* precision */ = le.Uint64(inBuf[16:24]) // Unused
   208  	flags := le.Uint16(inBuf[24:32])
   209  
   210  	var err sys.Errno
   211  	// subclockflags has only one flag defined:  subscription_clock_abstime
   212  	switch flags {
   213  	case 0: // relative time
   214  	case 1: // subscription_clock_abstime
   215  		err = sys.ENOTSUP
   216  	default: // subclockflags has only one flag defined.
   217  		err = sys.EINVAL
   218  	}
   219  
   220  	if err != 0 {
   221  		return 0, err
   222  	} else {
   223  		// https://linux.die.net/man/3/clock_settime says relative timers are
   224  		// unaffected. Since this function only supports relative timeout, we can
   225  		// skip name ID validation and use a single sleep function.
   226  
   227  		return time.Duration(timeout), 0
   228  	}
   229  }
   230  
   231  // writeEvent writes the event corresponding to the processed subscription.
   232  // https://github.com/WebAssembly/WASI/blob/snapshot-01/phases/snapshot/docs.md#-event-struct
   233  func writeEvent(outBuf []byte, evt *event) {
   234  	copy(outBuf, evt.userData)  // userdata
   235  	outBuf[8] = byte(evt.errno) // uint16, but safe as < 255
   236  	outBuf[9] = 0
   237  	le.PutUint32(outBuf[10:], uint32(evt.eventType))
   238  	// TODO: When FD events are supported, write outOffset+16
   239  }
   240
View as plain text