Method SetEffectiveWindow

pkg/jobmanager/cirbuf.go:37–63

SetEffectiveWindow changes the sync mode and effective window size for flow control. The windowSize is capped at the buffer size. When window shrinks: sync mode blocks new writes, async mode truncates old data to enforce limit. When window increases: blocked writers are woken up if space becomes ava

(syncMode bool, windowSize int)

// When window shrinks: sync mode blocks new writes, async mode truncates old data to enforce limit.
// When window increases: blocked writers are woken up if space becomes available.
func (cb *CirBuf) SetEffectiveWindow(syncMode bool, windowSize int) {
	cb.lock.Lock()
	defer cb.lock.Unlock()

	maxSize := len(cb.buf)
	if windowSize > maxSize {
		windowSize = maxSize
	}

	oldSyncMode := cb.syncMode
	oldWindowSize := cb.windowSize
	cb.windowSize = windowSize
	cb.syncMode = syncMode

	// In async mode, enforce window size by truncating buffer if needed
	if !syncMode && cb.count > windowSize {
		excess := cb.count - windowSize
		cb.readPos = (cb.readPos + excess) % maxSize
		cb.count = windowSize
	}

	// Only sync mode blocks writers, so only wake if we were in sync mode.
	// Wake when window grows (more space available) or switching to async (no longer blocking).
	if oldSyncMode && (windowSize > oldWindowSize || !syncMode) {
		cb.tryWakeWriter()
	}
}

// WriteAvailable attempts to write as much data as possible without blocking.
// Returns the number of bytes written and a channel to wait on if buffer is full (nil if not blocking).