RegionManagerImpl implements RegionManager { @Override public Region createRegion ( final FlowDef flow, final RegionExecPlan regionExecPlan ) { checkArgument( flow != null, "flow is null" ); checkState( !regions.containsKey( regionExecPlan.getRegionId() ), "Region %s is already created!", regionExecPlan.getRegionId() ); final int regionId = regionExecPlan.getRegionId(); final int replicaCount = regionExecPlan.getReplicaCount(); final int pipelineCount = regionExecPlan.getPipelineCount(); final RegionDef regionDef = regionExecPlan.getRegionDef(); final int regionOperatorCount = regionDef.getOperatorCount(); checkPipelineStartIndices( regionId, regionOperatorCount, regionExecPlan.getPipelineStartIndices() ); LOGGER.debug( "Creating components for regionId={} pipelineCount={} replicaCount={}", regionId, pipelineCount, replicaCount ); if ( regionDef.getRegionType() == PARTITIONED_STATEFUL ) { partitionService.createPartitionDistribution( regionId, replicaCount ); } final PipelineReplica[][] pipelineReplicas = new PipelineReplica[ pipelineCount ][ replicaCount ]; final SchedulingStrategy[] schedulingStrategies = new SchedulingStrategy[ regionOperatorCount ]; final UpstreamCtx[] upstreamCtxes = new UpstreamCtx[ regionOperatorCount ]; preInitializeRegionOperators( flow, regionDef, schedulingStrategies, upstreamCtxes ); for ( int pipelineIndex = 0; pipelineIndex < pipelineCount; pipelineIndex++ ) { final int pipelineId = regionExecPlan.getPipelineStartIndex( pipelineIndex ); final OperatorDef[] operatorDefs = regionExecPlan.getOperatorDefsByPipelineStartIndex( pipelineId ); final int pipelineOperatorCount = operatorDefs.length; final SchedulingStrategy[] pipelineSchedulingStrategies = new SchedulingStrategy[ pipelineOperatorCount ]; arraycopy( schedulingStrategies, pipelineId, pipelineSchedulingStrategies, 0, pipelineOperatorCount ); final int[] fusionStartIndices = findFusionStartIndices( pipelineSchedulingStrategies ); final int pipelineOperatorReplicaCount = fusionStartIndices.length; final OperatorReplica[][] operatorReplicas = new OperatorReplica[ replicaCount ][ pipelineOperatorReplicaCount ]; LOGGER.debug( "Initializing pipeline replica for regionId={} pipelineIndex={} with {} fused operators", regionId, pipelineIndex, pipelineOperatorReplicaCount ); final PipelineReplicaId[] pipelineReplicaIds = createPipelineReplicaIds( regionId, replicaCount, pipelineId ); final int forwardedKeySize = regionDef.getForwardedKeySize(); final PipelineReplicaMeter[] replicaMeters = new PipelineReplicaMeter[ replicaCount ]; for ( int replicaIndex = 0; replicaIndex < replicaCount; replicaIndex++ ) { replicaMeters[ replicaIndex ] = new PipelineReplicaMeter( config.getMetricManagerConfig().getTickMask(), pipelineReplicaIds[ replicaIndex ], operatorDefs[ 0 ] ); } for ( int i = 0; i < pipelineOperatorReplicaCount; i++ ) { final int fromOperatorIndex = fusionStartIndices[ i ]; final OperatorQueue[] operatorQueues = createOperatorQueues( flow, regionId, replicaCount, ( fromOperatorIndex == 0 ), operatorDefs[ fromOperatorIndex ], forwardedKeySize ); final TupleQueueDrainerPool[] drainerPools = createTupleQueueDrainerPools( regionId, replicaCount, ( fromOperatorIndex == 0 ), operatorDefs[ fromOperatorIndex ] ); final int toOperatorIndex = ( i < fusionStartIndices.length - 1 ) ? fusionStartIndices[ i + 1 ] : pipelineOperatorCount; final int fusedOperatorCount = toOperatorIndex - fromOperatorIndex; final OperatorDef[][] fusedOperatorDefs = new OperatorDef[ replicaCount ][ fusedOperatorCount ]; final InternalInvocationCtx[][] fusedInvocationCtxes = new InternalInvocationCtx[ replicaCount ][ fusedOperatorCount ]; for ( int j = fusedOperatorCount - 1; j >= 0; j-- ) { final OperatorDef operatorDef = operatorDefs[ fromOperatorIndex + j ]; final int inputPortCount = operatorDef.getInputPortCount(); final OperatorKVStore[] operatorKvStores = createOperatorKVStores( regionId, replicaCount, operatorDef ); for ( int replicaIndex = 0; replicaIndex < replicaCount; replicaIndex++ ) { fusedOperatorDefs[ replicaIndex ][ j ] = operatorDef; final Function<PartitionKey, KVStore> kvStoreSupplier = operatorKvStores[ replicaIndex ]::getKVStore; final OutputCollector outputCollector; if ( j == fusedOperatorCount - 1 ) { final TuplesImpl output = new TuplesImpl( operatorDef.getOutputPortCount() ); outputCollector = new DefaultOutputCollector( output ); } else { outputCollector = (OutputCollector) fusedInvocationCtxes[ replicaIndex ][ j + 1 ]; } final InternalInvocationCtx invocationCtx; if ( j == 0 ) { invocationCtx = new DefaultInvocationCtx( inputPortCount, kvStoreSupplier, outputCollector ); } else if ( operatorDef.getOperatorType() == PARTITIONED_STATEFUL ) { final List<String> partitionFieldNames = operatorDef.getPartitionFieldNames(); final PartitionKeyExtractor ext = partitionKeyExtractorFactory.createPartitionKeyExtractor( partitionFieldNames, forwardedKeySize ); invocationCtx = new FusedPartitionedInvocationCtx( inputPortCount, kvStoreSupplier, ext, outputCollector ); } else { invocationCtx = new FusedInvocationCtx( inputPortCount, kvStoreSupplier, outputCollector ); } fusedInvocationCtxes[ replicaIndex ][ j ] = invocationCtx; } } for ( int replicaIndex = 0; replicaIndex < replicaCount; replicaIndex++ ) { final DefaultInvocationCtx ctx = (DefaultInvocationCtx) fusedInvocationCtxes[ replicaIndex ][ 0 ]; operatorReplicas[ replicaIndex ][ i ] = new OperatorReplica( pipelineReplicaIds[ replicaIndex ], operatorQueues[ replicaIndex ], drainerPools[ replicaIndex ], replicaMeters[ replicaIndex ], ctx::createInputTuples, fusedOperatorDefs[ replicaIndex ], fusedInvocationCtxes[ replicaIndex ] ); } } for ( int replicaIndex = 0; replicaIndex < replicaCount; replicaIndex++ ) { LOGGER.debug( "Creating pipeline instance for regionId={} replicaIndex={} pipelineIndex={} pipelineStartIndex={}", regionId, replicaIndex, pipelineIndex, pipelineId ); final OperatorReplica[] pipelineOperatorReplicas = operatorReplicas[ replicaIndex ]; final OperatorQueue pipelineQueue = createPipelineQueue( flow, regionId, replicaIndex, pipelineOperatorReplicas ); pipelineReplicas[ pipelineIndex ][ replicaIndex ] = new PipelineReplica( pipelineReplicaIds[ replicaIndex ], pipelineOperatorReplicas, pipelineQueue, replicaMeters[ replicaIndex ] ); } } final Region region = new Region( regionExecPlan, schedulingStrategies, upstreamCtxes, pipelineReplicas ); regions.put( regionId, region ); return region; } @Inject  RegionManagerImpl( final JokerConfig config,
                               final PartitionService partitionService,
                               final OperatorKVStoreManager operatorKvStoreManager,
                               final OperatorQueueManager operatorQueueManager,
                               final PipelineTransformer pipelineTransformer,
                               final PartitionKeyExtractorFactory partitionKeyExtractorFactory ); @Override Region createRegion( final FlowDef flow, final RegionExecPlan regionExecPlan ); @Override void validatePipelineMergeParameters( final List<PipelineId> pipelineIds ); @Override Region mergePipelines( final List<PipelineId> pipelineIdsToMerge ); @Override void validatePipelineSplitParameters( final PipelineId pipelineId, final List<Integer> pipelineOperatorIndicesToSplit ); @Override Region splitPipeline( final PipelineId pipelineId, final List<Integer> pipelineOperatorIndicesToSplit ); @Override Region rebalanceRegion( final FlowDef flow, final int regionId, final int newReplicaCount ); @Override void releaseRegion( final int regionId );  }
@Test public void test_statelessRegion_singlePipeline_singleReplica_noInputConnection () { final FlowExample1 ex = new FlowExample1(); final List<RegionDef> regionDefs = regionDefFormer.createRegions( ex.flow ); final RegionDef regionDef = regionDefs.get( 0 ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, singletonList( 0 ), 1 ); final Region region = regionManager.createRegion( ex.flow, regionExecPlan ); assertNotNull( region ); final PipelineReplica[] pipelines = region.getReplicaPipelines( 0 ); assertEquals( 1, pipelines.length ); final PipelineReplica pipeline = pipelines[ 0 ]; assertEmptyOperatorQueue( pipeline, ex.operatorDef0.getInputPortCount() ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 0, 0 ), pipeline.id() ); assertEquals( 1, pipeline.getOperatorCount() ); final OperatorReplica operatorReplica = pipeline.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica, 0, ex.operatorDef0 ); assertEmptyOperatorQueue( operatorReplica ); assertNonBlockingTupleQueueDrainerPool( operatorReplica ); assertArrayEquals( new SchedulingStrategy[] { ScheduleWhenAvailable.INSTANCE }, region.getSchedulingStrategies() ); assertArrayEquals( new int[] { 0 }, region.getFusionStartIndices() ); assertPipelineReplicaMeter( pipeline ); }
@Test public void test_statelessRegion_singlePipeline_singleReplica_withInputConnection () { final FlowExample2 ex = new FlowExample2(); final List<RegionDef> regionDefs = regionDefFormer.createRegions( ex.flow ); final RegionDef regionDef = regionDefs.get( 1 ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, singletonList( 0 ), 1 ); final Region region = regionManager.createRegion( ex.flow, regionExecPlan ); assertNotNull( region ); final PipelineReplica[] pipelines = region.getReplicaPipelines( 0 ); assertEquals( 1, pipelines.length ); final PipelineReplica pipeline = pipelines[ 0 ]; assertDefaultOperatorQueue( pipeline, ex.operatorDef1.getInputPortCount() ); assertEmptyPipelineQueue( pipeline ); assertEquals( new PipelineReplicaId( region.getRegionId(), 0, 0 ), pipeline.id() ); assertEquals( 2, pipeline.getOperatorCount() ); final OperatorReplica operatorReplica = pipeline.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica, 0, ex.operatorDef1 ); assertOperatorDef( operatorReplica, 1, ex.operatorDef2 ); assertDefaultOperatorQueue( operatorReplica, ex.operatorDef1.getInputPortCount(), MULTI_THREADED ); assertBlockingTupleQueueDrainerPool( operatorReplica ); assertTrue( operatorReplica.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertTrue( operatorReplica.getInvocationCtx( 1 ) instanceof FusedInvocationCtx ); assertArrayEquals( new SchedulingStrategy[] { scheduleWhenTuplesAvailableOnDefaultPort( 1 ), scheduleWhenTuplesAvailableOnDefaultPort( 1 ) }, region.getSchedulingStrategies() ); assertArrayEquals( new int[] { 0 }, region.getFusionStartIndices() ); assertPipelineReplicaMeter( pipeline ); }
@Test public void test_partitionedStatefulRegion_twoPipelines_singleReplica_bothPipelinesStartWithStatelessOperator () { final FlowExample6 ex = new FlowExample6(); final List<RegionDef> regionDefs = regionDefFormer.createRegions( ex.flow ); final RegionDef regionDef = regionDefs.get( 1 ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, asList( 0, 2 ), 1 ); final Region region = regionManager.createRegion( ex.flow, regionExecPlan ); assertNotNull( region ); final PipelineReplica[] pipelines = region.getReplicaPipelines( 0 ); assertEquals( 2, pipelines.length ); final PipelineReplica pipeline0 = pipelines[ 0 ]; final PipelineReplica pipeline1 = pipelines[ 1 ]; assertDefaultOperatorQueue( pipeline0, ex.operatorDef1.getInputPortCount() ); assertEmptyPipelineQueue( pipeline0 ); assertDefaultOperatorQueue( pipeline1, ex.operatorDef3.getInputPortCount() ); assertEmptyPipelineQueue( pipeline0 ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 0, 0 ), pipeline0.id() ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 2, 0 ), pipeline1.id() ); assertEquals( 2, pipeline0.getOperatorCount() ); assertEquals( 1, pipeline0.getOperatorReplicaCount() ); assertEquals( 1, pipeline1.getOperatorCount() ); final OperatorReplica operatorReplica1 = pipeline0.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica1, 0, ex.operatorDef1 ); assertDefaultOperatorQueue( operatorReplica1, ex.operatorDef1.getInputPortCount(), MULTI_THREADED ); assertNull( kvStoreManager.getDefaultKVStore( region.getRegionId(), ex.operatorDef1.getId() ) ); assertBlockingTupleQueueDrainerPool( operatorReplica1 ); assertOperatorDef( operatorReplica1, 1, ex.operatorDef2 ); assertNull( tupleQueueManager.getPartitionedQueues( region.getRegionId(), ex.operatorDef2.getId() ) ); assertNotNull( kvStoreManager.getPartitionedKVStores( region.getRegionId(), ex.operatorDef2.getId() ) ); assertTrue( operatorReplica1.getInvocationCtx( 0 ) instanceof DefaultInvocationCtx ); assertTrue( operatorReplica1.getInvocationCtx( 1 ) instanceof FusedPartitionedInvocationCtx ); final OperatorReplica operatorReplica2 = pipeline1.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica2, 0, ex.operatorDef3 ); assertDefaultOperatorQueue( operatorReplica2, ex.operatorDef3.getInputPortCount(), MULTI_THREADED ); assertNull( kvStoreManager.getDefaultKVStore( region.getRegionId(), ex.operatorDef3.getId() ) ); assertBlockingTupleQueueDrainerPool( operatorReplica2 ); assertPipelineReplicaMeter( pipeline0 ); assertPipelineReplicaMeter( pipeline1 ); assertNotEquals( pipeline0.getMeter(), pipeline1.getMeter() ); }
@Test public void test_partitionedStatefulRegion_threePipeline_singleReplica_withStatelessOperatorFirst () { final FlowExample6 ex = new FlowExample6(); final List<RegionDef> regionDefs = regionDefFormer.createRegions( ex.flow ); final RegionDef regionDef = regionDefs.get( 1 ); final RegionExecPlan regionExecPlan = new RegionExecPlan( regionDef, asList( 0, 1, 2 ), 1 ); final Region region = regionManager.createRegion( ex.flow, regionExecPlan ); assertNotNull( region ); final PipelineReplica[] pipelines = region.getReplicaPipelines( 0 ); assertEquals( 3, pipelines.length ); final PipelineReplica pipeline0 = pipelines[ 0 ]; final PipelineReplica pipeline1 = pipelines[ 1 ]; final PipelineReplica pipeline2 = pipelines[ 2 ]; assertDefaultOperatorQueue( pipeline0, ex.operatorDef1.getInputPortCount() ); assertEmptyPipelineQueue( pipeline0 ); assertDefaultOperatorQueue( pipeline1, ex.operatorDef2.getInputPortCount() ); assertDefaultPipelineQueue( pipeline1 ); assertDefaultOperatorQueue( pipeline2, ex.operatorDef3.getInputPortCount() ); assertEmptyPipelineQueue( pipeline2 ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 0, 0 ), pipeline0.id() ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 1, 0 ), pipeline1.id() ); assertEquals( new PipelineReplicaId( regionDef.getRegionId(), 2, 0 ), pipeline2.id() ); assertEquals( 1, pipeline0.getOperatorCount() ); assertEquals( 1, pipeline1.getOperatorCount() ); assertEquals( 1, pipeline2.getOperatorCount() ); final OperatorReplica operatorReplica1 = pipeline0.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica1, 0, ex.operatorDef1 ); assertDefaultOperatorQueue( operatorReplica1, ex.operatorDef1.getInputPortCount(), MULTI_THREADED ); assertNull( kvStoreManager.getDefaultKVStore( region.getRegionId(), ex.operatorDef1.getId() ) ); assertBlockingTupleQueueDrainerPool( operatorReplica1 ); final OperatorReplica operatorReplica2 = pipeline1.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica2, 0, ex.operatorDef2 ); assertPartitionedOperatorQueue( operatorReplica2 ); assertNotNull( kvStoreManager.getPartitionedKVStores( region.getRegionId(), ex.operatorDef2.getId() ) ); assertNonBlockingTupleQueueDrainerPool( operatorReplica2 ); final OperatorReplica operatorReplica3 = pipeline2.getOperatorReplica( 0 ); assertOperatorDef( operatorReplica3, 0, ex.operatorDef3 ); assertDefaultOperatorQueue( operatorReplica3, ex.operatorDef3.getInputPortCount(), MULTI_THREADED ); assertNull( kvStoreManager.getDefaultKVStore( region.getRegionId(), ex.operatorDef3.getId() ) ); assertBlockingTupleQueueDrainerPool( operatorReplica3 ); assertPipelineReplicaMeter( pipeline0 ); assertPipelineReplicaMeter( pipeline1 ); assertPipelineReplicaMeter( pipeline2 ); assertNotEquals( pipeline0.getMeter(), pipeline1.getMeter() ); assertNotEquals( pipeline0.getMeter(), pipeline2.getMeter() ); assertNotEquals( pipeline1.getMeter(), pipeline2.getMeter() ); }