@Test public void testProcessorNumManaged() throws InjectionException { globalMetricHolder.getNumEventsMetric().updateValue( (long)EVENT_NUM_HIGH_THRES * 2); globalMetricHolder.getNumEventsMetric().updateValue( (long)EVENT_NUM_HIGH_THRES * 2); globalMetricHolder.getCpuSysUtilMetric().updateValue(0); handler.getPubSubEventHandler().onNext(new MetricUpdateEvent()); prevIncreaseNum = (int)(prevIncreaseNum * INCREASE_RATE); verify(eventProcessorManager, times(1)).increaseEventProcessors(prevIncreaseNum); globalMetricHolder.getNumEventsMetric().updateValue(0); globalMetricHolder.getNumEventsMetric().updateValue(0); handler.getPubSubEventHandler().onNext(new MetricUpdateEvent()); verify(eventProcessorManager, times(1)).increaseEventProcessors(prevIncreaseNum); globalMetricHolder.getNumEventsMetric().updateValue( (long)EVENT_NUM_HIGH_THRES * 2); globalMetricHolder.getNumEventsMetric().updateValue( (long)EVENT_NUM_HIGH_THRES * 2); handler.getPubSubEventHandler().onNext(new MetricUpdateEvent()); prevIncreaseNum = (int)(prevIncreaseNum * INCREASE_RATE); verify(eventProcessorManager, times(1)).increaseEventProcessors(prevIncreaseNum); globalMetricHolder.getNumEventsMetric().updateValue(0); globalMetricHolder.getNumEventsMetric().updateValue(0); globalMetricHolder.getNumEventsMetric().updateValue(0); handler.getPubSubEventHandler().onNext(new MetricUpdateEvent()); verify(eventProcessorManager, times(1)).decreaseEventProcessors(DECREASE_NUM); handler.getPubSubEventHandler().onNext(new MetricUpdateEvent()); verify(eventProcessorManager, times(2)).decreaseEventProcessors(DECREASE_NUM); }
@Override public void onNext(final MetricUpdateEvent metricUpdateEvent) { final double currCpuUtil = globalMetricHolder.getCpuSysUtilMetric().getEwmaValue(); final double currEventNum = globalMetricHolder.getNumEventsMetric().getEwmaValue(); if (currCpuUtil < cpuUtilLowThreshold) { if (currEventNum > eventNumHighThreshold) { final int increaseNum = (int)(prevIncreaseNum * increaseRate); eventProcessorManager.increaseEventProcessors(increaseNum); prevIncreaseNum = increaseNum; } else if (currEventNum < eventNumLowThreshold) { eventProcessorManager.decreaseEventProcessors(decreaseNum); prevIncreaseNum = 1; } } }
MISDEventProcessorNumAssigner implements EventProcessorNumAssigner { @Override public void onNext(final MetricUpdateEvent metricUpdateEvent) { final double currCpuUtil = globalMetricHolder.getCpuSysUtilMetric().getEwmaValue(); final double currEventNum = globalMetricHolder.getNumEventsMetric().getEwmaValue(); if (currCpuUtil < cpuUtilLowThreshold) { if (currEventNum > eventNumHighThreshold) { final int increaseNum = (int)(prevIncreaseNum * increaseRate); eventProcessorManager.increaseEventProcessors(increaseNum); prevIncreaseNum = increaseNum; } else if (currEventNum < eventNumLowThreshold) { eventProcessorManager.decreaseEventProcessors(decreaseNum); prevIncreaseNum = 1; } } } }
MISDEventProcessorNumAssigner implements EventProcessorNumAssigner { @Override public void onNext(final MetricUpdateEvent metricUpdateEvent) { final double currCpuUtil = globalMetricHolder.getCpuSysUtilMetric().getEwmaValue(); final double currEventNum = globalMetricHolder.getNumEventsMetric().getEwmaValue(); if (currCpuUtil < cpuUtilLowThreshold) { if (currEventNum > eventNumHighThreshold) { final int increaseNum = (int)(prevIncreaseNum * increaseRate); eventProcessorManager.increaseEventProcessors(increaseNum); prevIncreaseNum = increaseNum; } else if (currEventNum < eventNumLowThreshold) { eventProcessorManager.decreaseEventProcessors(decreaseNum); prevIncreaseNum = 1; } } } @Inject private  MISDEventProcessorNumAssigner(
      @Parameter(EventNumHighThreshold.class) final double eventNumHighThreshold,
      @Parameter(EventNumLowThreshold.class) final double eventNumLowThreshold,
      @Parameter(CpuUtilLowThreshold.class) final double cpuUtilLowThreshold,
      @Parameter(EventProcessorIncreaseRate.class) final double increaseRate,
      @Parameter(EventProcessorDecreaseNum.class) final int decreaseNum,
      final EventProcessorManager eventProcessorManager,
      final GlobalMetrics globalMetricHolder,
      final MistPubSubEventHandler pubSubEventHandler); }
MISDEventProcessorNumAssigner implements EventProcessorNumAssigner { @Override public void onNext(final MetricUpdateEvent metricUpdateEvent) { final double currCpuUtil = globalMetricHolder.getCpuSysUtilMetric().getEwmaValue(); final double currEventNum = globalMetricHolder.getNumEventsMetric().getEwmaValue(); if (currCpuUtil < cpuUtilLowThreshold) { if (currEventNum > eventNumHighThreshold) { final int increaseNum = (int)(prevIncreaseNum * increaseRate); eventProcessorManager.increaseEventProcessors(increaseNum); prevIncreaseNum = increaseNum; } else if (currEventNum < eventNumLowThreshold) { eventProcessorManager.decreaseEventProcessors(decreaseNum); prevIncreaseNum = 1; } } } @Inject private  MISDEventProcessorNumAssigner(
      @Parameter(EventNumHighThreshold.class) final double eventNumHighThreshold,
      @Parameter(EventNumLowThreshold.class) final double eventNumLowThreshold,
      @Parameter(CpuUtilLowThreshold.class) final double cpuUtilLowThreshold,
      @Parameter(EventProcessorIncreaseRate.class) final double increaseRate,
      @Parameter(EventProcessorDecreaseNum.class) final int decreaseNum,
      final EventProcessorManager eventProcessorManager,
      final GlobalMetrics globalMetricHolder,
      final MistPubSubEventHandler pubSubEventHandler); @Override void onNext(final MetricUpdateEvent metricUpdateEvent); }
MISDEventProcessorNumAssigner implements EventProcessorNumAssigner { @Override public void onNext(final MetricUpdateEvent metricUpdateEvent) { final double currCpuUtil = globalMetricHolder.getCpuSysUtilMetric().getEwmaValue(); final double currEventNum = globalMetricHolder.getNumEventsMetric().getEwmaValue(); if (currCpuUtil < cpuUtilLowThreshold) { if (currEventNum > eventNumHighThreshold) { final int increaseNum = (int)(prevIncreaseNum * increaseRate); eventProcessorManager.increaseEventProcessors(increaseNum); prevIncreaseNum = increaseNum; } else if (currEventNum < eventNumLowThreshold) { eventProcessorManager.decreaseEventProcessors(decreaseNum); prevIncreaseNum = 1; } } } @Inject private  MISDEventProcessorNumAssigner(
      @Parameter(EventNumHighThreshold.class) final double eventNumHighThreshold,
      @Parameter(EventNumLowThreshold.class) final double eventNumLowThreshold,
      @Parameter(CpuUtilLowThreshold.class) final double cpuUtilLowThreshold,
      @Parameter(EventProcessorIncreaseRate.class) final double increaseRate,
      @Parameter(EventProcessorDecreaseNum.class) final int decreaseNum,
      final EventProcessorManager eventProcessorManager,
      final GlobalMetrics globalMetricHolder,
      final MistPubSubEventHandler pubSubEventHandler); @Override void onNext(final MetricUpdateEvent metricUpdateEvent);  }