code: def _service_futures(self, futures, until=None):
          
          if until is None:
              timeout = None
          else:
              timeout = until - time.time()
              if timeout < 0:
                  timeout = 0
          try:
              response = self._q.get(timeout)
          except TimeoutError:
              raise TimeoutError(
                  "Timeout waiting for %s" % self._describe_futures(futures))
          if response is self._sentinel_stop:
              self._sentinel_stop = None
          elif response is self.STOP:
              if self._sentinel_stop is None:
                  # This is a stop we should listen to...
                  raise AbortedError(
                      "Aborted waiting for %s" % self._describe_futures(futures))
          elif isinstance(response, Update):
              # This is an update for a subscription
              if response.id in self._subscriptions:
                  func, args = self._subscriptions[response.id]
                  func(response.value, *args)
          elif isinstance(response, Return):
              future = self._futures.pop(response.id)
              del self._requests[future]
              self._pending_unsubscribes.pop(future, None)
              result = response.value
              future.set_result(result)
              try:
                  futures.remove(future)
              except ValueError:
                  pass
          elif isinstance(response, Error):
              future = self._futures.pop(response.id)
              del self._requests[future]
              future.set_exception(response.message)
              try:
                  futures.remove(future)
              except ValueError:
                  pass
              else:
                  raise response.message
docs: Args:
              futures (list): The futures to service
              until (float): Timestamp to wait until