sched_test.go 23 KB
Newer Older
mashun1's avatar
v1  
mashun1 committed
1
2
3
4
5
package server

import (
	"bytes"
	"context"
xuxzh1's avatar
init  
xuxzh1 committed
6
	"errors"
mashun1's avatar
v1  
mashun1 committed
7
8
9
10
11
	"log/slog"
	"os"
	"testing"
	"time"

xuxzh1's avatar
init  
xuxzh1 committed
12
13
	"github.com/stretchr/testify/require"

mashun1's avatar
v1  
mashun1 committed
14
15
	"github.com/ollama/ollama/api"
	"github.com/ollama/ollama/app/lifecycle"
xuxzh1's avatar
update  
xuxzh1 committed
16
	"github.com/ollama/ollama/discover"
mashun1's avatar
v1  
mashun1 committed
17
18
19
20
	"github.com/ollama/ollama/format"
	"github.com/ollama/ollama/llm"
)

xuxzh1's avatar
init  
xuxzh1 committed
21
func TestMain(m *testing.M) {
mashun1's avatar
v1  
mashun1 committed
22
23
	os.Setenv("OLLAMA_DEBUG", "1")
	lifecycle.InitLogging()
xuxzh1's avatar
init  
xuxzh1 committed
24
	os.Exit(m.Run())
mashun1's avatar
v1  
mashun1 committed
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
}

func TestInitScheduler(t *testing.T) {
	ctx, done := context.WithCancel(context.Background())
	defer done()
	s := InitScheduler(ctx)
	s.loadedMu.Lock()
	require.NotNil(t, s.loaded)
	s.loadedMu.Unlock()
}

func TestLoad(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer done()
	s := InitScheduler(ctx)
	var ggml *llm.GGML // value not used in tests
	req := &LlmRequest{
		ctx:             ctx,
		model:           &Model{ModelPath: "foo"},
		opts:            api.DefaultOptions(),
		successCh:       make(chan *runnerRef, 1),
		errCh:           make(chan error, 1),
xuxzh1's avatar
init  
xuxzh1 committed
47
		sessionDuration: &api.Duration{Duration: 2 * time.Second},
mashun1's avatar
v1  
mashun1 committed
48
49
	}
	// Fail to load model first
xuxzh1's avatar
update  
xuxzh1 committed
50
	s.newServerFn = func(gpus discover.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options, numParallel int) (llm.LlamaServer, error) {
xuxzh1's avatar
init  
xuxzh1 committed
51
		return nil, errors.New("something failed to load model blah")
mashun1's avatar
v1  
mashun1 committed
52
	}
xuxzh1's avatar
update  
xuxzh1 committed
53
	gpus := discover.GpuInfoList{}
xuxzh1's avatar
init  
xuxzh1 committed
54
55
	s.load(req, ggml, gpus, 0)
	require.Empty(t, req.successCh)
mashun1's avatar
v1  
mashun1 committed
56
57
	require.Len(t, req.errCh, 1)
	s.loadedMu.Lock()
xuxzh1's avatar
init  
xuxzh1 committed
58
	require.Empty(t, s.loaded)
mashun1's avatar
v1  
mashun1 committed
59
60
61
62
	s.loadedMu.Unlock()
	err := <-req.errCh
	require.Contains(t, err.Error(), "this model may be incompatible")

xuxzh1's avatar
init  
xuxzh1 committed
63
	server := &mockLlm{estimatedVRAM: 10, estimatedVRAMByGPU: map[string]uint64{}}
xuxzh1's avatar
update  
xuxzh1 committed
64
	s.newServerFn = func(gpus discover.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options, numParallel int) (llm.LlamaServer, error) {
mashun1's avatar
v1  
mashun1 committed
65
66
		return server, nil
	}
xuxzh1's avatar
init  
xuxzh1 committed
67
	s.load(req, ggml, gpus, 0)
mashun1's avatar
v1  
mashun1 committed
68
69
70
71
72
73
74
75
76
77
78
79
	select {
	case err := <-req.errCh:
		require.NoError(t, err)
	case resp := <-req.successCh:
		require.Equal(t, uint64(10), resp.estimatedVRAM)
		require.Equal(t, uint(1), resp.refCount)
		s.loadedMu.Lock()
		require.Len(t, s.loaded, 1)
		s.loadedMu.Unlock()
	}

	req.model.ModelPath = "dummy_model_path"
xuxzh1's avatar
init  
xuxzh1 committed
80
81
	server.waitResp = errors.New("wait failure")
	s.load(req, ggml, gpus, 0)
mashun1's avatar
v1  
mashun1 committed
82
83
84
85
	select {
	case err := <-req.errCh:
		require.Contains(t, err.Error(), "wait failure")
	case resp := <-req.successCh:
xuxzh1's avatar
init  
xuxzh1 committed
86
		t.Fatalf("unexpected success %v", resp)
mashun1's avatar
v1  
mashun1 committed
87
88
89
90
91
92
93
94
95
96
	}
	s.loadedMu.Lock()
	runner := s.loaded["dummy_model_path"]
	s.loadedMu.Unlock()
	require.NotNil(t, runner)
	require.Equal(t, uint(0), runner.refCount)
	time.Sleep(1 * time.Millisecond)
	require.Len(t, s.expiredCh, 1)
}

xuxzh1's avatar
init  
xuxzh1 committed
97
type reqBundle struct {
mashun1's avatar
v1  
mashun1 committed
98
99
100
101
102
103
104
	ctx     context.Context //nolint:containedctx
	ctxDone func()
	srv     *mockLlm
	req     *LlmRequest
	ggml    *llm.GGML
}

xuxzh1's avatar
update  
xuxzh1 committed
105
func (scenario *reqBundle) newServer(gpus discover.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options, numParallel int) (llm.LlamaServer, error) {
mashun1's avatar
v1  
mashun1 committed
106
107
108
	return scenario.srv, nil
}

xuxzh1's avatar
init  
xuxzh1 committed
109
110
111
func newScenarioRequest(t *testing.T, ctx context.Context, modelName string, estimatedVRAM uint64, duration *api.Duration) *reqBundle {
	b := &reqBundle{}
	b.ctx, b.ctxDone = context.WithCancel(ctx)
mashun1's avatar
v1  
mashun1 committed
112
113
114
	t.Helper()

	f, err := os.CreateTemp(t.TempDir(), modelName)
xuxzh1's avatar
init  
xuxzh1 committed
115
	require.NoError(t, err)
mashun1's avatar
v1  
mashun1 committed
116
117
	defer f.Close()

xuxzh1's avatar
init  
xuxzh1 committed
118
	require.NoError(t, llm.WriteGGUF(f, llm.KV{
mashun1's avatar
v1  
mashun1 committed
119
120
121
122
123
124
125
126
127
128
		"general.architecture":          "llama",
		"llama.context_length":          uint32(32),
		"llama.embedding_length":        uint32(4096),
		"llama.block_count":             uint32(1),
		"llama.attention.head_count":    uint32(32),
		"llama.attention.head_count_kv": uint32(32),
		"tokenizer.ggml.tokens":         []string{" "},
		"tokenizer.ggml.scores":         []float32{0},
		"tokenizer.ggml.token_type":     []int32{0},
	}, []llm.Tensor{
xuxzh1's avatar
init  
xuxzh1 committed
129
130
131
132
		{Name: "blk.0.attn.weight", Kind: uint32(0), Offset: uint64(0), Shape: []uint64{1, 1, 1, 1}, WriterTo: bytes.NewReader(make([]byte, 32))},
		{Name: "output.weight", Kind: uint32(0), Offset: uint64(0), Shape: []uint64{1, 1, 1, 1}, WriterTo: bytes.NewReader(make([]byte, 32))},
	}))
	require.NoError(t, err)
mashun1's avatar
v1  
mashun1 committed
133
134
135

	fname := f.Name()
	model := &Model{Name: modelName, ModelPath: fname}
xuxzh1's avatar
init  
xuxzh1 committed
136
	b.ggml, err = llm.LoadModel(model.ModelPath, 0)
mashun1's avatar
v1  
mashun1 committed
137
138
	require.NoError(t, err)

xuxzh1's avatar
init  
xuxzh1 committed
139
140
141
142
143
	if duration == nil {
		duration = &api.Duration{Duration: 5 * time.Millisecond}
	}
	b.req = &LlmRequest{
		ctx:             b.ctx,
mashun1's avatar
v1  
mashun1 committed
144
145
		model:           model,
		opts:            api.DefaultOptions(),
xuxzh1's avatar
init  
xuxzh1 committed
146
		sessionDuration: duration,
mashun1's avatar
v1  
mashun1 committed
147
148
149
		successCh:       make(chan *runnerRef, 1),
		errCh:           make(chan error, 1),
	}
xuxzh1's avatar
init  
xuxzh1 committed
150
151
	b.srv = &mockLlm{estimatedVRAM: estimatedVRAM, estimatedVRAMByGPU: map[string]uint64{"": estimatedVRAM}}
	return b
mashun1's avatar
v1  
mashun1 committed
152
153
}

xuxzh1's avatar
update  
xuxzh1 committed
154
155
func getGpuFn() discover.GpuInfoList {
	g := discover.GpuInfo{Library: "metal"}
xuxzh1's avatar
init  
xuxzh1 committed
156
157
	g.TotalMemory = 24 * format.GigaByte
	g.FreeMemory = 12 * format.GigaByte
xuxzh1's avatar
update  
xuxzh1 committed
158
	return []discover.GpuInfo{g}
xuxzh1's avatar
init  
xuxzh1 committed
159
}
mashun1's avatar
v1  
mashun1 committed
160

xuxzh1's avatar
update  
xuxzh1 committed
161
162
func getCpuFn() discover.GpuInfoList {
	g := discover.GpuInfo{Library: "cpu"}
xuxzh1's avatar
init  
xuxzh1 committed
163
164
	g.TotalMemory = 32 * format.GigaByte
	g.FreeMemory = 26 * format.GigaByte
xuxzh1's avatar
update  
xuxzh1 committed
165
	return []discover.GpuInfo{g}
xuxzh1's avatar
init  
xuxzh1 committed
166
}
mashun1's avatar
v1  
mashun1 committed
167

xuxzh1's avatar
init  
xuxzh1 committed
168
169
170
func TestRequestsSameModelSameRequest(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer done()
mashun1's avatar
v1  
mashun1 committed
171
	s := InitScheduler(ctx)
xuxzh1's avatar
init  
xuxzh1 committed
172
173
174
175
176
177
178
179
180
181
	s.getGpuFn = getGpuFn
	s.getCpuFn = getCpuFn
	a := newScenarioRequest(t, ctx, "ollama-model-1", 10, &api.Duration{Duration: 5 * time.Millisecond})
	b := newScenarioRequest(t, ctx, "ollama-model-1", 11, &api.Duration{Duration: 0})
	b.req.model = a.req.model
	b.ggml = a.ggml

	s.newServerFn = a.newServer
	slog.Info("a")
	s.pendingReqCh <- a.req
mashun1's avatar
v1  
mashun1 committed
182
183
184
	require.Len(t, s.pendingReqCh, 1)
	s.Run(ctx)
	select {
xuxzh1's avatar
init  
xuxzh1 committed
185
186
187
188
189
190
	case resp := <-a.req.successCh:
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, a.req.errCh)
	case err := <-a.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
191
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
192
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
193
194
195
	}

	// Same runner as first request due to not needing a reload
xuxzh1's avatar
init  
xuxzh1 committed
196
197
198
	s.newServerFn = b.newServer
	slog.Info("b")
	s.pendingReqCh <- b.req
mashun1's avatar
v1  
mashun1 committed
199
	select {
xuxzh1's avatar
init  
xuxzh1 committed
200
201
202
203
204
205
	case resp := <-b.req.successCh:
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, b.req.errCh)
	case err := <-b.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
206
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
		t.Fatal("timeout")
	}
}

func TestRequestsSimpleReloadSameModel(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer done()
	s := InitScheduler(ctx)
	s.getGpuFn = getGpuFn
	s.getCpuFn = getCpuFn
	a := newScenarioRequest(t, ctx, "ollama-model-1", 10, &api.Duration{Duration: 5 * time.Millisecond})
	b := newScenarioRequest(t, ctx, "ollama-model-1", 20, &api.Duration{Duration: 5 * time.Millisecond})
	tmpModel := *a.req.model
	b.req.model = &tmpModel
	b.ggml = a.ggml

	s.newServerFn = a.newServer
	slog.Info("a")
	s.pendingReqCh <- a.req
	require.Len(t, s.pendingReqCh, 1)
	s.Run(ctx)
	select {
	case resp := <-a.req.successCh:
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, a.req.errCh)
	case err := <-a.req.errCh:
		t.Fatal(err.Error())
	case <-ctx.Done():
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
237
238
239
	}

	// Trigger a reload
xuxzh1's avatar
init  
xuxzh1 committed
240
241
242
243
	s.newServerFn = b.newServer
	b.req.model.AdapterPaths = []string{"new"}
	slog.Info("b")
	s.pendingReqCh <- b.req
mashun1's avatar
v1  
mashun1 committed
244
245
	// finish first two requests, so model can reload
	time.Sleep(1 * time.Millisecond)
xuxzh1's avatar
init  
xuxzh1 committed
246
	a.ctxDone()
mashun1's avatar
v1  
mashun1 committed
247
	select {
xuxzh1's avatar
init  
xuxzh1 committed
248
249
250
251
252
253
	case resp := <-b.req.successCh:
		require.Equal(t, resp.llama, b.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, b.req.errCh)
	case err := <-b.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
254
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
255
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
256
	}
xuxzh1's avatar
init  
xuxzh1 committed
257
}
mashun1's avatar
v1  
mashun1 committed
258

xuxzh1's avatar
init  
xuxzh1 committed
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
func TestRequestsMultipleLoadedModels(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer done()
	s := InitScheduler(ctx)
	s.getGpuFn = getGpuFn
	s.getCpuFn = getCpuFn

	// Multiple loaded models
	a := newScenarioRequest(t, ctx, "ollama-model-3a", 1*format.GigaByte, nil)
	b := newScenarioRequest(t, ctx, "ollama-model-3b", 24*format.GigaByte, nil)
	c := newScenarioRequest(t, ctx, "ollama-model-4a", 30, nil)
	c.req.opts.NumGPU = 0                                       // CPU load, will be allowed
	d := newScenarioRequest(t, ctx, "ollama-model-3c", 30, nil) // Needs prior unloaded

	t.Setenv("OLLAMA_MAX_LOADED_MODELS", "1")
	s.newServerFn = a.newServer
	slog.Info("a")
	s.pendingReqCh <- a.req
	s.Run(ctx)
mashun1's avatar
v1  
mashun1 committed
278
	select {
xuxzh1's avatar
init  
xuxzh1 committed
279
280
281
282
283
284
	case resp := <-a.req.successCh:
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, a.req.errCh)
	case err := <-a.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
285
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
286
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
287
288
289
290
291
	}
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 1)
	s.loadedMu.Unlock()

xuxzh1's avatar
init  
xuxzh1 committed
292
293
294
295
	t.Setenv("OLLAMA_MAX_LOADED_MODELS", "0")
	s.newServerFn = b.newServer
	slog.Info("b")
	s.pendingReqCh <- b.req
mashun1's avatar
v1  
mashun1 committed
296
	select {
xuxzh1's avatar
init  
xuxzh1 committed
297
298
299
300
301
302
	case resp := <-b.req.successCh:
		require.Equal(t, resp.llama, b.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, b.req.errCh)
	case err := <-b.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
303
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
304
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
305
306
307
308
309
310
	}
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 2)
	s.loadedMu.Unlock()

	// This is a CPU load with NumGPU = 0 so it should load
xuxzh1's avatar
init  
xuxzh1 committed
311
312
313
	s.newServerFn = c.newServer
	slog.Info("c")
	s.pendingReqCh <- c.req
mashun1's avatar
v1  
mashun1 committed
314
	select {
xuxzh1's avatar
init  
xuxzh1 committed
315
316
317
318
319
320
	case resp := <-c.req.successCh:
		require.Equal(t, resp.llama, c.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, c.req.errCh)
	case err := <-c.req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
321
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
322
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
323
324
325
326
327
328
	}
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 3)
	s.loadedMu.Unlock()

	// Try to load a model that wont fit
xuxzh1's avatar
init  
xuxzh1 committed
329
330
	s.newServerFn = d.newServer
	slog.Info("d")
mashun1's avatar
v1  
mashun1 committed
331
332
333
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 3)
	s.loadedMu.Unlock()
xuxzh1's avatar
init  
xuxzh1 committed
334
	a.ctxDone() // Won't help since this one isn't big enough to make room
mashun1's avatar
v1  
mashun1 committed
335
	time.Sleep(2 * time.Millisecond)
xuxzh1's avatar
init  
xuxzh1 committed
336
	s.pendingReqCh <- d.req
mashun1's avatar
v1  
mashun1 committed
337
338
339
340
341
	// finish prior request, so new model can load
	time.Sleep(6 * time.Millisecond)
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 2)
	s.loadedMu.Unlock()
xuxzh1's avatar
init  
xuxzh1 committed
342
	b.ctxDone()
mashun1's avatar
v1  
mashun1 committed
343
	select {
xuxzh1's avatar
init  
xuxzh1 committed
344
345
346
347
	case resp := <-d.req.successCh:
		require.Equal(t, resp.llama, d.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, d.req.errCh)
mashun1's avatar
v1  
mashun1 committed
348
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
349
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
350
351
352
353
354
355
356
	}
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 2)
	s.loadedMu.Unlock()
}

func TestGetRunner(t *testing.T) {
xuxzh1's avatar
update  
xuxzh1 committed
357
	ctx, done := context.WithTimeout(context.Background(), 200*time.Millisecond)
mashun1's avatar
v1  
mashun1 committed
358
359
	defer done()

xuxzh1's avatar
init  
xuxzh1 committed
360
361
362
363
	a := newScenarioRequest(t, ctx, "ollama-model-1a", 10, &api.Duration{Duration: 2 * time.Millisecond})
	b := newScenarioRequest(t, ctx, "ollama-model-1b", 10, &api.Duration{Duration: 2 * time.Millisecond})
	c := newScenarioRequest(t, ctx, "ollama-model-1c", 10, &api.Duration{Duration: 2 * time.Millisecond})
	t.Setenv("OLLAMA_MAX_QUEUE", "1")
mashun1's avatar
v1  
mashun1 committed
364
	s := InitScheduler(ctx)
xuxzh1's avatar
init  
xuxzh1 committed
365
366
367
368
369
	s.getGpuFn = getGpuFn
	s.getCpuFn = getCpuFn
	s.newServerFn = a.newServer
	slog.Info("a")
	successCh1a, errCh1a := s.GetRunner(a.ctx, a.req.model, a.req.opts, a.req.sessionDuration)
mashun1's avatar
v1  
mashun1 committed
370
	require.Len(t, s.pendingReqCh, 1)
xuxzh1's avatar
init  
xuxzh1 committed
371
372
	slog.Info("b")
	successCh1b, errCh1b := s.GetRunner(b.ctx, b.req.model, b.req.opts, b.req.sessionDuration)
mashun1's avatar
v1  
mashun1 committed
373
	require.Len(t, s.pendingReqCh, 1)
xuxzh1's avatar
init  
xuxzh1 committed
374
	require.Empty(t, successCh1b)
mashun1's avatar
v1  
mashun1 committed
375
376
377
378
379
380
	require.Len(t, errCh1b, 1)
	err := <-errCh1b
	require.Contains(t, err.Error(), "server busy")
	s.Run(ctx)
	select {
	case resp := <-successCh1a:
xuxzh1's avatar
init  
xuxzh1 committed
381
382
383
384
385
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, errCh1a)
	case err := <-errCh1a:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
386
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
387
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
388
	}
xuxzh1's avatar
init  
xuxzh1 committed
389
	a.ctxDone() // Set "a" model to idle so it can unload
mashun1's avatar
v1  
mashun1 committed
390
391
392
393
	s.loadedMu.Lock()
	require.Len(t, s.loaded, 1)
	s.loadedMu.Unlock()

xuxzh1's avatar
init  
xuxzh1 committed
394
395
396
	c.req.model.ModelPath = "bad path"
	slog.Info("c")
	successCh1c, errCh1c := s.GetRunner(c.ctx, c.req.model, c.req.opts, c.req.sessionDuration)
mashun1's avatar
v1  
mashun1 committed
397
	// Starts in pending channel, then should be quickly processsed to return an error
xuxzh1's avatar
update  
xuxzh1 committed
398
	time.Sleep(50 * time.Millisecond) // Long enough for the "a" model to expire and unload
xuxzh1's avatar
init  
xuxzh1 committed
399
	require.Empty(t, successCh1c)
mashun1's avatar
v1  
mashun1 committed
400
	s.loadedMu.Lock()
xuxzh1's avatar
init  
xuxzh1 committed
401
	require.Empty(t, s.loaded)
mashun1's avatar
v1  
mashun1 committed
402
403
404
405
	s.loadedMu.Unlock()
	require.Len(t, errCh1c, 1)
	err = <-errCh1c
	require.Contains(t, err.Error(), "bad path")
xuxzh1's avatar
init  
xuxzh1 committed
406
	b.ctxDone()
mashun1's avatar
v1  
mashun1 committed
407
408
}

xuxzh1's avatar
update  
xuxzh1 committed
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
func TestExpireRunner(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer done()
	s := InitScheduler(ctx)
	req := &LlmRequest{
		ctx:             ctx,
		model:           &Model{ModelPath: "foo"},
		opts:            api.DefaultOptions(),
		successCh:       make(chan *runnerRef, 1),
		errCh:           make(chan error, 1),
		sessionDuration: &api.Duration{Duration: 2 * time.Minute},
	}

	var ggml *llm.GGML
	gpus := discover.GpuInfoList{}
	server := &mockLlm{estimatedVRAM: 10, estimatedVRAMByGPU: map[string]uint64{}}
	s.newServerFn = func(gpus discover.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options, numParallel int) (llm.LlamaServer, error) {
		return server, nil
	}
	s.load(req, ggml, gpus, 0)

	select {
	case err := <-req.errCh:
		if err != nil {
			t.Fatalf("expected no errors when loading, got '%s'", err.Error())
		}
	case resp := <-req.successCh:
		s.loadedMu.Lock()
		if resp.refCount != uint(1) || len(s.loaded) != 1 {
			t.Fatalf("expected a model to be loaded")
		}
		s.loadedMu.Unlock()
	}

	s.expireRunner(&Model{ModelPath: "foo"})

	s.finishedReqCh <- req
	s.processCompleted(ctx)

	s.loadedMu.Lock()
	if len(s.loaded) != 0 {
		t.Fatalf("expected model to be unloaded")
	}
	s.loadedMu.Unlock()
}

mashun1's avatar
v1  
mashun1 committed
455
456
457
458
459
460
// TODO - add one scenario that triggers the bogus finished event with positive ref count
func TestPrematureExpired(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer done()

	// Same model, same request
xuxzh1's avatar
init  
xuxzh1 committed
461
	scenario1a := newScenarioRequest(t, ctx, "ollama-model-1a", 10, nil)
mashun1's avatar
v1  
mashun1 committed
462
	s := InitScheduler(ctx)
xuxzh1's avatar
update  
xuxzh1 committed
463
464
	s.getGpuFn = func() discover.GpuInfoList {
		g := discover.GpuInfo{Library: "metal"}
mashun1's avatar
v1  
mashun1 committed
465
466
		g.TotalMemory = 24 * format.GigaByte
		g.FreeMemory = 12 * format.GigaByte
xuxzh1's avatar
update  
xuxzh1 committed
467
		return []discover.GpuInfo{g}
mashun1's avatar
v1  
mashun1 committed
468
469
470
471
472
473
474
475
	}
	s.newServerFn = scenario1a.newServer
	successCh1a, errCh1a := s.GetRunner(scenario1a.ctx, scenario1a.req.model, scenario1a.req.opts, scenario1a.req.sessionDuration)
	require.Len(t, s.pendingReqCh, 1)
	s.Run(ctx)
	select {
	case resp := <-successCh1a:
		require.Equal(t, resp.llama, scenario1a.srv)
xuxzh1's avatar
init  
xuxzh1 committed
476
477
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, errCh1a)
mashun1's avatar
v1  
mashun1 committed
478
479
480
481
482
		s.loadedMu.Lock()
		require.Len(t, s.loaded, 1)
		s.loadedMu.Unlock()
		slog.Info("sending premature expired event now")
		s.expiredCh <- resp // Shouldn't happen in real life, but make sure its safe
xuxzh1's avatar
init  
xuxzh1 committed
483
484
	case err := <-errCh1a:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
485
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
486
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
487
	}
xuxzh1's avatar
init  
xuxzh1 committed
488
	time.Sleep(scenario1a.req.sessionDuration.Duration)
mashun1's avatar
v1  
mashun1 committed
489
490
491
492
	scenario1a.ctxDone()
	time.Sleep(20 * time.Millisecond)
	require.LessOrEqual(t, len(s.finishedReqCh), 1)
	time.Sleep(10 * time.Millisecond)
xuxzh1's avatar
init  
xuxzh1 committed
493
	require.Empty(t, s.finishedReqCh)
mashun1's avatar
v1  
mashun1 committed
494
	s.loadedMu.Lock()
xuxzh1's avatar
init  
xuxzh1 committed
495
	require.Empty(t, s.loaded)
mashun1's avatar
v1  
mashun1 committed
496
497
498
499
500
501
502
503
504
505
506
507
508
	s.loadedMu.Unlock()

	// also shouldn't happen in real life
	s.finishedReqCh <- scenario1a.req
	time.Sleep(5 * time.Millisecond)
}

func TestUseLoadedRunner(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	req := &LlmRequest{
		ctx:             ctx,
		opts:            api.DefaultOptions(),
		successCh:       make(chan *runnerRef, 1),
xuxzh1's avatar
init  
xuxzh1 committed
509
		sessionDuration: &api.Duration{Duration: 2},
mashun1's avatar
v1  
mashun1 committed
510
511
	}
	finished := make(chan *LlmRequest)
xuxzh1's avatar
init  
xuxzh1 committed
512
513
	llm1 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{}}
	r1 := &runnerRef{llama: llm1, sessionDuration: 1, numParallel: 1}
mashun1's avatar
v1  
mashun1 committed
514
515
516
517
518
519
	req.useLoadedRunner(r1, finished)
	require.Equal(t, uint(1), r1.refCount)
	require.Equal(t, time.Duration(2), r1.sessionDuration)
	select {
	case success := <-req.successCh:
		require.Equal(t, r1, success)
xuxzh1's avatar
init  
xuxzh1 committed
520
521
	case err := <-req.errCh:
		t.Fatal(err.Error())
mashun1's avatar
v1  
mashun1 committed
522
	case <-ctx.Done():
xuxzh1's avatar
init  
xuxzh1 committed
523
		t.Fatal("timeout")
mashun1's avatar
v1  
mashun1 committed
524
525
526
527
528
529
530
531
532
	}
	done()
	fin := <-finished
	require.Equal(t, req, fin)
}

func TestUpdateFreeSpace(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()
xuxzh1's avatar
update  
xuxzh1 committed
533
	gpus := discover.GpuInfoList{
mashun1's avatar
v1  
mashun1 committed
534
535
536
537
538
539
540
541
542
543
544
545
546
		{
			Library: "a",
			ID:      "1",
		},
		{
			Library: "a",
			ID:      "2",
		},
	}
	gpus[0].TotalMemory = 1000
	gpus[0].FreeMemory = 900
	gpus[1].TotalMemory = 2000
	gpus[1].FreeMemory = 1900
xuxzh1's avatar
init  
xuxzh1 committed
547
548
549
550
	llm1 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{"1": 50, "2": 50}}
	llm2 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{"1": 125, "2": 75}}
	r1 := &runnerRef{llama: llm1, gpus: gpus, numParallel: 1}
	r2 := &runnerRef{llama: llm2, gpus: gpus, numParallel: 1}
mashun1's avatar
v1  
mashun1 committed
551
552
553
554
555
556
557
558

	s := InitScheduler(ctx)
	s.loadedMu.Lock()
	s.loaded["a"] = r1
	s.loaded["b"] = r2
	s.loadedMu.Unlock()

	s.updateFreeSpace(gpus)
xuxzh1's avatar
init  
xuxzh1 committed
559
560
561
562
563
564
565
	require.Equal(t, uint64(1000-50-125), gpus[0].FreeMemory)
	require.Equal(t, uint64(2000-50-75), gpus[1].FreeMemory)
}

func TestFilterGPUsWithoutLoadingModels(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()
xuxzh1's avatar
update  
xuxzh1 committed
566
	gpus := discover.GpuInfoList{
xuxzh1's avatar
init  
xuxzh1 committed
567
568
569
570
571
572
573
574
575
		{
			Library: "cuda",
			ID:      "0",
		},
		{
			Library: "cuda",
			ID:      "1",
		},
	}
xuxzh1's avatar
update  
xuxzh1 committed
576
	r1 := &runnerRef{gpus: discover.GpuInfoList{gpus[0]}, loading: true}
xuxzh1's avatar
init  
xuxzh1 committed
577
578
579
580
581
582
583
584
585
586

	s := InitScheduler(ctx)
	s.loadedMu.Lock()
	s.loaded["a"] = r1
	s.loadedMu.Unlock()

	tmp := s.filterGPUsWithoutLoadingModels(gpus)
	require.Len(t, tmp, 1)
	require.Equal(t, "1", tmp[0].ID)

xuxzh1's avatar
update  
xuxzh1 committed
587
	r1.gpus = discover.GpuInfoList{gpus[1]}
xuxzh1's avatar
init  
xuxzh1 committed
588
589
590
591
	tmp = s.filterGPUsWithoutLoadingModels(gpus)
	require.Len(t, tmp, 1)
	require.Equal(t, "0", tmp[0].ID)

xuxzh1's avatar
update  
xuxzh1 committed
592
	r1.gpus = discover.GpuInfoList{}
xuxzh1's avatar
init  
xuxzh1 committed
593
594
	tmp = s.filterGPUsWithoutLoadingModels(gpus)
	require.Len(t, tmp, 2)
mashun1's avatar
v1  
mashun1 committed
595
596
597
598
599
600
}

func TestFindRunnerToUnload(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()

xuxzh1's avatar
init  
xuxzh1 committed
601
602
	r1 := &runnerRef{refCount: 1, sessionDuration: 1, numParallel: 1}
	r2 := &runnerRef{sessionDuration: 2, numParallel: 1}
mashun1's avatar
v1  
mashun1 committed
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620

	s := InitScheduler(ctx)
	s.loadedMu.Lock()
	s.loaded["a"] = r1
	s.loaded["b"] = r2
	s.loadedMu.Unlock()

	resp := s.findRunnerToUnload()
	require.Equal(t, r2, resp)
	r2.refCount = 1
	resp = s.findRunnerToUnload()
	require.Equal(t, r1, resp)
}

func TestNeedsReload(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()

xuxzh1's avatar
init  
xuxzh1 committed
621
	llm := &mockLlm{estimatedVRAMByGPU: map[string]uint64{}}
mashun1's avatar
v1  
mashun1 committed
622
623
	do := api.DefaultOptions()
	runner := &runnerRef{
xuxzh1's avatar
init  
xuxzh1 committed
624
625
626
627
628
629
630
		model: &Model{
			AdapterPaths:   []string{"adapter1"},
			ProjectorPaths: []string{"projector1"},
		},
		Options:     &do,
		llama:       llm,
		numParallel: 1,
mashun1's avatar
v1  
mashun1 committed
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
	}
	req := &LlmRequest{
		model: &Model{
			AdapterPaths:   []string{"adapter2"},
			ProjectorPaths: []string{"projector2"},
		},
		opts: api.DefaultOptions(),
	}
	resp := runner.needsReload(ctx, req)
	require.True(t, resp)
	req.model.AdapterPaths = runner.model.AdapterPaths
	resp = runner.needsReload(ctx, req)
	require.True(t, resp)
	req.model.ProjectorPaths = runner.model.ProjectorPaths
	runner.loading = true
	req.opts.NumBatch = 1234
	resp = runner.needsReload(ctx, req)
	require.True(t, resp)
	req.opts.NumBatch = runner.Options.NumBatch
xuxzh1's avatar
init  
xuxzh1 committed
650
	llm.pingResp = errors.New("foo")
mashun1's avatar
v1  
mashun1 committed
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
	resp = runner.needsReload(ctx, req)
	require.True(t, resp)
	llm.pingResp = nil
	resp = runner.needsReload(ctx, req)
	require.False(t, resp)
	req.opts.NumGPU = 99
	resp = runner.needsReload(ctx, req)
	require.True(t, resp)
	req.opts.NumGPU = -1
	resp = runner.needsReload(ctx, req)
	require.False(t, resp)
}

func TestUnloadAllRunners(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()

xuxzh1's avatar
init  
xuxzh1 committed
668
669
	llm1 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{}}
	llm2 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{}}
mashun1's avatar
v1  
mashun1 committed
670
671
672
	s := InitScheduler(ctx)
	s.unloadAllRunners()

xuxzh1's avatar
init  
xuxzh1 committed
673
674
	r1 := &runnerRef{llama: llm1, numParallel: 1}
	r2 := &runnerRef{llama: llm2, numParallel: 1}
mashun1's avatar
v1  
mashun1 committed
675
676
677
678
679
680
681
682
683
684
685
686

	s.loadedMu.Lock()
	s.loaded["a"] = r1
	s.loaded["b"] = r2
	s.loadedMu.Unlock()
	s.unloadAllRunners()

	require.True(t, llm1.closeCalled)
	require.True(t, llm2.closeCalled)
}

func TestUnload(t *testing.T) {
xuxzh1's avatar
init  
xuxzh1 committed
687
688
689
	llm1 := &mockLlm{estimatedVRAMByGPU: map[string]uint64{}}
	r1 := &runnerRef{llama: llm1, numParallel: 1}
	r2 := &runnerRef{model: &Model{AdapterPaths: []string{"A"}}, numParallel: 1}
mashun1's avatar
v1  
mashun1 committed
690
691
692
693
694
695
	r1.unload()
	require.True(t, llm1.closeCalled)
	r2.unload()
	require.Nil(t, r2.model)
}

xuxzh1's avatar
init  
xuxzh1 committed
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
func TestAlreadyCanceled(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer done()
	dctx, done2 := context.WithCancel(ctx)
	done2()
	scenario1a := newScenarioRequest(t, dctx, "ollama-model-1", 10, &api.Duration{Duration: 0})
	s := InitScheduler(ctx)
	slog.Info("scenario1a")
	s.pendingReqCh <- scenario1a.req
	require.Len(t, s.pendingReqCh, 1)
	s.Run(ctx)
	time.Sleep(5 * time.Millisecond)
	require.Empty(t, s.pendingReqCh)
	require.Empty(t, scenario1a.req.errCh)
	require.Empty(t, scenario1a.req.successCh)
}

func TestHomogeneousGPUs(t *testing.T) {
	ctx, done := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer done()
	s := InitScheduler(ctx)

xuxzh1's avatar
update  
xuxzh1 committed
718
	s.getGpuFn = func() discover.GpuInfoList {
xuxzh1's avatar
init  
xuxzh1 committed
719
		// Set memory values to require the model to be spread
xuxzh1's avatar
update  
xuxzh1 committed
720
		gpus := []discover.GpuInfo{
xuxzh1's avatar
init  
xuxzh1 committed
721
722
723
724
725
726
727
728
729
730
731
			{Library: "cuda"},
			{Library: "rocm"},
		}
		gpus[0].TotalMemory = 1 * format.GibiByte
		gpus[0].FreeMemory = 256 * format.MebiByte
		gpus[1].TotalMemory = 1 * format.GibiByte
		gpus[1].FreeMemory = 256 * format.MebiByte
		return gpus
	}
	s.getCpuFn = getCpuFn
	a := newScenarioRequest(t, ctx, "ollama-model-1", 10, &api.Duration{Duration: 5 * time.Millisecond})
xuxzh1's avatar
update  
xuxzh1 committed
732
	s.newServerFn = func(gpus discover.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options, numParallel int) (llm.LlamaServer, error) {
xuxzh1's avatar
init  
xuxzh1 committed
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
		require.Len(t, gpus, 1)
		return a.newServer(gpus, model, ggml, adapters, projectors, opts, numParallel)
	}
	slog.Info("a")
	s.pendingReqCh <- a.req
	require.Len(t, s.pendingReqCh, 1)
	s.Run(ctx)
	select {
	case resp := <-a.req.successCh:
		require.Equal(t, resp.llama, a.srv)
		require.Empty(t, s.pendingReqCh)
		require.Empty(t, a.req.errCh)
	case err := <-a.req.errCh:
		t.Fatal(err.Error())
	case <-ctx.Done():
		t.Fatal("timeout")
	}
}

mashun1's avatar
v1  
mashun1 committed
752
type mockLlm struct {
xuxzh1's avatar
init  
xuxzh1 committed
753
754
755
756
757
758
759
760
761
762
763
764
765
766
	pingResp           error
	waitResp           error
	completionResp     error
	embeddingResp      []float32
	embeddingRespErr   error
	tokenizeResp       []int
	tokenizeRespErr    error
	detokenizeResp     string
	detonekizeRespErr  error
	closeResp          error
	closeCalled        bool
	estimatedVRAM      uint64
	estimatedTotal     uint64
	estimatedVRAMByGPU map[string]uint64
mashun1's avatar
v1  
mashun1 committed
767
768
769
770
771
772
773
}

func (s *mockLlm) Ping(ctx context.Context) error             { return s.pingResp }
func (s *mockLlm) WaitUntilRunning(ctx context.Context) error { return s.waitResp }
func (s *mockLlm) Completion(ctx context.Context, req llm.CompletionRequest, fn func(llm.CompletionResponse)) error {
	return s.completionResp
}
xuxzh1's avatar
init  
xuxzh1 committed
774
775

func (s *mockLlm) Embedding(ctx context.Context, input string) ([]float32, error) {
mashun1's avatar
v1  
mashun1 committed
776
777
	return s.embeddingResp, s.embeddingRespErr
}
xuxzh1's avatar
init  
xuxzh1 committed
778

mashun1's avatar
v1  
mashun1 committed
779
780
781
func (s *mockLlm) Tokenize(ctx context.Context, content string) ([]int, error) {
	return s.tokenizeResp, s.tokenizeRespErr
}
xuxzh1's avatar
init  
xuxzh1 committed
782

mashun1's avatar
v1  
mashun1 committed
783
784
785
func (s *mockLlm) Detokenize(ctx context.Context, tokens []int) (string, error) {
	return s.detokenizeResp, s.detonekizeRespErr
}
xuxzh1's avatar
init  
xuxzh1 committed
786

mashun1's avatar
v1  
mashun1 committed
787
788
789
790
func (s *mockLlm) Close() error {
	s.closeCalled = true
	return s.closeResp
}
xuxzh1's avatar
init  
xuxzh1 committed
791
792
793
func (s *mockLlm) EstimatedVRAM() uint64                  { return s.estimatedVRAM }
func (s *mockLlm) EstimatedTotal() uint64                 { return s.estimatedTotal }
func (s *mockLlm) EstimatedVRAMByGPU(gpuid string) uint64 { return s.estimatedVRAMByGPU[gpuid] }