StopIteration is not raised correctly with _iter=True

[ghost]

I found an odd case in which the return value of something like sh.ls('-a', _iter=True) is not behaving as a regular iterator. If I keep calling next on an expression with _iter=True everything looks okay. However, when I try to iterate over chunks using itertools.izip_longest(*([ iter(p) ] * 10)) (iterating over chunks of 10 elements), there is some point in which next is not raising StopIteration and instead hangs the Python REPL.

(sh-env)➜  test  touch {1..200}.txt
(sh-env)➜  test  python
>>> import sh
>>> import itertools
>>> p = sh.ls('-a', _iter=True)
>>> s = itertools.izip_longest(*([ iter(p) ] * 10))
>>> next(s)
(u'.\t 11.txt   140.txt  161.txt  182.txt  21.txt  42.txt  63.txt  84.txt\n', u'..\t 120.txt  141.txt  162.txt  183.txt  22.txt  43.txt  64.txt  85.txt\n', u'100.txt  121.txt  142.txt  163.txt  184.txt  23.txt  44.txt  65.txt  86.txt\n', u'101.txt  122.txt  143.txt  164.txt  185.txt  24.txt  45.txt  66.txt  87.txt\n', u'102.txt  123.txt  144.txt  165.txt  186.txt  25.txt  46.txt  67.txt  88.txt\n', u'103.txt  124.txt  145.txt  166.txt  187.txt  26.txt  47.txt  68.txt  89.txt\n', u'104.txt  125.txt  146.txt  167.txt  188.txt  27.txt  48.txt  69.txt  8.txt\n', u'105.txt  126.txt  147.txt  168.txt  189.txt  28.txt  49.txt  6.txt   90.txt\n', u'106.txt  127.txt  148.txt  169.txt  18.txt   29.txt  4.txt   70.txt  91.txt\n', u'107.txt  128.txt  149.txt  16.txt   190.txt  2.txt   50.txt  71.txt  92.txt\n')
>>> next(s)
(u'108.txt  129.txt  14.txt   170.txt  191.txt  30.txt  51.txt  72.txt  93.txt\n', u'109.txt  12.txt   150.txt  171.txt  192.txt  31.txt  52.txt  73.txt  94.txt\n', u'10.txt\t 130.txt  151.txt  172.txt  193.txt  32.txt  53.txt  74.txt  95.txt\n', u'110.txt  131.txt  152.txt  173.txt  194.txt  33.txt  54.txt  75.txt  96.txt\n', u'111.txt  132.txt  153.txt  174.txt  195.txt  34.txt  55.txt  76.txt  97.txt\n', u'112.txt  133.txt  154.txt  175.txt  196.txt  35.txt  56.txt  77.txt  98.txt\n', u'113.txt  134.txt  155.txt  176.txt  197.txt  36.txt  57.txt  78.txt  99.txt\n', u'114.txt  135.txt  156.txt  177.txt  198.txt  37.txt  58.txt  79.txt  9.txt\n', u'115.txt  136.txt  157.txt  178.txt  199.txt  38.txt  59.txt  7.txt\n', u'116.txt  137.txt  158.txt  179.txt  19.txt   39.txt  5.txt   80.txt\n')
>>> next(s)
^CTraceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/site-packages/sh.py", line 559, in next
    chunk = self.process._pipe_queue.get(True, 0.001)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/Queue.py", line 177, in get
    self.not_empty.wait(remaining)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/threading.py", line 352, in wait
    gotit = waiter.acquire(0)
KeyboardInterrupt
>>> next(s)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

The interesting thing is that after interrupting the last next operation, if I call next again, then I get a StopInteration.

This is how I think it should work:

>>> p = iter(range(24))
>>> s = itertools.izip_longest(*([ iter(p) ] * 10))
>>> next(s)
(0, 1, 2, 3, 4, 5, 6, 7, 8, 9)
>>> next(s)
(10, 11, 12, 13, 14, 15, 16, 17, 18, 19)
>>> next(s)
(20, 21, 22, 23, None, None, None, None, None, None)
>>> next(s)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

[amoffat]

Your process is already an iterator, by virtue of passing _iter=True when you launched the process, so no need to wrap it again in iter(). Also, the docs for itertools.izip_longest say that this is the behavior of the functions:

izip_longest('ABCD', 'xy', fillvalue='-') --> Ax By C- D-

It's designed to take all the args as (presumably?) unique iterators, pull one from each, and zip it up. The fact that you're using the same iterator 10 times is strange. It means izip_longest will attempt to pull a value from the first iterator, and it gets one, then tries to pull a value from the second iterator (which is also the first iterator!) and gets the second value of the first iterator...etc etc. For each iterator index N, you're getting the Nth value of that single iterator that you've duplicated N times. Is this a standard practice for chunking iterators?

Thinking about it this way, it makes sense why the code is locking up. Your resulting iterator (the one returned from izip_longest) can only yield a value IFF it has collected 10 values from the single iterator you passed in 10 times. Let's say your process would return 6 values but you're asking for 10. When izip_longest gets to iterator 7, the process raises StopIteration. To izip_longest, this is normal, as any iterator it is receiving could raise StopIteration when its done. So izip_longest merrily goes to iterator 8 and asks it to yield a value, and since you're using the same iterator for each argument, that iterator is already done... the process has ended and StopIteration was already raised previously.

So the bug here seems to be that multiple calls to next() on a process iterator where the process iterator has stopped, will block. Granted, you've found a very strange way of uncovering it :) It can be shown like this as well:

>>> p = sh.ls(_iter=True)
>>> next(p) # repeat this until StopIteration, then do it one more time

In the meantime, this should accomplish the same thing you're trying to do and works correctly with sh because it stops after the process says its done.

def chunk_iterator(it, times):
    chunks = []
    for piece in it:
        chunks.append(piece)
        if len(chunks) == times:
            yield chunks
            chunks = []
    yield chunks

[ghost]

Thank you for your response.

Is this a standard practice for chunking iterators?

I believe so. A similar piece of code is included in the recipes section of itertools (in the grouper function).

Yes, definitely calling next(p) in your simpler example repeatedly is producing the same behavior, but in that case, at least the first response is a StopIteration exception. Here you can see another example where instead of returning a list with some elements and filled with None, it just freezes (in this example, I think I created 25 files which are returned in 3 or 4 pieces by sh.ls):

>>> p = sh.ls(_iter=True)
>>> s = itertools.izip_longest(*([ iter(p) ] * 100))
>>> next(s)
^CTraceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/site-packages/sh.py", line 559, in next
    chunk = self.process._pipe_queue.get(True, 0.001)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/Queue.py", line 177, in get
    self.not_empty.wait(remaining)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/threading.py", line 359, in wait
    _sleep(delay)
KeyboardInterrupt

A regular iterator follows this behavior:

>>> p = iter(range(5))
>>> s = itertools.izip_longest(*([ p ] * 100))
>>> next(s)
(0, 1, 2, 3, 4, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None, None)
>>> next(s)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

Are you sure chunk_iterator is working? I'm getting this:

>>> p = sh.ls(_iter=True)
>>> next(chunk_iterator(p, 2))
[u'file1\tfile12\tfile15\tfile18\tfile20\tfile23\tfile3  file6  file9\n', u'file10\tfile13\tfile16\tfile19\tfile21\tfile24\tfile4  file7\n']
>>> next(chunk_iterator(p, 2))
[u'file11\tfile14\tfile17\tfile2\tfile22\tfile25\tfile5  file8\n']
>>> next(chunk_iterator(p, 2))
^CTraceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 3, in chunk_iterator
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/site-packages/sh.py", line 559, in next
    chunk = self.process._pipe_queue.get(True, 0.001)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/Queue.py", line 177, in get
    self.not_empty.wait(remaining)
  File "/home/user/anaconda/envs/sh-env/lib/python2.7/threading.py", line 359, in wait
    _sleep(delay)
KeyboardInterrupt

If I set a variable x = chunk_iterator(p,2), then it works:

>>> p = sh.ls(_iter=True)
>>> x = chunk_iterator(p, 2)
>>> next(x)
[u'file1\tfile12\tfile15\tfile18\tfile20\tfile23\tfile3  file6  file9\n', u'file10\tfile13\tfile16\tfile19\tfile21\tfile24\tfile4  file7\n']
>>> next(x)
[u'file11\tfile14\tfile17\tfile2\tfile22\tfile25\tfile5  file8\n']
>>> next(x)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
StopIteration

[amoffat]

>>> p = sh.ls(_iter=True)
>>> next(chunk_iterator(p, 2))
>>> next(chunk_iterator(p, 2))

Don't do that. It's creating multiple chunked iterator for the same underlying iterator. It's causing the problem of calling a completed process iterator more than once again.

>>> output_iterator = sh.ls(_iter=True)
>>> chunks = chunk_iterator(output_iterator, 2)
>>> next(chunks)
>>> next(chunks) # etc

That is the correct usage

[ghost]

It's causing the problem of calling a completed process iterator more than once again.

Yes, that's what I thought. Thanks.

[amoffat]

Sure thing, thanks for reporting the underlying bug. I'll leave this open until I fix it.

[amoffat]

This is on the release-1.2 branch and will go out with it, closing this issue

[ghost]

Great. Thank you!