[Python][Flight] pyarrow.flight.FlightServerBase use threading instead of multiprocessing?

[tianjiqx]

I tried to use arrow flight, but when testing the performance (do_exchange), I found that the cpu usage can only be around 100%, maybe it is the GIL problem of python multi-threading.

I didn't find anything about setting the thread pool for FlightServerBase, so it seems that if you want to improve performance, it seems that you can only use multiprocessing.Pool internally to handle the internal working logic.

Do you have any other suggestions? Thanks!

[tianjiqx]

I've searched for similar questions and maybe learned something, but anyone with more suggestions is welcome.

• [Python] [Flight] How does concurrency in Arrow Flight work? Where in the source code can I find it?
• Best practices to use Arrow Flight Server arrow#13300

[lidavidm]

Yes, if you have heavy internal processing, your only real choice is to use multiprocessing. Depending on what you want to do, you could also set the SO_REUSEPORT gRPC option and spawn many processes listening to the same port.

There's no option to set the thread pool because 1) it's provided by gRPC and 2) it wouldn't help anyways, you're still limited by the GIL.

[tianjiqx]

@lidavidm thank you for your reply. My program still have some IO. Considering the copy cost of multi-process communication, it may be better to start more FlightServer.

[lidavidm]

Yeah, it does make it hard to build a CPU-intensive service in Python.

Maybe the SO_REUSEPORT option can be an Arrow Cookbook recipe - is that a good solution here?

[tianjiqx]

Probably didn't understand how to use SO_REUSEPORT. Sometimes it is possible to bind to the same port, but the created clients always only connect to the same server.

@contextlib.contextmanager
def _init_grpc_port(grpc_port):
    """Initialize grpc port for multiprocessing."""
    # Reference code.
    # https://github.com/grpc/grpc/issues/17659

    socket = py_socket.socket(py_socket.AF_INET6, py_socket.SOCK_STREAM)
    # in fact, py_socket.SOL_SOCKET = 65535
    # print(py_socket.SOL_SOCKET)

    socket.setsockopt(py_socket.SOL_SOCKET, py_socket.SO_REUSEADDR, 1)
    if socket.getsockopt(py_socket.SOL_SOCKET, py_socket.SO_REUSEADDR) == 0:
        raise RuntimeError("Failed to set SO_REUSEADDR.")
    socket.setsockopt(py_socket.SOL_SOCKET, py_socket.SO_REUSEPORT, 1)
    if socket.getsockopt(py_socket.SOL_SOCKET, py_socket.SO_REUSEPORT) == 0:
        raise RuntimeError("Failed to set SO_REUSEPORT.")
    socket.bind(('', grpc_port))
    try:
        yield socket.getsockname()[1]
    finally:
        socket.close()

def main():
    with _init_grpc_port(5005) as port:
        pass

    for i in range(2):
        p = Process(target=server.start, args=(i, ))
        # if set time.sleep(1) then always throw "os_error":"Address already in use","syscall":"bind"
        p.start()

[lidavidm]

The option has to be passed through gRPC (the code above effectively does nothing). See generic_options and the gRPC documentation

I'll see about adding a code snippet in the cookbook

[abjidge]

@lidavidm Can you please share or document the code snippet for this.

[lidavidm]

Ah, hmm, this isn't overridable from Python. Do you want to file an issue on the main repo?

[abjidge]

@lidavidm, Thank you for quick response.

Yes, Will file an issue.
Is there any known workaround to use multiprocessing for a pyarrow flight server?

[lidavidm]

Workaround? What is the issue?

[abjidge]

I mean is there any other way to use pyarrow server with multiprocessing until the above issue get resolved?

[lidavidm]

Sorry, I don't understand. You can just use the multiprocessing module. Is there a problem when using it?