code: func (e *edge) unpark(incoming []pipe.Sender, updates, allPipes []pipe.Receiver, f *pipeFactory) {
  	// process all incoming changes
  	depChanged := false
  	for _, upt := range updates {
  		if changed := e.processUpdate(upt); changed {
  			depChanged = true
  		}
  	}
  
  	if depChanged {
  		// the dep responses had changes. need to reevaluate edge state
  		e.recalcCurrentState()
  	}
  
  	desiredState, done := e.respondToIncoming(incoming, allPipes)
  	if done {
  		return
  	}
  
  	cacheMapReq := false
  	// set up new outgoing requests if needed
  	if e.cacheMapReq == nil && (e.cacheMap == nil || len(e.cacheRecords) == 0) {
  		index := e.cacheMapIndex
  		e.cacheMapReq = f.NewFuncRequest(func(ctx context.Context) (interface{}, error) {
  			return e.op.CacheMap(ctx, index)
  		})
  		cacheMapReq = true
  	}
  
  	// execute op
  	if e.execReq == nil && desiredState == edgeStatusComplete {
  		if ok := e.execIfPossible(f); ok {
  			return
  		}
  	}
  
  	if e.execReq == nil {
  		if added := e.createInputRequests(desiredState, f, false); !added && !e.hasActiveOutgoing && !cacheMapReq {
  			logrus.Errorf("buildkit scheluding error: leaving incoming open. forcing solve. Please report this with BUILDKIT_SCHEDULER_DEBUG=1")
  			e.createInputRequests(desiredState, f, true)
  		}
  	}
  
  }
docs: // unpark is called by the scheduler with incoming requests and updates for
  // previous calls.
  // To avoid deadlocks and resource leaks this function needs to follow
  // following rules:
  // 1) this function needs to return unclosed outgoing requests if some incoming
  //    requests were not completed
  // 2) this function may not return outgoing requests if it has completed all
  //    incoming requests