Pub / Sub doesn't work with multiprocessing

[theacodes]

This code fails with a pickle error:

def multiprocessing_callback(message):
 print('Received message: {}'.format(message))
 message.ack()


def receive_messages_with_multiprocessing(
 project, topic_name, subscription_name):
 """Receives messages from a pull subscription using multiprocessing."""
 import concurrent.futures
 import functools
 import multiprocessing

 import google.cloud.pubsub_v1.subscriber.policy.thread

 # Create a process pool and a queue for sending messages. This will
 # be used by the subscriber policy to execute callbacks.
 executor = concurrent.futures.ProcessPoolExecutor()
 manager = multiprocessing.Manager()
 queue = manager.Queue()

 policy_factory = functools.partial(
 google.cloud.pubsub_v1.subscriber.policy.thread.Policy,
 executor=executor,
 queue=queue)
 subscriber = pubsub_v1.SubscriberClient(policy_class=policy_factory)
 subscription_path = subscriber.subscription_path(
 project, subscription_name)

 subscriber.subscribe(subscription_path, callback=multiprocessing_callback)

 # The subscriber is non-blocking, so we must keep the main thread from
 # exiting to allow it to process messages in the background.
 print('Listening for messages on {}'.format(subscription_path))
 while True:
 time.sleep(60)

[lukesneeringer]

Can you provide the traceback, too? ;-)

[theacodes]

It's pretty unhelpful:

Traceback (most recent call last):
 File "/Users/jonwayne/.pyenv/versions/3.6.0/lib/python3.6/multiprocessing/queues.py", line 241, in _feed
 obj = _ForkingPickler.dumps(obj)
 File "/Users/jonwayne/.pyenv/versions/3.6.0/lib/python3.6/multiprocessing/reduction.py", line 51, in dumps
 cls(buf, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

[danoscarmike]

Hi @lukesneeringer what's the next step here?

[michaelplaing]

This occurs when using the "spawn" context of multiprocessing. However the "fork" context will also fail (but silently) when using grpcio==0.1.6. A workaround for "fork" is to specify grpcio==0.1.4.

[dhermes]

So I can't reproduce the pickling error with google-cloud-pubsub==0.30.0, but it still fails (eventually) with a

E1219 12:19:30.755077995 16917 ssl_transport_security.cc:436] Corruption detected.
E1219 12:19:30.755141409 16917 ssl_transport_security.cc:412] error:100003fc:SSL routines:OPENSSL_internal:SSLV3_ALERT_BAD_RECORD_MAC
E1219 12:19:30.755162457 16917 secure_endpoint.cc:185] Decryption error: TSI_DATA_CORRUPTED