<commit_before>package nomad

import (
	"fmt"
	"time"

	"github.com/armon/go-metrics"
	"github.com/hashicorp/nomad/nomad/state"
	"github.com/hashicorp/nomad/nomad/structs"
	"github.com/hashicorp/raft"
)

// planApply is a long lived goroutine that reads plan allocations from
// the plan queue, determines if they can be applied safely and applies
// them via Raft.
//
// Naively, we could simply dequeue a plan, verify, apply and then respond.
// However, the plan application is bounded by the Raft apply time and
// subject to some latency. This creates a stall condition, where we are
// not evaluating, but simply waiting for a transaction to complete.
//
// To avoid this, we overlap verification with apply. This means once
// we've verified plan N we attempt to apply it. However, while waiting
// for apply, we begin to verify plan N+1 under the assumption that plan
// N has succeeded.
//
// In this sense, we track two parallel versions of the world. One is
// the pessimistic one driven by the Raft log which is replicated. The
// other is optimistic and assumes our transactions will succeed. In the
// happy path, this lets us do productive work during the latency of
// apply.
//
// In the unhappy path (Raft transaction fails), effectively we only
// wasted work during a time we would have been waiting anyways. However,
// in anticipation of this case we cannot respond to the plan until
// the Raft log is updated. This means our schedulers will stall,
// but there are many of those and only a single plan verifier.
//
func (s *Server) planApply() {
	for {
		// Pull the next pending plan, exit if we are no longer leader
		pending, err := s.planQueue.Dequeue(0)
		if err != nil {
			return
		}

		// Verify the evaluation is outstanding, and that the tokens match.
		token, ok := s.evalBroker.Outstanding(pending.plan.EvalID)
		if !ok {
			s.logger.Printf("[ERR] nomad: plan received for non-outstanding evaluation %s",
				pending.plan.EvalID)
			pending.respond(nil, fmt.Errorf("evaluation is not outstanding"))
			continue
		}
		if pending.plan.EvalToken != token {
			s.logger.Printf("[ERR] nomad: plan received for evaluation %s with wrong token",
				pending.plan.EvalID)
			pending.respond(nil, fmt.Errorf("evaluation token does not match"))
			continue
		}

		// Snapshot the state so that we have a consistent view of the world
		snap, err := s.fsm.State().Snapshot()
		if err != nil {
			s.logger.Printf("[ERR] nomad: failed to snapshot state: %v", err)
			pending.respond(nil, err)
			continue
		}

		// Evaluate the plan
		result, err := evaluatePlan(snap, pending.plan)
		if err != nil {
			s.logger.Printf("[ERR] nomad: failed to evaluate plan: %v", err)
			pending.respond(nil, err)
			continue
		}

		// Fast-path the response if there is nothing to do
		if result.IsNoOp() {
			pending.respond(result, nil)
			continue
		}

		// Dispatch the Raft transaction for the plan
		future, err := s.applyPlan(result, snap)
		if err != nil {
			s.logger.Printf("[ERR] nomad: failed to submit plan: %v", err)
			pending.respond(nil, err)
			continue
		}

		// Respond to the plan in async
		waitCh := make(chan struct{})
		go s.asyncPlanWait(waitCh, future, result, pending)
		<-waitCh
	}
}

// applyPlan is used to apply the plan result and to return the alloc index
func (s *Server) applyPlan(result *structs.PlanResult, snap *state.StateSnapshot) (raft.ApplyFuture, error) {
	req := structs.AllocUpdateRequest{}
	for _, updateList := range result.NodeUpdate {
		req.Alloc = append(req.Alloc, updateList...)
	}
	for _, allocList := range result.NodeAllocation {
		req.Alloc = append(req.Alloc, allocList...)<commit_msg>nomad: overlap plan evaluation with apply<commit_after>package nomad

import (
	"fmt"
	"time"

	"github.com/armon/go-metrics"
	"github.com/hashicorp/nomad/nomad/state"
	"github.com/hashicorp/nomad/nomad/structs"
	"github.com/hashicorp/raft"
)

// planApply is a long lived goroutine that reads plan allocations from
// the plan queue, determines if they can be applied safely and applies
// them via Raft.
//
// Naively, we could simply dequeue a plan, verify, apply and then respond.
// However, the plan application is bounded by the Raft apply time and
// subject to some latency. This creates a stall condition, where we are
// not evaluating, but simply waiting for a transaction to complete.
//
// To avoid this, we overlap verification with apply. This means once
// we've verified plan N we attempt to apply it. However, while waiting
// for apply, we begin to verify plan N+1 under the assumption that plan
// N has succeeded.
//
// In this sense, we track two parallel versions of the world. One is
// the pessimistic one driven by the Raft log which is replicated. The
// other is optimistic and assumes our transactions will succeed. In the
// happy path, this lets us do productive work during the latency of
// apply.
//
// In the unhappy path (Raft transaction fails), effectively we only
// wasted work during a time we would have been waiting anyways. However,
// in anticipation of this case we cannot respond to the plan until
// the Raft log is updated. This means our schedulers will stall,
// but there are many of those and only a single plan verifier.
//
func (s *Server) planApply() {
	// waitCh is used to track an outstanding application
	// while snap holds an optimistic state which includes
	// that plan application.
	var waitCh chan struct{}
	var snap *state.StateSnapshot

	for {
		// Pull the next pending plan, exit if we are no longer leader
		pending, err := s.planQueue.Dequeue(0)
		if err != nil {
			return
		}

		// Verify the evaluation is outstanding, and that the tokens match.
		token, ok := s.evalBroker.Outstanding(pending.plan.EvalID)
		if !ok {
			s.logger.Printf("[ERR] nomad: plan received for non-outstanding evaluation %s",
				pending.plan.EvalID)
			pending.respond(nil, fmt.Errorf("evaluation is not outstanding"))
			continue
		}
		if pending.plan.EvalToken != token {
			s.logger.Printf("[ERR] nomad: plan received for evaluation %s with wrong token",
				pending.plan.EvalID)
			pending.respond(nil, fmt.Errorf("evaluation token does not match"))
			continue
		}

		// Check if out last plan has completed
		select {
		case <-waitCh:
			waitCh = nil
			snap = nil
		default:
		}

		// Snapshot the state so that we have a consistent view of the world
		// if no snapshot is available
		if snap == nil {
			snap, err = s.fsm.State().Snapshot()
			if err != nil {
				s.logger.Printf("[ERR] nomad: failed to snapshot state: %v", err)
				pending.respond(nil, err)
				continue
			}
		}

		// Evaluate the plan
		result, err := evaluatePlan(snap, pending.plan)
		if err != nil {
			s.logger.Printf("[ERR] nomad: failed to evaluate plan: %v", err)
			pending.respond(nil, err)
			continue
		}

		// Fast-path the response if there is nothing to do
		if result.IsNoOp() {
			pending.respond(result, nil)
			continue
		}

		// Ensure any parallel apply is complete before
		// starting the next one. This also limits how out
		// of date our snapshot can be.
		if waitCh != nil {
			<-waitCh
			snap, err = s.fsm.State().Snapshot()
			if err != nil {
				s.logger.Printf("[ERR] nomad: failed to snapshot state: %v", err)
				pending.respond(nil, err)
				continue
			}
		}

		// Dispatch the Raft transaction for the plan
		future, err := s.applyPlan(result, snap)
		if err != nil {
			s.logger.Printf("[ERR] nomad: failed to submit plan: %v", err)
			pending.respond(nil, err)
			continue
		}

		// Respond to the plan in async
		waitCh = make(chan struct{})
		go s.asyncPlanWait(waitCh, future, result, pending)
	}
}

// applyPlan is used to apply the plan result and to return the alloc index
func (s *Server) applyPlan(result *structs.PlanResult, snap *state.StateSnapshot) (raft.ApplyFuture, error) {
	req := structs.AllocUpdateRequest{}
	for _, updateList := range result.NodeUpdate {
		req.Alloc = append(req.Alloc, updateList...)
	}
	for _, allocList := range result.NodeAllocation {
		req.Alloc = append(req.Alloc, allocList...)
{'fraction_non_alphanumeric': 0.10237591740266203, 'fraction_numerical': 0.0004975743251648215, 'mean_word_length': 3.327233584499462, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4290588', 'n_tokens_mistral': 2590, 'n_tokens_neox': 2381, 'n_words': 987}