[BUG] The API of `iterate_emit_until()` in Pegasus gets stuck in some cases

**Describe the bug**

The API of `iterate_emit_until()` in Pegasus will get stuck in some cases.

**To Reproduce**

Running the following query on LDBC dataset:
```
g.V().hasLabel("COMMENT").limit(1000000).out("0..2", "REPLYOF").out("REPLYOF").count()
```
where `out("0..2", "REPLYOF")` is implemented by `iterate_emit_until()`. it gets stuck, and the WARN info suggests:
`output[(5.0)] block pushing batch of [root] to channel[9] ;`

The dataflow plan looks like:
```
============ Build Dataflow ==============
Peers: 64
Operators:
1 source(1)
2 limit_partition(2)
3 limit_partition(3)
4 filter_map_PathStart(4)
5 switch(5)
6 flat_map_EdgeExpand(6)
7 feedback(7)
8 flat_map_EdgeExpand(8)
9 count_local(9)
10 count_global(10)
11 sink_stream(11)
Channels:
[1: (0.0) -> (1.0)]
[2: (1.0) -> (2.0)]
[3: (2.0) => (3.0)]
[4: (3.0) => (4.0)]
[5: (4.0) => (5.0)]
[6: (5.1) -> (6.0)]
[7: (6.0) => (7.0)]
[8: (7.0) -> (5.1)]
[9: (5.0) => (8.0)]
[10: (8.0) -> (9.0)]
[11: (9.0) => (10.0)]
[12: (10.0) -> (11.0)]
==========================================
```