High throughput I/O

[progrium]

I've been pushing the limits of Linux I/O in v86 and ended up making a shared memory "pipe" that gets ~50MB/s transfer speed. Say you have a v86 machine with 1G memory. Add the kernel arguments: mem=1008M memmap=16M$1008M and you can get a MessageChannel API into the emulator:

emulator.add_listener("emulator-ready", function() {
	const channel = new SharedMemoryChannel(emulator);
	const port = channel.getUserPort(); // a MessageChannel port
});

SharedMemoryChannel implementation

class SharedMemoryChannel {
    constructor(emulator, baseAddr = 0x3F000000, size = 0x1000000) { // 16MB
        this.emulator = emulator;
        this.baseAddr = baseAddr;
        this.size = size;
        
        // Memory layout
        this.CONTROL_SIZE = 0x1000;
        this.BUFFER_SIZE = (size - this.CONTROL_SIZE) / 2;
        
        // Control structure offsets (matching Go side)
        this.JS_GO_HEAD = 0;    // uint32 - JS writes here
        this.JS_GO_TAIL = 4;    // uint32 - Go reads from here
        this.GO_JS_HEAD = 8;    // uint32 - Go writes here
        this.GO_JS_TAIL = 12;   // uint32 - JS reads from here
        
        // Buffer offsets
        this.JS_GO_BUFFER = this.CONTROL_SIZE;
        this.GO_JS_BUFFER = this.CONTROL_SIZE + this.BUFFER_SIZE;
        
        // MessageChannel setup - port2 is for user code
        this.channel = new MessageChannel();
        this.port1 = this.channel.port1;
        this.port2 = this.channel.port2;
        
        // Handle messages from user (via port2) -> forward to Go
        this.port1.onmessage = (e) => this.sendToGo(e.data);
        
        // console.log("Starting polling for Go->JS messages");
        this.pollForMessages();
    }
    
    // Read uint32 little-endian from memory
    readUint32(offset) {
        const bytes = this.emulator.read_memory(this.baseAddr + offset, 4);
        return bytes[0] | (bytes[1] << 8) | (bytes[2] << 16) | (bytes[3] << 24);
    }
    
    // Write uint32 little-endian to memory
    writeUint32(offset, value) {
        const bytes = new Uint8Array(4);
        bytes[0] = value & 0xFF;
        bytes[1] = (value >> 8) & 0xFF;
        bytes[2] = (value >> 16) & 0xFF;
        bytes[3] = (value >> 24) & 0xFF;
        this.emulator.write_memory(bytes, this.baseAddr + offset);
    }
    
    pollForMessages() {
        // Tight polling loop for maximum throughput
        const poll = () => {
            // Read ALL available messages in one poll cycle
            let messagesRead = 0;
            const maxPerCycle = 1000; // Prevent infinite loop
            
            while (messagesRead < maxPerCycle) {
                // Read head and tail pointers
                const head = this.readUint32(this.GO_JS_HEAD);
                const tail = this.readUint32(this.GO_JS_TAIL);
                
                // Use unsigned subtraction to handle uint32 overflow correctly
                // >>> 0 ensures uint32 semantics in JavaScript
                const dataAvailable = ((head - tail) >>> 0);
                
                // Check if there's a message
                if (dataAvailable === 0) {
                    break; // No more messages
                }
                
                // Calculate buffer positions using modulo
                const tailPos = tail % this.BUFFER_SIZE;
                
                // Read message length at tail position (using modulo)
                if (tailPos + 4 > this.BUFFER_SIZE) {
                    // Not enough space for length header, skip to next cycle
                    const newTail = (Math.floor(tail / this.BUFFER_SIZE) + 1) * this.BUFFER_SIZE;
                    this.writeUint32(this.GO_JS_TAIL, newTail);
                    continue;
                }
                
                const lengthBytes = this.emulator.read_memory(
                    this.baseAddr + this.GO_JS_BUFFER + tailPos,
                    4
                );
                const msgLength = lengthBytes[0] | (lengthBytes[1] << 8) | 
                                (lengthBytes[2] << 16) | (lengthBytes[3] << 24);
                
                if (msgLength > 0 && msgLength < this.BUFFER_SIZE) {
                    // Check if full message fits in buffer
                    if (tailPos + 4 + msgLength <= this.BUFFER_SIZE) {
                        // Read message body
                        const messageBody = this.emulator.read_memory(
                            this.baseAddr + this.GO_JS_BUFFER + tailPos + 4,
                            msgLength
                        );
                        
                        // Update tail pointer (using logical position, not modulo)
                        const newTail = tail + 4 + msgLength;
                        this.writeUint32(this.GO_JS_TAIL, newTail);
                        
                        // Post to port1 so user's port2 receives it
                        this.port1.postMessage(messageBody.slice().buffer);
                        // console.log("outer: recv:", msgLength)

                        messagesRead++;
                    } else {
                        // Message doesn't fit at current position, must be a gap
                        const newTail = (Math.floor(tail / this.BUFFER_SIZE) + 1) * this.BUFFER_SIZE;
                        console.log(`Message doesn't fit (tailPos=${tailPos}, length=${msgLength}), skipping to ${newTail}`);
                        this.writeUint32(this.GO_JS_TAIL, newTail);
                    }
                } else {
                    // Invalid length, skip to next cycle
                    const newTail = (Math.floor(tail / this.BUFFER_SIZE) + 1) * this.BUFFER_SIZE;
                    // console.log(`Invalid message length: ${msgLength} at tailPos=${tailPos}, skipping to ${newTail}`);
                    this.writeUint32(this.GO_JS_TAIL, newTail);
                    break;
                }
            }
            
            // Immediate retry for maximum throughput
            setTimeout(poll, 0);
        };
        
        poll(); // Start polling
    }
    
    sendToGo(data) {
        let bytes;
        if (typeof data === 'string') {
            bytes = new TextEncoder().encode(data);
        } else if (data instanceof ArrayBuffer) {
            bytes = new Uint8Array(data);
        } else if (data instanceof Uint8Array) {
            bytes = data;
        } else {
            bytes = new TextEncoder().encode(JSON.stringify(data));
        }
        
        const messageSize = 4 + bytes.length;
        
        // Busy-wait for immediate retries with long timeout
        for (let busyTries = 0; busyTries < 10000; busyTries++) {
            let head = this.readUint32(this.JS_GO_HEAD);
            const tail = this.readUint32(this.JS_GO_TAIL);
            
            // Calculate available space (BufferSize - used space)
            const safetyMargin = 16384;
            const dataUsed = ((head - tail) >>> 0);
            let available = this.BUFFER_SIZE - dataUsed;
            
            if (available > safetyMargin) {
                available -= safetyMargin;
            } else {
                available = 0;
            }
            
            // Check if we have space
            if (messageSize <= available) {
                // Calculate buffer position using modulo
                let headPos = head % this.BUFFER_SIZE;
                
                // Check if message fits at current position
                if (headPos + messageSize > this.BUFFER_SIZE) {
                    // Need to skip gap - advance head to next buffer cycle
                    const newHead = (Math.floor(head / this.BUFFER_SIZE) + 1) * this.BUFFER_SIZE;
                    const tailPos = tail % this.BUFFER_SIZE;
                    
                    // Check if tail is far enough from start of buffer (using buffer positions only)
                    // This avoids incorrect comparisons when uint32 overflows
                    if (tailPos >= messageSize) {
                        // Safe to skip gap and write at beginning
                        head = newHead;
                        headPos = 0;
                    } else {
                        // Not enough space at beginning, retry
                        continue;
                    }
                }
                
                // Write the message
                const message = new Uint8Array(messageSize);
                message[0] = bytes.length & 0xFF;
                message[1] = (bytes.length >> 8) & 0xFF;
                message[2] = (bytes.length >> 16) & 0xFF;
                message[3] = (bytes.length >> 24) & 0xFF;
                message.set(bytes, 4);
                
                this.emulator.write_memory(message, this.baseAddr + this.JS_GO_BUFFER + headPos);

                // Update head pointer (using logical position, not modulo)
                const newHead = head + messageSize;
                this.writeUint32(this.JS_GO_HEAD, newHead);

                // console.log("outer: write:", messageSize)
                return; // Success!
            }
            
        }
        
        // After 100k tries (~few seconds), give up
        console.warn("Buffer full after 10k attempts - message dropped");
    }
    
    getUserPort() {
        return this.port2;  // User code uses port2 for bidirectional communication
    }
}

And then in Linux you can use mmap and /dev/mem and wrap them with an API similar to JavaScript. Here is a Go implementation that gives you an io.ReadWriteCloser that works like any streaming file/socket/pipe. Any writes to it will end up hitting your onmessage handler on the JS port object. Reads will come from data you postMessage on the JS port.

Go wrapper to access the memory pipe

package shm

import (
	"encoding/binary"
	"fmt"
	"os"
	"runtime"
	"sync/atomic"
	"syscall"
	"time"
	"unsafe"
)

const (
	SharedMemAddr = 0x3F000000
	SharedMemSize = 0x1000000 // 16MB

	ControlSize = 0x1000
	BufferSize  = (SharedMemSize - ControlSize) / 2

	// Control structure offsets
	JsGoHead    = 0
	JsGoTail    = 4
	GoJsHead    = 8
	GoJsTail    = 12
	StatusFlags = 16

	// Buffer offsets
	JsGoBuffer = ControlSize
	GoJsBuffer = ControlSize + BufferSize
)

type SharedChannel struct {
	mem []byte

	// Pointers to control values
	jsGoHead *uint32
	jsGoTail *uint32
	goJsHead *uint32
	goJsTail *uint32

	// Buffer slices
	jsGoBuffer []byte
	goJsBuffer []byte

	// Partial read state (for stream-like semantics)
	readBuffer []byte // holds current message being read
	readPos    int    // current position within readBuffer
}

func NewSharedChannel() (*SharedChannel, error) {
	file, err := os.OpenFile("/dev/mem", os.O_RDWR|os.O_SYNC, 0)
	if err != nil {
		return nil, fmt.Errorf("failed to open /dev/mem: %w", err)
	}
	defer file.Close()

	mem, err := syscall.Mmap(
		int(file.Fd()),
		SharedMemAddr,
		SharedMemSize,
		syscall.PROT_READ|syscall.PROT_WRITE,
		syscall.MAP_SHARED,
	)
	if err != nil {
		return nil, fmt.Errorf("mmap failed: %w", err)
	}

	ch := &SharedChannel{
		mem: mem,

		jsGoHead: (*uint32)(unsafe.Pointer(&mem[JsGoHead])),
		jsGoTail: (*uint32)(unsafe.Pointer(&mem[JsGoTail])),
		goJsHead: (*uint32)(unsafe.Pointer(&mem[GoJsHead])),
		goJsTail: (*uint32)(unsafe.Pointer(&mem[GoJsTail])),

		jsGoBuffer: mem[JsGoBuffer : JsGoBuffer+BufferSize],
		goJsBuffer: mem[GoJsBuffer : GoJsBuffer+BufferSize],
	}

	// fmt.Printf("SharedChannel initialized at 0x%x\n", SharedMemAddr)
	return ch, nil
}

func (ch *SharedChannel) Close() error {
	return syscall.Munmap(ch.mem)
}

// Read implements io.Reader - reads from JS->Go buffer
// Uses head/tail pointers for synchronization
// Implements stream-like semantics: if a message is larger than the provided buffer,
// it will be read in multiple calls until fully consumed.
func (ch *SharedChannel) Read(p []byte) (n int, err error) {
	// First check if we have data from a previous partial read
	if ch.readBuffer != nil && ch.readPos < len(ch.readBuffer) {
		// Copy what fits from the remaining data
		n = copy(p, ch.readBuffer[ch.readPos:])
		ch.readPos += n

		// If we've consumed the entire message, clear the buffer
		if ch.readPos >= len(ch.readBuffer) {
			ch.readBuffer = nil
			ch.readPos = 0
		}

		return n, nil
	}

	// No partial data, read a new message from shared memory
	waitCount := 0
	for {
		// Read head and tail pointers (use atomic reads to ensure visibility)
		head := atomic.LoadUint32(ch.jsGoHead)
		tail := atomic.LoadUint32(ch.jsGoTail)

		// Use unsigned subtraction to handle uint32 overflow correctly
		dataAvailable := head - tail // Works even if head < tail due to overflow

		// Check if there's a message
		if dataAvailable > 0 {
			if waitCount > 0 {
				// fmt.Printf("[SharedChannel.Read] Message available after %d waits (head=%d, tail=%d)\n",
				// 	waitCount, head, tail)
			}
			waitCount = 0

			// Calculate buffer positions using modulo
			tailPos := tail % BufferSize

			// Read message length at tail position (using modulo)
			if tailPos+4 > BufferSize {
				// Not enough space for length header, skip to next cycle
				newTail := ((tail / BufferSize) + 1) * BufferSize
				atomic.StoreUint32(ch.jsGoTail, newTail)
				continue
			}

			lengthBytes := ch.jsGoBuffer[tailPos : tailPos+4]
			length := binary.LittleEndian.Uint32(lengthBytes)

			// Check if we have a valid message
			if length > 0 && length < BufferSize {
				// Check if full message fits in buffer
				if tailPos+4+length <= BufferSize {
					// Read the ENTIRE message into internal buffer
					ch.readBuffer = make([]byte, length)
					copy(ch.readBuffer, ch.jsGoBuffer[tailPos+4:tailPos+4+length])
					ch.readPos = 0

					// Update tail pointer (using logical position, not modulo)
					newTail := tail + 4 + length
					atomic.StoreUint32(ch.jsGoTail, newTail)

					// fmt.Println("inner: read message:", length, "bytes, head:", head, "tail:", tail)

					// Now copy what fits into caller's buffer
					n = copy(p, ch.readBuffer)
					ch.readPos = n

					// If we've consumed the entire message, clear the buffer
					if ch.readPos >= len(ch.readBuffer) {
						ch.readBuffer = nil
						ch.readPos = 0
					}

					return n, nil
				} else {
					// Message doesn't fit at current position, must be a gap
					newTail := ((tail / BufferSize) + 1) * BufferSize
					fmt.Printf("Message doesn't fit (tailPos=%d, length=%d), skipping to %d\n", tailPos, length, newTail)
					atomic.StoreUint32(ch.jsGoTail, newTail)
				}
			} else {
				// Invalid length, skip to next cycle
				newTail := ((tail / BufferSize) + 1) * BufferSize
				// fmt.Printf("Invalid message length: %d at tailPos=%d, skipping to %d\n", length, tailPos, newTail)
				atomic.StoreUint32(ch.jsGoTail, newTail)
			}
		}

		// No message found, yield CPU and retry immediately
		waitCount++
		if waitCount%1000 == 0 {
			// fmt.Printf("[SharedChannel.Read] Waiting for message... (head=%d, tail=%d, waited %d times)\n",
			// 	atomic.LoadUint32(ch.jsGoHead), atomic.LoadUint32(ch.jsGoTail), waitCount)
		}
		time.Sleep(1 * time.Millisecond)
	}
}

// Write implements io.Writer - writes to Go->JS buffer
// Uses head/tail pointers for synchronization
// Returns error if buffer is full after timeout
func (ch *SharedChannel) Write(p []byte) (n int, err error) {
	messageSize := 4 + uint32(len(p))

	// Wait for space in buffer with timeout (max 1 second)
	deadline := time.Now().Add(1 * time.Second)
	attempts := 0

	for time.Now().Before(deadline) {
		head := atomic.LoadUint32(ch.goJsHead)
		tail := atomic.LoadUint32(ch.goJsTail)

		// Calculate available space (reserve some margin to prevent full buffer ambiguity)
		const safetyMargin = 16384 // Reserve 16KB (1% of 16MB buffer)
		var available uint32
		if head >= tail {
			// Space from head to end, plus from start to tail
			available = (BufferSize - head) + tail
		} else {
			// Space between tail and head (wrapped case)
			available = tail - head
		}

		// Subtract safety margin
		if available > safetyMargin {
			available -= safetyMargin
		} else {
			available = 0
		}

		// Check if message fits
		if messageSize <= available {
			// Calculate buffer position using modulo
			headPos := head % BufferSize

			// Check if message fits at current position
			if headPos+messageSize > BufferSize {
				// Need to skip gap - advance head to next buffer cycle
				newHead := ((head / BufferSize) + 1) * BufferSize
				tailPos := tail % BufferSize

				// Check if tail is far enough from start of buffer (using buffer positions only)
				// This avoids incorrect comparisons when uint32 overflows
				if tailPos >= messageSize {
					// Safe to skip gap and write at beginning
					head = newHead
					headPos = 0
				} else {
					// Not enough space at beginning, wait
					attempts++
					if attempts < 100 {
						runtime.Gosched()
					} else {
						time.Sleep(1 * time.Millisecond)
						attempts = 0
					}
					continue
				}
			}

			// We have space, write the message
			message := make([]byte, messageSize)
			binary.LittleEndian.PutUint32(message[:4], uint32(len(p)))
			copy(message[4:], p)

			// Write message at head position (using modulo)
			copy(ch.goJsBuffer[headPos:], message)

			// Memory barrier to ensure visibility
			runtime.Gosched()

			// Update head pointer to signal new message (logical position, not modulo)
			newHead := head + messageSize
			atomic.StoreUint32(ch.goJsHead, newHead)

			// fmt.Println("inner: write:", len(p), "head:", head, "tail:", tail)

			return len(p), nil
		}

		// Buffer full, yield and retry
		attempts++
		if attempts < 100 {
			runtime.Gosched()
		} else {
			time.Sleep(1 * time.Millisecond)
			attempts = 0
		}
	}

	// Timeout - buffer still full
	return 0, fmt.Errorf("write timeout: buffer full after 1s")
}