1

I have a program where I have to pull files from over the network (p4 print pulls files from version control server and prints to stdout). Because the network and IO was the biggest bottleneck and I am trying to use asyncio. I tried using the standard asyncio.subprocess.PIPE, but because I have multiple subprocesses I keep getting deadlocks. The solution I want to try is to make a new file and have stdout write to there.

Here are some of the errors I got

Attempt 2: Error "OSError: [Errno 9] Bad file descriptor"

async def _subprocess_wrapper(self, path):
    async with self.sem:
        _, write = os.pipe()
        proc = await asyncio.create_subprocess_exec(
            'p4', 'print', '-q', path,
            stdout=write,
            stderr=write
        )
        status = await proc.wait()
        file = os.fdopen(write, 'r')
        txt  = file.read()
        os.close(write)
        os.close(_)
        return status, txt

Attempt 3: Error "AttributeError: 'NoneType' object has no attribute 'read'"

async def _subprocess_wrapper(self, path):
    async with self.sem:
        _, write = os.pipe()
        proc = await asyncio.create_subprocess_exec(
            'p4', 'print', '-q', path,
            stdout=write,
            stderr=write
        )
        status = await proc.wait()
        if status != 0:
            txt = await proc.stderr.read()
        else:
            txt = await proc.stdout.read()
        os.close(write)
        os.close(_)
        return status, txt.decode()

Any help would be appreciated

2
  • 2
    I strongly recommend to return to the standard asyncio and to try to identify and fix the deadlock. In asyncio all I/O simply must be either non-blocking or managed by the event loop. This loop uses select (or poll) to identify file descriptors (network sockets, pipes) which are ready to read/write and transfers data between these FDs and buffers. The application code for I/O doing await interacts with these data buffers, and not with the descriptors. Using a direct pipe I/O like you are trying to do simply does not fit into asyncio.
    – VPfB
    Commented Aug 18, 2022 at 16:08
  • 1
    The problem you have is reading the pipes in blocking way. See this answer (and question) on how to do this properly: stackoverflow.com/a/57736190/2538382 Commented Aug 21, 2022 at 15:13

1 Answer 1

2

I gave up on trying use my own pipe and changed my wait() to communicate, as per the docs...
[wait] can deadlock when using stdout=PIPE or stderr=PIPE and the child process generates so much output that it blocks waiting for the OS pipe buffer to accept more data. Use the communicate() method when using pipes to avoid this condition

My working code

async def _subprocess_wrapper(self, path):
    async with self.sem:
        proc = await asyncio.create_subprocess_exec(
            'p4', 'print', '-q', path,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await proc.communicate()
        txt = stdout if proc.returncode == 0 else stderr
        return proc.returncode, txt.decode()

If anyone knows if there is a better way to make this scale I would appreciate the insight

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.