Stop pollProgress from spinning after the input stream closes

[Flo4604]

TL;DR

When using Bidirectional(false) and Restate.Run its possible to burn CPU due to pollProgress when using Bidirectional(false)

---

When doing 12 concurrent invocations that each sleep 4 seconds we spend 40.84s of CPU across 28.4M DoProgress calls, and the sdk calls NotifyInputClosed 28,476,268 times instead of once per run.

The fix is setting readChan to nil once input closes, which stops the select L110 to burn the cpu and taking the same workload to 0.04s at identical wall clock.

readInputLoop closes readChan on EOF, and a closed channel is always ready to receive, so the select took that case every time instead of waiting for the run closure.

7ff0fa6 (#86) introduced this when EOF went from a sentinel value to closing the channel,
with the Bidirectional (default) mode it keeps the request body open, so readChan never closes and the bug never happens. Thats probably why it never got found

Blocking here is safe as with input closed the shared core v7.0.3 (async_results.rs:128) returns WaitingExternalProgress only while a run in the awaited set is still executing, and that run always completes on runClosureCompletions.
Otherwise it suspends, so a wakeup always remains.

A different approach would be to listen to the shared cores waiting_input: false field but internal/proto/internal.proto:79 declares waiting_external_progress as Empty so we cant read that in the go sdk yet.
It could be passed trough instead and let the select wait on readChan only when waiting_input is true.
Happy to do that version if that is preferred

Reproduction:

Start a Restate server, then run this service:

package main

import (
	"context"
	"log"
	"syscall"
	"time"

	restate "github.com/restatedev/sdk-go"
	"github.com/restatedev/sdk-go/server"
)

type Spinner struct{}

func (Spinner) Sleep(ctx restate.Context, _ restate.Void) (string, error) {
	return restate.Run(ctx, func(ctx restate.RunContext) (string, error) {
		time.Sleep(4 * time.Second)
		return "done", nil
	})
}

// Print how many cores this process is using, once a second.
func reportCPU() {
	cpu := func() float64 {
		var ru syscall.Rusage
		syscall.Getrusage(syscall.RUSAGE_SELF, &ru)
		return float64(ru.Utime.Sec) + float64(ru.Utime.Usec)/1e6 +
			float64(ru.Stime.Sec) + float64(ru.Stime.Usec)/1e6
	}
	for prev := cpu(); ; {
		time.Sleep(time.Second)
		now := cpu()
		log.Printf("cores busy: %.2f", now-prev)
		prev = now
	}
}

func main() {
	go reportCPU()
	// Bidirectional(false) is required
	srv := server.NewRestate().
		Bidirectional(false).
		Bind(restate.Reflect(Spinner{}))
	log.Fatal(srv.Start(context.Background(), ":9080"))
}

# if Restate runs in a container, use host.docker.internal instead of localhost
curl localhost:9070/deployments -H 'content-type: application/json' -d '{"uri":"http://localhost:9080"}'

for i in $(seq 8); do curl -s localhost:8080/Spinner/Sleep -X POST & done

On main the service holds about one core per call for the full 4 seconds:

cores busy: 0.00
cores busy: 7.49     <- 8 calls arrive
cores busy: 7.74
cores busy: 7.73
cores busy: 7.77
cores busy: 0.28     <- closures finish
cores busy: 0.00

With the fix it never goes above 0.02.

Also checked:

• 267 of 267 conformance tests, all 8 suites.
• Bidirectional mode is identical before and after.
• Suspension paths under -race: timer, Run then timer then Run, awakeable, concurrent mixed load. All pass.
• Only one state could hang instead of spin: select entered with input closed and no run in flight. Counted across the whole suite, it never happens.

Links

• Go SDK: pollProgress / DoProgress loop dominates CPU during long-running restate.Run closures #139 looks like the same bug on v0.24.0
• [WIP] Fix read channel issue #86 introduced the regression.

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[Flo4604]

I have read the CLA Document and I hereby sign the CLA

[slinkydeveloper]

This sounds right, thanks for fixing it! Once I merge i'll do a patch release

[slinkydeveloper]

Thank you! Release 1.0.4