solver: pipe implementation utilizes generics for better typing

This updates the pipe library to use generics for the request payload
and the status value. This allows the solver to put in explicit types
rather than rely on type casting from interfaces which helps with type
safety and understandability.

The status value used by the solver uses the `any` type instead of an
explicit type because the `unpark` method takes a generic list of pipes
and the different pipes have different result types. We can likely
change this in the future or create a discriminated union for the
types that can be used in this package. That is left for future work
because at least the request payload is typed now.

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

Author: jsternberg

solver/edge.go

@@ -29,7 +29,7 @@ func newEdge(ed Edge, op activeOp, index *edgeIndex) *edge {
 	e := &edge{
 		edge:               ed,
 		op:                 op,
-		depRequests:        map[pipe.Receiver]*dep{},
+		depRequests:        map[pipeReceiver]*dep{},
 		keyMap:             map[string]struct{}{},
 		cacheRecords:       map[string]*CacheRecord{},
 		cacheRecordsLoaded: map[string]struct{}{},
@@ -65,14 +65,14 @@ type edge struct {
 	op   activeOp
 
 	edgeState
-	depRequests map[pipe.Receiver]*dep
+	depRequests map[pipeReceiver]*dep
 	deps        []*dep
 
-	cacheMapReq        pipe.Receiver
+	cacheMapReq        pipeReceiver
 	cacheMapDone       bool
 	cacheMapIndex      int
 	cacheMapDigests    []digest.Digest
-	execReq            pipe.Receiver
+	execReq            pipeReceiver
 	execCacheLoad      bool
 	err                error
 	cacheRecords       map[string]*CacheRecord
@@ -99,11 +99,11 @@ type edge struct {
 
 // dep holds state for a dependant edge
 type dep struct {
-	req pipe.Receiver
+	req pipeReceiver
 	edgeState
 	index             Index
 	keyMap            map[string]*CacheKey
-	slowCacheReq      pipe.Receiver
+	slowCacheReq      pipeReceiver
 	slowCacheComplete bool
 	slowCacheFoundKey bool
 	slowCacheKey      *ExportableCacheKey
@@ -122,7 +122,7 @@ func newDep(i Index) *dep {
 
 // edgePipe is a pipe for requests between two edges
 type edgePipe struct {
-	*pipe.Pipe
+	*pipe.Pipe[*edgeRequest, any]
 	From, Target *edge
 	mu           sync.Mutex
 }
@@ -198,21 +198,21 @@ func (e *edge) isComplete() bool {
 }
 
 // finishIncoming finalizes the incoming pipe request
-func (e *edge) finishIncoming(req pipe.Sender) {
+func (e *edge) finishIncoming(req pipeSender) {
 	err := e.err
 	if req.Request().Canceled && err == nil {
 		err = context.Canceled
 	}
 	if e.debug {
-		bklog.G(context.TODO()).Debugf("finishIncoming %s %v %#v desired=%s", e.edge.Vertex.Name(), err, e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
+		bklog.G(context.TODO()).Debugf("finishIncoming %s %v %#v desired=%s", e.edge.Vertex.Name(), err, e.edgeState, req.Request().Payload.desiredState)
 	}
 	req.Finalize(&e.edgeState, err)
 }
 
 // updateIncoming updates the current value of incoming pipe request
-func (e *edge) updateIncoming(req pipe.Sender) {
+func (e *edge) updateIncoming(req pipeSender) {
 	if e.debug {
-		bklog.G(context.TODO()).Debugf("updateIncoming %s %#v desired=%s", e.edge.Vertex.Name(), e.edgeState, req.Request().Payload.(*edgeRequest).desiredState)
+		bklog.G(context.TODO()).Debugf("updateIncoming %s %#v desired=%s", e.edge.Vertex.Name(), e.edgeState, req.Request().Payload.desiredState)
 	}
 	req.Update(&e.edgeState)
 }
@@ -353,7 +353,7 @@ func (e *edge) skipPhase2FastCache(dep *dep) bool {
 //     requests were not completed
 //  2. this function may not return outgoing requests if it has completed all
 //     incoming requests
-func (e *edge) unpark(incoming []pipe.Sender, updates, allPipes []pipe.Receiver, f *pipeFactory) {
+func (e *edge) unpark(incoming []pipeSender, updates, allPipes []pipeReceiver, f *pipeFactory) {
 	// process all incoming changes
 	depChanged := false
 	for _, upt := range updates {
@@ -414,7 +414,7 @@ func (e *edge) markFailed(f *pipeFactory, err error) {
 }
 
 // processUpdate is called by unpark for every updated pipe request
-func (e *edge) processUpdate(upt pipe.Receiver) (depChanged bool) {
+func (e *edge) processUpdate(upt pipeReceiver) (depChanged bool) {
 	// response for cachemap request
 	if upt == e.cacheMapReq && upt.Status().Completed {
 		if err := upt.Status().Err; err != nil {
@@ -719,7 +719,7 @@ func (e *edge) recalcCurrentState() {
 
 // respondToIncoming responds to all incoming requests. completing or
 // updating them when possible
-func (e *edge) respondToIncoming(incoming []pipe.Sender, allPipes []pipe.Receiver) (edgeStatusType, bool) {
+func (e *edge) respondToIncoming(incoming []pipeSender, allPipes []pipeReceiver) (edgeStatusType, bool) {
 	// detect the result state for the requests
 	allIncomingCanComplete := true
 	desiredState := e.state
@@ -731,7 +731,7 @@ func (e *edge) respondToIncoming(incoming []pipe.Sender, allPipes []pipe.Receive
 		for _, req := range incoming {
 			if !req.Request().Canceled {
 				allCanceled = false
-				if r := req.Request().Payload.(*edgeRequest); desiredState < r.desiredState {
+				if r := req.Request().Payload; desiredState < r.desiredState {
 					desiredState = r.desiredState
 					if e.hasActiveOutgoing || r.desiredState == edgeStatusComplete || r.currentKeys == len(e.keys) {
 						allIncomingCanComplete = false
@@ -757,7 +757,7 @@ func (e *edge) respondToIncoming(incoming []pipe.Sender, allPipes []pipe.Receive
 		}
 
 		// can close all but one requests
-		var leaveOpen pipe.Sender
+		var leaveOpen pipeSender
 		for _, req := range incoming {
 			if !req.Request().Canceled {
 				leaveOpen = req
@@ -784,7 +784,7 @@ func (e *edge) respondToIncoming(incoming []pipe.Sender, allPipes []pipe.Receive
 
 	// update incoming based on current state
 	for _, req := range incoming {
-		r := req.Request().Payload.(*edgeRequest)
+		r := req.Request().Payload
 		if req.Request().Canceled {
 			e.finishIncoming(req)
 		} else if !e.hasActiveOutgoing && e.state >= r.desiredState {
@@ -803,7 +803,7 @@ func (e *edge) createInputRequests(desiredState edgeStatusType, f *pipeFactory,
 
 	// initialize deps state
 	if e.deps == nil {
-		e.depRequests = make(map[pipe.Receiver]*dep)
+		e.depRequests = make(map[pipeReceiver]*dep)
 		e.deps = make([]*dep, 0, len(e.edge.Vertex.Inputs()))
 		for i := range e.edge.Vertex.Inputs() {
 			e.deps = append(e.deps, newDep(Index(i)))
@@ -842,13 +842,13 @@ func (e *edge) createInputRequests(desiredState edgeStatusType, f *pipeFactory,
 		if dep.state < desiredStateDep {
 			addNew := true
 			if dep.req != nil && !dep.req.Status().Completed {
-				if dep.req.Request().(*edgeRequest).desiredState != desiredStateDep {
+				if dep.req.Request().desiredState != desiredStateDep {
 					if e.debug {
 						bklog.G(context.TODO()).
 							WithField("edge_vertex_name", e.edge.Vertex.Name()).
 							WithField("edge_vertex_digest", e.edge.Vertex.Digest()).
 							WithField("dep_index", dep.index).
-							WithField("dep_req_desired_state", dep.req.Request().(*edgeRequest).desiredState).
+							WithField("dep_req_desired_state", dep.req.Request().desiredState).
 							WithField("dep_desired_state", desiredStateDep).
 							WithField("dep_state", dep.state).
 							Debug("cancel input request")
@@ -860,7 +860,7 @@ func (e *edge) createInputRequests(desiredState edgeStatusType, f *pipeFactory,
 							WithField("edge_vertex_name", e.edge.Vertex.Name()).
 							WithField("edge_vertex_digest", e.edge.Vertex.Digest()).
 							WithField("dep_index", dep.index).
-							WithField("dep_req_desired_state", dep.req.Request().(*edgeRequest).desiredState).
+							WithField("dep_req_desired_state", dep.req.Request().desiredState).
 							WithField("dep_desired_state", desiredStateDep).
 							WithField("dep_state", dep.state).
 							Debug("skip input request based on existing request")
@@ -1062,3 +1062,13 @@ func withSelector(keys []ExportableCacheKey, selector digest.Digest) []CacheKeyW
 	}
 	return out
 }
+
+type (
+	pipeRequest  = pipe.Request[*edgeRequest]
+	pipeSender   = pipe.Sender[*edgeRequest, any]
+	pipeReceiver = pipe.Receiver[*edgeRequest, any]
+)
+
+func newPipe(req pipeRequest) *pipe.Pipe[*edgeRequest, any] {
+	return pipe.New[*edgeRequest, any](req)
+}