import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
protected void logStageScheduleRates() {
    int totalStages = GraphManager.countStages(gm);
       for(int i=1;i<=totalStages;i++) {
           PronghornStage s = GraphManager.getStage(gm, i);
           if (null != s) {
               
               Object rate = GraphManager.getNota(gm, i, GraphManager.SCHEDULE_RATE, null);
               if (null == rate) {
                   logger.debug("{} is running without breaks",s);
               } else  {
                   logger.debug("{} is running at rate of {}",s,rate);
               }
           }
           
       }
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static GraphManager cloneStagesWithNotaKey(GraphManager m, Object key) {
	GraphManager clone = new GraphManager();
	//register each stage
	int i = m.stageIdToStage.length;
	while (--i>=0) {
		PronghornStage stage = m.stageIdToStage[i];
		if (null!=stage) {
			//copy this stage if it has the required key
			if (m != getNota(m, stage, key, m)) {
				copyStage(m, clone, stage);
				copyNotasForStage(m, clone, stage);
			}
		}
	}
	return clone;
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static PronghornStage getStageWithNotaKey(GraphManager m, Object key, int ordinal) {
   
   int i = m.stageIdToStage.length;
   while (--i>=0) {
       PronghornStage stage = m.stageIdToStage[i];
       if (null!=stage) {
           //count this stage if it has the required key
           if (null != getNota(m, stage, key, null)) {
               if (--ordinal<=0) {
                   return stage;
               }
           }
       }
   }
   throw new UnsupportedOperationException("Invalid configuration. Unable to find requested ordinal "+ordinal);
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static GraphManager cloneStagesWithNotaKeyValue(GraphManager m, Object key, Object value) {
	GraphManager clone = new GraphManager();
	//register each stage
	int i = m.stageIdToStage.length;
	while (--i>=0) {
		PronghornStage stage = m.stageIdToStage[i];
		if (null!=stage) {
			//copy this stage if it has the required key
			if (value.equals(getNota(m, stage, key, null))) {
				copyStage(m, clone, stage);
				copyNotasForStage(m, clone, stage);
			}
		}
	}
	return clone;
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static boolean isInputLocal(int startIdx,
									int stopIdx, 
		                            GraphManager gm, 
		                            PronghornStage[] stages, 
		                            int[] script,
		                            int goalId) {
	//scan for an output which matches this goal Id
	
	for(int i = startIdx; i<=stopIdx; i++) {
		int stageId = stages[script[i]].stageId;
		int outC = GraphManager.getOutputPipeCount(gm, stageId);
		for(int k = 1; k <= outC; k++) {
			if (goalId == GraphManager.getOutputPipe(gm, stageId, k).id) {
				return true;
			}
		}
	}
	return false;
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static void copyStage(GraphManager m, GraphManager clone, PronghornStage stage) {
	int stageId = beginStageRegister(clone, stage);
	
	int idx;
	int ringId;
	
	idx = m.stageIdToInputsBeginIdx[stageId];
	while (-1 != (ringId=m.multInputIds[idx++])) {	
		assert(0==Pipe.contentRemaining(m.pipeIdToPipe[ringId]));
		regInput(clone, m.pipeIdToPipe[ringId], stageId);					
	}				
	
	idx = m.stageIdToOutputsBeginIdx[stageId];
	while (-1 != (ringId=m.multOutputIds[idx++])) {					
		assert(0==Pipe.contentRemaining(m.pipeIdToPipe[ringId]));
		regOutput(clone, m.pipeIdToPipe[ringId], stageId);					
	}		
	
	endStageRegister(clone, stage);
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static void register(GraphManager gm, PronghornStage stage, Pipe[] inputs, Pipe[] outputs) {
	
	synchronized(gm.lock) {
		int stageId = beginStageRegister(gm, stage);
		setStateToNew(gm, stageId);
		
		int i=0;
		int limit = inputs.length;
		while (i<limit) {
			regInput(gm,inputs,stageId,i,inputs[i++]);
		}
		
		//loop over outputs
		i = 0;
		limit = outputs.length;
		while (i<limit) {
			regOutput(gm, outputs, stageId, i, outputs[i++]);
		}
		
		endStageRegister(gm, stage);
								
	}
	
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static PronghornStage getOutputStage(GraphManager m, int ordinal) {
    int count = 0;
    int i = m.stageIdToStage.length;
    while (--i>=0) {
        if (null!=m.stageIdToStage[i]) {                
            //an input stage is one that has no input ring buffers
            if (-1 == m.multOutputIds[m.stageIdToOutputsBeginIdx[m.stageIdToStage[i].stageId]]) {
                if (!stageForMonitorData(m, m.stageIdToStage[i])) {
                    if (++count==ordinal) {
                        return m.stageIdToStage[i];
                    }
                }
            }
        }
    }
    throw new UnsupportedOperationException("Invalid configuration. Unable to find requested output ordinal "+ordinal);
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static void populateRanks(GraphManager m, Map<Object, StringBuilder> ranks, PronghornStage stage,
		String stageId) {
	if (ranks!=null && m.cachedRanks==null) {
		//thes rank keys are cached
		Object rankKey = getNota(m, stage.stageId, GraphManager.DOT_RANK_NAME, null);
		if (rankKey!=null) {
			
			//{ rank=same, b, c, d }
			StringBuilder b = ranks.get(rankKey);
			if (null==b) {
				b = new StringBuilder("{ rank=same");
				ranks.put(rankKey, b);
			}
			
			b.append(" \"").append(stageId).append("\",");
			
		}
	}
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static void enableBatching(GraphManager gm) {
	int j = gm.pipeIdToPipe.length;
	while (--j>=0) {
		Pipe ring = gm.pipeIdToPipe[j];
		//never enable batching on the monitor rings
		if (null!=ring && !ringHoldsMonitorData(gm, ring) ) {
			
			int ringId1 = ring.id;
			int stageId1 = GraphManager.getRingConsumerId(gm, ringId1);
			if (stageId1>=0) {
				if (PronghornStage.supportsBatchedRelease(gm.stageIdToStage[stageId1])) { 
					Pipe.setMaxReleaseBatchSize(ring);
				}
			}
			
			int ringId = ring.id;
			int stageId = GraphManager.getRingProducerId(gm, ringId);
			if (stageId>=0) {
				if (PronghornStage.supportsBatchedPublish(gm.stageIdToStage[stageId])) {
					Pipe.setMaxPublishBatchSize(ring);
				}
			}
		}
	}
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
/**
 * Start with ordinal selection of input stages then ordinal selection of each output ring there after.
 * TODO: do generic return that extends pronghornStage
 * @param m
 * @param path
 */
public static PronghornStage findStageByPath(GraphManager m, int ... path) {
	
	int ordinal = path[0];
	int i = 0;
    int limit = m.stageIdToStage.length;
	while (i<limit) {
		if (null!=m.stageIdToStage[i]) {				
			//an input stage is one that has no input ring buffers
			if (-1 == m.multInputIds[m.stageIdToInputsBeginIdx[m.stageIdToStage[i].stageId]]) {
				if (--ordinal<=0) {
					//starting from 1 find this path
					return findStageByPath(m, m.stageIdToStage[i], 1, path);
				}
			}
		}
		i++;
	}	
	throw new UnsupportedOperationException("Unable to find ordinal input stage of "+path[0]);
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private void createSchedulers(GraphManager graphManager, PronghornStage[][] stageArrays) {

	/////////////
    //for each array of stages create a scheduler
    ///////////// 
    ntsArray = new NonThreadScheduler[threadCount];

    int k = stageArrays.length;
    int ntsIdx = 0;
    while (--k >= 0) {
    	if (null!=stageArrays[k]) {
    		
    		if (logger.isDebugEnabled()) {
    			logger.debug("{} Single thread for group {}", ntsIdx, Arrays.toString(stageArrays[k]) );
    		}
    		PronghornStage pronghornStage = stageArrays[k][stageArrays[k].length-1];
			String name = pronghornStage.stageId+":"+pronghornStage.getClass().getSimpleName()+"...";
    		
    		
    		ntsArray[ntsIdx++]=new NonThreadScheduler(graphManager, stageArrays[k], name, true);	    		     
    	}
    }
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static void accumWhenZero(GraphManager graphManager, int stageId, long duration) {
	PronghornStage stage = getStage(graphManager, stageId);
	
	if ((stage instanceof PipeCleanerStage) ||
		(stage instanceof ReplicatorStage) ) {
		
		//these can be very fast and should not be logged.
		
	} else {
	
		int x = totalZeroDurations.incrementAndGet();
						
		if (Integer.numberOfLeadingZeros(x-1)!=Integer.numberOfLeadingZeros(x)) {
			if (duration<0) {
				logger.info("Bad duration {}",duration);
			} else {
				logger.info("Warning: the OS has measured stages taking zero ms {} times. "
					+ "Most recent case is for {}.", x, stage);
			}
		}
		
		graphManager.stageRunNS[stageId] += defaultDurationWhenZero;
	}
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public static RunningStdDev stdDevPipesPerStage(GraphManager m) {
	   
	   if (null == m.stdDevPipes) {
		 
		   m.stdDevPipes = new RunningStdDev();
		   int i = m.stageIdToStage.length;
	       while (--i>=0) {
	           PronghornStage stage = m.stageIdToStage[i];
	           if (null!=stage) {
	        	  
	        	  int sample =  
	        	  getInputPipeCount(m, stage.stageId)+
	        	  getOutputPipeCount(m, stage.stageId);
	        	  
	        	  RunningStdDev.sample(m.stdDevPipes, sample);
	        	  
	           }
	       }		       
	   }
      return m.stdDevPipes;
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private void add(PronghornStage[] pronghornStages, PronghornStage stage, final int root, final GraphManager graphManager, final IntHashTable rootsTable) {
	int i = 0;
	while (i<pronghornStages.length && pronghornStages[i]!=null) {
		if (pronghornStages[i]==stage) {
			return;//already added
		}
		i++;
	}
	//now add the new stage at index i
	pronghornStages[i]=stage;
	
	//Recursively add the ones under the same root.
	
	int outputCount = GraphManager.getOutputPipeCount(graphManager, stage.stageId);
	for(int r = 1; r<=outputCount; r++) {
		Pipe outputPipe = GraphManager.getOutputPipe(graphManager, stage, r);
		
		int consumerId = GraphManager.getRingConsumerId(graphManager, outputPipe.id);
		//this exists and has the same root so add it
		if (consumerId>=0 && rootId(consumerId, rootsTable)==root) {
			add(pronghornStages, GraphManager.getStage(graphManager, consumerId), root, graphManager, rootsTable);
		}
	}
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static int[] buildProducersList(int count, int idx, final GraphManager graphManager, PronghornStage[] stages) {
	
		//skip over the non producers
		while (idx<stages.length) {
			
			if (null!=GraphManager.getNota(graphManager, stages[idx].stageId, GraphManager.PRODUCER, null) || 
				(0==GraphManager.getInputPipeCount(graphManager, stages[idx])) ) {
				int[] result = buildProducersList(count+1, idx+1, graphManager, stages);    		
				result[count] = idx;    		
				return result;
			}
			
			idx++;
		}
		
		return new int[count];
		
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
private static long runStageWithRate(GraphManager graphManager, long nearestNextRun, int s, long rate, PronghornStage stage, NonThreadScheduler that) {
	//check time and only run if valid
	long start = System.nanoTime();
	                    		
	long nextRun = that.lastRun[s]+rate;
	long nsDelay = nextRun - start;
	if (nsDelay<=0) {
		//logger.info("running stage {}",stage);
		run(that.graphManager, stage, that);
		that.lastRun[s] = start;
		
		long now = System.nanoTime();
		GraphManager.accumRunTimeNS(graphManager, stage.stageId, now-start, now);
					
		nearestNextRun = Math.min(nearestNextRun, start+rate);
	} else {    
		//logger.info("skipped stage {}",stage);
		nearestNextRun = Math.min(nearestNextRun, nextRun);
	}
	return nearestNextRun;
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public int weight(Pipe p) {
	
	if (weights[p.id]==0) {
	
		int result = (int)p.config().slabBits();
		
		//returns the max pipe length from this pipe or any of the pipes that feed its producer.
		//if this value turns out to be large then we should probably not join these two stages.
		
		int producerId = GraphManager.getRingProducerId(graphManager, p.id);		
		if (producerId>=0) {
			PronghornStage producer = GraphManager.getStage(graphManager, producerId);			
			int count = GraphManager.getInputPipeCount(graphManager, producer);
			while (--count>=0) {
				Pipe inputPipe = GraphManager.getInputPipe(graphManager, producer, count);				
				result = Math.max(result, inputPipe.config().slabBits());
			}
		} else {
			//no producer found, an external thread must be pushing data into this, there is nothing to combine it with				
		}
		weights[p.id] = result;
	} 
	return weights[p.id];
	
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
public ReactiveManagerPipeConsumer(Object obj, ReactiveOperators operators, Pipe[] inputs) {
	
	this.obj = obj;
	this.inputs = inputs;
	assert(PronghornStage.noNulls(inputs));
	this.operators = new ReactiveOperator[inputs.length];
	
	int i = inputs.length;
	while (--i>=0) {
		this.operators[i] = operators.getOperator(inputs[i]);
	}
}
 

import com.ociweb.pronghorn.stage.PronghornStage; //导入依赖的package包/类
@Override
public void startup() {
	super.startup();
	percentileValues = new int[Pipe.totalPipes()+1];
	trafficValues = new long[Pipe.totalPipes()+1];
	
	int i = inputs.length;
	pctFull = new short[i];
	hists = new Histogram[i];
	while (--i>=0) {
		hists[i] = new Histogram(10000,2); 
	}
			
	position = inputs.length;
	
	observedPipeId = new int[inputs.length];
	Arrays.fill(observedPipeId, -1);
	observedPipeBytesAllocated = new long[inputs.length];
	observedPipeName = new String[inputs.length];
	
	int j = inputs.length;
	while (--j>=0) {
		int stageId = GraphManager.getRingProducerStageId(graphManager, inputs[j].id);	
           PronghornStage producer = GraphManager.getStage(graphManager, stageId);
           if (producer instanceof PipeMonitorStage) {
           	PipeMonitorStage p = (PipeMonitorStage)producer;
           	
           	observedPipeId[j] = p.getObservedPipeId();
           	observedPipeBytesAllocated[j] = p.getObservedPipeBytesAllocated();
           	observedPipeName[j] = p.getObservedPipeName();
           	
           }
           
	}
	
	
}