Method Start

flyteplugins/go/tasks/pluginmachinery/workqueue/queue.go:185–251

(ctx context.Context)

}

func (q *queue) Start(ctx context.Context) error {
	q.wlock.Lock()
	defer q.wlock.Unlock()

	if q.started {
		return fmt.Errorf("queue already started")
	}

	for i := 0; i < q.workers; i++ {
		go func(ctx context.Context) {
			for {
				select {
				case <-ctx.Done():
					logger.Debug(ctx, "Context cancelled. Shutting down.")
					return
				default:
					item, shutdown := q.queue.Get()
					if shutdown {
						logger.Debug(ctx, "Work queue is shutting down.")
						return
					}

					wrapperV := item.Clone()
					wrapper := &wrapperV
					ws := wrapper.status
					var err error

					func() {
						defer func() {
							if e, ok := recover().(error); ok {
								logger.Errorf(ctx, "Worker panic'd while processing item [%v]. Error: %v", wrapper.id, e)
								err = e
							}
						}()

						ctxWithFields := contextWithValues(ctx, wrapper.logFields)
						ws, err = q.processor.Process(ctxWithFields, wrapper.payload)
					}()

					if err != nil {
						q.metrics.ProcessorErrors.Inc()

						wrapper.retryCount++
						wrapper.err = err
						if wrapper.retryCount >= uint(q.maxRetries) {
							logger.Debugf(ctx, "WorkItem [%v] exhausted all retries. Last Error: %v.",
								wrapper.ID(), err)
							wrapper.status = WorkStatusFailed
							ws = WorkStatusFailed
							q.index.Add(wrapper)
							continue
						}
					}

					wrapper.status = ws
					q.index.Add(wrapper)
					if !ws.IsTerminal() {
						q.queue.Add(wrapper)