Skip to content

[Bug] Program hanging when input is throttling by maxsize and exception raised in stage #110

@chengjinluo

Description

@chengjinluo

Describe the bug
Please refer to the code example.

After some investigation, I found all process workers exits successfully.
But the initial worker thread which feed the input numbers into queue stuck on the block call of multiprocessing.Queue.put, and it can not get the exception raised by stopit.

Manually edit use_thread default value to False in pypeln.process.worker.start_workers will be a workaround (will start initial worker as a process instead of thread, and it will be terminated by signal).

It seems that pypeln.thread.Queue.IterableQueue has a different implementation, it will try put operation with timeout so that the thread is able to get the exception raised by stopit.

Minimal code to reproduce

import pypeln as pl

def proc(x):
    print(x, flush=True)
    if x == 10:
        assert False

if __name__ == '__main__':
    stage = pl.process.map(proc, list(range(100)), maxsize=4, workers=4)
    list(stage)

Expected behavior
Program exits with exception.

Library Info
os: ubuntu 22.04
python=3.10.12
pypeln=0.4.9

Screenshots
If applicable, add screenshots to help explain your problem.

Additional context
Add any other context about the problem here.

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions