Pub / Sub Publisher: Async publish breaks around 1000 messages

[anorth2]

Am unsure if this is a bug (can't find documentation explaining this behavior).
I can send up to 999 pubsub messages and wait for results before continuing, but if I try to send 1000, none of them complete. However, if I run in debug mode, they will complete (because I have a breakpoint in the code allowing more execution time).

It exhibits this failure both in the main thread, or calling Publisher.publish from a thread.

2. OS type and version
   MacOS High Sierra

3. Python version and virtual environment information python --version
   Python 3.6.3

4. google-cloud-python version pip show google-cloud, pip show google-<service> or pip freeze

google-api-core==0.1.2
google-auth==1.2.1
google-cloud-core==0.28.0
google-cloud-pubsub==0.29.2
google-gax==0.15.16
googleapis-common-protos==1.5.3
grpc-google-iam-v1==0.11.4

5. Stacktrace if available

None, hanging, no exception.

6. Steps to reproduce

Publish 1000 messages storing their futures. None of the futures will complete.

7. Code example
   Goal here is to send specified number of messages per second and sleep.

 def publish_function(self):
 # save all futures so we can wait for them to complete
 # Callbacks receive the future as their only argument, as defined in
 # the Future interface.
 def callback(future):
 message_id = future.result()
 self.logger.debug(f"Sent message: {message_id}")

 # The callback is added once you get the future. If you add a callback
 # and the future is already done, it will simply be executed immediately.
 future = self.publisher.publish(self.topic_path, data=self.message)
 print(future)
 future.add_done_callback(callback)
 self.results.append(future)

 def publish(self, total_messages, messages_per_second):
 i = 0
 sleep_time = 1
 if messages_per_second < 1:
 sleep_time = float(sleep_time) / float(messages_per_second)
 logger.debug(f"Sleep time is: {sleep_time}")
 while i < total_messages and not self.exit:
 start = time.time()
 # number of times to send per second
 # if less than one, then the sleep timer handles the wait. We just send one message here
 for j in range(int(math.ceil(messages_per_second))):
 self.publish_function()
 i += 1
 self.next_id()
 # exit inner loop if we hit the total to send
 if i >= total_messages or self.exit:
 self.stop()
 break

 # apply result to all futures in self.results. This blocks until the future it's called on completes
 # list forces the map to execute since map returns an iterable


 list(map(futures.Future.result, self.results)) <<<<<<<<<<<--------- ***HANGS HERE***

 end = time.time()
 logger.info(
 f"Took {end - start} to send {messages_per_second} messages in thread {current_thread().getName()}")
 logger.debug(f"If greater than 0 seconds, sleeping for {sleep_time - (end - start)}")
 if sleep_time - (end - start) > 0:
 time.sleep(sleep_time - (end - start))

[dhermes]

@anorth2 I'm fairly certain it's a bug. I ran into it yesterday when reproducing a subscriber issue that required a large backlog of published messages. I got as high as 1375 messages through before failure, but not consistently.

Thanks for filing.

[anorth2]

@dhermes Also wanted to ask, are there expected side effects from calling publisher.publish from a thread?

if I use the following code list(map(Future.result, running_futures)) (where the Future is from from concurrent.futures._base import Future)
where running_futures is a list of threads each executing publisher.publish, I get an error from the publisher where len(message_ids) != len(futures)

[dhermes]

@anorth2 To provide a quick answer: I don't know. Since I've started the "hardening" process on the highly concurrent Pub / Sub implementation (i.e. 0.28.x and beyond) I have only worked on the subscriber. But I hope to know more later this week.

[dhermes]

I have a reproducible sample that I am debugging here: https://github.com/dhermes/google-cloud-pubsub-performance/tree/master/publish-many