| 252 | } |
| 253 | |
| 254 | func (m *Manager) runTask(t *model.Task) { |
| 255 | // update task state |
| 256 | tCtx, tc := context.WithCancel(m.ctx) |
| 257 | ti := &taskItem{ |
| 258 | ctx: tCtx, |
| 259 | cancel: tc, |
| 260 | task: t, |
| 261 | } |
| 262 | m.taskMap[t.ID] = ti |
| 263 | |
| 264 | // set task state to running |
| 265 | t.Updated = time.Now().Unix() |
| 266 | t.State = model.TaskRunning |
| 267 | |
| 268 | err := model.UpdateTask(m.db, t) |
| 269 | if err != nil { |
| 270 | // update task failed, try start again in next round |
| 271 | return |
| 272 | } |
| 273 | |
| 274 | log.Debugf("task scheduled to run: %v", t.LogData()) |
| 275 | |
| 276 | m.wg.Add(1) |
| 277 | |
| 278 | go func() { |
| 279 | defer func() { |
| 280 | // panic recover |
| 281 | if r := recover(); r != nil { |
| 282 | ti.err = fmt.Errorf("%v", r) |
| 283 | } |
| 284 | |
| 285 | // send finish trigger |
| 286 | select { |
| 287 | case m.finishCh <- ti.task.ID: |
| 288 | case <-m.ctx.Done(): |
| 289 | } |
| 290 | |
| 291 | m.wg.Done() |
| 292 | }() |
| 293 | |
| 294 | h, ok := m.handleMap[ti.task.Type] |
| 295 | if !ok { |
| 296 | // invalid task |
| 297 | ti.err = errors.Errorf("task %d type %d is invalid", ti.task.ID, ti.task.Type) |
| 298 | return |
| 299 | } |
| 300 | |
| 301 | result, err := h(tCtx, m.config, m.db, ti.task) |
| 302 | ti.result = result |
| 303 | if err != nil { |
| 304 | // task is failed |
| 305 | ti.err = errors.Wrapf(err, "execute task %d failed", ti.task.ID) |
| 306 | } |
| 307 | }() |
| 308 | } |
| 309 | |
| 310 | func (m *Manager) cleanupTask(t *taskItem) { |
| 311 | // collect result and save to database |