mirror of
https://github.com/ollama/ollama.git
synced 2026-09-21 13:38:14 -05:00
fix data races in progress and sched (#18319)
progress: join the render loop in stop and do the final writes after the goroutine exits, so Stop/StopAndClear cannot race an in-flight render on the shared bufio.Writer. sched: read the unload-mutable fields in runnerRef.LogValue only under a successful refMu.TryLock and omit them when contended, since slog resolves it on goroutines that may already hold refMu.
This commit is contained in:
@@ -32,10 +32,13 @@ type Progress struct {
|
||||
stopOnce sync.Once
|
||||
// done is closed to tell the render loop to exit.
|
||||
done chan struct{}
|
||||
// loopDone is closed by the render loop when it exits; stop waits on it so
|
||||
// no in-flight render can race the writes in Stop and StopAndClear.
|
||||
loopDone chan struct{}
|
||||
}
|
||||
|
||||
func NewProgress(w io.Writer) *Progress {
|
||||
p := &Progress{w: bufio.NewWriter(w), done: make(chan struct{})}
|
||||
p := &Progress{w: bufio.NewWriter(w), done: make(chan struct{}), loopDone: make(chan struct{})}
|
||||
go p.start()
|
||||
return p
|
||||
}
|
||||
@@ -49,6 +52,9 @@ func (p *Progress) stop() (bool, int) {
|
||||
stopped = true
|
||||
})
|
||||
|
||||
// Wait for the render loop to exit so no render can race the writes below.
|
||||
<-p.loopDone
|
||||
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
@@ -74,12 +80,12 @@ func (p *Progress) Stop() bool {
|
||||
}
|
||||
|
||||
func (p *Progress) StopAndClear() bool {
|
||||
stopped, pos := p.stop()
|
||||
defer p.w.Flush()
|
||||
|
||||
fmt.Fprint(p.w, "\033[?25l")
|
||||
defer fmt.Fprint(p.w, "\033[?25h")
|
||||
|
||||
stopped, pos := p.stop()
|
||||
if stopped {
|
||||
// clear all progress lines
|
||||
for i := range pos {
|
||||
@@ -144,6 +150,7 @@ func (p *Progress) renderLocked() {
|
||||
func (p *Progress) start() {
|
||||
ticker := time.NewTicker(100 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
defer close(p.loopDone)
|
||||
|
||||
for {
|
||||
select {
|
||||
|
||||
+11
-5
@@ -1498,6 +1498,9 @@ func (s *Scheduler) waitForVRAMRecovery(runner *runnerRef, runners []ml.Filtered
|
||||
return finished
|
||||
}
|
||||
|
||||
// LogValue may run from goroutines that already hold refMu (see the scheduler
|
||||
// debug logs), so the unload-mutable fields are read only under TryLock and
|
||||
// omitted when the lock is contended.
|
||||
func (runner *runnerRef) LogValue() slog.Value {
|
||||
if runner == nil {
|
||||
return slog.StringValue("nil")
|
||||
@@ -1507,24 +1510,27 @@ func (runner *runnerRef) LogValue() slog.Value {
|
||||
modelID = runner.modelKey
|
||||
}
|
||||
attrs := []slog.Attr{}
|
||||
if runner.refMu.TryLock() {
|
||||
if runner.model != nil {
|
||||
attrs = append(attrs, slog.String("name", runner.model.Name))
|
||||
}
|
||||
if len(runner.gpus) > 0 {
|
||||
attrs = append(attrs,
|
||||
slog.Any("inference", runner.gpus),
|
||||
slog.Any("inference", slices.Clone(runner.gpus)),
|
||||
)
|
||||
}
|
||||
attrs = append(attrs, slog.Int("pid", runner.pid))
|
||||
if runner.Options != nil {
|
||||
attrs = append(attrs, slog.Int("num_ctx", runner.Options.NumCtx))
|
||||
}
|
||||
runner.refMu.Unlock()
|
||||
}
|
||||
attrs = append(attrs,
|
||||
slog.String("size", format.HumanBytes2(runner.totalSize)),
|
||||
slog.String("vram", format.HumanBytes2(runner.vramSize)),
|
||||
slog.Int("parallel", runner.numParallel),
|
||||
slog.Int("pid", runner.pid),
|
||||
slog.String("model", modelID),
|
||||
)
|
||||
if runner.Options != nil {
|
||||
attrs = append(attrs, slog.Int("num_ctx", runner.Options.NumCtx))
|
||||
}
|
||||
return slog.GroupValue(attrs...)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user