Name and order the build queue code

[onechat] / memory.go

diff --git a/memory.go b/memory.go
index c3e9f454177837e6352f35b098a1dc53a0d73ad9..6937726b36de7998e18617d964ccc8fe77157e8f 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -39,6 +39,10 @@ type Task struct {
 // start. It never builds the views again from the log, because a new view
 // differs from the live one and so loses the prompt cache.
 //
 // start. It never builds the views again from the log, because a new view
 // differs from the live one and so loses the prompt cache.
 //

+// Each write goes to the operating system at once, and a syncer makes the
+// writes durable by group commit in the background. Flush waits until the
+// writes so far are durable.
+//

 // The summarizer builds the nodes in the queue in the background. The start
 // reads the indexes once to fill the queue again. After that, only a new
 // message or a new node adds work. Memory keeps the nodes of the views and the
 // The summarizer builds the nodes in the queue in the background. The start
 // reads the indexes once to fill the queue again. After that, only a new
 // message or a new node adds work. Memory keeps the nodes of the views and the


@@ -48,10 +52,11 @@ type Memory struct {
        mutex      sync.Mutex
        root       string
        log        *Log
        mutex      sync.Mutex
        root       string
        log        *Log

+       syncer     *syncer

        tree       *Tree
        chat       *View
        cview      *View
        tree       *Tree
        chat       *View
        cview      *View

-       queue      *queue
+       queue      *BuildQueue

        summarizer Summarizer
        ctx        context.Context
        cancel     context.CancelFunc
        summarizer Summarizer
        ctx        context.Context
        cancel     context.CancelFunc


@@ -61,8 +66,8 @@ type Memory struct {
 
 // viewFile is the content of view.json.
 type viewFile struct {
 
 // viewFile is the content of view.json.
 type viewFile struct {

-       Chat       viewState `json:"chat"`
-       Compaction viewState `json:"compaction"`
+       Chat       ViewState `json:"chat"`
+       Compaction ViewState `json:"compaction"`

 }
 
 // openMemory opens the chat in root and starts the builds that are ready. The
 }
 
 // openMemory opens the chat in root and starts the builds that are ready. The


@@ -77,13 +82,18 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
        m := &Memory{
                root:       root,
                log:        log,
        m := &Memory{
                root:       root,
                log:        log,

-               queue:      newQueue(),
+               queue:      newBuildQueue(),

                summarizer: summarizer,
                changed:    make(chan struct{}),
        }
 
        m.ctx, m.cancel = context.WithCancel(ctx)
 
                summarizer: summarizer,
                changed:    make(chan struct{}),
        }
 
        m.ctx, m.cancel = context.WithCancel(ctx)
 

+       m.syncer, err = openSyncer(root)
+       if err != nil {
+               return nil, errors.Join(err, m.Close())
+       }
+

        m.tree, err = openTree(ctx, root, log.Len(), clock)
        if err != nil {
                return nil, errors.Join(err, m.Close())
        m.tree, err = openTree(ctx, root, log.Len(), clock)
        if err != nil {
                return nil, errors.Join(err, m.Close())


@@ -105,6 +115,7 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
        }
 
        m.mutex.Lock()
        }
 
        m.mutex.Lock()

+       m.syncer.Written()

        m.dispatch()
        m.mutex.Unlock()
 
        m.dispatch()
        m.mutex.Unlock()
 


@@ -115,8 +126,8 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
 func (m *Memory) load() error {
        data, err := os.ReadFile(m.viewPath())
        if errors.Is(err, os.ErrNotExist) && m.log.Len() == 0 {
 func (m *Memory) load() error {
        data, err := os.ReadFile(m.viewPath())
        if errors.Is(err, os.ErrNotExist) && m.log.Len() == 0 {

-               m.chat = newView(chatHigh, chatLow)
-               m.cview = newView(compactionHigh, compactionLow)
+               m.chat = newView(viewChatHigh, viewChatLow)
+               m.cview = newView(viewCompactionHigh, viewCompactionLow)

 
                return nil
        }
 
                return nil
        }


@@ -135,12 +146,12 @@ func (m *Memory) load() error {
                return fmt.Errorf("decode view: %w", err)
        }
 
                return fmt.Errorf("decode view: %w", err)
        }
 

-       m.chat, err = restoreView(file.Chat, chatHigh, chatLow, m.tree)
+       m.chat, err = restoreView(file.Chat, viewChatHigh, viewChatLow, m.tree)

        if err != nil {
                return fmt.Errorf("chat view: %w", err)
        }
 
        if err != nil {
                return fmt.Errorf("chat view: %w", err)
        }
 

-       m.cview, err = restoreView(file.Compaction, compactionHigh, compactionLow, m.tree)
+       m.cview, err = restoreView(file.Compaction, viewCompactionHigh, viewCompactionLow, m.tree)

        if err != nil {
                return fmt.Errorf("compaction view: %w", err)
        }
        if err != nil {
                return fmt.Errorf("compaction view: %w", err)
        }


@@ -227,8 +238,8 @@ func (m *Memory) catchUp() error {
        return m.save()
 }
 
        return m.save()
 }
 

-// Close stops the builds, waits for them, closes the files of the chat, and
-// releases its lock.
+// Close stops the builds, waits for them, makes the writes durable, closes the
+// files of the chat, and releases its lock.

 func (m *Memory) Close() error {
        m.cancel()
        m.workers.Wait()
 func (m *Memory) Close() error {
        m.cancel()
        m.workers.Wait()


@@ -237,8 +248,12 @@ func (m *Memory) Close() error {
        defer m.mutex.Unlock()
 
        var err error
        defer m.mutex.Unlock()
 
        var err error

+       if m.syncer != nil {
+               err = m.syncer.Close()
+       }
+

        if m.tree != nil {
        if m.tree != nil {

-               err = m.tree.Close()
+               err = errors.Join(err, m.tree.Close())

        }
 
        return errors.Join(err, m.log.Close())
        }
 
        return errors.Join(err, m.log.Close())


@@ -255,6 +270,8 @@ func (m *Memory) Append(kind, text string) (Message, error) {
                return Message{}, err
        }
 
                return Message{}, err
        }
 

+       defer m.syncer.Written()
+

        err = m.addMessage(message)
        if err != nil {
                return Message{}, err
        err = m.addMessage(message)
        if err != nil {
                return Message{}, err


@@ -457,6 +474,7 @@ func (m *Memory) run(key nodeKey, task Task) {
                        Text:  text,
                        Size:  len(text),
                })
                        Text:  text,
                        Size:  len(text),
                })

+               m.syncer.Written()

        }
 
        if err != nil {
        }
 
        if err != nil {


@@ -490,6 +508,11 @@ func (m *Memory) Wait(ctx context.Context) error {
        }
 }
 
        }
 }
 

+// Flush returns when every write before the call is durable.
+func (m *Memory) Flush(ctx context.Context) error {
+       return m.syncer.Flush(ctx)
+}
+

 // save writes the views to view.json.
 func (m *Memory) save() error {
        data, err := json.Marshal(viewFile{
 // save writes the views to view.json.
 func (m *Memory) save() error {
        data, err := json.Marshal(viewFile{


@@ -520,6 +543,39 @@ func (m *Memory) View() string {
        return m.chat.Render()
 }
 
        return m.chat.Render()
 }
 

+// ViewLines returns the lines of the chat view, each with its newline, without
+// the leaf lines of the messages in skip.
+func (m *Memory) ViewLines(skip map[int]bool) []string {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.chat.LinesWithout(skip)
+}
+
+// Len returns the number of messages in the log.
+func (m *Memory) Len() int {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.log.Len()
+}
+
+// Message reads the message with the given ID.
+func (m *Memory) Message(id int) (Message, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.log.Message(id)
+}
+
+// Node reads the node (l, i) if it is built.
+func (m *Memory) Node(level, index int) (Node, bool, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       return m.tree.Node(level, index)
+}
+

 // Context returns the compaction view for the build of node (l, i). It ends
 // before the message of a leaf, or after the last message of a merge, and it
 // stops at the first line that is not built.
 // Context returns the compaction view for the build of node (l, i). It ends
 // before the message of a leaf, or after the last message of a merge, and it
 // stops at the first line that is not built.