<commit_before>package sensitivity

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/intelsdi-x/athena/pkg/executor"
	"github.com/intelsdi-x/athena/pkg/snap"
	"github.com/intelsdi-x/athena/pkg/utils/err_collection"
	"github.com/intelsdi-x/swan/pkg/experiment/phase"
	"github.com/pkg/errors"
)

// measurementPhase performs a measurement for given loadPointIndex.
// User of this Phase is responsible to orchestrate this phase for each loadPoint to
// fulfill the LoadPointsCount.
type measurementPhase struct {
	namePrefix string
	// Latency Sensitivity (Production) workload.
	pr LauncherSessionPair
	// Workload Generator for Latency Sensitivity task.
	lgForPr LoadGeneratorSessionPair
	// Aggressors (Best Effort) tasks to stress LC task.
	// It can be empty in case of baseline phase.
	bes []LauncherSessionPair
	// Measurement duration in [s].
	loadDuration time.Duration
	// Number of load points to test.
	loadPointsCount int
	// Number of repetitions
	repetitions int
	// Current measurement's load point.
	currentLoadPointIndex int

	// Shared reference for measurement targetQPS resulted from Tuning Phase.
	PeakLoad *int

	activeSnapSessions       []snap.SessionHandle
	activeLaunchersTasks     []executor.TaskHandle
	activeLoadGeneratorTasks []executor.TaskHandle
}

// Returns measurement name.
func (m *measurementPhase) Name() string {
	return m.namePrefix + "_measurement_for_loadpoint_id_" +
		strconv.Itoa(m.currentLoadPointIndex)
}

// Returns number of repetitions.
func (m *measurementPhase) Repetitions() int {
	return m.repetitions
}

// Gets current loadPoint from linear function y = a * x where `x` is loadPointIndex.
func (m *measurementPhase) getLoadPoint() int {
	// Since we know that the function is satisfied
	// when TargetLoadPoint = a * loadPointsCount, we can calculate `a` parameter.
	a := float64(*m.PeakLoad) / float64(m.loadPointsCount)
	x := float64(m.currentLoadPointIndex)

	return int(a * x)
}

func (m *measurementPhase) clean() (err error) {
	var errCollection errcollection.ErrorCollection

	// Cleaning and stopping active Launchers' tasks.
	for _, task := range m.activeLaunchersTasks {
		err = task.Stop()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while stopping task"))

			// Don't clean when stop failed.
			continue
		}

		err = task.Clean()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while cleaning task"))
		}
	}
	m.activeLaunchersTasks = []executor.TaskHandle{}

	// Cleaning only active LoadGenerators' tasks.
	for _, task := range m.activeLoadGeneratorTasks {
		err = task.Clean()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while cleaning task"))
		}
	}
	m.activeLoadGeneratorTasks = []executor.TaskHandle{}

	// Stopping only active Snap sessions.
	for _, snapSession := range m.activeSnapSessions {
		log.Debug("Waiting for snap session to complete it's work. ", snapSession)
		err = snapSession.Wait()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while waiting for Snap session to complete it's work"))<commit_msg>Give enough time before killing snap sessions (#418)<commit_after>package sensitivity

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	log "github.com/Sirupsen/logrus"
	"github.com/intelsdi-x/athena/pkg/executor"
	"github.com/intelsdi-x/athena/pkg/snap"
	"github.com/intelsdi-x/athena/pkg/utils/err_collection"
	"github.com/intelsdi-x/swan/pkg/experiment/phase"
	"github.com/pkg/errors"
)

// measurementPhase performs a measurement for given loadPointIndex.
// User of this Phase is responsible to orchestrate this phase for each loadPoint to
// fulfill the LoadPointsCount.
type measurementPhase struct {
	namePrefix string
	// Latency Sensitivity (Production) workload.
	pr LauncherSessionPair
	// Workload Generator for Latency Sensitivity task.
	lgForPr LoadGeneratorSessionPair
	// Aggressors (Best Effort) tasks to stress LC task.
	// It can be empty in case of baseline phase.
	bes []LauncherSessionPair
	// Measurement duration in [s].
	loadDuration time.Duration
	// Number of load points to test.
	loadPointsCount int
	// Number of repetitions
	repetitions int
	// Current measurement's load point.
	currentLoadPointIndex int

	// Shared reference for measurement targetQPS resulted from Tuning Phase.
	PeakLoad *int

	activeSnapSessions       []snap.SessionHandle
	activeLaunchersTasks     []executor.TaskHandle
	activeLoadGeneratorTasks []executor.TaskHandle
}

// Returns measurement name.
func (m *measurementPhase) Name() string {
	return m.namePrefix + "_measurement_for_loadpoint_id_" +
		strconv.Itoa(m.currentLoadPointIndex)
}

// Returns number of repetitions.
func (m *measurementPhase) Repetitions() int {
	return m.repetitions
}

// Gets current loadPoint from linear function y = a * x where `x` is loadPointIndex.
func (m *measurementPhase) getLoadPoint() int {
	// Since we know that the function is satisfied
	// when TargetLoadPoint = a * loadPointsCount, we can calculate `a` parameter.
	a := float64(*m.PeakLoad) / float64(m.loadPointsCount)
	x := float64(m.currentLoadPointIndex)

	return int(a * x)
}

func (m *measurementPhase) clean() (err error) {
	var errCollection errcollection.ErrorCollection

	// Cleaning and stopping active Launchers' tasks.
	for _, task := range m.activeLaunchersTasks {
		err = task.Stop()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while stopping task"))

			// Don't clean when stop failed.
			continue
		}

		err = task.Clean()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while cleaning task"))
		}
	}
	m.activeLaunchersTasks = []executor.TaskHandle{}

	// Cleaning only active LoadGenerators' tasks.
	for _, task := range m.activeLoadGeneratorTasks {
		err = task.Clean()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while cleaning task"))
		}
	}
	m.activeLoadGeneratorTasks = []executor.TaskHandle{}

	// Give enough time for executors to cleanup, before stopping the session.
	// It is workaround for metrics from publisher of loadgenerator session not always available in the storage.
	// See the description of SCE-876.
	time.Sleep(10 * time.Second)

	// Stopping only active Snap sessions.
	for _, snapSession := range m.activeSnapSessions {
		log.Debug("Waiting for snap session to complete it's work. ", snapSession)
		err = snapSession.Wait()
		if err != nil {
			errCollection.Add(errors.Wrap(err, "error while waiting for Snap session to complete it's work"))
{'fraction_non_alphanumeric': 0.10027726432532348, 'fraction_numerical': 0.0030807147258163892, 'mean_word_length': 4.140934283452098, 'pattern_counts': {'":': 0, '<': 3, '<?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': '3735866', 'n_tokens_mistral': 2100, 'n_tokens_neox': 1976, 'n_words': 612}