sched.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703
  1. package server
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "log/slog"
  7. "reflect"
  8. "runtime"
  9. "sort"
  10. "strings"
  11. "sync"
  12. "time"
  13. "github.com/ollama/ollama/api"
  14. "github.com/ollama/ollama/envconfig"
  15. "github.com/ollama/ollama/format"
  16. "github.com/ollama/ollama/gpu"
  17. "github.com/ollama/ollama/llm"
  18. )
  19. type LlmRequest struct {
  20. ctx context.Context //nolint:containedctx
  21. model *Model
  22. opts api.Options
  23. sessionDuration time.Duration
  24. successCh chan *runnerRef
  25. errCh chan error
  26. schedAttempts uint
  27. }
  28. type Scheduler struct {
  29. pendingReqCh chan *LlmRequest
  30. finishedReqCh chan *LlmRequest
  31. expiredCh chan *runnerRef
  32. unloadedCh chan interface{}
  33. loaded map[string]*runnerRef
  34. loadedMu sync.Mutex
  35. loadFn func(req *LlmRequest, ggml *llm.GGML, gpus gpu.GpuInfoList)
  36. newServerFn func(gpus gpu.GpuInfoList, model string, ggml *llm.GGML, adapters []string, projectors []string, opts api.Options) (llm.LlamaServer, error)
  37. getGpuFn func() gpu.GpuInfoList
  38. getCpuFn func() gpu.GpuInfoList
  39. reschedDelay time.Duration
  40. }
  41. var ErrMaxQueue = fmt.Errorf("server busy, please try again. maximum pending requests exceeded")
  42. func InitScheduler(ctx context.Context) *Scheduler {
  43. sched := &Scheduler{
  44. pendingReqCh: make(chan *LlmRequest, envconfig.MaxQueuedRequests),
  45. finishedReqCh: make(chan *LlmRequest, envconfig.MaxQueuedRequests),
  46. expiredCh: make(chan *runnerRef, envconfig.MaxQueuedRequests),
  47. unloadedCh: make(chan interface{}, envconfig.MaxQueuedRequests),
  48. loaded: make(map[string]*runnerRef),
  49. newServerFn: llm.NewLlamaServer,
  50. getGpuFn: gpu.GetGPUInfo,
  51. getCpuFn: gpu.GetCPUInfo,
  52. reschedDelay: 250 * time.Millisecond,
  53. }
  54. sched.loadFn = sched.load
  55. return sched
  56. }
  57. // context must be canceled to decrement ref count and release the runner
  58. func (s *Scheduler) GetRunner(c context.Context, model *Model, opts api.Options, sessionDuration time.Duration) (chan *runnerRef, chan error) {
  59. // allocate a large enough kv cache for all parallel requests
  60. if opts.NumCtx < 4 {
  61. opts.NumCtx = 4
  62. }
  63. opts.NumCtx *= envconfig.NumParallel
  64. req := &LlmRequest{
  65. ctx: c,
  66. model: model,
  67. opts: opts,
  68. sessionDuration: sessionDuration,
  69. successCh: make(chan *runnerRef),
  70. errCh: make(chan error, 1),
  71. }
  72. select {
  73. case s.pendingReqCh <- req:
  74. default:
  75. req.errCh <- ErrMaxQueue
  76. }
  77. return req.successCh, req.errCh
  78. }
  79. // Returns immediately, spawns go routines for the scheduler which will shutdown when ctx is done
  80. func (s *Scheduler) Run(ctx context.Context) {
  81. slog.Debug("starting llm scheduler")
  82. go func() {
  83. s.processPending(ctx)
  84. }()
  85. go func() {
  86. s.processCompleted(ctx)
  87. }()
  88. }
  89. func (s *Scheduler) processPending(ctx context.Context) {
  90. for {
  91. select {
  92. case <-ctx.Done():
  93. slog.Debug("shutting down scheduler pending loop")
  94. return
  95. case pending := <-s.pendingReqCh:
  96. // Block other requests until we get this pending request running
  97. pending.schedAttempts++
  98. if pending.ctx.Err() != nil {
  99. slog.Debug("pending request cancelled or timed out, skipping scheduling")
  100. continue
  101. }
  102. for {
  103. var runnerToExpire *runnerRef
  104. s.loadedMu.Lock()
  105. runner := s.loaded[pending.model.ModelPath]
  106. loadedCount := len(s.loaded)
  107. s.loadedMu.Unlock()
  108. if runner != nil {
  109. if runner.needsReload(ctx, pending) {
  110. runnerToExpire = runner
  111. } else {
  112. // Runner is usable, return it
  113. pending.useLoadedRunner(runner, s.finishedReqCh)
  114. break
  115. }
  116. } else if envconfig.MaxRunners > 0 && loadedCount >= envconfig.MaxRunners {
  117. slog.Debug("max runners achieved, unloading one to make room", "runner_count", loadedCount)
  118. runnerToExpire = s.findRunnerToUnload()
  119. } else {
  120. // Either no models are loaded or below envconfig.MaxRunners
  121. // Get a refreshed GPU list
  122. var gpus gpu.GpuInfoList
  123. if pending.opts.NumGPU == 0 {
  124. gpus = s.getCpuFn()
  125. } else {
  126. gpus = s.getGpuFn()
  127. }
  128. // Load model for fitting
  129. ggml, err := llm.LoadModel(pending.model.ModelPath)
  130. if err != nil {
  131. pending.errCh <- err
  132. break
  133. }
  134. // Evaluate if the model will fit in the available system memory, or if we should unload a model first
  135. if len(gpus) == 1 && gpus[0].Library == "cpu" {
  136. if loadedCount == 0 {
  137. slog.Debug("cpu mode with first model, loading")
  138. s.loadFn(pending, ggml, gpus)
  139. break
  140. }
  141. runnerToExpire = s.maybeFindCPURunnerToUnload(pending, ggml, gpus)
  142. if runnerToExpire == nil {
  143. slog.Debug("cpu mode with available system memory or first model, loading")
  144. s.loadFn(pending, ggml, gpus)
  145. break
  146. }
  147. // else we need to expire a runner
  148. } else if loadedCount == 0 {
  149. // No models loaded. Load the model but prefer the best fit.
  150. slog.Debug("loading first model", "model", pending.model.ModelPath)
  151. g := pickBestFitGPUs(pending, ggml, gpus)
  152. if g != nil {
  153. gpus = g
  154. }
  155. s.loadFn(pending, ggml, gpus)
  156. break
  157. }
  158. if runnerToExpire == nil {
  159. // More than one loaded model, so we have to see if the
  160. // new one fits
  161. //
  162. // We want to avoid loading on any GPUs that have other
  163. // models still loading on them to avoid potential races
  164. // with VRAM consumption ramping up during load
  165. availGpus := s.filterGPUsWithoutLoadingModels(gpus)
  166. // Update free memory from currently loaded models
  167. s.updateFreeSpace(availGpus)
  168. fitGpus := pickBestFitGPUs(pending, ggml, availGpus)
  169. if fitGpus != nil {
  170. slog.Debug("new model fits with existing models, loading")
  171. s.loadFn(pending, ggml, fitGpus)
  172. break
  173. }
  174. // We couldn't find a set of GPUs to fully load the new
  175. // model. If no other models are loading (both GPU lists
  176. // are the same) then we need to unload another model to
  177. // make room
  178. if len(availGpus) < len(gpus) {
  179. // There are other requests pending, and this one
  180. // needs more time, so put it on the back of the
  181. // queue so that we might satisfy other pending
  182. // requests that aren't blocked
  183. go func() {
  184. // Process in a go routine to avoid deadlocking
  185. // the scheduler if our queue is full
  186. slog.Debug("delaying scheduling while other models finish loading", "attempts", pending.schedAttempts, "model", pending.model.ModelPath)
  187. time.Sleep(s.reschedDelay)
  188. s.pendingReqCh <- pending
  189. }()
  190. break
  191. }
  192. runnerToExpire = s.findRunnerToUnload()
  193. }
  194. }
  195. if runnerToExpire == nil {
  196. // Shouildn't happen
  197. slog.Error("runner to expire was nil!")
  198. continue
  199. }
  200. // Trigger an expiration to unload once it's done
  201. runnerToExpire.refMu.Lock()
  202. slog.Debug("resetting model to expire immediately to make room", "modelPath", runnerToExpire.modelPath, "refCount", runnerToExpire.refCount)
  203. if runnerToExpire.expireTimer != nil {
  204. runnerToExpire.expireTimer.Stop()
  205. runnerToExpire.expireTimer = nil
  206. }
  207. runnerToExpire.sessionDuration = 0
  208. if runnerToExpire.refCount <= 0 {
  209. s.expiredCh <- runnerToExpire
  210. }
  211. runnerToExpire.refMu.Unlock()
  212. // Wait for the unload to happen
  213. // Note: at this point we're queueing up all incoming requests, even if they were for
  214. // a different model that's loaded and not scheduled to be removed.
  215. slog.Debug("waiting for pending requests to complete and unload to occur", "modelPath", runnerToExpire.modelPath)
  216. select {
  217. case <-ctx.Done():
  218. slog.Debug("shutting down scheduler pending loop")
  219. return
  220. case <-s.unloadedCh:
  221. slog.Debug("unload completed", "modelPath", runnerToExpire.modelPath)
  222. continue
  223. }
  224. }
  225. case <-s.unloadedCh:
  226. // An unload request when there are no pending request can be ignored
  227. slog.Debug("ignoring unload event with no pending requests")
  228. }
  229. }
  230. }
  231. func (s *Scheduler) processCompleted(ctx context.Context) {
  232. // Process completed requests, expired timers, and unloading models
  233. for {
  234. select {
  235. case <-ctx.Done():
  236. slog.Debug("shutting down scheduler completed loop")
  237. return
  238. case finished := <-s.finishedReqCh:
  239. s.loadedMu.Lock()
  240. runner := s.loaded[finished.model.ModelPath]
  241. s.loadedMu.Unlock()
  242. if runner == nil {
  243. slog.Error("finished request signal received after model unloaded", "modelPath", finished.model.ModelPath)
  244. continue
  245. }
  246. runner.refMu.Lock()
  247. runner.refCount--
  248. if runner.refCount <= 0 {
  249. if runner.sessionDuration <= 0 {
  250. slog.Debug("runner with zero duration has gone idle, expiring to unload", "modelPath", runner.modelPath)
  251. if runner.expireTimer != nil {
  252. runner.expireTimer.Stop()
  253. runner.expireTimer = nil
  254. }
  255. s.expiredCh <- runner
  256. } else if runner.expireTimer == nil {
  257. slog.Debug("runner with non-zero duration has gone idle, adding timer", "modelPath", runner.modelPath, "duration", runner.sessionDuration)
  258. runner.expireTimer = time.AfterFunc(runner.sessionDuration, func() {
  259. slog.Debug("timer expired, expiring to unload", "modelPath", runner.modelPath)
  260. runner.refMu.Lock()
  261. defer runner.refMu.Unlock()
  262. if runner.expireTimer != nil {
  263. runner.expireTimer.Stop()
  264. runner.expireTimer = nil
  265. }
  266. s.expiredCh <- runner
  267. })
  268. runner.expiresAt = time.Now().Add(runner.sessionDuration)
  269. } else {
  270. slog.Debug("runner with non-zero duration has gone idle, resetting timer", "modelPath", runner.modelPath, "duration", runner.sessionDuration)
  271. runner.expireTimer.Reset(runner.sessionDuration)
  272. runner.expiresAt = time.Now().Add(runner.sessionDuration)
  273. }
  274. }
  275. slog.Debug("after processing request finished event", "modelPath", runner.modelPath, "refCount", runner.refCount)
  276. runner.refMu.Unlock()
  277. case runner := <-s.expiredCh:
  278. slog.Debug("runner expired event received", "modelPath", runner.modelPath)
  279. runner.refMu.Lock()
  280. if runner.refCount > 0 {
  281. // Shouldn't happen, but safeguard to ensure no leaked runners
  282. slog.Debug("expired event with positive ref count, retrying", "modelPath", runner.modelPath, "refCount", runner.refCount)
  283. go func(runner *runnerRef) {
  284. // We can't unload yet, but want to as soon as the current request completes
  285. // So queue up another expired event
  286. time.Sleep(10 * time.Millisecond)
  287. s.expiredCh <- runner
  288. }(runner)
  289. runner.refMu.Unlock()
  290. continue
  291. }
  292. s.loadedMu.Lock()
  293. slog.Debug("got lock to unload", "modelPath", runner.modelPath)
  294. finished := runner.waitForVRAMRecovery()
  295. runner.unload()
  296. delete(s.loaded, runner.modelPath)
  297. s.loadedMu.Unlock()
  298. slog.Debug("runner released", "modelPath", runner.modelPath)
  299. runner.refMu.Unlock()
  300. <-finished
  301. slog.Debug("sending an unloaded event", "modelPath", runner.modelPath)
  302. s.unloadedCh <- struct{}{}
  303. }
  304. }
  305. }
  306. // Complete the pending request and send the runner back to the requester
  307. // Wires up a finished event after the request context is completed
  308. // Updates session duration, and resets expiration timer
  309. func (pending *LlmRequest) useLoadedRunner(runner *runnerRef, finished chan *LlmRequest) {
  310. runner.refMu.Lock()
  311. defer runner.refMu.Unlock()
  312. runner.refCount++
  313. if runner.expireTimer != nil {
  314. runner.expireTimer.Stop()
  315. runner.expireTimer = nil
  316. }
  317. runner.sessionDuration = pending.sessionDuration
  318. pending.successCh <- runner
  319. go func() {
  320. <-pending.ctx.Done()
  321. slog.Debug("context for request finished")
  322. finished <- pending
  323. }()
  324. }
  325. func (s *Scheduler) load(req *LlmRequest, ggml *llm.GGML, gpus gpu.GpuInfoList) {
  326. llama, err := s.newServerFn(gpus, req.model.ModelPath, ggml, req.model.AdapterPaths, req.model.ProjectorPaths, req.opts)
  327. if err != nil {
  328. // some older models are not compatible with newer versions of llama.cpp
  329. // show a generalized compatibility error until there is a better way to
  330. // check for model compatibility
  331. if errors.Is(llm.ErrUnsupportedFormat, err) || strings.Contains(err.Error(), "failed to load model") {
  332. err = fmt.Errorf("%v: this model may be incompatible with your version of Ollama. If you previously pulled this model, try updating it by running `ollama pull %s`", err, req.model.ShortName)
  333. }
  334. slog.Info("NewLlamaServer failed", "model", req.model.ModelPath, "error", err)
  335. req.errCh <- err
  336. return
  337. }
  338. runner := &runnerRef{
  339. model: req.model,
  340. modelPath: req.model.ModelPath,
  341. llama: llama,
  342. Options: &req.opts,
  343. sessionDuration: req.sessionDuration,
  344. gpus: gpus,
  345. estimatedVRAM: llama.EstimatedVRAM(),
  346. estimatedTotal: llama.EstimatedTotal(),
  347. loading: true,
  348. refCount: 1,
  349. }
  350. runner.refMu.Lock()
  351. s.loadedMu.Lock()
  352. s.loaded[req.model.ModelPath] = runner
  353. slog.Info("loaded runners", "count", len(s.loaded))
  354. s.loadedMu.Unlock()
  355. go func() {
  356. defer runner.refMu.Unlock()
  357. if err = llama.WaitUntilRunning(req.ctx); err != nil {
  358. slog.Error("error loading llama server", "error", err)
  359. runner.refCount--
  360. req.errCh <- err
  361. slog.Debug("triggering expiration for failed load", "model", runner.modelPath)
  362. s.expiredCh <- runner
  363. return
  364. }
  365. slog.Debug("finished setting up runner", "model", req.model.ModelPath)
  366. runner.loading = false
  367. go func() {
  368. <-req.ctx.Done()
  369. slog.Debug("context for request finished")
  370. s.finishedReqCh <- req
  371. }()
  372. req.successCh <- runner
  373. }()
  374. }
  375. func (s *Scheduler) updateFreeSpace(allGpus gpu.GpuInfoList) {
  376. type predKey struct {
  377. Library string
  378. ID string
  379. }
  380. predMap := map[predKey]uint64{} // Sum up the total predicted usage per GPU for all runners
  381. s.loadedMu.Lock()
  382. for _, r := range s.loaded {
  383. r.refMu.Lock()
  384. if r.llama != nil {
  385. for _, gpu := range allGpus {
  386. predMap[predKey{gpu.Library, gpu.ID}] += r.llama.EstimatedVRAMByGPU(gpu.ID)
  387. }
  388. } else {
  389. slog.Warn("unexpected nil runner reference, memory prediction may be incorrect")
  390. }
  391. r.refMu.Unlock()
  392. }
  393. s.loadedMu.Unlock()
  394. // Now that we've summed up all the GPU usage predictions across all the loaded runners, update the gpu list
  395. for i := range allGpus {
  396. if p, ok := predMap[predKey{allGpus[i].Library, allGpus[i].ID}]; ok {
  397. slog.Debug("gpu reported", "gpu", allGpus[i].ID, "library", allGpus[i].Library, "available", format.HumanBytes2(allGpus[i].FreeMemory))
  398. if p > allGpus[i].TotalMemory {
  399. // Shouldn't happen
  400. slog.Warn("predicted usage exceeds VRAM", "gpu", allGpus[i].ID, "totalMemory", allGpus[i].TotalMemory, "predicted", p)
  401. allGpus[i].FreeMemory = 0
  402. } else if (allGpus[i].TotalMemory - p) < allGpus[i].FreeMemory { // predicted free is smaller than reported free, use it
  403. // TODO maybe we should just always trust our numbers, since cuda's free memory reporting is laggy
  404. // and we might unload models we didn't actually need to. The risk is if some other GPU intensive app is loaded
  405. // after we start our first runner, then we'll never acount for that, so picking the smallest free value seems prudent.
  406. allGpus[i].FreeMemory = allGpus[i].TotalMemory - p
  407. }
  408. slog.Info("updated VRAM based on existing loaded models", "gpu", allGpus[i].ID, "library", allGpus[i].Library, "total", format.HumanBytes2(allGpus[i].TotalMemory), "available", format.HumanBytes2(allGpus[i].FreeMemory))
  409. }
  410. }
  411. }
  412. // While models are loading the VRAM consumption numbers will be indeterminate, so we have
  413. // to avoid scheduling another model on the same GPU(s) that haven't stabilized.
  414. // This routine returns the set of GPUs that do not have an active loading model.
  415. // If all GPUs have loading models, an empty list will be returned (not a single CPU entry)
  416. func (s *Scheduler) filterGPUsWithoutLoadingModels(allGpus gpu.GpuInfoList) gpu.GpuInfoList {
  417. ret := append(gpu.GpuInfoList{}, allGpus...)
  418. s.loadedMu.Lock()
  419. defer s.loadedMu.Unlock()
  420. for _, runner := range s.loaded {
  421. if runner.loading {
  422. slog.Debug("overlapping loads detected", "gpus", runner.gpus, "model", runner.modelPath)
  423. for _, busyGPU := range runner.gpus {
  424. for i := range ret {
  425. if ret[i].ID == busyGPU.ID {
  426. ret = append(ret[:i], ret[i+1:]...)
  427. break
  428. }
  429. }
  430. }
  431. }
  432. }
  433. return ret
  434. }
  435. // TODO consolidate sched_types.go
  436. type runnerRef struct {
  437. refMu sync.Mutex
  438. // refCond sync.Cond // Signaled on transition from 1 -> 0 refCount
  439. refCount uint // prevent unloading if > 0
  440. // unloading bool // set to true when we are trying to unload the runner
  441. llama llm.LlamaServer
  442. loading bool // True only during initial load, then false forever
  443. gpus gpu.GpuInfoList // Recorded at time of provisioning
  444. estimatedVRAM uint64
  445. estimatedTotal uint64
  446. sessionDuration time.Duration
  447. expireTimer *time.Timer
  448. expiresAt time.Time
  449. model *Model
  450. modelPath string
  451. *api.Options
  452. }
  453. // The refMu must already be held when calling unload
  454. func (runner *runnerRef) unload() {
  455. if runner.expireTimer != nil {
  456. runner.expireTimer.Stop()
  457. runner.expireTimer = nil
  458. }
  459. if runner.llama != nil {
  460. runner.llama.Close()
  461. }
  462. runner.model = nil
  463. runner.llama = nil
  464. runner.Options = nil
  465. runner.gpus = nil
  466. }
  467. func (runner *runnerRef) needsReload(ctx context.Context, req *LlmRequest) bool {
  468. slog.Debug("evaluating already loaded", "model", req.model.ModelPath)
  469. runner.refMu.Lock()
  470. defer runner.refMu.Unlock()
  471. timeout := 10 * time.Second
  472. if runner.loading {
  473. timeout = 2 * time.Minute // Initial load can take a long time for big models on slow systems...
  474. }
  475. if runner.Options == nil {
  476. return true
  477. }
  478. // Don't reload runner if num_gpu=-1 was provided
  479. optsExisting := runner.Options.Runner
  480. optsNew := req.opts.Runner
  481. if optsNew.NumGPU < 0 {
  482. optsExisting.NumGPU = -1
  483. optsNew.NumGPU = -1
  484. }
  485. ctx, cancel := context.WithTimeout(ctx, timeout)
  486. defer cancel()
  487. if !reflect.DeepEqual(runner.model.AdapterPaths, req.model.AdapterPaths) || // have the adapters changed?
  488. !reflect.DeepEqual(runner.model.ProjectorPaths, req.model.ProjectorPaths) || // have the projectors changed?
  489. !reflect.DeepEqual(optsExisting, optsNew) || // have the runner options changed?
  490. runner.llama.Ping(ctx) != nil {
  491. return true
  492. }
  493. return false
  494. }
  495. // Free memory reporting on GPUs can lag for a while even after the runner
  496. // exits, so we have to keep checking until we see the available memory recover,
  497. // otherwise subsequent model loads will get far less layers loaded or worse
  498. // case, may completely fall back to CPU mode.
  499. // This routine must be called before the runner unloads so it can establish
  500. // a before and after GPU memory allocation. The returned channel
  501. // will be notified when we're done waiting, or have timed out and should
  502. // proceed anyway
  503. func (runner *runnerRef) waitForVRAMRecovery() chan interface{} {
  504. finished := make(chan interface{}, 1)
  505. // CPU or Metal don't need checking, so no waiting required
  506. // windows can page VRAM, only cuda currently can report accurate used vram usage
  507. if len(runner.gpus) == 0 ||
  508. (len(runner.gpus) == 1 && (runner.gpus[0].Library == "cpu" || runner.gpus[0].Library == "metal")) ||
  509. (runtime.GOOS == "windows" && runner.gpus[0].Library != "cuda") {
  510. finished <- struct{}{}
  511. return finished
  512. }
  513. start := time.Now()
  514. // Establish a baseline before we unload
  515. gpusBefore := gpu.GetGPUInfo()
  516. var totalMemoryBefore, freeMemoryBefore uint64
  517. for _, gpu := range gpusBefore {
  518. totalMemoryBefore += gpu.TotalMemory
  519. freeMemoryBefore += gpu.FreeMemory
  520. }
  521. go func() {
  522. expiresAt := start.Add(5 * time.Second) // typical convergence is 0.5-1.5s
  523. ticker := time.NewTicker(250 * time.Millisecond)
  524. defer ticker.Stop()
  525. for {
  526. <-ticker.C
  527. if time.Now().After(expiresAt) {
  528. slog.Warn("gpu VRAM usage didn't recover within timeout", "seconds", time.Since(start).Seconds(), "model", runner.modelPath)
  529. finished <- struct{}{}
  530. }
  531. // Query GPUs, look for free to go back up
  532. gpusNow := gpu.GetGPUInfo()
  533. var totalMemoryNow, freeMemoryNow uint64
  534. for _, gpu := range gpusNow {
  535. totalMemoryNow += gpu.TotalMemory
  536. freeMemoryNow += gpu.FreeMemory
  537. }
  538. // If we're within ~80% of the estimated memory usage recovered, bail out
  539. if float32(freeMemoryNow-freeMemoryBefore) > float32(runner.estimatedVRAM)*0.8 {
  540. slog.Debug(fmt.Sprintf("gpu VRAM free memory converged after %0.2f seconds", time.Since(start).Seconds()), "model", runner.modelPath)
  541. finished <- struct{}{}
  542. return
  543. }
  544. }
  545. }()
  546. return finished
  547. }
  548. type ByDuration []*runnerRef
  549. func (a ByDuration) Len() int { return len(a) }
  550. func (a ByDuration) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
  551. func (a ByDuration) Less(i, j int) bool {
  552. // uint64 to turn negative time (never unload) to largest
  553. return uint64(a[i].sessionDuration) < uint64(a[j].sessionDuration)
  554. }
  555. // TODO - future consideration to pick runners based on size
  556. // type BySize []*runnerRef
  557. // func (a BySize) Len() int { return len(a) }
  558. // func (a BySize) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
  559. // func (a BySize) Less(i, j int) bool { return a[i].estimatedVRAM < a[j].estimatedVRAM }
  560. // pickBestFitGPUs will try to find the optimal placement of the model in the available GPUs where the model fully fits
  561. // If the model can not be fit fully within the available GPU(s) nil is returned
  562. func pickBestFitGPUs(req *LlmRequest, ggml *llm.GGML, gpus gpu.GpuInfoList) gpu.GpuInfoList {
  563. var estimatedVRAM uint64
  564. for _, gl := range gpus.ByLibrary() {
  565. var ok bool
  566. sgl := append(make(gpu.GpuInfoList, 0, len(gl)), gl...)
  567. // TODO - potentially sort by performance capability, existing models loaded, etc.
  568. // Note: at present, this will favor more VRAM over faster GPU speed in mixed setups
  569. sort.Sort(sort.Reverse(gpu.ByFreeMemory(sgl)))
  570. // First attempt to fit the model into a single GPU
  571. if !envconfig.SchedSpread {
  572. for _, g := range sgl {
  573. if ok, estimatedVRAM = llm.PredictServerFit([]gpu.GpuInfo{g}, ggml, req.model.AdapterPaths, req.model.ProjectorPaths, req.opts); ok {
  574. slog.Debug("new model will fit in available VRAM in single GPU, loading", "model", req.model.ModelPath, "gpu", g.ID, "available", g.FreeMemory, "required", format.HumanBytes2(estimatedVRAM))
  575. return []gpu.GpuInfo{g}
  576. }
  577. }
  578. }
  579. // TODO future refinements
  580. // - if multiple Libraries, see if any single GPU in any Library will fit
  581. // - try subsets of GPUs instead of just falling back to 1 or all in a family
  582. // Now try all the GPUs
  583. if ok, estimatedVRAM = llm.PredictServerFit(sgl, ggml, req.model.AdapterPaths, req.model.ProjectorPaths, req.opts); ok {
  584. slog.Debug("new model will fit in available VRAM, loading", "model", req.model.ModelPath, "library", sgl[0].Library, "required", format.HumanBytes2(estimatedVRAM))
  585. return sgl
  586. }
  587. }
  588. return nil
  589. }
  590. // findRunnerToUnload finds a runner to unload to make room for a new model
  591. func (s *Scheduler) findRunnerToUnload() *runnerRef {
  592. s.loadedMu.Lock()
  593. runnerList := make([]*runnerRef, 0, len(s.loaded))
  594. for _, r := range s.loaded {
  595. runnerList = append(runnerList, r)
  596. }
  597. s.loadedMu.Unlock()
  598. if len(runnerList) == 0 {
  599. slog.Debug("no loaded runner to unload")
  600. return nil
  601. }
  602. // In the future we can enhance the algorithm to be smarter about picking the optimal runner to unload
  603. // e.g., if we have multiple options, will one make room for the request?
  604. sort.Sort(ByDuration(runnerList))
  605. // First try to find a runner that's already idle
  606. for _, runner := range runnerList {
  607. runner.refMu.Lock()
  608. rc := runner.refCount
  609. runner.refMu.Unlock()
  610. if rc == 0 {
  611. slog.Debug("found an idle runner to unload")
  612. return runner
  613. }
  614. }
  615. // None appear idle, just wait for the one with the shortest duration
  616. slog.Debug("no idle runners, picking the shortest duration", "count", len(runnerList))
  617. return runnerList[0]
  618. }
  619. func (s *Scheduler) unloadAllRunners() {
  620. s.loadedMu.Lock()
  621. defer s.loadedMu.Unlock()
  622. for model, runner := range s.loaded {
  623. if runner.llama != nil {
  624. slog.Debug("shutting down runner", "model", model)
  625. runner.llama.Close()
  626. }
  627. }
  628. }
  629. // If other runners are loaded, make sure the pending request will fit in system memory
  630. // If not, pick a runner to unload, else return nil and the request can be loaded
  631. func (s *Scheduler) maybeFindCPURunnerToUnload(req *LlmRequest, ggml *llm.GGML, gpus gpu.GpuInfoList) *runnerRef {
  632. slog.Debug("evaluating if CPU model load will fit in available system memory")
  633. estimate := llm.EstimateGPULayers(gpus, ggml, req.model.ProjectorPaths, req.opts)
  634. if estimate.TotalSize <= gpus[0].FreeMemory {
  635. slog.Debug("cpu inference mode, model fits in available system memory", "model", format.HumanBytes2(estimate.TotalSize), "available", format.HumanBytes2(gpus[0].FreeMemory))
  636. return nil
  637. }
  638. // TODO - optimization: try to find CPU only runners first, or partial offloads with enough in system memory to make room
  639. return s.findRunnerToUnload()
  640. }