python gcp pubsub async publish breaks under load or does not send messages

I am trying to run a few load tests for a GCP PubSub-> Filebeat -> ElasticSearch pipeline but started running into an issue with my Python load generating app: I can't send more than ~2-8K events per second, as at that point my async Python publisher threads start failing with the error: RuntimeError: can't start new thread

Stacktrace:

in grpc._cython.cygrpc._call
File "src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi", line 62, in grpc._cython.cygrpc._get_metadata
File "src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi", line 28, in grpc._cython.cygrpc._spawn_callback_async
File "src/python/grpcio/grpc/_cython/_cygrpc/credentials.pyx.pxi", line 19, in grpc._cython.cygrpc._spawn_callback_in_thread
File "src/python/grpcio/grpc/_cython/_cygrpc/fork_posix.pyx.pxi", line 117, in grpc._cython.cygrpc.ForkManagedThread.start
File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 852, in start
_start_new_thread(self._bootstrap, ())
RuntimeError: can't start new thread

Here is the gist of the setup:

def main():
... 
batch_settings = types.BatchSettings(
    max_messages=100,  # default 100
    max_bytes=1024,  # default 1 MB
    max_latency=1,  # default 10 ms
)
global publisher
publisher = pubsub_v1.PublisherClient(batch_settings)

...

jobs = []
for thread_number in range(num_threads):
    t = multiprocessing.Process(target=generate_and_publish_events,
                                args=(run_id, num_iterations, thread_number, iter_batch_size, sleep_seconds))
    jobs.append(t)
    t.start()  # new child process is started at this point, it has its own execution flow

for curr_job in jobs:  # wait for all processes to finish
    curr_job.join()

And here are a few scenarios I tried - and their [unsatisfactory :-( ] results: For all scenarios, I ran the following load profile:

• gcp batch size: 100 events
• 20 threads , each doing:
• • 1000 events per batch (or iteration as I call, to distinguish from the gcp's internal pubsub batch size)
• • 2 iterations
• • 1 second sleep between each iteration

Scenario 1: Publish all events with a registered callback, wait for all to complete:

def pubsub_callback(future):
    message_id = future.result()
    #print('Message published: {}'.format(message_id))


def publish_events_to_pubsub(gcp_events):
    publish_futures = []
    for event in gcp_events:
        publish_future = publisher.publish(topic_path, event)
        publish_future.add_done_callback(pubsub_callback)
        publish_futures.append(publish_future)

    futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

Results: 2K events are published to PubSub, after that - client fails with above errors

Scenario 2: Publish all events without a callback, wait for all to complete:

def publish_events_to_pubsub(gcp_events):
    publish_futures = []
    for event in gcp_events:
        publish_future = publisher.publish(topic_path, event)
        publish_futures.append(publish_future)

    futures.wait(publish_futures, return_when=futures.ALL_COMPLETED)

Two variations:

Scenario 2A: GCP batch size = 100

Scenario 2B: GCP batch size = 1000

Results: (same for 2A and 2B): 8K events are published to PubSub, after that - client fails with above errors

Scenario 3: Publish events without a callback and do not wait for the futures completion:

def publish_events_to_pubsub(gcp_events):
    for event in gcp_events:
        publish_future = publisher.publish(topic_path, event)

Results: NO events are published at all!

Unfortunately, I could not find any documentation on how the Python GCP pubsub client works internally (beyond the "Hello World"-type of examples) - so would appreciate if someone could shed some light on why I'm hitting this ceiling and then those errors. It seems like there are tons of threads being created internally by the client, which eventually hits some thread limit per Python process (?), but I could not find any info on how to control that...

Thank you!! Marina

UPDATE:

The "RuntimeError: can't start new thread" error was thrown when the above load test was run from PyCharm IDE.

When I ran the test from a command line directly - the errors were different (but were still happening after about 8-10K events were sent):

E0921 11:59:14.043220000 123146771587072 wakeup_fd_pipe.cc:39]         pipe creation failed (24): Too many open files
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 296, in wait
waiter.acquire()
E0921 15:59:13.995942000 123146493063168 ev_poll_posix.cc:944]           pollset_work: {"created":"@1663775953.995920000","description":"Too   many open files","errno":24,"file":"src/core/lib/iomgr  /wakeup_fd_pipe.cc","file_line":40,"os_error":"Too many open files","syscall":"pipe"}
E0921 15:59:14.035606000 123147020726272 wakeup_fd_pipe.cc:39]         pipe creation failed (24): Too many open files