Fixes, updates and refactor.

Author: Strabox

.gitignore

@@ -1,3 +1,4 @@
+/in
 /out
 /.idea
 *.exe

RequestStream.js

@@ -24,22 +24,24 @@ const logJsonFeederTag = "JS-FEEDER"
 
 // jsonFeeder generates a stream of user requests reading from a json file.
 type jsonFeeder struct {
-collector *metrics.Collector // Metrics collector that collects system level metrics.
-reqInjectionNode sync.Map // Map of ContainerID<->NodeIndex.
-systemTotalResources types.Resources // Caravela's maximum resources.
-randomGenerator *rand.Rand // Pseudo-random generator.
-simConfigs *configuration.Configuration // Simulator's configurations.
-caravelaConfigs *caravelaConfigs.Configuration // Caravela's configurations.
+collector *metrics.Collector // Metrics collector that collects system level metrics.
+containerInjectionNode sync.Map // Map of ContainerID<->NodeIndex.
+currentRequests sync.Map // Map of RequestID<->ContainerID.
+systemTotalResources types.Resources // Caravela's maximum resources.
+randomGenerator *rand.Rand // Pseudo-random generator.
+simConfigs *configuration.Configuration // Simulator's configurations.
+caravelaConfigs *caravelaConfigs.Configuration // Caravela's configurations.
 }
 
 // newJsonFeeder creates a new json feeder.
 func newJsonFeeder(simConfigs *configuration.Configuration, caravelaConfigs *caravelaConfigs.Configuration, rngSeed int64) (Feeder, error) {
 return &jsonFeeder{
-collector: nil,
-reqInjectionNode: sync.Map{},
-randomGenerator: rand.New(caravelaUtil.NewSourceSafe(rand.NewSource(rngSeed))),
-simConfigs: simConfigs,
-caravelaConfigs: caravelaConfigs,
+collector: nil,
+containerInjectionNode: sync.Map{},
+currentRequests: sync.Map{},
+randomGenerator: rand.New(caravelaUtil.NewSourceSafe(rand.NewSource(rngSeed))),
+simConfigs: simConfigs,
+caravelaConfigs: caravelaConfigs,
 }, nil
 }
 
@@ -57,18 +59,11 @@ func (j *jsonFeeder) Start(ticksChannel <-chan chan RequestTask) {
 CPUs float64 `json:"CPU request"`
 Memory float64 `json:"memory request"`
 }
+const maxJsonFiles = 18
+jsonFileCounter := 0
 tick := 0
 
-fileReader, err := os.Open("RequestStream.js")
-if err != nil {
-panic(fmt.Errorf("json feeder invalid request stream file: %s", err))
-}
-
-jsonRequestStream := json.NewDecoder(fileReader)
-_, err = jsonRequestStream.Token() // Read open bracket
-if err != nil {
-panic(err)
-}
+jsonRequestStream, currentJsonFile := j.getJsonRequestStream(fmt.Sprintf("in/Stream_%d.js", jsonFileCounter))
 
 for {
 select {
@@ -87,67 +82,104 @@ func (j *jsonFeeder) Start(ticksChannel <-chan chan RequestTask) {
 
 requestID := strconv.FormatInt(reqJson.JobID, 10)
 reqResources := j.generateRequestResources(reqJson.CPUs, reqJson.Memory)
-tickCpusAcc += reqResources.CPUs
-tickMemoryAcc += reqResources.Memory
 
 if reqJson.EventType == 1 && !j.requestExists(requestID) { // Deploy container request.
 
+tickCpusAcc += reqResources.CPUs
+tickMemoryAcc += reqResources.Memory
+j.currentRequests.Store(requestID, nil)
+
 newTickChan <- func(nodeIndex int, injectedNode *node.Node, currentTime time.Duration) {
-requestCtx := context.WithValue(context.Background(), types.RequestIDKey, requestID)
-j.collector.CreateRunRequest(nodeIndex, requestID, reqResources, currentTime)
-contStatus, err := injectedNode.SubmitContainers(
-requestCtx,
-[]types.ContainerConfig{{
-ImageKey: util.RandomName(),
-Name: util.RandomName(),
-PortMappings: caravela.EmptyPortMappings(),
-Args: caravela.EmptyContainerArgs(),
-Resources: reqResources,
-GroupPolicy: types.SpreadGroupPolicy,
-}})
-if err == nil {
-j.reqInjectionNode.Store(contStatus[0].ContainerID, injectedNode)
-j.collector.RunRequestSucceeded()
+if _, exist := j.currentRequests.Load(requestID); !exist {
+requestCtx := context.WithValue(context.Background(), types.RequestIDKey, requestID)
+j.collector.CreateRunRequest(nodeIndex, requestID, reqResources, currentTime)
+contStatus, err := injectedNode.SubmitContainers(
+requestCtx,
+[]types.ContainerConfig{{
+ImageKey: util.RandomName(),
+Name: util.RandomName(),
+PortMappings: caravela.EmptyPortMappings(),
+Args: caravela.EmptyContainerArgs(),
+Resources: reqResources,
+GroupPolicy: types.SpreadGroupPolicy,
+}})
+if err == nil {
+j.containerInjectionNode.Store(contStatus[0].ContainerID, injectedNode)
+j.currentRequests.Store(requestID, contStatus[0].ContainerID)
+j.collector.RunRequestSucceeded()
+}
+j.collector.ArchiveRunRequest(requestID)
 }
-j.collector.ArchiveRunRequest(requestID)
 }
 
 } else if (reqJson.EventType == 2 || reqJson.EventType == 3 || reqJson.EventType == 4 ||
 reqJson.EventType == 5 || reqJson.EventType == 6) && j.requestExists(requestID) { // Stop container request.
 
-j.reqInjectionNode.Range(func(key, value interface{}) bool {
-containerID, _ := key.(string)
-injectionNode, _ := value.(*node.Node)
-newTickChan <- func(_ int, _ *node.Node, _ time.Duration) {
-err := injectionNode.StopContainers(context.Background(), []string{containerID})
-if err != nil {
-//util.Log.Infof(util.LogTag(logRandFeederTag)+"Stop container FAILED, err: %s", err)
-}
-j.reqInjectionNode.Delete(containerID)
+contID, exist := j.currentRequests.Load(requestID)
+containerID, ok := contID.(string)
+if !exist || !ok {
+j.currentRequests.Delete(requestID)
+continue
+}
+
+injNode, exist := j.containerInjectionNode.Load(containerID)
+if !exist {
+continue
+//panic(fmt.Errorf("no mapping between request and container, %s", containerID))
+}
+injectionNode, ok := injNode.(*node.Node)
+if !ok {
+panic("node to *node.Node")
+}
+
+newTickChan <- func(_ int, _ *node.Node, _ time.Duration) {
+err := injectionNode.StopContainers(context.Background(), []string{containerID})
+if err != nil {
+//util.Log.Infof(util.LogTag(logRandFeederTag)+"Stop container FAILED, err: %s", err)
 }
-return false
-})
+j.currentRequests.Delete(requestID)
+j.containerInjectionNode.Delete(containerID)
+}
 
 }
 
 }
 
-if !jsonRequestStream.More() {
-fileReader.Close() // Close the request file.
-close(newTickChan) // No more user requests for this tick.
+if !jsonRequestStream.More() && (jsonFileCounter == maxJsonFiles) {
+fmt.Println("ASDDDDDDDDDDDDDDDDDDDDDDDD")
+currentJsonFile.Close() // Close the request file.
+close(newTickChan) // No more user requests for this tick.
 return
+} else if !jsonRequestStream.More() {
+jsonFileCounter++
+currentJsonFile.Close() // Close the request file.
+jsonRequestStream, currentJsonFile = j.getJsonRequestStream(fmt.Sprintf("in/Stream_%d.js", jsonFileCounter))
 }
 
 close(newTickChan) // No more user requests for this tick.
 } else { // Simulator closed ticks channel.
-fileReader.Close() // Close the request file.
-return // Stop feeding engine
+currentJsonFile.Close() // Close the request file.
+return  // Stop feeding engine
 }
 }
 tick++
 }
 }
 
+func (j *jsonFeeder) getJsonRequestStream(filePath string) (*json.Decoder, *os.File) {
+fileReader, err := os.Open(filePath)
+if err != nil {
+panic(fmt.Errorf("json feeder invalid request stream file: %s", err))
+}
+
+jsonRequestStream := json.NewDecoder(fileReader)
+_, err = jsonRequestStream.Token() // Read open bracket
+if err != nil {
+panic(err)
+}
+return jsonRequestStream, fileReader
+}
+
 // generateRequestResources ...
 func (j *jsonFeeder) generateRequestResources(normalizedCpus, normalizedMemory float64) types.Resources {
 cpuClasses := make([]int, len(j.caravelaConfigs.ResourcesPartitions().CPUClasses))
@@ -195,6 +227,6 @@ func (j *jsonFeeder) ratioSystemResources(cpus, memory int) float64 {
 
 // requestExists verifies if a request was already injected in the request stream.
 func (j *jsonFeeder) requestExists(requestID string) bool {
-_, ok := j.reqInjectionNode.Load(requestID)
-return ok
+_, exist := j.currentRequests.Load(requestID)
+return exist
 }

engine/feeder/json_feeder.go

@@ -1,6 +1,6 @@
-NumberOfNodes = 100000
+NumberOfNodes = 10000
 TickInterval = "20s"
-MaxTicks = 350
+MaxTicks = 90
 Multithread = true
 DiscoveryBackends = ["chord-single-offer", "chord-random"]
 RequestFeeder = "json" # random, json