Merged in measureClassificationOverhead (pull request #12)

MeasureClassificationOverhead

Approved-by: Akash Kothawale <akothaw1@binghamton.edu>

logging/def/clsfnTaskDistOverhead.go

@@ -0,0 +1,11 @@
+package logging
+
+type ClsfnTaskDistOverheadLogger struct {
+	loggerObserverImpl
+}
+
+func (col ClsfnTaskDistOverheadLogger) Log(message string) {
+	// Logging the overhead of classifying tasks in the scheduling window and determining the distribution
+	//      of light power consuming and heavy power consuming tasks.
+	col.logObserverSpecifics[clsfnTaskDistOverheadLogger].logFile.Println(message)
+}

logging/def/logType.go

@@ -15,6 +15,7 @@ var (
 	PCP                     = messageNametoMessageType("PCP")
 	DEG_COL                 = messageNametoMessageType("DEG_COL")
 	SPS                     = messageNametoMessageType("SPS")
+	CLSFN_TASKDIST_OVERHEAD = messageNametoMessageType("CLSFN_TASKDIST_OVERHEAD")
 )
 
 // Text colors for the different types of log messages.

logging/def/logger.go

@@ -20,6 +20,7 @@ func newLogger() *LoggerDriver {
 			PCP:         true,
 			DEG_COL:     true,
 			SPS:         true,
+			CLSFN_TASKDIST_OVERHEAD: true,
 		},
 	}
 	return logger

logging/def/loggerFactory.go

@@ -13,6 +13,7 @@ const (
 	pcpLogger                   = "pcp-logger"
 	degColLogger                = "degCol-logger"
 	spsLogger                   = "schedPolicySwitch-logger"
+	clsfnTaskDistOverheadLogger = "classificationOverhead-logger"
 )
 
 // Logger class factory
@@ -22,6 +23,7 @@ var Loggers map[string]loggerObserver = map[string]loggerObserver{
 	pcpLogger:                   nil,
 	degColLogger:                nil,
 	spsLogger:                   nil,
+	clsfnTaskDistOverheadLogger: nil,
 }
 
 // Logger options to help initialize loggers
@@ -43,6 +45,7 @@ func withLoggerSpecifics(prefix string) loggerOption {
 			pcpLogger:                   &specifics{},
 			degColLogger:                &specifics{},
 			spsLogger:                   &specifics{},
+			clsfnTaskDistOverheadLogger: &specifics{},
 		}
 		l.(*loggerObserverImpl).setLogFilePrefix(prefix)
 		l.(*loggerObserverImpl).setLogFile()
@@ -70,6 +73,9 @@ func attachAllLoggers(lg *LoggerDriver, startTime time.Time, prefix string) {
 	Loggers[spsLogger] = &SchedPolicySwitchLogger{
 		loggerObserverImpl: *loi,
 	}
+	Loggers[clsfnTaskDistOverheadLogger] = &ClsfnTaskDistOverheadLogger{
+		loggerObserverImpl: *loi,
+	}
 
 	for _, lmt := range GetLogMessageTypes() {
 		switch lmt {
@@ -89,6 +95,8 @@ func attachAllLoggers(lg *LoggerDriver, startTime time.Time, prefix string) {
 			lg.attach(DEG_COL, Loggers[degColLogger])
 		case SPS.String():
 			lg.attach(SPS, Loggers[spsLogger])
+		case CLSFN_TASKDIST_OVERHEAD.String():
+			lg.attach(CLSFN_TASKDIST_OVERHEAD, Loggers[clsfnTaskDistOverheadLogger])
 		}
 	}
 }

logging/def/loggerObservers.go

@@ -84,6 +84,15 @@ func (loi *loggerObserverImpl) setLogFilePrefix(prefix string) {
 		schedPolicySwitchLogFilePrefix = loi.logDirectory + "/" + schedPolicySwitchLogFilePrefix
 	}
 	loi.logObserverSpecifics[spsLogger].logFilePrefix = schedPolicySwitchLogFilePrefix
+
+	// Setting logFilePrefix for clsfnTaskDist logger.
+	// Execution time of every call to def.GetTaskDistribution(...) would be recorded and logged in this file.
+	// The overhead would be logged in microseconds.
+	clsfnTaskDistOverheadLogFilePrefix := prefix + "_classificationOverhead.log"
+	if loi.logDirectory != "" {
+		clsfnTaskDistOverheadLogFilePrefix = loi.logDirectory + "/" + clsfnTaskDistOverheadLogFilePrefix
+	}
+	loi.logObserverSpecifics[clsfnTaskDistOverheadLogger].logFilePrefix = clsfnTaskDistOverheadLogFilePrefix
 }
 
 func (loi *loggerObserverImpl) setLogDirectory(dirName string) {

schedulers/base.go

@@ -404,3 +404,8 @@ func (s *BaseScheduler) LogSchedPolicySwitch(taskDist float64, name string, next
 		s.Log(elecLogDef.GENERAL, fmt.Sprintf("Switching... TaskDistribution[%d] ==> %s", taskDist, name))
 	}
 }
+
+func (s *BaseScheduler) LogClsfnAndTaskDistOverhead(overhead time.Duration) {
+	// Logging the overhead in microseconds.
+	s.Log(elecLogDef.CLSFN_TASKDIST_OVERHEAD, fmt.Sprintf("%f", float64(overhead.Nanoseconds())/1000.0))
+}

schedulers/electronScheduler.go

@@ -5,6 +5,7 @@ import (
 	elecLogDef "bitbucket.org/sunybingcloud/elektron/logging/def"
 	mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
 	sched "github.com/mesos/mesos-go/api/v0/scheduler"
+	"time"
 )
 
 // Implements mesos scheduler.
@@ -72,4 +73,6 @@ type ElectronScheduler interface {
 	LogTaskStatusUpdate(status *mesos.TaskStatus)
 	// Log Scheduling policy switches (if any)
 	LogSchedulingPolicySwitch()
+	// Log the computation overhead of classifying tasks in the scheduling window.
+	LogClsfnAndTaskDistOverhead(overhead time.Duration)
 }

schedulers/first-fit.go

@@ -6,7 +6,6 @@ import (
 	"bitbucket.org/sunybingcloud/elektron/utilities/offerUtils"
 	mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
 	sched "github.com/mesos/mesos-go/api/v0/scheduler"
-	"log"
 )
 
 // Decides if to take an offer or not

schedulers/schedPolicy.go

@@ -4,6 +4,7 @@ import (
 	"bitbucket.org/sunybingcloud/electron/def"
 	mesos "github.com/mesos/mesos-go/api/v0/mesosproto"
 	sched "github.com/mesos/mesos-go/api/v0/scheduler"
+	"time"
 )
 
 type SchedPolicyContext interface {
@@ -46,8 +47,13 @@ func (bsps *baseSchedPolicyState) SwitchIfNecessary(spc SchedPolicyContext) {
 		// The next scheduling policy will schedule at max schedWindowSize number of tasks.
 		baseSchedRef.schedWindowSize, baseSchedRef.numTasksInSchedWindow =
 			baseSchedRef.schedWindowResStrategy.Apply(func() interface{} { return baseSchedRef.tasks })
+                if baseSchedRef.schedWindowSize > 0 {
+			// Record overhead to classify the tasks in the scheduling window and using the classification results
+			// 	to determine the distribution of low power consuming and high power consuming tasks.
+			startTime := time.Now()
 			// Determine the distribution of tasks in the new scheduling window.
 			taskDist, err := def.GetTaskDistributionInWindow(baseSchedRef.schedWindowSize, baseSchedRef.tasks)
+			baseSchedRef.LogClsfnAndTaskDistOverhead(time.Now().Sub(startTime))
 			// If no resource offers have been received yet, and
 			// 	the name of the first scheduling policy to be deployed is provided,
 			// 	we switch to this policy regardless of the task distribution.
@@ -109,6 +115,9 @@ func (bsps *baseSchedPolicyState) SwitchIfNecessary(spc SchedPolicyContext) {
 			baseSchedRef.SwitchSchedPol(SchedPolicies[switchToPolicyName])
 			// Resetting the number of tasks scheduled.
 			bsps.numTasksScheduled = 0
+		} else {
+			// There is no need to switch the scheduling policy as there aren't any tasks in the window.
+		}
 	}
 }