func (cp *compiler) pipeline(n *parse.Pipeline) op {
	ops := cp.forms(n.Forms)
	p := n.Begin()

	return func(ec *evalCtx) {
		var nextIn *port

		errors := make([]Error, len(ops))
		finished := make(chan bool, len(ops))

		// For each form, create a dedicated evalCtx and run asynchronously
		for i, op := range ops {
			newEc := ec.fork(fmt.Sprintf("form op %v", op))
			if i > 0 {
				newEc.ports[0] = nextIn
			}
			if i < len(ops)-1 {
				// Each internal port pair consists of a (byte) pipe pair and a
				// channel.
				// os.Pipe sets O_CLOEXEC, which is what we want.
				reader, writer, e := os.Pipe()
				if e != nil {
					ec.errorf(p, "failed to create pipe: %s", e)
				}
				ch := make(chan Value, pipelineChanBufferSize)
				newEc.ports[1] = &port{
					f: writer, ch: ch, closeF: true, closeCh: true}
				nextIn = &port{
					f: reader, ch: ch, closeF: true, closeCh: false}
			}
			thisOp := op
			thisError := &errors[i]
			go func() {
				(*thisError).inner = newEc.peval(thisOp)
				newEc.closePorts()
				finished <- true
			}()
		}
		// Wait for all forms to finish
		for i := 0; i < len(ops); i++ {
			<-finished
		}
		if !allok(errors) {
			if len(errors) == 1 {
				throw(errors[0].inner)
			} else {
				throw(multiError{errors})
			}
		}
	}
}

func (cp *compiler) pipeline(n *parse.Pipeline) Op {
	ops := cp.forms(n.Forms)
	p := n.Begin()

	return func(ec *EvalCtx) {
		var nextIn *Port

		errorChans := make([]chan Error, len(ops))

		// For each form, create a dedicated evalCtx and run asynchronously
		for i, op := range ops {
			newEc := ec.fork(fmt.Sprintf("form op %v", op))
			if i > 0 {
				newEc.ports[0] = nextIn
			}
			if i < len(ops)-1 {
				// Each internal port pair consists of a (byte) pipe pair and a
				// channel.
				// os.Pipe sets O_CLOEXEC, which is what we want.
				reader, writer, e := os.Pipe()
				if e != nil {
					ec.errorf(p, "failed to create pipe: %s", e)
				}
				ch := make(chan Value, pipelineChanBufferSize)
				newEc.ports[1] = &Port{
					File: writer, Chan: ch, CloseFile: true, CloseChan: true}
				nextIn = &Port{
					File: reader, Chan: ch, CloseFile: true, CloseChan: false}
			}
			thisOp := op
			errorChans[i] = make(chan Error)
			thisErrorChan := errorChans[i]
			go func() {
				err := newEc.PEval(thisOp)
				// Logger.Printf("closing ports of %s", newEc.context)
				ClosePorts(newEc.ports)
				thisErrorChan <- Error{err}
			}()
		}

		intCh := make(chan os.Signal)
		signal.Notify(intCh, syscall.SIGINT)
		interrupted := make(chan struct{})
		cancel := make(chan struct{}, 1)
		go func() {
			// When SIGINT is received, sleep for InterruptDeadline before the
			// closing interrupted channel.
			select {
			case <-intCh:
			case <-cancel:
				return
			}
			select {
			case <-time.After(InterruptDeadline):
			case <-cancel:
				return
			}
			close(interrupted)
		}()

		// Wait for all forms to finish and collect error returns, unless an
		// interrupt was received and the form didn't quit within
		// InterruptDeadline.
		errors := make([]Error, len(ops))
		for i, errorChan := range errorChans {
			select {
			case errors[i] = <-errorChan:
			case <-interrupted:
				errors[i] = Error{ErrStillRunning}
			}
		}

		// Make sure the SIGINT listener exits.
		close(cancel)
		signal.Stop(intCh)

		// Make sure I am in foreground.
		if PutInForeground && sys.IsATTY(0) {
			err := sys.Tcsetpgrp(0, syscall.Getpgrp())
			if err != nil {
				throw(err)
			}
		}

		if !allok(errors) {
			if len(errors) == 1 {
				throw(errors[0].inner)
			} else {
				throw(multiError{errors})
			}
		}
	}
}

func (cp *compiler) pipelineOp(n *parse.Pipeline) Op {
	return Op{cp.pipeline(n), n.Begin(), n.End()}
}

func (cp *compiler) pipeline(n *parse.Pipeline) OpFunc {
	ops := cp.formOps(n.Forms)

	return func(ec *EvalCtx) {
		bg := n.Background
		if bg {
			ec = ec.fork("background job " + n.SourceText())
			ec.intCh = nil
			ec.background = true

			if ec.Editor != nil {
				// TODO: Redirect output in interactive mode so that the line
				// editor does not get messed up.
			}
		}

		nforms := len(ops)

		var wg sync.WaitGroup
		wg.Add(nforms)
		errors := make([]Error, nforms)
		var verdict bool

		var nextIn *Port

		// For each form, create a dedicated evalCtx and run asynchronously
		for i, op := range ops {
			newEc := ec.fork(fmt.Sprintf("form op %v", op))
			if i > 0 {
				newEc.ports[0] = nextIn
			}
			if i < nforms-1 {
				// Each internal port pair consists of a (byte) pipe pair and a
				// channel.
				// os.Pipe sets O_CLOEXEC, which is what we want.
				reader, writer, e := os.Pipe()
				if e != nil {
					throwf("failed to create pipe: %s", e)
				}
				ch := make(chan Value, pipelineChanBufferSize)
				newEc.ports[1] = &Port{
					File: writer, Chan: ch, CloseFile: true, CloseChan: true}
				nextIn = &Port{
					File: reader, Chan: ch, CloseFile: true, CloseChan: false}
			}
			thisOp := op
			thisError := &errors[i]
			isLast := i == nforms-1
			go func() {
				err := newEc.PEval(thisOp)
				// Logger.Printf("closing ports of %s", newEc.context)
				ClosePorts(newEc.ports)
				if isLast {
					verdict = newEc.verdict
				}
				*thisError = Error{err}
				wg.Done()
			}()
		}

		if bg {
			// Background job, wait for form termination asynchronously.
			go func() {
				wg.Wait()
				msg := "job " + n.SourceText() + " finished"
				if !allok(errors) {
					msg += ", errors = " + makeCompositeError(errors).Error()
				}
				if !verdict {
					msg += ", pred = false"
				}
				if ec.Editor != nil {
					m := ec.Editor.ActiveMutex()
					m.Lock()
					defer m.Unlock()

					if ec.Editor.Active() {
						ec.Editor.Notify("%s", msg)
					} else {
						ec.ports[2].File.WriteString(msg)
					}
				} else {
					ec.ports[2].File.WriteString(msg)
				}
			}()
		} else {
			wg.Wait()
			maybeThrow(makeCompositeError(errors))
			ec.verdict = verdict
		}
	}
}