runner.go: Retry decoding after defragmentation if needed

Fragmentation of the KV cache can occur due to cache shifting or
different sequences getting processed. Decode uses a heuristic to
decide if it should defrag. However, this heuristic isn't 100%
accurate, so decoding can sometimes fail by surprise.

For these cases, if decode indicates that there is no KV cache space,
we should defrag and then try again.
This commit is contained in:
Jesse Gross 2024-11-19 11:00:41 -08:00 committed by Jesse Gross
parent 5f68fcab12
commit 7121dfa309
3 changed files with 50 additions and 6 deletions

View file

@ -10,7 +10,38 @@ import (
"github.com/ollama/ollama/api" "github.com/ollama/ollama/api"
) )
func TestLongInputContext(t *testing.T) {
// Setting NUM_PARALLEL to 1 ensures the allocated context is exactly what
// we asked for and there is nothing extra that we could spill over into
t.Setenv("OLLAMA_NUM_PARALLEL", "1")
// Longer needed for small footprint GPUs
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
// Set up the test data
req := api.GenerateRequest{
Model: "llama2",
Prompt: "Oh, dont speak to me of Austria. Perhaps I dont understand things, but Austria never has wished, and does not wish, for war. She is betraying us! Russia alone must save Europe. Our gracious sovereign recognizes his high vocation and will be true to it. That is the one thing I have faith in! Our good and wonderful sovereign has to perform the noblest role on earth, and he is so virtuous and noble that God will not forsake him. He will fulfill his vocation and crush the hydra of revolution, which has become more terrible than ever in the person of this murderer and villain! We alone must avenge the blood of the just one.... Whom, I ask you, can we rely on?... England with her commercial spirit will not and cannot understand the Emperor Alexanders loftiness of soul. She has refused to evacuate Malta. She wanted to find, and still seeks, some secret motive in our actions. What answer did Novosíltsev get? None. The English have not understood and cannot understand the self-abnegation of our Emperor who wants nothing for himself, but only desires the good of mankind. And what have they promised? Nothing! And what little they have promised they will not perform! Prussia has always declared that Buonaparte is invincible, and that all Europe is powerless before him.... And I dont believe a word that Hardenburg says, or Haugwitz either. This famous Prussian neutrality is just a trap. I have faith only in God and the lofty destiny of our adored monarch. He will save Europe! What country is this referring to?",
Stream: &stream,
Options: map[string]interface{}{
"temperature": 0,
"seed": 123,
"num_ctx": 128,
},
}
client, _, cleanup := InitServerConnection(ctx, t)
defer cleanup()
if err := PullIfMissing(ctx, client, req.Model); err != nil {
t.Fatalf("PullIfMissing failed: %v", err)
}
DoGenerate(ctx, t, client, req, []string{"russia", "germany", "france", "england", "austria", "prussia"}, 120*time.Second, 10*time.Second)
}
func TestContextExhaustion(t *testing.T) { func TestContextExhaustion(t *testing.T) {
// Setting NUM_PARALLEL to 1 ensures the allocated context is exactly what
// we asked for and there is nothing extra that we could spill over into
t.Setenv("OLLAMA_NUM_PARALLEL", "1")
// Longer needed for small footprint GPUs // Longer needed for small footprint GPUs
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel() defer cancel()

View file

@ -157,9 +157,7 @@ type Context struct {
numThreads int numThreads int
} }
func (c *Context) KvCacheClear() { var ErrKvCacheFull = errors.New("could not find a kv cache slot")
C.llama_kv_cache_clear(c.c)
}
func (c *Context) Decode(batch *Batch) error { func (c *Context) Decode(batch *Batch) error {
// Positive return values does not mean a fatal error, but rather a warning. // Positive return values does not mean a fatal error, but rather a warning.
@ -173,7 +171,7 @@ func (c *Context) Decode(batch *Batch) error {
} }
if code > 0 { if code > 0 {
return fmt.Errorf("could not find a KV slot for the batch - try reducing the size of the batch or increase the context. code: %d", code) return ErrKvCacheFull
} }
return nil return nil
@ -195,6 +193,14 @@ func (c *Context) KvCacheSeqCp(srcSeqId int, dstSeqId int, p0 int, p1 int) {
C.llama_kv_cache_seq_cp(c.c, C.int(srcSeqId), C.int(dstSeqId), C.int(p0), C.int(p1)) C.llama_kv_cache_seq_cp(c.c, C.int(srcSeqId), C.int(dstSeqId), C.int(p0), C.int(p1))
} }
func (c *Context) KvCacheClear() {
C.llama_kv_cache_clear(c.c)
}
func (c *Context) KvCacheDefrag() {
C.llama_kv_cache_defrag(c.c)
}
// Get the embeddings for a sequence id // Get the embeddings for a sequence id
func (c *Context) GetEmbeddingsSeq(seqId int) []float32 { func (c *Context) GetEmbeddingsSeq(seqId int) []float32 {
embeddings := unsafe.Pointer(C.llama_get_embeddings_seq(c.c, C.int(seqId))) embeddings := unsafe.Pointer(C.llama_get_embeddings_seq(c.c, C.int(seqId)))

View file

@ -425,10 +425,17 @@ func (s *Server) processBatch(tokenBatch *llama.Batch, embedBatch *llama.Batch)
s.lc.SetCrossAttention(crossAttention) s.lc.SetCrossAttention(crossAttention)
err := s.lc.Decode(batch) err := s.lc.Decode(batch)
if err != nil {
if errors.Is(err, llama.ErrKvCacheFull) {
slog.Debug("defragmenting kv cache")
s.cache.lc.KvCacheDefrag()
err = s.lc.Decode(batch)
}
if err != nil { if err != nil {
slog.Error("failed to decode batch", "error", err) slog.Error("failed to decode batch", "error", err)
return return
} }
}
if crossAttention { if crossAttention {
// synchronize state to ensure the cross attention batch is complete. // synchronize state to ensure the cross attention batch is complete.