coro: demand-gate native syscall compensation

benchmark/coro_core/README.md

@@ -34,6 +34,13 @@ pollute the core artifact. Its modes are:
 
 - `file`: one persistent temporary file, with cache-hot 4 KiB
   seek/write/seek/read round trips;
+- `file-syscall`: the same physical file transaction using direct
+  `syscall.Seek`, `syscall.Write`, and `syscall.Read` calls inside the measured
+  loop, isolating worker-boundary cost from the `os.File`/`internal/poll`
+  wrapper chain;
+- `pipe-block`: a scheduler-progress gate in which a direct pipe read blocks
+  the sole current M while a sleeping goroutine must be run by a compensation
+  M, wake on its timer, and write the byte which releases the reader;
 - `tcp`: one persistent loopback TCP connection, with 4 KiB request/echo round
   trips between two goroutines.
 

benchmark/coro_core/testdata/io_workload/main.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"syscall"
 	"time"
 )
 
@@ -84,6 +85,87 @@ func fileRoundTrip(count, rounds int) int {
 	return checksum
 }
 
+// fileSyscallRoundTrip performs the same persistent-file transaction as
+// fileRoundTrip but deliberately bypasses os.File and internal/poll inside the
+// measured loop. Keeping both modes in one source fixture separates the
+// compiler/runtime worker boundary from coroutine frames introduced by the
+// standard-library wrapper chain without changing the physical syscalls.
+func fileSyscallRoundTrip(count, rounds int) int {
+	file, err := os.CreateTemp("", "llgo-coro-benchmark-syscall-*")
+	if err != nil {
+		panic(err)
+	}
+	path := file.Name()
+	defer os.Remove(path)
+	defer file.Close()
+
+	fd := int(file.Fd())
+	payload := make([]byte, payloadSize)
+	readback := make([]byte, payloadSize)
+	fillPayload(payload)
+	checksum := 0
+	for round := range rounds {
+		for index := range count {
+			if _, err := syscall.Seek(fd, 0, 0); err != nil {
+				panic(err)
+			}
+			if n, err := syscall.Write(fd, payload); err != nil || n != len(payload) {
+				panic("short syscall file write")
+			}
+			if _, err := syscall.Seek(fd, 0, 0); err != nil {
+				panic(err)
+			}
+			if n, err := syscall.Read(fd, readback); err != nil || n != len(readback) {
+				panic("short syscall file read")
+			}
+			checksum += int(readback[(round*count+index)%len(readback)])
+		}
+	}
+	return checksum
+}
+
+// blockingPipeRoundTrip is a scheduler-progress gate, not a throughput mode.
+// The reader enters a genuinely blocking syscall on the current M. With
+// GOMAXPROCS=1, the buffered armed handoff first proves that the writer has
+// parked on its timer. Only a compensation M can then service that timer and
+// issue the write which releases the reader.
+func blockingPipeRoundTrip(count, rounds int) int {
+	fds := make([]int, 2)
+	if err := syscall.Pipe(fds); err != nil {
+		panic(err)
+	}
+	defer syscall.Close(fds[0])
+	defer syscall.Close(fds[1])
+
+	readback := []byte{0}
+	checksum := 0
+	for round := range rounds {
+		for index := range count {
+			value := byte(round*count + index + 1)
+			armed := make(chan struct{}, 1)
+			done := make(chan error, 1)
+			go func() {
+				armed <- struct{}{}
+				time.Sleep(time.Millisecond)
+				written, err := syscall.Write(fds[1], []byte{value})
+				if err == nil && written != 1 {
+					err = syscall.EIO
+				}
+				done <- err
+			}()
+			<-armed
+			if read, err := syscall.Read(fds[0], readback); err != nil || read != 1 {
+				panic("blocking pipe read failed")
+			}
+			if err := <-done; err != nil {
+				panic(err)
+			}
+			checksum += int(readback[0])
+		}
+	}
+	return checksum
+}
+
 func writeFull(conn *net.TCPConn, payload []byte) {
 	for len(payload) != 0 {
 		n, err := conn.Write(payload)
@@ -146,7 +228,7 @@ func tcpRoundTrip(count, rounds int) int {
 
 func main() {
 	if len(os.Args) != 4 {
-		panic("usage: io_workload <file|tcp> <count> <rounds>")
+		panic("usage: io_workload <file|file-syscall|pipe-block|tcp> <count> <rounds>")
 	}
 	mode := os.Args[1]
 	count, ok := parsePositive(os.Args[2])
@@ -163,6 +245,10 @@ func main() {
 	switch mode {
 	case "file":
 		result = fileRoundTrip(count, rounds)
+	case "file-syscall":
+		result = fileSyscallRoundTrip(count, rounds)
+	case "pipe-block":
+		result = blockingPipeRoundTrip(count, rounds)
 	case "tcp":
 		result = tcpRoundTrip(count, rounds)
 	default:

cl/coro_worker.go

@@ -34,6 +34,7 @@ const (
 	coroHostOperationDeadlineResumeHookV1 = "__llgo_coro_host_operation_deadline_resume_v1"
 	coroOSThreadLockedHookV1              = "__llgo_coro_os_thread_locked_v1"
 	coroOSThreadForeignCallHookV1         = "__llgo_coro_os_thread_foreign_call_v1"
+	coroNativeSyscallCallHookV1           = "__llgo_coro_native_syscall_call_v1"
 )
 
 const (
@@ -225,6 +226,7 @@ func (p *context) compileCoroHostOperation(
 		physicalWords,
 		keepaliveSlots,
 		&coroHostWordOperationV1{metadata: metadata},
+		false,
 	)
 	return b.Aggregate(
 		p.type_(results, llssa.InGo),
@@ -245,9 +247,11 @@ func (p *context) compileCoroWorkerWordCall(
 	args []llssa.Expr,
 	keepaliveSlots []llssa.Expr,
 	host *coroHostWordOperationV1,
+	nativeSyscall bool,
 ) coroWorkerWordResultV1 {
 	body := p.requireCoroWorkerBody(b)
 	if function.IsNil() || host == nil && traceTarget.IsNil() || host != nil && !traceTarget.IsNil() ||
+		host != nil && nativeSyscall ||
 		len(args) > coroWorkerMaxArgsV1 ||
 		host != nil && len(host.metadata) != 0 &&
 			len(host.metadata) != coroHostOperationDeadlineMetadataWordsV1 {
@@ -301,34 +305,34 @@ func (p *context) compileCoroWorkerWordCall(
 			resumeHook = coroHostOperationDeadlineResumeHookV1
 		}
 	}
-	state := b.Alloc(p.prog.RuntimeType(stateType), false)
 	r1 := b.Alloc(p.prog.Uintptr(), false)
 	r2 := b.Alloc(p.prog.Uintptr(), false)
 	errno := b.Alloc(p.prog.Uintptr(), false)
 	zero := p.prog.Zero(p.prog.Uintptr())
-	physicalArgs := make([]llssa.Expr, 0, 7+len(metadata)+coroWorkerMaxArgsV1)
-	physicalArgs = append(physicalArgs,
-		body.task,
-		body.coro.Handle(),
-		b.Convert(b.Prog.VoidPtr(), body.header),
-		b.Convert(b.Prog.VoidPtr(), state),
-		function,
-	)
-	if host == nil {
-		physicalArgs = append(physicalArgs, traceTarget)
-	}
 	argcValue := p.prog.IntVal(uint64(len(args)), p.prog.Uint32())
-	physicalArgs = append(physicalArgs, argcValue)
-	physicalArgs = append(physicalArgs, metadata...)
-	for index := 0; index < coroWorkerMaxArgsV1; index++ {
-		if index < len(args) {
-			physicalArgs = append(physicalArgs, args[index])
-		} else {
-			physicalArgs = append(physicalArgs, zero)
-		}
-	}
 
 	emitPark := func(worker llssa.Builder) {
+		state := worker.Alloc(p.prog.RuntimeType(stateType), false)
+		physicalArgs := make([]llssa.Expr, 0, 7+len(metadata)+coroWorkerMaxArgsV1)
+		physicalArgs = append(physicalArgs,
+			body.task,
+			body.coro.Handle(),
+			worker.Convert(worker.Prog.VoidPtr(), body.header),
+			worker.Convert(worker.Prog.VoidPtr(), state),
+			function,
+		)
+		if host == nil {
+			physicalArgs = append(physicalArgs, traceTarget)
+		}
+		physicalArgs = append(physicalArgs, argcValue)
+		physicalArgs = append(physicalArgs, metadata...)
+		for index := 0; index < coroWorkerMaxArgsV1; index++ {
+			if index < len(args) {
+				physicalArgs = append(physicalArgs, args[index])
+			} else {
+				physicalArgs = append(physicalArgs, zero)
+			}
+		}
 		body.emitCoroParkOperation(p, worker, coroParkOperation{
 			prepare: func(active llssa.Builder, _, _ uint32) llssa.Expr {
 				return active.Prog.BoolVal(true)
@@ -360,17 +364,7 @@ func (p *context) compileCoroWorkerWordCall(
 		return coroWorkerWordResultV1{r1: b.Load(r1), r2: b.Load(r2), errno: b.Load(errno)}
 	}
 
-	lockedHook := p.pkg.NewFunc(coroOSThreadLockedHookV1, coroOSThreadLockedSignature(), llssa.InC)
-	locked := b.Call(lockedHook.Expr, body.task)
-	directBlock := b.Func.MakeBlock()
-	workerBlock := b.Func.MakeBlock()
 	join := b.Func.MakeBlock()
-	b.If(locked, directBlock, workerBlock)
-
-	b.SetBlockEx(directBlock, llssa.AtEnd, false)
-	direct := p.pkg.NewFunc(
-		coroOSThreadForeignCallHookV1, coroOSThreadForeignCallSignature(), llssa.InC,
-	)
 	directArgs := make([]llssa.Expr, 0, 4+coroWorkerMaxArgsV1+3)
 	directArgs = append(directArgs, body.task, function, traceTarget, argcValue)
 	for index := 0; index < coroWorkerMaxArgsV1; index++ {
@@ -381,33 +375,47 @@ func (p *context) compileCoroWorkerWordCall(
 		}
 	}
 	directArgs = append(directArgs, r1, r2, errno)
-	directStatus := b.Call(direct.Expr, directArgs...)
-	directNormal := b.Func.MakeBlock()
-	directMemoryFault := b.Func.MakeBlock()
-	directDivideFault := b.Func.MakeBlock()
-	directInvalid := b.Func.MakeBlock()
-	directDispatch := b.Switch(directStatus, directInvalid)
-	directDispatch.Case(p.prog.IntVal(coroWorkerResumeSuccessV1, p.prog.Uint32()), directNormal)
-	directDispatch.Case(p.prog.IntVal(coroWorkerResumeFaultMemoryV1, p.prog.Uint32()), directMemoryFault)
-	directDispatch.Case(p.prog.IntVal(coroWorkerResumeFaultDivideV1, p.prog.Uint32()), directDivideFault)
-	directDispatch.End(b)
-	b.SetBlockEx(directMemoryFault, llssa.AtEnd, false)
-	p.compileCoroTerminalFault(b, coroFaultNilV1)
-	b.SetBlockEx(directDivideFault, llssa.AtEnd, false)
-	p.compileCoroTerminalFault(b, coroFaultIntegerDivideByZeroV1)
-	b.SetBlockEx(directInvalid, llssa.AtEnd, false)
-	b.Unreachable()
-	b.SetBlockEx(directNormal, llssa.AtEnd, false)
-	b.Jump(join)
-
-	b.SetBlockEx(workerBlock, llssa.AtEnd, false)
-	emitPark(b)
-	b.Jump(join)
+	emitDirect := func(hookName string) {
+		direct := p.pkg.NewFunc(
+			hookName, coroOSThreadForeignCallSignature(), llssa.InC,
+		)
+		directStatus := b.Call(direct.Expr, directArgs...)
+		directNormal := b.Func.MakeBlock()
+		directMemoryFault := b.Func.MakeBlock()
+		directDivideFault := b.Func.MakeBlock()
+		directInvalid := b.Func.MakeBlock()
+		directDispatch := b.Switch(directStatus, directInvalid)
+		directDispatch.Case(p.prog.IntVal(coroWorkerResumeSuccessV1, p.prog.Uint32()), directNormal)
+		directDispatch.Case(p.prog.IntVal(coroWorkerResumeFaultMemoryV1, p.prog.Uint32()), directMemoryFault)
+		directDispatch.Case(p.prog.IntVal(coroWorkerResumeFaultDivideV1, p.prog.Uint32()), directDivideFault)
+		directDispatch.End(b)
+		b.SetBlockEx(directMemoryFault, llssa.AtEnd, false)
+		p.compileCoroTerminalFault(b, coroFaultNilV1)
+		b.SetBlockEx(directDivideFault, llssa.AtEnd, false)
+		p.compileCoroTerminalFault(b, coroFaultIntegerDivideByZeroV1)
+		b.SetBlockEx(directInvalid, llssa.AtEnd, false)
+		b.Unreachable()
+		b.SetBlockEx(directNormal, llssa.AtEnd, false)
+		b.Jump(join)
+	}
+	if nativeSyscall {
+		emitDirect(coroNativeSyscallCallHookV1)
+	} else {
+		lockedHook := p.pkg.NewFunc(coroOSThreadLockedHookV1, coroOSThreadLockedSignature(), llssa.InC)
+		locked := b.Call(lockedHook.Expr, body.task)
+		directBlock := b.Func.MakeBlock()
+		workerBlock := b.Func.MakeBlock()
+		b.If(locked, directBlock, workerBlock)
+		b.SetBlockEx(directBlock, llssa.AtEnd, false)
+		emitDirect(coroOSThreadForeignCallHookV1)
+		b.SetBlockEx(workerBlock, llssa.AtEnd, false)
+		emitPark(b)
+		b.Jump(join)
+	}
 	b.SetBlockContinuation(join)
-	// The worker queue deliberately contains only copied uintptr words. Keep
-	// every independently proved typed owner live until the physical completion
-	// acknowledgement has selected this normal resume path; llvm.fake.use emits
-	// no machine code but forces CoroSplit to retain the values in the frame.
+	// Keep every independently proved typed owner live until direct return or
+	// worker completion has selected this normal path; llvm.fake.use emits no
+	// machine code but forces CoroSplit to retain the values in the frame.
 	p.emitCoroKeepaliveSlots(b, keepaliveSlots)
 	return coroWorkerWordResultV1{r1: b.Load(r1), r2: b.Load(r2), errno: b.Load(errno)}
 }
@@ -520,7 +528,7 @@ func (p *context) compileCoroWorkerSyscall(
 		compiled[index] = p.compileValue(b, argument)
 	}
 	keepaliveSlots := p.compileCoroCallKeepaliveSlots(b, direct)
-	result := p.compileCoroWorkerWordCall(b, compiled[0], compiled[0], compiled[1:], keepaliveSlots, nil)
+	result := p.compileCoroWorkerWordCall(b, compiled[0], compiled[0], compiled[1:], keepaliveSlots, nil, true)
 	errnoValue := p.filterSyscallErrno(b, result.r1, result.errno, convention)
 	return b.Aggregate(p.type_(results, llssa.InGo), result.r1, result.r2, errnoValue)
 }

cl/coro_worker_cgo.go

@@ -875,6 +875,7 @@ func (p *context) compileCoroWorkerCgoErrnoCall(
 		[]llssa.Expr{b.Convert(p.prog.Uintptr(), record)},
 		nil,
 		nil,
+		false,
 	)
 	b.KeepAlive(record)
 	p.cgoRet = b.LoadKnownNonNil(b.FieldAddr(record, shape.resultField))
@@ -927,6 +928,7 @@ func (p *context) compileCoroWorkerCgoTransaction(
 		[]llssa.Expr{b.Convert(p.prog.Uintptr(), record)},
 		keepaliveSlots,
 		nil,
+		false,
 	)
 	b.KeepAlive(record)
 	if shape.result == nil {

cl/coro_worker_foreign.go

@@ -1270,6 +1270,7 @@ func (p *context) compileCoroWorkerForeignTransaction(
 		[]llssa.Expr{b.Convert(p.prog.Uintptr(), record)},
 		keepaliveSlots,
 		nil,
+		false,
 	)
 	// The native queue carries the record address as an opaque uintptr. This
 	// post-acknowledgement use forces CoroSplit to retain the complete typed