Method run

src/go/plugin/framework/functions/manager.go:156–211

(ctx context.Context, quitCh chan struct{})

}

func (m *Manager) run(ctx context.Context, quitCh chan struct{}) {
	parser := newInputParser()
	m.scheduler = newKeyScheduler(m.queueSize)
	m.observeSchedulerPending()
	var workersWG sync.WaitGroup

	m.startWorkers(&workersWG)

	// Wake any enqueue() blocked on a full scheduler when the parent context
	// is canceled. Necessary because the reader loop below may itself be
	// blocked inside dispatchInvocation -> scheduler.enqueue waiting for
	// space, and would otherwise miss the ctx.Done signal.
	stopWatcher := make(chan struct{})
	defer close(stopWatcher)
	go func() {
		select {
		case <-ctx.Done():
			if m.scheduler != nil {
				m.scheduler.stop()
			}
		case <-stopWatcher:
		}
	}()

	for {
		select {
		case <-ctx.Done():
			m.shutdown(false, true, quitCh, &workersWG)
			return
		case line, ok := <-m.input.lines():
			if !ok {
				m.shutdown(false, true, quitCh, &workersWG)
				return
			}
			event, err := parser.parseEvent(line)
			if err != nil {
				m.Warningf("parse function: %v ('%s')", err, line)
				continue
			}

			switch event.kind {
			case inputEventNone, inputEventProgress:
				continue
			case inputEventQuit:
				m.shutdown(true, true, quitCh, &workersWG)
				return
			case inputEventCancel:
				m.handleCancelEvent(event)
				continue
			case inputEventCall:
				m.observeFunctionCall()
				m.dispatchInvocation(ctx, event.fn)
			}
		}
	}
}

func (m *Manager) startWorkers(workersWG *sync.WaitGroup) {