Merge pull request moby#5323 from jsternberg/generic-pipes

solver: pipe implementation utilizes generics for better typing

Author: tonistiigi

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)
+}