序

本文主要研究一下storm的IWaitStrategy

IWaitStrategy

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/IWaitStrategy.java

public interface IWaitStrategy {static IWaitStrategy createBackPressureWaitStrategy(Map<String, Object> topologyConf) {IWaitStrategy producerWaitStrategy =ReflectionUtils.newInstance((String) topologyConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));producerWaitStrategy.prepare(topologyConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);return producerWaitStrategy;}void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation);/*** Implementations of this method should be thread-safe (preferably no side-effects and lock-free)* <p>* Supports static or dynamic backoff. Dynamic backoff relies on idleCounter to estimate how long caller has been idling.* <p>* <pre>* <code>*  int idleCounter = 0;*  int consumeCount = consumeFromQ();*  while (consumeCount==0) {*     idleCounter = strategy.idle(idleCounter);*     consumeCount = consumeFromQ();*  }* </code>* </pre>** @param idleCounter managed by the idle method until reset* @return new counter value to be used on subsequent idle cycle*/int idle(int idleCounter) throws InterruptedException;enum WAIT_SITUATION {SPOUT_WAIT, BOLT_WAIT, BACK_PRESSURE_WAIT}}

• 这个接口提供了一个工厂方法，默认是读取topology.backpressure.wait.strategy参数值，创建producerWaitStrategy，并使用WAIT_SITUATION.BACK_PRESSURE_WAIT初始化
• WAIT_SITUATION一共有三类，分别是SPOUT_WAIT, BOLT_WAIT, BACK_PRESSURE_WAIT
• 该接口定义了int idle(int idleCounter)方法，用于static或dynamic backoff

SpoutExecutor

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java

public class SpoutExecutor extends Executor {private static final Logger LOG = LoggerFactory.getLogger(SpoutExecutor.class);private final IWaitStrategy spoutWaitStrategy;private final IWaitStrategy backPressureWaitStrategy;private final AtomicBoolean lastActive;private final MutableLong emittedCount;private final MutableLong emptyEmitStreak;private final SpoutThrottlingMetrics spoutThrottlingMetrics;private final boolean hasAckers;private final SpoutExecutorStats stats;private final BuiltinMetrics builtInMetrics;SpoutOutputCollectorImpl spoutOutputCollector;private Integer maxSpoutPending;private List<ISpout> spouts;private List<SpoutOutputCollector> outputCollectors;private RotatingMap<Long, TupleInfo> pending;private long threadId = 0;public SpoutExecutor(final WorkerState workerData, final List<Long> executorId, Map<String, String> credentials) {super(workerData, executorId, credentials, ClientStatsUtil.SPOUT);this.spoutWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_SPOUT_WAIT_STRATEGY));this.spoutWaitStrategy.prepare(topoConf, WAIT_SITUATION.SPOUT_WAIT);this.backPressureWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));this.backPressureWaitStrategy.prepare(topoConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);//......}//......
}

• 这里创建了两个watiStrategy，一个是spoutWaitStrategy，一个是backPressureWaitStrategy
• spoutWaitStrategy读取的是topology.spout.wait.strategy参数，在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive
• backPressureWaitStrategy读取的是topology.backpressure.wait.strategy参数，在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive

BoltExecutor

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java

public class BoltExecutor extends Executor {private static final Logger LOG = LoggerFactory.getLogger(BoltExecutor.class);private final BooleanSupplier executeSampler;private final boolean isSystemBoltExecutor;private final IWaitStrategy consumeWaitStrategy;       // employed when no incoming dataprivate final IWaitStrategy backPressureWaitStrategy;  // employed when outbound path is congestedprivate final BoltExecutorStats stats;private final BuiltinMetrics builtInMetrics;private BoltOutputCollectorImpl outputCollector;public BoltExecutor(WorkerState workerData, List<Long> executorId, Map<String, String> credentials) {super(workerData, executorId, credentials, ClientStatsUtil.BOLT);this.executeSampler = ConfigUtils.mkStatsSampler(topoConf);this.isSystemBoltExecutor = (executorId == Constants.SYSTEM_EXECUTOR_ID);if (isSystemBoltExecutor) {this.consumeWaitStrategy = makeSystemBoltWaitStrategy();} else {this.consumeWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BOLT_WAIT_STRATEGY));this.consumeWaitStrategy.prepare(topoConf, WAIT_SITUATION.BOLT_WAIT);}this.backPressureWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_STRATEGY));this.backPressureWaitStrategy.prepare(topoConf, WAIT_SITUATION.BACK_PRESSURE_WAIT);this.stats = new BoltExecutorStats(ConfigUtils.samplingRate(this.getTopoConf()),ObjectReader.getInt(this.getTopoConf().get(Config.NUM_STAT_BUCKETS)));this.builtInMetrics = new BuiltinBoltMetrics(stats);}private static IWaitStrategy makeSystemBoltWaitStrategy() {WaitStrategyPark ws = new WaitStrategyPark();Map<String, Object> conf = new HashMap<>();conf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 5000);ws.prepare(conf, WAIT_SITUATION.BOLT_WAIT);return ws;}//......
}

• 这里创建了两个IWaitStrategy，一个是consumeWaitStrategy，一个是backPressureWaitStrategy
• consumeWaitStrategy在非SystemBoltExecutor的情况下读取的是topology.bolt.wait.strategy参数，在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive；如果是SystemBoltExecutor则使用的是WaitStrategyPark策略
• backPressureWaitStrategy读取的是读取的是topology.backpressure.wait.strategy参数，在defaults.yaml里头值为org.apache.storm.policy.WaitStrategyProgressive

WaitStrategyPark

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/WaitStrategyPark.java

public class WaitStrategyPark implements IWaitStrategy {private long parkTimeNanoSec;public WaitStrategyPark() { // required for instantiation via reflection. must call prepare() thereafter}// Convenience alternative to prepare() for use in Testspublic WaitStrategyPark(long microsec) {parkTimeNanoSec = microsec * 1_000;}@Overridepublic void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation) {if (waitSituation == WAIT_SITUATION.SPOUT_WAIT) {parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PARK_MICROSEC));} else if (waitSituation == WAIT_SITUATION.BOLT_WAIT) {parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC));} else if (waitSituation == WAIT_SITUATION.BACK_PRESSURE_WAIT) {parkTimeNanoSec = 1_000 * ObjectReader.getLong(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PARK_MICROSEC));} else {throw new IllegalArgumentException("Unknown wait situation : " + waitSituation);}}@Overridepublic int idle(int idleCounter) throws InterruptedException {if (parkTimeNanoSec == 0) {return 1;}LockSupport.parkNanos(parkTimeNanoSec);return idleCounter + 1;}
}

• 该策略使用的是LockSupport.parkNanos(parkTimeNanoSec)方法

WaitStrategyProgressive

storm-2.0.0/storm-client/src/jvm/org/apache/storm/policy/WaitStrategyProgressive.java

/*** A Progressive Wait Strategy* <p> Has three levels of idling. Stays in each level for a configured number of iterations before entering the next level.* Level 1 - No idling. Returns immediately. Stays in this level for `level1Count` iterations. Level 2 - Calls LockSupport.parkNanos(1).* Stays in this level for `level2Count` iterations Level 3 - Calls Thread.sleep(). Stays in this level until wait situation changes.** <p>* The initial spin can be useful to prevent downstream bolt from repeatedly sleeping/parking when the upstream component is a bit* relatively slower. Allows downstream bolt can enter deeper wait states only if the traffic to it appears to have reduced.* <p>*/
public class WaitStrategyProgressive implements IWaitStrategy {private int level1Count;private int level2Count;private long level3SleepMs;@Overridepublic void prepare(Map<String, Object> conf, WAIT_SITUATION waitSituation) {if (waitSituation == WAIT_SITUATION.SPOUT_WAIT) {level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL1_COUNT));level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL2_COUNT));level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_SPOUT_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));} else if (waitSituation == WAIT_SITUATION.BOLT_WAIT) {level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL1_COUNT));level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL2_COUNT));level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_BOLT_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));} else if (waitSituation == WAIT_SITUATION.BACK_PRESSURE_WAIT) {level1Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL1_COUNT));level2Count = ObjectReader.getInt(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL2_COUNT));level3SleepMs = ObjectReader.getLong(conf.get(Config.TOPOLOGY_BACKPRESSURE_WAIT_PROGRESSIVE_LEVEL3_SLEEP_MILLIS));} else {throw new IllegalArgumentException("Unknown wait situation : " + waitSituation);}}@Overridepublic int idle(int idleCounter) throws InterruptedException {if (idleCounter < level1Count) {                     // level 1 - no waiting++idleCounter;} else if (idleCounter < level1Count + level2Count) { // level 2 - parkNanos(1L)++idleCounter;LockSupport.parkNanos(1L);} else {                                      // level 3 - longer idling with Thread.sleep()Thread.sleep(level3SleepMs);}return idleCounter;}
}

• WaitStrategyProgressive是一个渐进式的wait strategy，它分为3个level的idling
• level 1是no idling，立刻返回；在level 1经历了level1Count的次数之后进入level 2
• level 2使用的是LockSupport.parkNanos(1)，在level 2经历了level2Count次数之后进入level 3
• level 3使用的是Thread.sleep(level3SleepMs)，在wait situation改变的时候跳出
• 不同的WAIT_SITUATION读取不同的LEVEL1_COUNT、LEVEL2_COUNT、LEVEL3_SLEEP_MILLIS参数，对于spout，它们的默认值分别为0、0、1；对于bolt它们的默认值分别为1、1000、1；对于back pressure，它们的默认值分别为1、1000、1

SpoutExecutor.call

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java

    @Overridepublic Callable<Long> call() throws Exception {init(idToTask, idToTaskBase);return new Callable<Long>() {final int recvqCheckSkipCountMax = getSpoutRecvqCheckSkipCount();int recvqCheckSkips = 0;int swIdleCount = 0; // counter for spout wait strategyint bpIdleCount = 0; // counter for back pressure wait strategyint rmspCount = 0;@Overridepublic Long call() throws Exception {int receiveCount = 0;if (recvqCheckSkips++ == recvqCheckSkipCountMax) {receiveCount = receiveQueue.consume(SpoutExecutor.this);recvqCheckSkips = 0;}long currCount = emittedCount.get();boolean reachedMaxSpoutPending = (maxSpoutPending != 0) && (pending.size() >= maxSpoutPending);boolean isActive = stormActive.get();if (!isActive) {inactiveExecute();return 0L;}if (!lastActive.get()) {lastActive.set(true);activateSpouts();}boolean pendingEmitsIsEmpty = tryFlushPendingEmits();boolean noEmits = true;long emptyStretch = 0;if (!reachedMaxSpoutPending && pendingEmitsIsEmpty) {for (int j = 0; j < spouts.size(); j++) { // in critical path. don't use iterators.spouts.get(j).nextTuple();}noEmits = (currCount == emittedCount.get());if (noEmits) {emptyEmitStreak.increment();} else {emptyStretch = emptyEmitStreak.get();emptyEmitStreak.set(0);}}if (reachedMaxSpoutPending) {if (rmspCount == 0) {LOG.debug("Reached max spout pending");}rmspCount++;} else {if (rmspCount > 0) {LOG.debug("Ended max spout pending stretch of {} iterations", rmspCount);}rmspCount = 0;}if (receiveCount > 1) {// continue without idlingreturn 0L;}if (!pendingEmits.isEmpty()) { // then facing backpressurebackPressureWaitStrategy();return 0L;}bpIdleCount = 0;if (noEmits) {spoutWaitStrategy(reachedMaxSpoutPending, emptyStretch);return 0L;}swIdleCount = 0;return 0L;}private void backPressureWaitStrategy() throws InterruptedException {long start = Time.currentTimeMillis();if (bpIdleCount == 0) { // check avoids multiple log msgs when in a idle loopLOG.debug("Experiencing Back Pressure from downstream components. Entering BackPressure Wait.");}bpIdleCount = backPressureWaitStrategy.idle(bpIdleCount);spoutThrottlingMetrics.skippedBackPressureMs(Time.currentTimeMillis() - start);}private void spoutWaitStrategy(boolean reachedMaxSpoutPending, long emptyStretch) throws InterruptedException {emptyEmitStreak.increment();long start = Time.currentTimeMillis();swIdleCount = spoutWaitStrategy.idle(swIdleCount);if (reachedMaxSpoutPending) {spoutThrottlingMetrics.skippedMaxSpoutMs(Time.currentTimeMillis() - start);} else {if (emptyStretch > 0) {LOG.debug("Ending Spout Wait Stretch of {}", emptyStretch);}}}// returns true if pendingEmits is emptyprivate boolean tryFlushPendingEmits() {for (AddressedTuple t = pendingEmits.peek(); t != null; t = pendingEmits.peek()) {if (executorTransfer.tryTransfer(t, null)) {pendingEmits.poll();} else { // to avoid reordering of emits, stop at first failurereturn false;}}return true;}};}

• spout维护了pendingEmits队列，即emit没有成功或者等待emit的队列，同时也维护了pending的RotatingMap，即等待ack的tuple的id及数据
• spout从topology.max.spout.pending读取TOPOLOGY_MAX_SPOUT_PENDING配置，计算maxSpoutPending=ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 0) * idToTask.size()，默认为null，即maxSpoutPending为0
• spout在!reachedMaxSpoutPending && pendingEmitsIsEmpty的条件下才调用nextTuple发送数据；在pendingEmits不为空的时候触发backPressureWaitStrategy；在noEmits((currCount == emittedCount.get()))时触发spoutWaitStrategy
• 在每次调用call的时候，在调用nextTuple之间记录currCount = emittedCount.get()；如果有调用nextTuple的话，则会在SpoutOutputCollectorImpl的emit或emitDirect等方法更新emittedCount；之后用noEmits=(currCount == emittedCount.get())判断是否有发射数据
• spout维护了bpIdleCount以及swIdleCount，分别用于backPressureWaitStrategy.idle(bpIdleCount)、spoutWaitStrategy.idle(swIdleCount)

BoltExecutor.call

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java

    @Overridepublic Callable<Long> call() throws Exception {init(idToTask, idToTaskBase);return new Callable<Long>() {int bpIdleCount = 0;int consumeIdleCounter = 0;private final ExitCondition tillNoPendingEmits = () -> pendingEmits.isEmpty();@Overridepublic Long call() throws Exception {boolean pendingEmitsIsEmpty = tryFlushPendingEmits();if (pendingEmitsIsEmpty) {if (bpIdleCount != 0) {LOG.debug("Ending Back Pressure Wait stretch : {}", bpIdleCount);}bpIdleCount = 0;int consumeCount = receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits);if (consumeCount == 0) {if (consumeIdleCounter == 0) {LOG.debug("Invoking consume wait strategy");}consumeIdleCounter = consumeWaitStrategy.idle(consumeIdleCounter);if (Thread.interrupted()) {throw new InterruptedException();}} else {if (consumeIdleCounter != 0) {LOG.debug("Ending consume wait stretch : {}", consumeIdleCounter);}consumeIdleCounter = 0;}} else {if (bpIdleCount == 0) { // check avoids multiple log msgs when spinning in a idle loopLOG.debug("Experiencing Back Pressure. Entering BackPressure Wait. PendingEmits = {}", pendingEmits.size());}bpIdleCount = backPressureWaitStrategy.idle(bpIdleCount);}return 0L;}// returns true if pendingEmits is emptyprivate boolean tryFlushPendingEmits() {for (AddressedTuple t = pendingEmits.peek(); t != null; t = pendingEmits.peek()) {if (executorTransfer.tryTransfer(t, null)) {pendingEmits.poll();} else { // to avoid reordering of emits, stop at first failurereturn false;}}return true;}};}

• bolt executor同样也维护了pendingEmits，在pendingEmits不为空的时候，触发backPressureWaitStrategy.idle(bpIdleCount)
• 在pendingEmits为空时，根据receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits)返回的consumeCount，若为0则触发consumeWaitStrategy.idle(consumeIdleCounter)
• bolt executor维护了bpIdleCount及consumeIdleCounter，分别用于backPressureWaitStrategy.idle(bpIdleCount)以及consumeWaitStrategy.idle(consumeIdleCounter)

小结

• spout和bolt的executor里头都用到了backPressureWaitStrategy，读取的是topology.backpressure.wait.strategy参数(for any producer (spout/bolt/transfer thread) when the downstream Q is full)，使用的实现类为org.apache.storm.policy.WaitStrategyProgressive，在下游component的recv queue满的时候使用的背压策略；具体是使用pendingEmits队列来判断，spout或bolt的call方法里头每次判断pendingEmitsIsEmpty都是调用tryFlushPendingEmits，先尝试发送数据，如果下游成功接收，则pendingEmits队列为空，通过这种机制来动态判断下游负载，决定是否触发backpressure
• spout使用的spoutWaitStrategy，读取的是topology.spout.wait.strategy参数(employed when there is no data to produce)，使用的实现类为org.apache.storm.policy.WaitStrategyProgressive，在没有数据发射的时候使用；具体是使用emittedCount来判断
• bolt使用的consumeWaitStrategy，在非SystemBoltExecutor的情况下读取的是topology.bolt.wait.strategy参数(employed when there is no data in its receive buffer to process)，使用的实现类为org.apache.storm.policy.WaitStrategyProgressive，在receive buffer没有数据处理的时候使用；具体是使用receiveQueue.consume(BoltExecutor.this, tillNoPendingEmits)返回的consumeCount来判断
• spout与bolt不同的还有一点就是spout除了pendingEmitsIsEmpty还多了一个reachedMaxSpoutPending参数，来判断是否继续产生数据，bolt则使用pendingEmitsIsEmpty来判断是否可以继续消费数据
• IWaitStrategy除了WaitStrategyProgressive实现，还有WaitStrategyPark实现，该策略在bolt是SystemBolt的情况下使用

doc

• IWaitStrategy
• WaitStrategyProgressive
• WaitStrategyPark