Method run

cmd/cql-proxy/task/task.go:171–252

()

}

func (m *Manager) run() {
	defer m.wg.Done()

	for {
		select {
		case <-m.ctx.Done():
			// kill pending tasks
			for _, t := range m.taskMap {
				t.cancel()
			}

			return
		case wi := <-m.waitCh:
			if _, ok := m.taskMap[wi.id]; ok {
				m.waitMap[wi.id] = append(m.waitMap[wi.id], wi)
			} else {
				// trigger directly
				if wi.ch != nil {
					select {
					case <-wi.ch:
					default:
						close(wi.ch)
					}
				}
			}
		case id := <-m.killCh:
			// kill task
			if t, ok := m.taskMap[id]; ok {
				t.cancel()
			}
		case id := <-m.finishCh:
			// finish task
			if t, ok := m.taskMap[id]; ok {
				m.cleanupTask(t)
			}
		case tsk := <-m.newCh:
			// new task
			if _, ok := m.taskMap[tsk.ID]; !ok {
				m.runTask(tsk)
			}
		case <-time.After(10 * time.Second):
			// poll database for existing task
			tasks, err := model.ListIncompleteTask(m.db, MaxTaskPerRound)

			if err != nil {
				continue
			}

			for _, t := range tasks {
				switch t.State {
				case model.TaskWaiting:
					// start job
					if _, ok := m.taskMap[t.ID]; !ok {
						m.runTask(t)
					}
				case model.TaskRunning:
					// check running state
					if _, ok := m.taskMap[t.ID]; !ok {