Odd Hanging Behaviour on OrderedWorkers Based On Payload Size

[lycovian]

I'm seeing hangs using OrderedWorker if the payload used in put is longer than a certain size. Here is an example that will randomly hang on my machine if a single character is added to the payload:

from __future__ import print_function
import random
import mpipe

class OWorker(mpipe.OrderedWorker):
    def doTask(self, value):
        return value

if __name__ == '__main__':

    stage_get = mpipe.Stage(OWorker, 2)
    p = mpipe.Pipeline(stage_get)

    j=95
    if random.randint(0,1):
        print("HANGS")
        j += 1
    else:
        print("WORKS")

    for i in list(range(10)):
        d = {'payload': 'a'*j}
        p.put(d)

    p.put(None)

    for r in p.results():
        print(r)

Tested hanging behavior on OS X using Python 2.7.11 and 3.5.1. Latest version of mpipe installed via pip install mpipe.

[vmlaker]

@lycovian, unfortunately i'm not able to duplicate the result--it's not hanging on my platform with either 95 or 96 character string (Python 2.7.12 and 3.5.2, mpipe version 1.0.8, Ubuntu 16.04.) I'm not seeing anything wrong with your code, either.

[jleoirab]

Hi all, I've been noticing this issue on my end as well. My case is somewhat similar. I was working with Python 3.6 on OSX. I was trying to send about 64bytes and OrderedWorker seemed to hang while trying to put to its output tube.

I didn't do much investigation but I felt that I should speak out for anyone having this issue. I looked at this which lead me to believe that there might be and issue with the TubeP implementation that would cause it to block. However, I didn't find anything out of the ordinary.

Long story short, I just switched to the TubeQ implementation by overriding the getTubeClass for my workers and everything seemed to work. Again, I'm not sure of what exactly is wrong with TubeP but switching away from it seemed to fix my issue.