sync_dispatch.go Raw History Folder

1package syncdispatch
2
3import (
4	"sync"
5	"time"
6)
7
8// SyncDispatchStatus describes how a synchronous event-loop dispatch finished.
9type SyncDispatchStatus string
10
11const (
12	// SyncDispatchInline means the callback ran immediately because the caller
13	// already owned the target main context.
14	SyncDispatchInline SyncDispatchStatus = "inline"
15	// SyncDispatchCompleted means the callback was dispatched and completed before
16	// the configured timeout elapsed.
17	SyncDispatchCompleted SyncDispatchStatus = "completed"
18	// SyncDispatchTimedOut means the callback did not start before the timeout and
19	// was canceled before running side effects.
20	SyncDispatchTimedOut SyncDispatchStatus = "timed_out"
21	// SyncDispatchCompletedAfterTimeout means the callback started before the
22	// timeout and completed after the timeout elapsed. Callers must treat it as
23	// completed because side effects have run.
24	SyncDispatchCompletedAfterTimeout SyncDispatchStatus = "completed_after_timeout"
25	// SyncDispatchQueuedAfterTimeout means the callback did not start before the
26	// timeout, but the dispatch was intentionally left queued so it may run later.
27	SyncDispatchQueuedAfterTimeout SyncDispatchStatus = "queued_after_timeout"
28	// SyncDispatchSkipped means required inputs were missing, so no callback ran.
29	SyncDispatchSkipped SyncDispatchStatus = "skipped"
30)
31
32// SyncDispatchOptions configures RunSynchronousDispatch.
33type SyncDispatchOptions struct {
34	// Label identifies the caller in diagnostics. The helper does not log by
35	// itself, but callers can include this label when reporting the returned
36	// result.
37	Label string
38	// Timeout bounds how long the caller waits for dispatched work. Non-positive
39	// durations fall back to DefaultSyncDispatchTimeout.
40	Timeout time.Duration
41	// IsOwner returns whether the caller already owns the target main context.
42	IsOwner func() bool
43	// Dispatch schedules a callback on the target main context.
44	Dispatch func(func())
45	// AllowLateStartAfterTimeout keeps queued work runnable when the timeout fires
46	// before Dispatch starts the callback. Use this only for cleanup work where
47	// canceling would leak resources; decision callbacks should leave this false.
48	AllowLateStartAfterTimeout bool
49}
50
51// SyncDispatchResult is returned by RunSynchronousDispatch.
52type SyncDispatchResult struct {
53	Label   string
54	Status  SyncDispatchStatus
55	Elapsed time.Duration
56}
57
58// Completed reports whether the callback completed before returning.
59func (r SyncDispatchResult) Completed() bool {
60	return r.Status == SyncDispatchInline || r.Status == SyncDispatchCompleted || r.Status == SyncDispatchCompletedAfterTimeout
61}
62
63// DefaultSyncDispatchTimeout is a conservative upper bound for synchronous
64// waits on a target event loop. It is long enough for normal idle dispatch but
65// short enough to avoid converting a stalled loop into an unbounded caller
66// deadlock.
67const DefaultSyncDispatchTimeout = 2 * time.Second
68
69const (
70	dispatchStateInit int32 = iota
71	dispatchStateStarted
72	dispatchStateCompleted
73	dispatchStateTimedOut
74)
75
76// RunSynchronousDispatch executes fn inline when the target main context is
77// already owned by the caller; otherwise it schedules fn with opts.Dispatch and
78// waits for completion up to opts.Timeout.
79//
80// If the timeout fires before the dispatched callback starts, the default
81// behavior is to skip the callback when the main loop eventually services it. If
82// AllowLateStartAfterTimeout is true, queued work is left runnable and the
83// result reports queued_after_timeout instead. If the callback has already
84// started, Go cannot preempt it safely; this helper waits for it to finish and
85// reports completed_after_timeout so callers do not fail closed while side
86// effects are still possible.
87func RunSynchronousDispatch(opts SyncDispatchOptions, fn func()) SyncDispatchResult {
88	start := time.Now()
89	result := SyncDispatchResult{Label: opts.Label, Status: SyncDispatchSkipped}
90	if fn == nil {
91		result.Elapsed = time.Since(start)
92		return result
93	}
94	if opts.IsOwner != nil && opts.IsOwner() {
95		fn()
96		result.Status = SyncDispatchInline
97		result.Elapsed = time.Since(start)
98		return result
99	}
100	if opts.Dispatch == nil {
101		result.Elapsed = time.Since(start)
102		return result
103	}
104
105	timeout := opts.Timeout
106	if timeout <= 0 {
107		timeout = DefaultSyncDispatchTimeout
108	}
109
110	deadline := start.Add(timeout)
111	done := make(chan struct{})
112	state := dispatchStateInit
113	var stateMu sync.Mutex
114	loadState := func() int32 {
115		stateMu.Lock()
116		defer stateMu.Unlock()
117		return state
118	}
119	markCallbackStarted := func() bool {
120		stateMu.Lock()
121		defer stateMu.Unlock()
122		if state != dispatchStateInit {
123			return false
124		}
125		// Check deadline and transition to dispatchStateStarted under one lock so
126		// the timer cannot mark dispatchStateTimedOut after this callback is allowed
127		// to start. Late-start cleanup opts out through AllowLateStartAfterTimeout.
128		if !opts.AllowLateStartAfterTimeout && time.Now().After(deadline) {
129			state = dispatchStateTimedOut
130			return false
131		}
132		state = dispatchStateStarted
133		return true
134	}
135	markCallbackCompleted := func() {
136		stateMu.Lock()
137		defer stateMu.Unlock()
138		state = dispatchStateCompleted
139	}
140	go func() {
141		opts.Dispatch(func() {
142			if !markCallbackStarted() {
143				close(done)
144				return
145			}
146			defer close(done)
147			fn()
148			markCallbackCompleted()
149		})
150	}()
151
152	timer := time.NewTimer(timeout)
153	defer timer.Stop()
154	select {
155	case <-done:
156		if loadState() == dispatchStateTimedOut {
157			result.Status = SyncDispatchTimedOut
158		} else {
159			result.Status = SyncDispatchCompleted
160		}
161	case <-timer.C:
162		stateMu.Lock()
163		currentState := state
164		if opts.AllowLateStartAfterTimeout && currentState == dispatchStateInit {
165			// Cleanup work may remain queued after timer expiry; decision callbacks
166			// instead fall through and mark dispatchStateTimedOut while still unstarted.
167			stateMu.Unlock()
168			result.Status = SyncDispatchQueuedAfterTimeout
169			result.Elapsed = time.Since(start)
170			return result
171		}
172		if currentState == dispatchStateInit {
173			state = dispatchStateTimedOut
174			stateMu.Unlock()
175			result.Status = SyncDispatchTimedOut
176			result.Elapsed = time.Since(start)
177			return result
178		}
179		stateMu.Unlock()
180		// dispatchStateStarted means the worker already owns fn side effects; wait on
181		// done so the caller reports completed-after-timeout only after completion.
182		<-done
183		if loadState() == dispatchStateTimedOut {
184			result.Status = SyncDispatchTimedOut
185		} else {
186			result.Status = SyncDispatchCompletedAfterTimeout
187		}
188	}
189	result.Elapsed = time.Since(start)
190	return result
191}
sync_dispatch.go RawHistoryFolder

sync_dispatch.go Raw History Folder
