Queue closing does not affect sync .put() calls in waiting state

[kc41]

trafficstars

Hi! I found some potentially unexpected behaviour of queue closing. If thread producer blocks on attempt to sync put to queue and we close queue in another control thread, thread producer will wait forever. I suppose that expected behaviour should be a RuntimeError in sync put() method on queue closing. What do you think about it?

Here is a code to reproduce this situation:

import asyncio
import logging
from concurrent.futures.thread import ThreadPoolExecutor
from queue import Queue

import janus

logging.basicConfig(format='%(threadName)-12s: %(message)s', level=logging.DEBUG)


async def main(tpe):
    hybrid_q = janus.Queue(maxsize=1)

    def some_long_job(q: Queue):
        logging.info("Job is running")
        for i in range(int(1e6)):
            try:
                logging.info("Putting to q: %s", i)
                q.put(f"item_{i}")
                logging.info("Putting to q done: %s", i)
            except Exception as ex:
                logging.exception(ex)
                raise

    job = asyncio.ensure_future(asyncio.get_event_loop().run_in_executor(tpe, some_long_job, hybrid_q.sync_q))
    await asyncio.sleep(0.5)

    job.cancel()
    logging.info("Closing queue")
    hybrid_q.close()

    logging.info("Waiting q to be closed")
    await hybrid_q.wait_closed()
    logging.info("Queue was closed")


if __name__ == '__main__':
    tpe = ThreadPoolExecutor(thread_name_prefix="TPE_")
    asyncio.run(main(tpe))
    logging.info("Shutting down TPE")
    tpe.shutdown(wait=True)
    logging.info("TPE was shut down")

[asvetlov]

Thanks for the question.

I'm not sure if the exception raising is good for this case: it just means that every q.put() should be wrapped in try/except because literally every call may raise RuntimeError. It looks very annoying.

[dplusic]

What about notifying q._sync_not_full in q.close()? We may get RuntimeError for closing as usual if q._check_closing() is called after _sync_not_full.wait() in sync_q.put().

[asvetlov]

Fixed by #267

[dplusic]

@asvetlov How could #267 fix this?

[asvetlov]

Ooops. Sorry, you are right. Hard day for me.

[asvetlov]

Please feel free to propose a pull request.