title: Processing of google pubsub messages slows down (python)
body: <p>Since switching to the newest python libs, which are threading/callback based we've been hitting slowness between our pubsub producers and consumers. We're relatively new to google's pubsub, and we're wondering if others have experienced similar issues after the recent library changes, or know of settings we may have missed.</p>
  
  <p>We see an unexpected slow down from the time we push out messages to the time they're consumed by 3 workers (in python).  Our handlers take a small number of milliseconds to process each request, and we also changed our code to call <code>message.ack()</code> before running our handler. E.g. <code>self.sub_client.subscribe(subscription_path, callback=self.message_callback)</code>.  These messages aren't duplicates. When we enqueue them we record the the time in msecs to get an idea how long they're in the queue.</p>
  
  <pre><code>for pod in worker-staging-deployment-1003989621-2mx0n worker-staging-deployment-1003989621-b6llt worker-staging-deployment-1003989621-lx4gq; do echo == $pod ==; kubectl logs $pod -c fra-worker | grep 'ACK start'; done
  == fra-worker-staging-deployment-1003989621-2mx0n ==                                        
  [2017-09-25 23:29:03,987] {pubsub.py:147} INFO - ACK start: 22 ms for 1506382143.88 (0.10699987411499023 secs)                                                                                                                                                                                                                                                                                                      
  [2017-09-25 23:29:04,966] {pubsub.py:147} INFO - ACK start: 3 ms for 1506382144.767 (0.19900012016296387 secs)
  [2017-09-25 23:29:14,708] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382144.219 (10.488999843597412 secs)
  [2017-09-25 23:29:17,706] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382147.229 (10.476999998092651 secs)
  [2017-09-25 23:29:37,767] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382144.782 (32.984999895095825 secs)
  [2017-09-25 23:30:00,649] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382146.257 (54.39199995994568 secs)
  == fra-worker-staging-deployment-1003989621-b6llt ==
  [2017-09-25 23:29:04,083] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382143.957 (0.12599992752075195 secs)
  [2017-09-25 23:29:05,261] {pubsub.py:147} INFO - ACK start: 3 ms for 1506382144.916 (0.3450000286102295 secs)
  [2017-09-25 23:29:15,703] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382144.336 (11.367000102996826 secs)                                                                                                                                                                                                                                                                                                       
  [2017-09-25 23:29:25,630] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382143.812 (21.818000078201294 secs)
  [2017-09-25 23:29:38,706] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382144.49 (34.21600008010864 secs)
  [2017-09-25 23:30:01,752] {pubsub.py:147} INFO - ACK start: 3 ms for 1506382146.696 (55.055999994277954 secs)                                                                                                                                                                                                                                                                                                       
  == fra-worker-staging-deployment-1003989621-lx4gq ==                                                                                                                                                                                                                                                                                                                                                                
  [2017-09-25 23:29:03,342] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382142.889 (0.4530000686645508 secs)                                                                                                                                               
  [2017-09-25 23:29:04,955] {pubsub.py:147} INFO - ACK start: 2 ms for 1506382143.907 (1.0469999313354492 secs)                                                                                                                                                                                                                   
  [2017-09-25 23:29:14,704] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382143.888 (10.815999984741211 secs)                                                                                                                                                                                                                                                                                                       
  [2017-09-25 23:29:17,705] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382147.205 (10.5 secs)                                                                                                                                                                                                                                                                                                                     
  [2017-09-25 23:29:37,767] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382144.197 (33.5699999332428 secs)                                            
  [2017-09-25 23:29:59,733] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382144.269 (55.46399998664856 secs)                                                                                                                                                                                                                                                                                                        
  [2017-09-25 23:31:18,870] {pubsub.py:147} INFO - ACK start: 1 ms for 1506382146.924 (131.94599986076355 secs)                                                                                                                                                                                   
  </code></pre>
  
  <p>Initially it looks like messages are taking a short amount of time from enqueuing to reading, but then they start coming later and later as if there's a 10s, 32s, 55s delay.  (and these's aren't duplicates so this is not retry logic due to failed acks).</p>
  
  <p>We wrote a small test which works quickly for a small number of senders and messages, but once we bump up the messages to 1500 and senders to 3 we see that the publish call often returns a future with an exception result <code>"PublishError('Some messages were not successfully published.</code>  The results show around 500 messages/sec but an error rate of >10% of <code>publish()</code> calls throw this exception</p>
  
  <pre><code>Done in 2929 ms, 512.12 qps (154 10.3%)
  Done in 2901 ms, 517.06 qps (165 11.0%)
  Done in 2940 ms, 510.20 qps (217 14.5%)
  </code></pre>
  
  <p>Although our senders finish in 3 seconds (these run in parallel) the workers are getting messages which were enqueued over 20 secs ago</p>
  
  <pre><code>Got message {'tstamp': '1506557436.988', 'msg': 'msg#393@982'} 20.289 sec
  </code></pre>
  
  <p>Here's the worker/listener:</p>
  
  <pre><code>import time
  
  from google.api.core.exceptions import RetryError as core_RetryError
  from google.cloud import pubsub_v1
  from google.cloud.pubsub_v1.subscriber.policy import thread
  from google.gax.errors import RetryError as gax_RetryError
  import grpc
  
  from core.utils import b64json_decode, b64json_encode, Timer
  
  
  TOPIC = 'pubsub-speed-test'
  NOTIFY_PROJECT = '== OUR PROJECT =='
  
  
  def receive(message):
      decoded = b64json_decode(message.data)
      message.ack()
      took = time.time() - float(decoded.get('tstamp', 0))
      print(f'Got message {decoded} {took:0.3f} sec')
  
  
  if __name__ == '__main__':
      client = pubsub_v1.SubscriberClient()
      topic_path = client.topic_path(NOTIFY_PROJECT, TOPIC)
      subs_path = client.subscription_path(NOTIFY_PROJECT, 'pubsub-worker')
  
      try:
          create_subscription(subs_path, topic_path)
      except Exception:
          pass
      print(f'Subscription: topic={topic_path} subscription={subs_path}')
  
      timer = Timer()
      client.subscribe(subs_path, callback=receive)
      time.sleep(120)
  </code></pre>
  
  <p>And the sender/publisher:</p>
  
  <pre><code>import os
  import time
  import concurrent.futures
  
  from google.api.core.exceptions import RetryError as core_RetryError
  from google.cloud import pubsub_v1
  from google.cloud.pubsub_v1.subscriber.policy import thread
  from google.gax.errors import RetryError as gax_RetryError
  import grpc
  
  from core.utils import b64json_decode, b64json_encode, Timer
  
  TOPIC = 'pubsub-speed-test'
  NOTIFY_PROJECT = '== OUR PROJECT =='
  
  
  def publish(topic_path, message, client):
      tstamp = f'{time.time():0.3f}'
      data = {'tstamp': tstamp, 'msg': message}
      future = client.publish(topic_path, b64json_encode(data, raw=True))
      future.add_done_callback(lambda x: print(f'Publishing done callback: {data}'))
      return future
  
  
  if __name__ == '__main__':
      client = pubsub_v1.PublisherClient()
      topic_path = client.topic_path(NOTIFY_PROJECT, TOPIC)
  
      num = 1500
      pid = os.getpid()
      fs = []
      timer = Timer()
      for i in range(0, num):
          f = publish(topic_path, f'msg#{i}@{pid}', client)
          fs.append(f)
      print(f'Launched {len(fs)} futures in {timer.get_msecs()} ms')
  
      good = bad = 0
      for future in fs:
          try:
              data = future.result()
              # print(f'result: {data}')
              good += 1
          except Exception as exc:
              print(f'generated an exception: {exc} ({exc!r})')
              bad += 1
      took_ms = timer.get_msecs()
      pct = bad / num
      print(f'Done in {took_ms} ms, {num / took_ms * 1000:0.2f} qps ({bad} {pct:0.1%})')
  </code></pre>
  
  <p>Here's the Timer class from our core.utils:</p>
  
  <pre><code>####################
  # Time / Timing
  ####################
  
  
  def utcnow():
      """Time now with tzinfo, mainly for mocking in unittests"""
      return arrow.utcnow()
  
  
  def relative_time():
      """Relative time for finding timedeltas depening on your python version"""
      if sys.version_info[0] &gt;= 3:
          return time.perf_counter()
      else:
          return time.time()
  
  
  class Timer:
      def __init__(self):
          self.reset()
  
      def reset(self):
          self.start_time = relative_time()
  
      def get_msecs(self):
          return int((relative_time() - self.start_time) * 1000)
  
      def get_secs(self):
          return int((relative_time() - self.start_time))
  </code></pre>
  
  <p>Also, in our main code, we occasionally see IOErrors that the threads don't seem to recover from (in addition to DEADLINE_EXCEEDED which can be ignored).  To cope with that we've wrapped the Policy which let's us catch some exceptions and restart the clients as needed (though we're not sure that's working well)</p>
  
  <pre><code>class OurPolicy(thread.Policy):
      _exception_caught = None
  
      def __init__(self, *args, **kws):
          logger.info(f'Initializing our PubSub Policy Wrapper')  # noqa                                                                                                                                                                                                                                              
          return super(OurPolicy, self).__init__(*args, **kws)
  
      def on_exception(self, exc):
          # If this is DEADLINE_EXCEEDED, then we want to retry by returning None instead of raise-ing                                                                                                                                                                                                                
          deadline_exceeded = grpc.StatusCode.DEADLINE_EXCEEDED
          code_value = getattr(exc, 'code', lambda: None)()
          logger.error(f'Caught Exception in PubSub Policy Wrapper: code={code_value} exc={exc}')
          if code_value == deadline_exceeded:
              return
          OurPolicy._exception_caught = exc
          # will just raise exc                                                                                                                                                                                                                                                                                       
          return super(OurPolicy, self).on_exception(exc)
  
  [...later...]
  
                  while True:
                      time.sleep(1)
                      if OurPolicy._exception_caught:
                          exc = OurPolicy._exception_caught
                          OurPolicy._exception_caught = None
                          raise exc
  </code></pre>
  
  <p>And our versions:</p>
  
  <pre><code>pip freeze | grep goog
  gapic-google-cloud-pubsub-v1==0.15.4
  google-auth==1.0.2
  google-cloud-core==0.27.1
  google-cloud-pubsub==0.28.2
  google-gax==0.15.14
  googleapis-common-protos==1.5.2
  grpc-google-iam-v1==0.11.1
  proto-google-cloud-pubsub-v1==0.15.4
  </code></pre>