import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
public StageScheduler createScheduler(final MsgRuntime runtime) {
			
	final StageScheduler scheduler = 
			 runtime.builder.threadLimit>0 ?				
	         StageScheduler.defaultScheduler(gm, 
	        		 runtime.builder.threadLimit, 
	        		 runtime.builder.threadLimitHard) :
	         StageScheduler.defaultScheduler(gm);

	Runtime.getRuntime().addShutdownHook(new Thread() {
		public void run() {
			scheduler.shutdown();
			scheduler.awaitTermination(getShutdownSeconds(), TimeUnit.SECONDS);
		}
	});
	
	
	return scheduler;
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
public void runGraph() {
	//logger.info("run graph");
	
	
	MonitorConsoleStage.attach(gm);
	
	StageScheduler scheduler = new ThreadPerStageScheduler(gm);
	
	Runtime.getRuntime().addShutdownHook(new Thread() {
	    public void run() {
	    	scheduler.shutdown();
	    	scheduler.awaitTermination(3, TimeUnit.SECONDS);
	    }
	});
	
	scheduler.startup();		
	//logger.info("running graph");
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Override
public StageScheduler createScheduler(MsgRuntime runtime) {

    if (isInUnitTest) {
                  
        //NOTE: need to consider different schedulers in the future.
       return new NonThreadScheduler(gm);
    } else {
       return super.createScheduler(runtime);
    }
          
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
public StageScheduler getScheduler() {
    return scheduler;
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Test
public void testLineReaderRollover() {
	//Tests many different primary ring sizes to force rollover at different points.
	//Checks that every run produces the same results as the previous run.
	if ("arm".equals(System.getProperty("os.arch"))) {
   		assertTrue(true);
   	}
   	
   	else {		
	final ByteArrayOutputStream baos = new ByteArrayOutputStream();
	
	byte[] last = null;
	
	ByteBuffer data = generateCVSData(21);		    
	int dataSize = data.limit();
	
	
	int t = 10;
	while (--t>=4) {
		data.position(0);
		data.limit(dataSize);
		
		PipeConfig linesRingConfigLocal = new PipeConfig(RawDataSchema.instance, 1<<t, 1<<20);	
		
		final Pipe linesRing = new Pipe(linesRingConfigLocal);		
		GraphManager gm = new GraphManager();
		LineSplitterByteBufferStage lineSplitter = new LineSplitterByteBufferStage(gm, data, linesRing);
		
		baos.reset();
		ToOutputStreamStage reader = new ToOutputStreamStage(gm,linesRing,baos, false);
		
		StageScheduler ss = new ThreadPerStageScheduler(gm);
		ss.startup();
		
		ss.awaitTermination(1, TimeUnit.MINUTES);
		
		//return the position to the beginning to run the test again
		
		byte[] results = baos.toByteArray();
		if (null!=last) {
			int j = 0;
			while (j<last.length && j<results.length) {
				if (last[j]!=results[j]) {
					System.err.println("missmatch found at:"+j);						
					break;
				}
				j++;
			}		
			assertEquals(last.length,results.length);
			assertTrue("Missed on "+t+" vs "+(t+1)+" at idx"+mismatchAt(last,results)+"\n",
					   Arrays.equals(last,  results));
		}
					
		last = Arrays.copyOf(results, results.length);
	
	}
	
   	}
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Test
public <M extends MatrixSchema<M>> void testRowsToColSplit() {
	
	int rows=10;
	int columns=6;
	MatrixSchema<M> schema = BuildMatrixCompute.buildSchema(rows, columns, MatrixTypes.Integers);
	ColumnSchema<M> cs = new ColumnSchema<M>(schema);
	RowSchema<M> rs = new RowSchema<M>(schema);

	GraphManager gm = new GraphManager();		
	PipeConfig<RowSchema<M>> rowConfig = new PipeConfig<RowSchema<M>>(rs, schema.getRows());		
	PipeConfig<ColumnSchema<M>> columnConfig = new PipeConfig<ColumnSchema<M>>(cs, 2);
	
	Pipe[] intputAsColumns = new Pipe[schema.getColumns()];
	
	Pipe<RowSchema<M>> inputRows = new Pipe<RowSchema<M>>(rowConfig); 

	int t = intputAsColumns.length;
	while (--t>=0) {
		intputAsColumns[t]=new Pipe<ColumnSchema<M>>(columnConfig);
	}
	
	
	new RowsToColumnRouteStage(gm, inputRows, intputAsColumns);
	int i = schema.getColumns();
	ByteArrayOutputStream[] targets = new ByteArrayOutputStream[i];
	PronghornStage[] watch = new PronghornStage[i];
	while (--i>=0) {
		targets[i] = new ByteArrayOutputStream();
		watch[i] =new ConsoleJSONDumpStage<>(gm, intputAsColumns[i], new PrintStream(targets[i]));
	}
	
	
	
	
	//StageScheduler scheduler = new ThreadPerStageScheduler(gm);
	//int targetThreadCount = 6;
	StageScheduler scheduler = new ThreadPerStageScheduler(gm);
			//new FixedThreadsScheduler(gm,targetThreadCount); //TODO: do not enable until the thread grouping is balanced in FixedThreadScheduler
	
	scheduler.startup();	
			
	for(int c=0;c<schema.getRows();c++) {
		while (!Pipe.hasRoomForWrite(inputRows)) {
			Thread.yield();
		}
		
		Pipe.addMsgIdx(inputRows, schema.rowId);		
		for(int r=0;r<schema.getColumns();r++) {
			Pipe.addIntValue(c, inputRows);
		}
		Pipe.confirmLowLevelWrite(inputRows, Pipe.sizeOf(inputRows, schema.rowId));
		Pipe.publishWrites(inputRows);	
		
	}
	Pipe.publishEOF(inputRows);
		
	i = schema.getColumns();
	while (--i>=0) {
		GraphManager.blockUntilStageBeginsShutdown(gm, watch[i], 500);//timeout in ms
	}
	
	scheduler.awaitTermination(2, TimeUnit.SECONDS);
	
	i = schema.getColumns();
	while (--i>=0) {
		String actualText = new String(targets[i].toByteArray());
		
		//System.out.println(actualText);
		
		assertTrue(actualText.contains("{\"1\":0}"));
		assertTrue(actualText.contains("{\"5\":4}"));
		assertTrue(actualText.contains("{\"10\":9}"));
		
		assertTrue(actualText.indexOf("{\"1\":0}") > actualText.indexOf("{\"1\":1}"));
		
	}
	
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Test
public <M extends MatrixSchema<M>> void testcolToRowsSplit() {
	
	int rows=10;
	int columns=6;
	MatrixSchema schema = BuildMatrixCompute.buildSchema(rows, columns, MatrixTypes.Integers);
	ColumnSchema<M> cs = new ColumnSchema<M>(schema);
	RowSchema<M> rs = new RowSchema<M>(schema);

	GraphManager gm = new GraphManager();		
	PipeConfig<RowSchema<M>> rowConfig = new PipeConfig<RowSchema<M>>(rs, schema.getRows());		
	PipeConfig<ColumnSchema<M>> columnConfig = new PipeConfig<ColumnSchema<M>>(cs, 2);
	
	Pipe<ColumnSchema<M>>[] columnsPipes = new Pipe[schema.getColumns()];
	
	int i = columnsPipes.length;
	while (--i>=0) {
		columnsPipes[i] = new Pipe<ColumnSchema<M>>(columnConfig);
	}
	
	Pipe<RowSchema<M>> rowsPipe = new Pipe<RowSchema<M>>(rowConfig); 
	
	
	new ColumnsToRowsStage(gm, columnsPipes, rowsPipe);
	ByteArrayOutputStream capture = new ByteArrayOutputStream();
	ConsoleJSONDumpStage<RowSchema<M>> watch = new ConsoleJSONDumpStage<>(gm, rowsPipe, new PrintStream(capture));
	
	
	StageScheduler scheduler = new ThreadPerStageScheduler(gm);
	scheduler.startup();

	int c = columnsPipes.length;
	while (--c>=0) {
		
		int size = Pipe.addMsgIdx(columnsPipes[c], schema.columnId);
		int r = rows;
		while (--r >= 0) {
			Pipe.addIntValue(r, columnsPipes[c]);				
		}
		Pipe.confirmLowLevelWrite(columnsPipes[c], size);
		Pipe.publishWrites(columnsPipes[c]);
					
		
	}
	
	GraphManager.blockUntilStageBeginsShutdown(gm, watch, 500);
	
	String actualText = new String(capture.toByteArray());
	//System.out.println(actualText);
	
	assertTrue(actualText.contains("{\"1\":0}"));
	assertTrue(actualText.contains("{\"5\":4}"));
	assertTrue(actualText.contains("{\"6\":9}"));
	
	assertTrue(actualText.indexOf("{\"1\":0}") > actualText.indexOf("{\"1\":1}"));
	
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
private void run() {
    
    
    StageScheduler scheduler = new ThreadPerStageScheduler(gm);
    scheduler.startup();
    
    Scanner scan = new Scanner(System.in);
    System.out.println("press enter to exit");
    scan.hasNextLine();//blocks until enter 
           
    System.out.println("exiting...");
    
    scheduler.shutdown();
    scheduler.awaitTermination(100, TimeUnit.MILLISECONDS);
    
    stage.printExitStatus();
            
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
private void run() {
    
    GraphManager gm = new GraphManager();
    
    buildApplicationGraph(gm);            
    
    StageScheduler scheduler = new ThreadPerStageScheduler(gm);  //TODO: Something cool would be a scheduler that runs in the netty event loop.
    scheduler.startup();
    
    //Call this method if we want to shut down early
    try {
        Thread.sleep(1000*60*60*12);//run for 12 hours then shutdown
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    scheduler.shutdown();   
    
    scheduler.awaitTermination(2, TimeUnit.SECONDS); //shutdown process should not take longer than 2 seconds or something is wrong
            
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Ignore
    public void twoGeneratorsTest() {
        
        FieldReferenceOffsetManager from = buildFROM();        
        assertTrue(null!=from);
        
        PipeConfig busConfig = new PipeConfig(new MessageSchemaDynamic(from), 10, 64);
        
        GraphManager gm = new GraphManager();
        
        
//simple test with no split        
        PronghornStage generator1 = new TestGenerator(gm, seed, iterations, pipe(busConfig));        
        PronghornStage generator2 = new TestGenerator(gm, seed, iterations, pipe(busConfig));   
        
        PronghornStage validateResults = new TestValidator(gm, 
                                                getOutputPipe(gm, generator1), 
                                                getOutputPipe(gm, generator2));
               
        
        //start the timer       
        final long start = System.currentTimeMillis();
        
        GraphManager.enableBatching(gm);
        
        StageScheduler scheduler = new ThreadPerStageScheduler(GraphManager.cloneAll(gm));        
        scheduler.startup();        
        
        //blocks until all the submitted runnables have stopped
       

        //this timeout is set very large to support slow machines that may also run this test.
        boolean cleanExit = scheduler.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);
      
        long duration = System.currentTimeMillis()-start;
        
        
    }
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Ignore
    public void splitterTest() {
        
        FieldReferenceOffsetManager from = buildFROM();        
        assertTrue(null!=from);
        
        PipeConfig busConfig = new PipeConfig(new MessageSchemaDynamic(from), 10, 64);
        
        GraphManager gm = new GraphManager();
        
        int seed = 420;
        int iterations = 10;

        
//simple test using split
        PronghornStage generator = new TestGenerator(gm, seed, iterations, pipe(busConfig));        
        ReplicatorStage splitter = new ReplicatorStage(gm, getOutputPipe(gm, generator), pipe(busConfig.grow2x()), pipe(busConfig.grow2x()));       
        PronghornStage validateResults = new TestValidator(gm, getOutputPipe(gm, splitter, 2), getOutputPipe(gm, splitter, 1));
  
        
        
        //start the timer       
        final long start = System.currentTimeMillis();
        
      //  GraphManager.enableBatching(gm);
        
        StageScheduler scheduler = new ThreadPerStageScheduler(GraphManager.cloneAll(gm));        
        scheduler.startup();        
        
        //blocks until all the submitted runnables have stopped
       
        //this timeout is set very large to support slow machines that may also run this test.
        boolean cleanExit = scheduler.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);
      
        long duration = System.currentTimeMillis()-start;
        
        
    }
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
private void run(final int maxPipes, GraphManager graphManager,	MQTTStage[] outputStages) {
	
	//Add monitoring
	//MonitorConsoleStage.attach(graphManager);
			
	//Enable batching
	//GraphManager.enableBatching(graphManager);
	
	StageScheduler scheduler = new ThreadPerStageScheduler(graphManager);
	
	long start = System.currentTimeMillis();
	scheduler.startup();		 
	 
	Scanner scan = new Scanner(System.in);
	System.out.println("press enter to exit");
	scan.hasNextLine();
	
	System.out.println("exiting...");
	scheduler.shutdown();
			
	long TIMEOUT_SECONDS = 10;
	scheduler.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);

	long duration = System.currentTimeMillis() - start;
	
	showTotals(maxPipes, outputStages, duration);
}
 

import com.ociweb.pronghorn.stage.scheduling.StageScheduler; //导入依赖的package包/类
@Test
public void runtTest() {
	
	InputStream demoFileStream = TestPipeline.class.getResourceAsStream("/exampleMessages.csv");    
	assert(null!=demoFileStream);
	
	byte[] data = null;
	try {
		data = new byte[demoFileStream.available()];
		demoFileStream.read(data);
	} catch (IOException e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
	//System.out.println(Arrays.toString(data));
	
	ByteBuffer byteBuffer = ByteBuffer.wrap(data);
	
	PipeConfig linesRingBufferConfig = new PipeConfig((byte)6,(byte)15,null, RawDataSchema.instance);
	PipeConfig messagesConfig = new PipeConfig((byte)6,(byte)15,null,  new MessageSchemaDynamic(MQTTFROM.from));
	
	Pipe linesRingBuffer = new Pipe(linesRingBufferConfig);
	Pipe messagesRingBuffer = new Pipe(messagesConfig);
	
	
	GraphManager graphManager = new GraphManager();
	LineSplitterByteBufferStage lineSplitterStage = new LineSplitterByteBufferStage(graphManager, byteBuffer, linesRingBuffer);
	
	int maxClientsBits = 10;
	int base = 1;
	String server = "";
	String clientPrefix = "";
	MessageCSVStage csvStage = new MessageCSVStage(graphManager, linesRingBuffer, messagesRingBuffer, maxClientsBits, base, server, clientPrefix);
	
	DumpCheckStage dumpStage = new DumpCheckStage(graphManager, messagesRingBuffer);

	StageScheduler scheduler = new ThreadPerStageScheduler(GraphManager.cloneAll(graphManager));
	scheduler.startup();

	long TIMEOUT_SECONDS = 2;
	boolean cleanExit = scheduler.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS);
	
	
	
	
	
}