StartProcess starts a process in the container. When a container isn't available (i.e. first time invoking or the container has exited) or cfg.Rollback is set, this method will start a new container and run the process in it. Otherwise, this method starts a new process in the existing container.
(pid string, resultCtx *build.ResultHandle, cfg *build.InvokeConfig)
| 98 | // this method will start a new container and run the process in it. Otherwise, this method starts a new process in the |
| 99 | // existing container. |
| 100 | func (m *Manager) StartProcess(pid string, resultCtx *build.ResultHandle, cfg *build.InvokeConfig) (*Process, error) { |
| 101 | // Get the target result to invoke a container from |
| 102 | var ctr *build.Container |
| 103 | if a := m.container.Load(); a != nil { |
| 104 | ctr = a.(*build.Container) |
| 105 | } |
| 106 | if cfg.Rollback || ctr == nil || ctr.IsUnavailable() { |
| 107 | go m.CancelRunningProcesses() |
| 108 | // (Re)create a new container if this is rollback or first time to invoke a process. |
| 109 | if ctr != nil { |
| 110 | go ctr.Cancel() // Finish the existing container |
| 111 | } |
| 112 | var err error |
| 113 | ctr, err = build.NewContainer(context.TODO(), resultCtx, cfg) |
| 114 | if err != nil { |
| 115 | return nil, errors.Errorf("failed to create container %v", err) |
| 116 | } |
| 117 | m.container.Store(ctr) |
| 118 | } |
| 119 | // [client(ForwardIO)] <-forwarder(switchable)-> [out] <-pipe-> [in] <- [process] |
| 120 | in, out := ioset.Pipe() |
| 121 | f := ioset.NewForwarder() |
| 122 | f.PropagateStdinClose = false |
| 123 | f.SetOut(&out) |
| 124 | |
| 125 | // Register process |
| 126 | ctx, cancel := context.WithCancelCause(context.TODO()) |
| 127 | var cancelOnce sync.Once |
| 128 | processCancelFunc := func() { |
| 129 | cancelOnce.Do(func() { |
| 130 | cancel(errors.WithStack(context.Canceled)) |
| 131 | f.Close() |
| 132 | in.Close() |
| 133 | out.Close() |
| 134 | }) |
| 135 | } |
| 136 | p := &Process{ |
| 137 | inEnd: f, |
| 138 | invokeConfig: cfg, |
| 139 | processCancel: processCancelFunc, |
| 140 | errCh: make(chan error), |
| 141 | } |
| 142 | m.processes.Store(pid, p) |
| 143 | go func() { |
| 144 | var err error |
| 145 | if err = ctr.Exec(ctx, cfg, in.Stdin, in.Stdout, in.Stderr); err != nil { |
| 146 | logrus.Debugf("process error: %v", err) |
| 147 | } |
| 148 | logrus.Debugf("finished process %s %v", pid, cfg.Entrypoint) |
| 149 | m.processes.Delete(pid) |
| 150 | processCancelFunc() |
| 151 | p.errCh <- err |
| 152 | }() |
| 153 | |
| 154 | return p, nil |
| 155 | } |
| 156 | |
| 157 | type ProcessInfo struct { |