python grpc server with multiprocessing fails

[p1c2u]

What version of gRPC and what language are you using?

grpc 1.13.0 with python 3.6.5

What operating system (Linux, Windows, …) and version?

CentOS 7 (Linux 3.10.0-862.3.2.el7.x86_64)

What runtime / compiler are you using (e.g. python version or version of gcc)

CPython 3.6.5

What did you do?

server = grpc.server(futures.ProcessPoolExecutor(max_workers=4))
service_pb2_grpc.add_FindPortServicer_to_server(FindPort(), server)
server.add_insecure_port('[::]:' + port)
server.start()

What did you expect to see?

Based on doc https://github.com/grpc/grpc/blob/master/doc/fork_support.md
gRPC server should run on 4 processes

What did you see instead?

Server crash

8156 tcp_server_posix.cc:210]    Failed accept4: Invalid argument
Traceback (most recent call last):
  File "/usr/lib64/python3.6/multiprocessing/queues.py", line 234, in _feed
    obj = _ForkingPickler.dumps(obj)
  File "/usr/lib64/python3.6/multiprocessing/reduction.py", line 52, in dumps
    cls(buf, protocol).dump(obj)
  File "stringsource", line 2, in grpc._cython.cygrpc.RequestCallEvent.__reduce_cython__
TypeError: no default __reduce__ due to non-trivial __cinit__

Make sure you include information that can help us debug (full error message, exception listing, stack trace, logs).

Anything else we should know about your project / environment?

[mehrdada]

@ericgribkoff is this a supported use-case?

[ericgribkoff]

No, this is not supported. futures.ProcessPoolExecutor forks in response to submitted tasks, meaning after the server starts (gRPC Python sends tasks to the executor to handle incoming RPCs). This is not compatible with gRPC's fork support on the server-side. You will need to pre-fork, meaning fork your subprocesses and then have each one (separately) start their own gRPC server instance.

What did you expect to see?

Based on doc https://github.com/grpc/grpc/blob/master/doc/fork_support.md
gRPC server should run on 4 processes

@p1c2u Thanks for pointing this out. I see that our documentation could be a bit confusing if you're looking at using fork with gRPC servers: we should update that document to make it more clear that the fork-support there is for client-side usage. (cc @kpayson64)

[mohit-chawla]

[PYTHON] @ericgribkoff , I have ran into a use case where i am using a gRPC for a server side CPU bound task (and need to use python multi processing as there isn't an option in pure Python). Can you please suggest a solution here as i see no support for server side for fork ?

I am facing grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.INTERNAL, Received RST_STREAM with error code 2)>

@evanj How did you deal with #15334 (comment) ?

[kriticism]

@ericgribkoff , @evanj : I'm facing the same issue as @mohit-chawla

[yogin16]

+1

[ericgribkoff]

The following two approaches should help with combining gRPC Python servers and the fork() syscall (e.g., using Python's multiprocessing library). The main constraint that must be satisified is that your process must only invoke the fork() syscall before you create your gRPC server(s).

Option 1:

If your environment supports the SO_REUSEPORT socket option, you can run multiple copies of your gRPC server in individual processes started via multiprocessing.Process, and all of the servers can listen on the same port. SO_REUSEPORT is set by default if you build gRPC from source on Linux, but it is not available in the manylinux spec used by the binaries we distribute on pip, so depending on your environment you may have to resort to option 2.

This type of pre-fork + SO_REUSEPORT would look something like the following:

def startGrpcServer():
    server = grpc.server()
    server.add_insecure_port('[::]:50051')
    server.start()

for i in range(5):
  p = multiprocessing.Process(target=startGrpcServer)
  p.start()

Option 2:

Run a single gRPC Python server, but offload all CPU-intensive work to a multiprocessing.Pool started before you create the gRPC server.

This would look like the following:

pool = multiprocessing.Pool(processes=4)

# The implementation of your RPC method. This will run in the main process, 
# but the work will be done asynchronously by the pool of pre-forked workers.
def rpcMethod(self, request, context):
  result = pool.apply_async(someExpensiveFunction(request)
  return result.get(timeout=1)

server = grpc.server()
server.add_insecure_port('[::]:50051')
server.start()

[ferrouswheel]

Did this behaviour change recently?

I have a grpc server invoking tensorflow, but tensorflow doesn't have anyway to explicitly let go of GPU memory.

Thus my solution, when receiving a grpc request, was to spawn a new process to do the work, then let it terminate after the work was done.

I was pretty sure I had this working with grpc earlier in 2018, but now I get Failed accept4: Invalid argument... 🤔

(I can work around it in an awkward way, but I'd like to clarify if I'm going mad)

[ericgribkoff]

@ferrouswheel gRPC servers using fork() (without following this with an immediate exec()) to process incoming requests has never been supported. Depending on a variety of factors (including the version of gRPC, what polling engine was in use) it might have worked sometimes, but probably not often enough to appear functional through any non-trivial amount of usage.

However, if you were spawning a new process via fork()+exec(), that should have worked - and should continue to work.