dap: implement first pass at breakpoints

Implement the first iteration of breakpoints. When a breakpoint is set,
it starts unverified. When a thread begins evaluation, it tries to see
if a breakpoint corresponds to one of the parsed instructions and will
verify it.

Breakpoints work when continue is used.

At the current moment, setting breakpoints while a thread is currently
running doesn't work. Breakpoints are rechecked each time execution is
about to restart.

Signed-off-by: Jonathan A. Sternberg <[email protected]>

Author: jsternberg

dap/adapter.go

@@ -7,13 +7,17 @@ import (
 	"fmt"
 	"io"
 	"path"
+	"path/filepath"
+	"slices"
 	"sync"
 	"sync/atomic"
 
 	"github.com/docker/buildx/build"
 	"github.com/docker/buildx/dap/common"
 	"github.com/google/go-dap"
 	gateway "github.com/moby/buildkit/frontend/gateway/client"
+	"github.com/moby/buildkit/solver/pb"
+	"github.com/opencontainers/go-digest"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
@@ -33,8 +37,9 @@ type Adapter[C LaunchConfig] struct {
 	threadsMu    sync.RWMutex
 	nextThreadID int
 
-	sourceMap sourceMap
-	idPool    *idPool
+	breakpointMap *breakpointMap
+	sourceMap     sourceMap
+	idPool        *idPool
 }
 
 func New[C LaunchConfig]() *Adapter[C] {
@@ -45,6 +50,7 @@ func New[C LaunchConfig]() *Adapter[C] {
 		evaluateReqCh: make(chan *evaluateRequest),
 		threads:       make(map[int]*thread),
 		nextThreadID:  1,
+		breakpointMap: newBreakpointMap(),
 		idPool:        new(idPool),
 	}
 	d.srv = NewServer(d.dapHandler())
@@ -151,14 +157,7 @@ func (d *Adapter[C]) Next(c Context, req *dap.NextRequest, resp *dap.NextRespons
 }
 
 func (d *Adapter[C]) SetBreakpoints(c Context, req *dap.SetBreakpointsRequest, resp *dap.SetBreakpointsResponse) error {
-	// TODO: implement breakpoints
-	for range req.Arguments.Breakpoints {
-		// Fail to create all breakpoints that were requested.
-		resp.Body.Breakpoints = append(resp.Body.Breakpoints, dap.Breakpoint{
-			Verified: false,
-			Message:  "breakpoints unsupported",
-		})
-	}
+	resp.Body.Breakpoints = d.breakpointMap.Set(req.Arguments.Source.Path, req.Arguments.Breakpoints)
 	return nil
 }
 
@@ -212,10 +211,11 @@ func (d *Adapter[C]) newThread(ctx Context, name string) (t *thread) {
 	d.threadsMu.Lock()
 	id := d.nextThreadID
 	t = &thread{
-		id:        id,
-		name:      name,
-		sourceMap: &d.sourceMap,
-		idPool:    d.idPool,
+		id:            id,
+		name:          name,
+		sourceMap:     &d.sourceMap,
+		breakpointMap: d.breakpointMap,
+		idPool:        d.idPool,
 	}
 	d.threads[t.id] = t
 	d.nextThreadID++
@@ -468,3 +468,100 @@ func (s *sourceMap) Get(fname string) ([]byte, bool) {
 	}
 	return v.([]byte), true
 }
+
+type breakpointMap struct {
+	byPath map[string][]dap.Breakpoint
+	mu     sync.RWMutex
+
+	nextID atomic.Int64
+}
+
+func newBreakpointMap() *breakpointMap {
+	return &breakpointMap{
+		byPath: make(map[string][]dap.Breakpoint),
+	}
+}
+
+func (b *breakpointMap) Set(fname string, sbps []dap.SourceBreakpoint) (breakpoints []dap.Breakpoint) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	prev := b.byPath[fname]
+	for _, sbp := range sbps {
+		index := slices.IndexFunc(prev, func(e dap.Breakpoint) bool {
+			return sbp.Line >= e.Line && sbp.Line <= e.EndLine && sbp.Column >= e.Column && sbp.Column <= e.EndColumn
+		})
+
+		var bp dap.Breakpoint
+		if index >= 0 {
+			bp = prev[index]
+		} else {
+			bp = dap.Breakpoint{
+				Id:        int(b.nextID.Add(1)),
+				Line:      sbp.Line,
+				EndLine:   sbp.Line,
+				Column:    sbp.Column,
+				EndColumn: sbp.Column,
+			}
+		}
+		breakpoints = append(breakpoints, bp)
+	}
+	b.byPath[fname] = breakpoints
+	return breakpoints
+}
+
+func (b *breakpointMap) Intersect(ctx Context, src *pb.Source, ws string) map[digest.Digest]int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	digests := make(map[digest.Digest]int)
+
+NEXT_DIGEST:
+	for dgst, locs := range src.Locations {
+		for _, loc := range locs.Locations {
+			if len(loc.Ranges) == 0 {
+				continue
+			}
+			r := loc.Ranges[0]
+
+			info := src.Infos[loc.SourceIndex]
+			fname := filepath.Join(ws, info.Filename)
+
+			bps := b.byPath[fname]
+			if len(bps) == 0 {
+				// No breakpoints for this file.
+				continue
+			}
+
+			for i, bp := range bps {
+				if !overlaps(r, &bp) {
+					continue
+				}
+
+				if !bp.Verified {
+					bp.Line = int(r.Start.Line)
+					bp.EndLine = int(r.End.Line)
+					bp.Column = int(r.Start.Character)
+					bp.EndColumn = int(r.End.Character)
+					bp.Verified = true
+
+					ctx.C() <- &dap.BreakpointEvent{
+						Event: dap.Event{Event: "breakpoint"},
+						Body: dap.BreakpointEventBody{
+							Reason:     "changed",
+							Breakpoint: bp,
+						},
+					}
+					bps[i] = bp
+				}
+				digests[digest.Digest(dgst)] = bp.Id
+				continue NEXT_DIGEST
+			}
+		}
+	}
+	return digests
+}
+
+func overlaps(r *pb.Range, bp *dap.Breakpoint) bool {
+	return r.Start.Line <= int32(bp.Line) && r.Start.Character <= int32(bp.Column) && r.End.Line >= int32(bp.EndLine) && r.End.Character >= int32(bp.EndColumn)
+}

dap/thread.go

@@ -23,8 +23,9 @@ type thread struct {
 	name string
 
 	// Persistent state from the adapter.
-	idPool    *idPool
-	sourceMap *sourceMap
+	idPool        *idPool
+	sourceMap     *sourceMap
+	breakpointMap *breakpointMap
 
 	// Inputs to the evaluate call.
 	c          gateway.Client
@@ -36,6 +37,7 @@ type thread struct {
 	def  *llb.Definition
 	ops  map[digest.Digest]*pb.Op
 	head digest.Digest
+	bps  map[digest.Digest]int
 
 	// Runtime state for the evaluate call.
 	regions         []*region
@@ -80,15 +82,18 @@ func (t *thread) Evaluate(ctx Context, c gateway.Client, ref gateway.Reference,
 	}
 
 	for {
+		if step == stepContinue {
+			t.setBreakpoints(ctx)
+		}
 		pos, err := t.seekNext(ctx, step)
 
-		reason, desc := t.needsDebug(pos, step, err)
-		if reason == "" {
+		event := t.needsDebug(pos, step, err)
+		if event.Reason == "" {
 			return err
 		}
 
 		select {
-		case step = <-t.pause(ctx, err, reason, desc):
+		case step = <-t.pause(ctx, err, event):
 		case <-ctx.Done():
 			return context.Cause(ctx)
 		}
@@ -100,7 +105,11 @@ func (t *thread) init(ctx Context, c gateway.Client, ref gateway.Reference, meta
 	t.ref = ref
 	t.meta = meta
 	t.sourcePath = inputs.ContextPath
-	return t.createRegions(ctx)
+
+	if err := t.getLLBState(ctx); err != nil {
+		return err
+	}
+	return t.createRegions()
 }
 
 func (t *thread) reset() {
@@ -111,17 +120,23 @@ func (t *thread) reset() {
 	t.ops = nil
 }
 
-func (t *thread) needsDebug(target digest.Digest, step stepType, err error) (reason, desc string) {
+func (t *thread) needsDebug(target digest.Digest, step stepType, err error) (e dap.StoppedEventBody) {
 	if err != nil {
-		reason = "exception"
-		desc = "Encountered an error during result evaluation"
-	} else if target != "" && step == stepNext {
-		reason = "step"
+		e.Reason = "exception"
+		e.Description = "Encountered an error during result evaluation"
+	} else if step == stepNext && target != "" {
+		e.Reason = "step"
+	} else if step == stepContinue {
+		if id, ok := t.bps[target]; ok {
+			e.Reason = "breakpoint"
+			e.Description = "Paused on breakpoint"
+			e.HitBreakpointIds = []int{id}
+		}
 	}
 	return
 }
 
-func (t *thread) pause(c Context, err error, reason, desc string) <-chan stepType {
+func (t *thread) pause(c Context, err error, event dap.StoppedEventBody) <-chan stepType {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
@@ -140,13 +155,10 @@ func (t *thread) pause(c Context, err error, reason, desc string) <-chan stepTyp
 		}
 	}
 
+	event.ThreadId = t.id
 	c.C() <- &dap.StoppedEvent{
 		Event: dap.Event{Event: "stopped"},
-		Body: dap.StoppedEventBody{
-			Reason:      reason,
-			Description: desc,
-			ThreadId:    t.id,
-		},
+		Body:  event,
 	}
 	return t.paused
 }
@@ -232,6 +244,10 @@ func (t *thread) getLLBState(ctx Context) error {
 	return err
 }
 
+func (t *thread) setBreakpoints(ctx Context) {
+	t.bps = t.breakpointMap.Intersect(ctx, t.def.Source, t.sourcePath)
+}
+
 func (t *thread) findBacklinks() map[digest.Digest]map[digest.Digest]struct{} {
 	backlinks := make(map[digest.Digest]map[digest.Digest]struct{})
 	for dgst := range t.ops {
@@ -249,11 +265,7 @@ func (t *thread) findBacklinks() map[digest.Digest]map[digest.Digest]struct{} {
 	return backlinks
 }
 
-func (t *thread) createRegions(ctx Context) error {
-	if err := t.getLLBState(ctx); err != nil {
-		return err
-	}
-
+func (t *thread) createRegions() error {
 	// Find the links going from inputs to their outputs.
 	// This isn't represented in the LLB graph but we need it to ensure
 	// an op only has one child and whether we are allowed to visit a node.
@@ -363,8 +375,11 @@ func (t *thread) seekNext(ctx Context, step stepType) (digest.Digest, error) {
 	}
 
 	target := t.head
-	if step == stepNext {
-		target = t.nextDigest()
+	switch step {
+	case stepNext:
+		target = t.nextDigest(nil)
+	case stepContinue:
+		target = t.continueDigest()
 	}
 
 	if target == "" {
@@ -394,11 +409,27 @@ func (t *thread) seek(ctx Context, target digest.Digest) (digest.Digest, error)
 	return t.curPos, err
 }
 
-func (t *thread) nextDigest() digest.Digest {
+func (t *thread) nextDigest(fn func(digest.Digest) bool) digest.Digest {
+	isValid := func(dgst digest.Digest) bool {
+		// Skip this digest because it has no locations in the source file.
+		if loc, ok := t.def.Source.Locations[string(dgst)]; !ok || len(loc.Locations) == 0 {
+			return false
+		}
+
+		// If a custom function has been set for validation, use it.
+		return fn == nil || fn(dgst)
+	}
+
 	// If we have no position, automatically select the first step.
 	if t.curPos == "" {
 		r := t.regions[len(t.regions)-1]
-		return r.digests[0]
+		if isValid(r.digests[0]) {
+			return r.digests[0]
+		}
+
+		// We cannot use the first position. Treat the first position as our
+		// current position so we can iterate.
+		t.curPos = r.digests[0]
 	}
 
 	// Look up the region associated with our current position.
@@ -426,15 +457,26 @@ func (t *thread) nextDigest() digest.Digest {
 		}
 
 		next := r.digests[i]
-		if loc, ok := t.def.Source.Locations[string(next)]; !ok || len(loc.Locations) == 0 {
-			// Skip this digest because it has no locations in the source file.
+		if !isValid(next) {
 			i++
 			continue
 		}
 		return next
 	}
 }
 
+func (t *thread) continueDigest() digest.Digest {
+	if len(t.bps) == 0 {
+		return t.head
+	}
+
+	fn := func(dgst digest.Digest) bool {
+		_, ok := t.bps[dgst]
+		return ok
+	}
+	return t.nextDigest(fn)
+}
+
 func (t *thread) solve(ctx context.Context, target digest.Digest) (gateway.Reference, error) {
 	if target == t.head {
 		return t.ref, nil