update log

server/dispatchTask.go

@@ -29,10 +29,11 @@ func newDispatchTask(w *Worker, task *odysseus.TaskContent) *dispatchTask {
 		result: make(chan *omanager.SubmitTaskResult, 1),
 	}
 }
-func (d *dispatchTask) dispatched(errCh chan error, wm *WorkerManager) {
+func (d *dispatchTask) dispatched(wm *WorkerManager) {
 	tm := time.NewTicker(time.Second * 5)
 	defer tm.Stop()
 	d.setStatus(TASK_WAIT_ACK)
+	t1 := time.Now()
 
 	select {
 	case ackMsg, ok := <-d.ack:
@@ -46,6 +47,7 @@ func (d *dispatchTask) dispatched(errCh chan error, wm *WorkerManager) {
 				"bootup-time":  msg.BootUpTime,
 				"queue-time":   msg.QueueWaitTime,
 				"execute-time": msg.ExecuteTime,
+				"ackcost":      time.Since(t1).Milliseconds(),
 			}).Debug("got ack message")
 			if !msg.CanExecute {
 				d.errCh <- errors.New("worker can't execute task")
@@ -56,8 +58,6 @@ func (d *dispatchTask) dispatched(errCh chan error, wm *WorkerManager) {
 		}
 	case <-tm.C:
 		d.errCh <- errors.New("ack timeout")
-	case err := <-errCh:
-		d.errCh <- err
 	}
 
 	if d.status == TASK_ACKED {

server/workerManager.go

@@ -348,9 +348,9 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 			msg.Message = taskMsg
 			errCh := make(chan error, 1)
 
-			go dtask.dispatched(errCh, wm)
 			callback = func(err error) bool {
 				if err == nil {
+					go dtask.dispatched(wm)
 					// add task to cache.
 					worker.recentTask.Add(task.TaskId, dtask)
 					if task.TaskKind == odysseus.TaskKind_ComputeTask {
@@ -365,7 +365,7 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 					"worker addr": worker.workerAddr,
 				}).Info("dispatch task to worker")
 				if err != nil {
-					errCh <- err
+					dtask.errCh <- err
 				}
 				return true
 			}