For a sliding window with size = 10 and slide = 4, the StreamSlicer does not get out of the while loop when processing the first tuple.
|
while (windowManager.hasFixedWindows() && te > min_next_edge_ts) { |
|
if (min_next_edge_ts >= 0) |
|
sliceManager.appendSlice(min_next_edge_ts, new Slice.Fixed()); |
|
min_next_edge_ts = calculateNextFixedEdge(te); |
|
} |
Reason: For its first call, assignNextWindowStart receives Long.MAX_VALUE as t_c and returns Long.MIN_VALUE because of the slide value 4. The variable min_next_edge_ts remains Long.MIN_VALUE and is again set to Long.MAX_VALUE in the next iteration of the loop.
|
private long calculateNextFixedEdge(long te) { |
|
// next_edge will be the last edge |
|
long current_min_edge = min_next_edge_ts == Long.MIN_VALUE ? Long.MAX_VALUE : min_next_edge_ts; |
|
long t_c = Math.max(te - this.windowManager.getMaxLateness(), current_min_edge); |
|
long edge = Long.MAX_VALUE; |
|
for (ContextFreeWindow tw : this.windowManager.getContextFreeWindows()) { |
|
if (tw.getWindowMeasure() == WindowMeasure.Time) { |
|
//long newNextEdge = t_c + tw.getSize() - (t_c) % tw.getSize(); |
|
long newNextEdge = tw.assignNextWindowStart(t_c); |
|
edge = Math.min(newNextEdge, edge); |
|
} |
|
} |
|
return edge; |
|
} |