René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Pubsub: Consider a blocking publisher to prevent OOMing","articleBody":"Publishing medium-sized (32 kiB) messages in a loop causes the process's memory usage to quickly run out. I set a 2 GiB memory limit, and the process runs out of memory after publishing about 20000 messages. I believe there are two issues:\r\n\r\n1. There is no limit to the number of queued bytes in the process.\r\n2. The client library keeps 3 copies of the messages and retains them until the batch \"monitor\" thread finishes.\r\n\r\n\r\nAs far as I can see, there no good way to use the current client library to publish large batches of messages without HUGE memory consumption. I believe this can be fixed in two ways:\r\n\r\n#### Reduce memory of the current library:\r\n\r\n1. Release messages when the batch is done: In `thread.Batch._commit`, set `self._messages` to `None` after it is published. This *seems* to make memory consumption better, but I didn't test it super carefully.\r\n\r\n2. Eliminate an extra copy: Change thread.Batch to build a PubsubRequest, and pass *that* into the gapic.PublisherClient, rather than a list of messages that then need to be copied. This reduces one copy of messages, which seems to make memory consumption better (and possibly make things faster? again not tested carefully)\r\n\r\n#### Back Pressure\r\n\r\nChange the publisher to keep a queue of batches, and block after there are more than N in flight at once.\r\n\r\nI implemented a client that takes the blocking queue approach. It can publish this workload without ever exceeding 300 MiB of memory. Only 3 in-flight batches were necessary to have the process be CPU bound when running in Google Cloud, so I don't think the queue doesn't even need to be too large.\r\n\r\n\r\n\r\n#### Environment details\r\n\r\n*OS*: Linux, ContainerOS (GKE), Container is Debian9 (using distroless)\r\n*Python*: 3.5.3\r\n*API*: google-cloud-python 0.41.0\r\n\r\n\r\n\r\n#### Steps to reproduce\r\n\r\n1. Package the following program in a Docker container.\r\n2. Run it in Kubernetes with a memory limit of 2 GiB\r\n3. Watch it blow up when you run it.\r\n\r\n\r\n#### Code example\r\n\r\n```python\r\n#!/usr/bin/env python3\r\n'''\r\nCreate the topic and a subscription:\r\n\r\ngcloud pubsub topics create deleteme_publish_memleak_test\r\ngcloud pubsub subscriptions create deleteme_memleak_subscription --topic=deleteme_publish_memleak_test\r\n'''\r\n\r\nfrom google.cloud import pubsub_v1\r\nimport argparse\r\nimport google.auth\r\nimport logging\r\nimport time\r\n\r\n\r\ndef main():\r\n    parser = argparse.ArgumentParser(description='mass publish to pub/sub')\r\n    parser.add_argument('--topic',\r\n                        default='deleteme_publish_memleak_test',\r\n                        help='topic to publish to')\r\n    parser.add_argument('--messages',\r\n                        type=int,\r\n                        default=200000,\r\n                        help='number of messages to publish')\r\n    parser.add_argument('--bytes_per_message',\r\n                        type=int,\r\n                        default=32 * 1024,\r\n                        help='bytes per message to publish')\r\n    parser.add_argument('--log_every_n', type=int, default=2000, help='log every N messages')\r\n    args = parser.parse_args()\r\n\r\n    credentials, project_id = google.auth.default()\r\n    publish_client = pubsub_v1.PublisherClient(credentials=credentials)\r\n    topic_path = publish_client.topic_path(project_id, args.topic)\r\n\r\n    logging.basicConfig(\r\n        level=logging.DEBUG,\r\n        format='%(asctime)s %(levelname)s %(name)s: %(message)s',\r\n    )\r\n    logging.info('publishing %d messages to topic %s; %d bytes per message; %.1f total MiB',\r\n                 args.messages, topic_path, args.bytes_per_message,\r\n                 (args.messages * args.bytes_per_message) / 1024. / 1024.)\r\n\r\n    start = time.time()\r\n    for i in range(args.messages):\r\n        message = (i % 256).to_bytes(1, byteorder='little') * args.bytes_per_message\r\n        publish_client.publish(topic_path, message)\r\n        if i % args.log_every_n == 0:\r\n            logging.info('published %d messages', i)\r\n\r\n    end = time.time()\r\n    logging.info('done publishing in %f s', end - start)\r\n\r\n\r\nif __name__ == '__main__':\r\n    main()\r\n```\r\n","author":{"url":"https://github.com/evanj","@type":"Person","name":"evanj"},"datePublished":"2019-05-16T21:39:24.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":6},"url":"https://github.com/16/python-pubsub/issues/16"}