//  The state manager task maintains the state and handles requests from
//  clients for snapshots:
545260
func state_manager() {
	kvmap := make(map[string]*kvsimple.Kvmsg)

	pipe, _ := zmq.NewSocket(zmq.PAIR)
	pipe.Connect("inproc://pipe")
	pipe.SendMessage("READY")
	snapshot, _ := zmq.NewSocket(zmq.ROUTER)
	snapshot.Bind("tcp://*:5556")

	poller := zmq.NewPoller()
	poller.Add(pipe, zmq.POLLIN)
	poller.Add(snapshot, zmq.POLLIN)
	sequence := int64(0) //  Current snapshot version number
LOOP:
	for {
		polled, err := poller.Poll(-1)
		if err != nil {
			break //  Context has been shut down
		}
		for _, item := range polled {
			switch socket := item.Socket; socket {
			case pipe:
				//  Apply state update from main thread
				kvmsg, err := kvsimple.RecvKvmsg(pipe)
				if err != nil {
					break LOOP //  Interrupted
				}
				sequence, _ = kvmsg.GetSequence()
				kvmsg.Store(kvmap)
			case snapshot:
				//  Execute state snapshot request
				msg, err := snapshot.RecvMessage(0)
				if err != nil {
					break LOOP //  Interrupted
				}
				identity := msg[0]
				//  Request is in second frame of message
				request := msg[1]
				if request != "ICANHAZ?" {
					fmt.Println("E: bad request, aborting")
					break LOOP
				}
				//  Send state snapshot to client

				//  For each entry in kvmap, send kvmsg to client
				for _, kvmsg := range kvmap {
					snapshot.Send(identity, zmq.SNDMORE)
					kvmsg.Send(snapshot)
				}

				// Give client some time to deal with it.
				// This reduces the risk that the client won't see
				// the END message, but it doesn't eliminate the risk.
				time.Sleep(100 * time.Millisecond)

				//  Now send END message with sequence number
				fmt.Printf("Sending state shapshot=%d\n", sequence)
				snapshot.Send(identity, zmq.SNDMORE)
				kvmsg := kvsimple.NewKvmsg(sequence)
				kvmsg.SetKey("KTHXBAI")
				kvmsg.SetBody("")
				kvmsg.Send(snapshot)
			}
		}
	}
}
1112912
func VerifyAllTasksDone(ctx context.Context, ts *topo.Server, uuid string) error {
	return verifyAllTasksState(ctx, ts, uuid, workflowpb.TaskState_TaskDone)
}
216045
func NewFailureTaskState(signature *Signature, err string) *TaskState {
	return &TaskState{
		TaskUUID: signature.UUID,
		State:    StateFailure,
		Error:    err,
	}
}
304841
func (tr *TaskRunner) persistLocalState() error {
	tr.stateLock.RLock()
	defer tr.stateLock.RUnlock()

	return tr.stateDB.PutTaskRunnerLocalState(tr.allocID, tr.taskName, tr.localState)
}
1364474
func (b *Backend) updateState(taskState *tasks.TaskState) error {
	encoded, err := json.Marshal(taskState)
	if err != nil {
		return err
	}

	return b.getClient().Set(&gomemcache.Item{
		Key:        taskState.TaskUUID,
		Value:      encoded,
		Expiration: b.getExpirationTimestamp(),
	})
}
653992
func (b *Backend) GroupTaskStates(groupUUID string, groupTaskCount int) ([]*tasks.TaskState, error) {
	groupMeta, err := b.getGroupMeta(groupUUID)
	if err != nil {
		return nil, err
	}

	return b.getStates(groupMeta.TaskUUIDs...)
}
1004275
func (c oauthClient) Task(taskID uint) (wl.Task, error) {
	url := fmt.Sprintf(
		"%s/tasks/%d",
		c.apiURL,
		taskID,
	)

	req, err := c.newGetRequest(url)
	if err != nil {
		return wl.Task{}, err
	}

	resp, err := c.do(req)
	if err != nil {
		return wl.Task{}, err
	}

	if resp.StatusCode != http.StatusOK {
		return wl.Task{}, fmt.Errorf("Unexpected response code %d - expected %d", resp.StatusCode, http.StatusOK)
	}

	task := transportTask{}
	err = json.NewDecoder(resp.Body).Decode(&task)
	if err != nil {
		return wl.Task{}, err
	}
	return taskFromTransport(task)
}
943365
func (b *Backend) PurgeState(taskUUID string) error {
	input := &dynamodb.DeleteItemInput{
		Key: map[string]*dynamodb.AttributeValue{
			"TaskUUID": {
				N: aws.String(taskUUID),
			},
		},
		TableName: aws.String(b.cnf.DynamoDB.TaskStatesTable),
	}
	_, err := b.client.DeleteItem(input)

	if err != nil {
		return err
	}
	return nil
}
969463
func (b *Backend) GetState(taskUUID string) (*tasks.TaskState, error) {
	declareQueueArgs := amqp.Table{
		// Time in milliseconds
		// after that message will expire
		"x-message-ttl": int32(b.getExpiresIn()),
		// Time after that the queue will be deleted.
		"x-expires": int32(b.getExpiresIn()),
	}
	conn, channel, _, _, _, err := b.Connect(
		b.GetConfig().Broker,
		b.GetConfig().TLSConfig,
		b.GetConfig().AMQP.Exchange,     // exchange name
		b.GetConfig().AMQP.ExchangeType, // exchange type
		taskUUID,                        // queue name
		false,                           // queue durable
		true,                            // queue delete when unused
		taskUUID,                        // queue binding key
		nil,                             // exchange declare args
		declareQueueArgs,                // queue declare args
		nil,                             // queue binding args
	)
	if err != nil {
		return nil, err
	}
	defer b.Close(channel, conn)

	d, ok, err := channel.Get(
		taskUUID, // queue name
		false,    // multiple
	)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, errors.New("No state ready")
	}

	d.Ack(false)

	state := new(tasks.TaskState)
	decoder := json.NewDecoder(bytes.NewReader([]byte(d.Body)))
	decoder.UseNumber()
	if err := decoder.Decode(state); err != nil {
		log.ERROR.Printf("Failed to unmarshal task state: %s", string(d.Body))
		log.ERROR.Print(err)
		return nil, err
	}

	return state, nil
}
838096
func (tr *TaskRunner) UpdateState(state string, event *structs.TaskEvent) {
	tr.stateLock.Lock()
	defer tr.stateLock.Unlock()

	if event != nil {
		tr.logger.Trace("setting task state", "state", state, "event", event.Type)

		// Append the event
		tr.appendEvent(event)
	}

	// Update the state
	if err := tr.updateStateImpl(state); err != nil {
		// Only log the error as we persistence errors should not
		// affect task state.
		tr.logger.Error("error persisting task state", "error", err, "event", event, "state", state)
	}

	// Notify the alloc runner of the transition
	tr.stateUpdater.TaskStateUpdated()
}
848154
func (d *Downloader) runStateSync(s *stateSync) *stateSync {
	var (
		active   = make(map[string]*stateReq) // Currently in-flight requests
		finished []*stateReq                  // Completed or failed requests
		timeout  = make(chan *stateReq)       // Timed out active requests
	)
	defer func() {
		// Cancel active request timers on exit. Also set peers to idle so they're
		// available for the next sync.
		for _, req := range active {
			req.timer.Stop()
			req.peer.SetNodeDataIdle(len(req.items))
		}
	}()
	// Run the state sync.
	go s.run()
	defer s.Cancel()

	// Listen for peer departure events to cancel assigned tasks
	peerDrop := make(chan *peerConnection, 1024)
	peerSub := s.d.peers.SubscribePeerDrops(peerDrop)
	defer peerSub.Unsubscribe()

	for {
		// Enable sending of the first buffered element if there is one.
		var (
			deliverReq   *stateReq
			deliverReqCh chan *stateReq
		)
		if len(finished) > 0 {
			deliverReq = finished[0]
			deliverReqCh = s.deliver
		}

		select {
		// The stateSync lifecycle:
		case next := <-d.stateSyncStart:
			return next

		case <-s.done:
			return nil

		// Send the next finished request to the current sync:
		case deliverReqCh <- deliverReq:
			// Shift out the first request, but also set the emptied slot to nil for GC
			copy(finished, finished[1:])
			finished[len(finished)-1] = nil
			finished = finished[:len(finished)-1]

		// Handle incoming state packs:
		case pack := <-d.stateCh:
			// Discard any data not requested (or previously timed out)
			req := active[pack.PeerId()]
			if req == nil {
				log.Debug("Unrequested node data", "peer", pack.PeerId(), "len", pack.Items())
				continue
			}
			// Finalize the request and queue up for processing
			req.timer.Stop()
			req.response = pack.(*statePack).states

			finished = append(finished, req)
			delete(active, pack.PeerId())

		// Handle dropped peer connections:
		case p := <-peerDrop:
			// Skip if no request is currently pending
			req := active[p.id]
			if req == nil {
				continue
			}
			// Finalize the request and queue up for processing
			req.timer.Stop()
			req.dropped = true

			finished = append(finished, req)
			delete(active, p.id)

		// Handle timed-out requests:
		case req := <-timeout:
			// If the peer is already requesting something else, ignore the stale timeout.
			// This can happen when the timeout and the delivery happens simultaneously,
			// causing both pathways to trigger.
			if active[req.peer.id] != req {
				continue
			}
			// Move the timed out data back into the download queue
			finished = append(finished, req)
			delete(active, req.peer.id)

		// Track outgoing state requests:
		case req := <-d.trackStateReq:
			// If an active request already exists for this peer, we have a problem. In
			// theory the trie node schedule must never assign two requests to the same
			// peer. In practice however, a peer might receive a request, disconnect and
			// immediately reconnect before the previous times out. In this case the first
			// request is never honored, alas we must not silently overwrite it, as that
			// causes valid requests to go missing and sync to get stuck.
			if old := active[req.peer.id]; old != nil {
				log.Warn("Busy peer assigned new state fetch", "peer", old.peer.id)

				// Make sure the previous one doesn't get siletly lost
				old.timer.Stop()
				old.dropped = true

				finished = append(finished, old)
			}
			// Start a timer to notify the sync loop if the peer stalled.
			req.timer = time.AfterFunc(req.timeout, func() {
				select {
				case timeout <- req:
				case <-s.done:
					// Prevent leaking of timer goroutines in the unlikely case where a
					// timer is fired just before exiting runStateSync.
				}
			})
			active[req.peer.id] = req
		}
	}
}
691947
func (b *Backend) PurgeState(taskUUID string) error {
	return b.getClient().Delete(taskUUID)
}
534674
func managementTask(c *asyncConnHdl, ctl workerCtl) (sig *interrupt_code, te *taskStatus) {
	/* TODO:
	connected:
		clearChannels();
		startWokers();
	disconnected:
		?
	on_fault:
		disconnect();
		goto disconnected;
	on_exit:
		?
	*/
	//	log.Println("MGR: do task ...")
	select {
	case stat := <-c.feedback:
		// do the shutdown for now -- TODO: try reconnect
		if stat.event == faulted || stat.event == quit_processed {
			if stat.event == faulted {
				// REVU - pretty please TODO do the customized log
				log.Printf("<INFO> - %s (manager task) FAULT EVENT ", c)
			}
			// REVU - pretty please TODO do the customized log
			//			log.Printf("<INFO> %s - (manager task) SHUTTING DOWN ...", c)
			c.shutdown <- true

			// REVU - pretty please TODO do the customized log
			//			log.Printf("<INFO> %s - (manager task) RAISING SIGNAL STOP ...", c)
			go func() { c.reqProcCtl <- stop }()
			go func() { c.rspProcCtl <- stop }()
			go func() { c.heartbeatCtl <- stop }()
			go func() { c.managerCtl <- stop }()
		}
	case s := <-ctl:
		return &s, &ok_status
	}

	return nil, &ok_status
}
711444
func (c *CheckpointWriter) UpdateTask(taskID string, status workflowpb.TaskState, err error) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	errorMessage := ""
	if err != nil {
		errorMessage = err.Error()
	}

	t := c.checkpoint.Tasks[taskID]
	t.State = status
	t.Error = errorMessage
	return c.saveLocked()
}
174287
func (b *Backend) updateState(signature *tasks.Signature, update bson.M) error {
	update = bson.M{"$set": update}
	_, err := b.tasksCollection.UpdateOne(context.Background(), bson.M{"_id": signature.UUID}, update, options.Update().SetUpsert(true))
	return err
}
1029398
func (q *taskPipeline) queueManager() {
	defer close(q.shutdown)
	defer close(q.newBackupTasks)

	for {
		q.queueCond.L.Lock()
		for q.queue.Front() == nil {
			q.queueCond.Wait()

			select {
			case <-q.quit:
				// Exit only after the queue has been fully drained.
				if q.queue.Len() == 0 {
					q.queueCond.L.Unlock()
					log.Debugf("Revoked state pipeline flushed.")
					return
				}

			case <-q.forceQuit:
				q.queueCond.L.Unlock()
				log.Debugf("Revoked state pipeline force quit.")
				return

			default:
			}
		}

		// Pop the first element from the queue.
		e := q.queue.Front()
		task := q.queue.Remove(e).(*backupTask)
		q.queueCond.L.Unlock()

		select {

		// Backup task submitted to dispatcher. We don't select on quit to
		// ensure that we still drain tasks while shutting down.
		case q.newBackupTasks <- task:

		// Force quit, return immediately to allow the client to exit.
		case <-q.forceQuit:
			log.Debugf("Revoked state pipeline force quit.")
			return
		}
	}
}
303280
func (s *service) State(ctx context.Context, r *taskAPI.StateRequest) (*taskAPI.StateResponse, error) {
	p, err := s.getProcess(r.ExecID)
	if err != nil {
		return nil, err
	}
	st, err := p.Status(ctx)
	if err != nil {
		return nil, err
	}
	status := task.StatusUnknown
	switch st {
	case "created":
		status = task.StatusCreated
	case "running":
		status = task.StatusRunning
	case "stopped":
		status = task.StatusStopped
	case "paused":
		status = task.StatusPaused
	case "pausing":
		status = task.StatusPausing
	}
	sio := p.Stdio()
	return &taskAPI.StateResponse{
		ID:         p.ID(),
		Bundle:     s.container.Bundle,
		Pid:        uint32(p.Pid()),
		Status:     status,
		Stdin:      sio.Stdin,
		Stdout:     sio.Stdout,
		Stderr:     sio.Stderr,
		Terminal:   sio.Terminal,
		ExitStatus: uint32(p.ExitStatus()),
		ExitedAt:   p.ExitedAt(),
	}, nil
}