<commit_before>package com.neverwinterdp.scribengin.dataflow.tracking;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import com.beust.jcommander.Parameter;
import com.neverwinterdp.message.TrackingWindowRegistry;
import com.neverwinterdp.message.TrackingWindowReport;
import com.neverwinterdp.registry.txevent.TXEvent;
import com.neverwinterdp.scribengin.ScribenginClient;
import com.neverwinterdp.scribengin.dataflow.Dataflow;
import com.neverwinterdp.scribengin.dataflow.DataflowClient;
import com.neverwinterdp.scribengin.dataflow.DataflowEvent;
import com.neverwinterdp.scribengin.dataflow.DataflowSubmitter;
import com.neverwinterdp.scribengin.shell.ScribenginShell;
import com.neverwinterdp.util.text.DateUtil;
import com.neverwinterdp.util.text.TabularFormater;
import com.neverwinterdp.vm.VMDescriptor;
import com.neverwinterdp.vm.client.VMClient;

public class TrackingWithSimulationLauncher extends TrackingLauncher {
  @Parameter(names = "--simulation-report-period", description = "")
  long simulationReportPeriod = 5000;
  
  @Parameter(names = "--simulation-period", description = "")
  long simulationPeriod = 30000;
  
  @Parameter(names = "--simulation-max", description = "")
  int simulationMax = 2;
  
  boolean simulateKill = false;
  
  List<SimulationLog> simulationLogs = new ArrayList<>();
  
  public TrackingWithSimulationLauncher setSimulateKill() {
    simulateKill = true;
    return this;
  }
  
  @Override
  public void execute(ScribenginShell shell, TrackingDataflowBuilder dflBuilder) throws Exception {
    VMClient vmClient = shell.getVMClient();
    submitVMGenerator(vmClient, dflBuilder);
    
    Dataflow<TrackingMessage, TrackingMessage> dfl = dflBuilder.buildDataflow();
    DataflowSubmitter submitter = submitDataflow(shell, dfl);
    DataflowClient dflClient = submitter.getDataflowClient(60000);
    
    submitVMValidator(vmClient, dflBuilder);
    
    String reportPath = dflBuilder.getTrackingConfig().getReportPath();
    for(int i = 0; i < simulationMax; i++) {
      long stopTime = System.currentTimeMillis() + simulationPeriod;
      long reportTime = 0;
      while(System.currentTimeMillis() < stopTime) {
        Thread.sleep(1000);
        reportTime +=  1000;
        if(reportTime >= simulationReportPeriod) {
          shell.execute(
              "plugin com.neverwinterdp.scribengin.dataflow.tracking.TrackingMonitor" +
              "  --dataflow-id " + dfl.getDataflowId() + " --report-path " + reportPath
          );
          reportTime = 0;
          shell.console().println(SimulationLog.toFormattedText(simulationLogs));
        }
      }
      
//      stopStartDataflow(shell, dflBuilder);
      if(i % 2 == 0) {
        killWorker(dflClient, dflBuilder.getDataflowId());
      } else {
        stopStartDataflow(shell, dflBuilder);
      }
    }
  }
  
  
  public void stopStartDataflow(ScribenginShell shell, TrackingDataflowBuilder dflBuilder) throws Exception {
    SimulationLog log = new SimulationLog("stop-start-dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Stop Dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    ScribenginShell scribenginShell = (ScribenginShell) shell;
    ScribenginClient scribenginClient = scribenginShell.getScribenginClient();
    String dataflowId = dflBuilder.getDataflowId();
    DataflowClient dflClient = scribenginClient.getDataflowClient(dataflowId);
    TXEvent stopEvent = new TXEvent("stop", DataflowEvent.Stop);
    dflClient.getDataflowRegistry().getMasterRegistry().getMaserEventBroadcaster().broadcast(stopEvent);
    
    TrackingWindowRegistry mtRegistry = dflClient.getDataflowRegistry().getMessageTrackingRegistry();
    while(true) {
      TrackingWindowReport report  = mtRegistry.getReport();
      System.err.println("Stop: tracking count = " + report.getTrackingCount());
      if(mtRegistry.getProgressCommitWindows().size() == 0 && report.getTrackingCount() > 0) break;
      Thread.sleep(1000);
    }
    shell.execute("dataflow wait-for-status --dataflow-id "  + dataflowId + " --status TERMINATED --timeout 90000 --report-period 10000") ;
    
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Start Dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    Dataflow<TrackingMessage, TrackingMessage> dfl = dflBuilder.buildDataflow();<commit_msg>fix tracking<commit_after>package com.neverwinterdp.scribengin.dataflow.tracking;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import com.beust.jcommander.Parameter;
import com.neverwinterdp.message.TrackingWindowRegistry;
import com.neverwinterdp.message.TrackingWindowReport;
import com.neverwinterdp.registry.txevent.TXEvent;
import com.neverwinterdp.scribengin.ScribenginClient;
import com.neverwinterdp.scribengin.dataflow.Dataflow;
import com.neverwinterdp.scribengin.dataflow.DataflowClient;
import com.neverwinterdp.scribengin.dataflow.DataflowEvent;
import com.neverwinterdp.scribengin.dataflow.DataflowSubmitter;
import com.neverwinterdp.scribengin.shell.ScribenginShell;
import com.neverwinterdp.util.text.DateUtil;
import com.neverwinterdp.util.text.TabularFormater;
import com.neverwinterdp.vm.VMDescriptor;
import com.neverwinterdp.vm.client.VMClient;

public class TrackingWithSimulationLauncher extends TrackingLauncher {
  @Parameter(names = "--simulation-report-period", description = "")
  long simulationReportPeriod = 5000;
  
  @Parameter(names = "--simulation-period", description = "")
  long simulationPeriod = 30000;
  
  @Parameter(names = "--simulation-max", description = "")
  int simulationMax = 2;
  
  boolean simulateKill = false;
  
  List<SimulationLog> simulationLogs = new ArrayList<>();
  
  public TrackingWithSimulationLauncher setSimulateKill() {
    simulateKill = true;
    return this;
  }
  
  @Override
  public void execute(ScribenginShell shell, TrackingDataflowBuilder dflBuilder) throws Exception {
    VMClient vmClient = shell.getVMClient();
    submitVMGenerator(vmClient, dflBuilder);
    
    Dataflow<TrackingMessage, TrackingMessage> dfl = dflBuilder.buildDataflow();
    DataflowSubmitter submitter = submitDataflow(shell, dfl);
    DataflowClient dflClient = submitter.getDataflowClient(60000);
    
    submitVMValidator(vmClient, dflBuilder);
    
    String reportPath = dflBuilder.getTrackingConfig().getReportPath();
    for(int i = 0; i < simulationMax; i++) {
      long stopTime = System.currentTimeMillis() + simulationPeriod;
      long reportTime = 0;
      while(System.currentTimeMillis() < stopTime) {
        Thread.sleep(1000);
        reportTime +=  1000;
        if(reportTime >= simulationReportPeriod) {
          shell.execute(
              "plugin com.neverwinterdp.scribengin.dataflow.tracking.TrackingMonitor" +
              "  --dataflow-id " + dfl.getDataflowId() + " --report-path " + reportPath
          );
          reportTime = 0;
          shell.console().println(SimulationLog.toFormattedText(simulationLogs));
        }
      }
      
//      stopStartDataflow(shell, dflBuilder);
      if(i % 2 == 0) {
        killWorker(dflClient, dflBuilder.getDataflowId());
      } else {
        stopStartDataflow(shell, dflBuilder);
      }
    }
  }
  
  
  public void stopStartDataflow(ScribenginShell shell, TrackingDataflowBuilder dflBuilder) throws Exception {
    SimulationLog log = new SimulationLog("stop-start-dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Stop Dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    ScribenginShell scribenginShell = (ScribenginShell) shell;
    ScribenginClient scribenginClient = scribenginShell.getScribenginClient();
    String dataflowId = dflBuilder.getDataflowId();
    DataflowClient dflClient = scribenginClient.getDataflowClient(dataflowId);
    TXEvent stopEvent = new TXEvent("stop", DataflowEvent.Stop);
    dflClient.getDataflowRegistry().getMasterRegistry().getMaserEventBroadcaster().broadcast(stopEvent);
    
    TrackingWindowRegistry mtRegistry = dflClient.getDataflowRegistry().getMessageTrackingRegistry();
    while(true) {
      TrackingWindowReport report  = mtRegistry.getReport();
      int commitWindowLeft = mtRegistry.getProgressCommitWindows().size();
      System.err.println("Stop: tracking count = " + report.getTrackingCount() + ",  commitWindowLeft = " + commitWindowLeft);
      if(mtRegistry.getProgressCommitWindows().size() == 0 && report.getTrackingCount() > 0) break;
      Thread.sleep(1000);
    }
    shell.execute("dataflow wait-for-status --dataflow-id "  + dataflowId + " --status TERMINATED --timeout 90000 --report-period 10000") ;
    
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Start Dataflow");
    System.err.println("--------------------------------------------------------------------------------------");
    Dataflow<TrackingMessage, TrackingMessage> dfl = dflBuilder.buildDataflow();
{'fraction_non_alphanumeric': 0.17415254237288136, 'fraction_numerical': 0.009322033898305085, 'mean_word_length': 2.9869087837837838, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, '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': '1572066', 'n_tokens_mistral': 2686, 'n_tokens_neox': 2618, 'n_words': 431}