dap: add stack traces with next and continue functionality

It is now possible to send next and continue as separate signals. When
executing a build, the debug adapter will divide the LLB graph into
regions. Each region corresponds to an uninterrupted chain of
instructions. It will also record which regions depend on which other
ones.

This determines the execution order and it also determines what the
stack traces look like.

When continue is used, we will attempt to evaluate the last leaf node
(the head). If we push next, we will determine which digest would be the
next one to be processed.

In the future, this will be used to also support breakpoints.

Signed-off-by: Jonathan A. Sternberg <jonathan.sternberg@docker.com>

Author: jsternberg

dap/adapter.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"
 
 	"github.com/docker/buildx/build"
 	"github.com/google/go-dap"
@@ -28,6 +29,8 @@ type Adapter[T any] struct {
 	threads      map[int]*thread
 	threadsMu    sync.RWMutex
 	nextThreadID int
+
+	idPool *idPool
 }
 
 func New[T any](cfg *build.InvokeConfig) *Adapter[T] {
@@ -38,6 +41,7 @@ func New[T any](cfg *build.InvokeConfig) *Adapter[T] {
 		evaluateReqCh: make(chan *evaluateRequest),
 		threads:       make(map[int]*thread),
 		nextThreadID:  1,
+		idPool:        new(idPool),
 	}
 	if cfg != nil {
 		d.cfg = *cfg
@@ -131,7 +135,16 @@ func (d *Adapter[T]) Continue(c Context, req *dap.ContinueRequest, resp *dap.Con
 	t := d.threads[req.Arguments.ThreadId]
 	d.threadsMu.RUnlock()
 
-	t.Resume(c)
+	t.Continue(c)
+	return nil
+}
+
+func (d *Adapter[T]) Next(c Context, req *dap.NextRequest, resp *dap.NextResponse) error {
+	d.threadsMu.RLock()
+	t := d.threads[req.Arguments.ThreadId]
+	d.threadsMu.RUnlock()
+
+	t.Next(c)
 	return nil
 }
 
@@ -182,7 +195,7 @@ func (d *Adapter[T]) launch(c Context) {
 			started := c.Go(func(c Context) {
 				defer d.deleteThread(c, t)
 				defer close(req.errCh)
-				req.errCh <- t.Evaluate(c, req.c, req.ref, req.meta, d.cfg)
+				req.errCh <- t.Evaluate(c, req.c, req.ref, req.meta)
 			})
 
 			if !started {
@@ -197,8 +210,9 @@ func (d *Adapter[T]) newThread(ctx Context, name string) (t *thread) {
 	d.threadsMu.Lock()
 	id := d.nextThreadID
 	t = &thread{
-		id:   id,
-		name: name,
+		id:     id,
+		name:   name,
+		idPool: d.idPool,
 	}
 	d.threads[t.id] = t
 	d.nextThreadID++
@@ -307,7 +321,7 @@ func (d *Adapter[T]) StackTrace(c Context, req *dap.StackTraceRequest, resp *dap
 		return errors.Errorf("no such thread: %d", req.Arguments.ThreadId)
 	}
 
-	resp.Body.StackFrames = t.StackFrames()
+	resp.Body.StackFrames = t.StackTrace()
 	return nil
 }
 
@@ -341,6 +355,7 @@ func (d *Adapter[T]) dapHandler() Handler {
 		Initialize:        d.Initialize,
 		Launch:            d.Launch,
 		Continue:          d.Continue,
+		Next:              d.Next,
 		SetBreakpoints:    d.SetBreakpoints,
 		ConfigurationDone: d.ConfigurationDone,
 		Disconnect:        d.Disconnect,
@@ -349,85 +364,6 @@ func (d *Adapter[T]) dapHandler() Handler {
 	}
 }
 
-type thread struct {
-	id   int
-	name string
-
-	paused chan struct{}
-	rCtx   *build.ResultHandle
-	mu     sync.Mutex
-}
-
-func (t *thread) Evaluate(ctx Context, c gateway.Client, ref gateway.Reference, meta map[string][]byte, cfg build.InvokeConfig) error {
-	err := ref.Evaluate(ctx)
-	if reason, desc := t.needsDebug(cfg, err); reason != "" {
-		rCtx := build.NewResultHandle(ctx, c, ref, meta, err)
-
-		select {
-		case <-t.pause(ctx, rCtx, reason, desc):
-		case <-ctx.Done():
-			t.Resume(ctx)
-			return context.Cause(ctx)
-		}
-	}
-	return err
-}
-
-func (t *thread) needsDebug(cfg build.InvokeConfig, err error) (reason, desc string) {
-	if !cfg.NeedsDebug(err) {
-		return
-	}
-
-	if err != nil {
-		reason = "exception"
-		desc = "Encountered an error during result evaluation"
-	} else {
-		reason = "pause"
-		desc = "Result evaluation completed"
-	}
-	return
-}
-
-func (t *thread) pause(c Context, rCtx *build.ResultHandle, reason, desc string) <-chan struct{} {
-	if t.paused == nil {
-		t.paused = make(chan struct{})
-	}
-	t.rCtx = rCtx
-
-	c.C() <- &dap.StoppedEvent{
-		Event: dap.Event{Event: "stopped"},
-		Body: dap.StoppedEventBody{
-			Reason:      reason,
-			Description: desc,
-			ThreadId:    t.id,
-		},
-	}
-	return t.paused
-}
-
-func (t *thread) Resume(c Context) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-
-	if t.paused == nil {
-		return
-	}
-
-	if t.rCtx != nil {
-		t.rCtx.Done()
-		t.rCtx = nil
-	}
-
-	close(t.paused)
-	t.paused = nil
-}
-
-// TODO: return a suitable stack frame for the thread.
-// For now, just returns nothing.
-func (t *thread) StackFrames() []dap.StackFrame {
-	return []dap.StackFrame{}
-}
-
 func (d *Adapter[T]) Out() io.Writer {
 	return &adapterWriter[T]{d}
 }
@@ -454,3 +390,15 @@ func (d *adapterWriter[T]) Write(p []byte) (n int, err error) {
 	}
 	return n, nil
 }
+
+type idPool struct {
+	next atomic.Int64
+}
+
+func (p *idPool) Get() int64 {
+	return p.next.Add(1)
+}
+
+func (p *idPool) Put(x int64) {
+	// noop
+}

dap/handler.go

@@ -51,6 +51,7 @@ type Handler struct {
 	Disconnect        HandlerFunc[*dap.DisconnectRequest, *dap.DisconnectResponse]
 	Terminate         HandlerFunc[*dap.TerminateRequest, *dap.TerminateResponse]
 	Continue          HandlerFunc[*dap.ContinueRequest, *dap.ContinueResponse]
+	Next              HandlerFunc[*dap.NextRequest, *dap.NextResponse]
 	Restart           HandlerFunc[*dap.RestartRequest, *dap.RestartResponse]
 	Threads           HandlerFunc[*dap.ThreadsRequest, *dap.ThreadsResponse]
 	StackTrace        HandlerFunc[*dap.StackTraceRequest, *dap.StackTraceResponse]

dap/server.go

@@ -117,6 +117,8 @@ func (s *Server) handleMessage(c Context, m dap.Message) (dap.ResponseMessage, e
 		return s.h.Terminate.Do(c, req)
 	case *dap.ContinueRequest:
 		return s.h.Continue.Do(c, req)
+	case *dap.NextRequest:
+		return s.h.Next.Do(c, req)
 	case *dap.RestartRequest:
 		return s.h.Restart.Do(c, req)
 	case *dap.ThreadsRequest: