// Path: src/main/java/storm/benchmark/lib/reducer/Reducer.java
// public interface Reducer<V> extends Serializable {
//   public V reduce(V v1, V v2);
//   public V zero();
//   public boolean isZero(V v);
// }


import backtype.storm.utils.MutableObject;
import org.apache.log4j.Logger;
import storm.benchmark.lib.reducer.Reducer;
import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
  }

  public void add(K obj, V val) {
    slots.add(obj, val, headSlot);
  }

  public Map<K, V> reduceThenAdvanceWindow() {
    Map<K, V> reduced = slots.reduceByKey();
    slots.wipeSlot(tailSlot);
    slots.wipeZeros();
    advanceHead();
    return reduced;
  }

  private void advanceHead() {
    headSlot = tailSlot;
    tailSlot = slotAfter(tailSlot);
  }

  private int slotAfter(int slot) {
    return (slot + 1) % windowLengthInSlots;
  }

  public static class Slots<K, V> implements Serializable {

    private static final long serialVersionUID = 4858185737378394432L;
    private static final Logger LOG = Logger.getLogger(Slots.class);

    private final Map<K, MutableObject[]> objToValues = new HashMap<K, MutableObject[]>();
    private final int numSlots;
// Path: src/main/java/storm/benchmark/lib/reducer/Reducer.java
// public interface Reducer<V> extends Serializable {
//   public V reduce(V v1, V v2);
//   public V zero();
//   public boolean isZero(V v);
// }


// Path: src/main/java/storm/benchmark/tools/SlidingWindow.java
import backtype.storm.utils.MutableObject;
import org.apache.log4j.Logger;
import storm.benchmark.lib.reducer.Reducer;
import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

  }

  public void add(K obj, V val) {
    slots.add(obj, val, headSlot);
  }

  public Map<K, V> reduceThenAdvanceWindow() {
    Map<K, V> reduced = slots.reduceByKey();
    slots.wipeSlot(tailSlot);
    slots.wipeZeros();
    advanceHead();
    return reduced;
  }

  private void advanceHead() {
    headSlot = tailSlot;
    tailSlot = slotAfter(tailSlot);
  }

  private int slotAfter(int slot) {
    return (slot + 1) % windowLengthInSlots;
  }

  public static class Slots<K, V> implements Serializable {

    private static final long serialVersionUID = 4858185737378394432L;
    private static final Logger LOG = Logger.getLogger(Slots.class);

    private final Map<K, MutableObject[]> objToValues = new HashMap<K, MutableObject[]>();
    private final int numSlots;
    private final Reducer<V> reducer;