code: func (s *scheduler) newPipe(target, from *edge, req pipe.Request) *pipe.Pipe {
  	p := &edgePipe{
  		Pipe:   pipe.New(req),
  		Target: target,
  		From:   from,
  	}
  
  	s.signal(target)
  	if from != nil {
  		p.OnSendCompletion = func() {
  			p.mu.Lock()
  			defer p.mu.Unlock()
  			s.signal(p.From)
  		}
  		s.outgoing[from] = append(s.outgoing[from], p)
  	}
  	s.incoming[target] = append(s.incoming[target], p)
  	p.OnReceiveCompletion = func() {
  		p.mu.Lock()
  		defer p.mu.Unlock()
  		s.signal(p.Target)
  	}
  	return p.Pipe
  }
docs: // newPipe creates a new request pipe between two edges