Some condition causing EOFError, BrokenPipe error

[lzamparo]

Hi,

This is in relation to my use of this package for feeding a training loop for word2vec, but I think the problem lies here, so I'm posting the issue here. Here's an associated gist with the relevant trace and code snippets.

Based on the traceback, here's what I think is happening:

1. the producer queue of macrobatch_queue eventually becomes unable to put((signal_examples, noise_examples))into the underlying _work_queue of the macrobatch_queue producer. This causes the BrokenPipeError
2. the iterable for the producer queue of macrobatch_queue is somehow still open when generate_dataset_parallel finishes, leading to an EOFError (I think, see here)

So, I'm going to try closing macrobatch_consumer after the final for loop in the generate_dataset_parallel to see if it resolves [2], and inserting a short wait after the put in generate_dataset_worker to see if it fixes [1] (see here)

Have you ever seen this behaviour in your own experiments? Thanks in advance,

[enewe101]

@lzamparo I've updated this to fix a rare bug due to a race condition during initialization of the queue. I don't know if that is the cause of your issue (have not been able to reproduce it yet). Would you pip install -U theano-word2vec (which should force updating of iterable-queue as well), and tell me if you still get the error?

[lzamparo]

@enewe101 thanks! Though after further experimentation, I think the source of my error is just that I was overloading the macrobatch_queue. So, my explanation of [2] was wrong :/

I eventually just counted (via logging) the number of times I enqueue to macrobatch_queue versus dequeue from same, and there was a ~10x difference. After explicitly slowing down the generate_dataset_workers and spawning fewer of them, I don't see this error on my computation anymore. Will certainly update the package, and try my original scenario.

[lzamparo]

I really wish somewhere in the Python docs for queue.Queue (i.e Manager.Queue) there was some discussion of the underlying size limits of a maxsize=0 queue.

[enewe101]

OK. I think that if you're correct, then this behavior, on putting onto a full IterbleQueue.Producer endpoint, is not ideal, so I'm keeping this open. There may be other things going on though.

I'm glad that you figured out a reasonably proximal cause in your case, and a workaround!

BTW, are you calling the word2vec.word2vec() function, or calling other classes manually?

[lzamparo]

I am using word2vec.word2vec() for training, since it seems really convenient. Though for validation I'm calling classes manually (populating them from previously serialized models).

[lzamparo]

I think that if you're correct, then this behavior, on putting onto a full IterbleQueue.Producer endpoint, is not ideal, so I'm keeping this open.

I'm not sure what the best policy is for dealing with this, though my preference is to delegate this to user code. In some cases, it might be fine to block until there is space to put. In other cases, it might be better to throw an exception, or even block to increase the maxsize and allow the putting to succeed.

Maybe example code for using IterableQueue could either get a return value for Producer.put, indicating success / failure (queue is full), or wrap in a try / except block, indicating that the except handling is up to the user?