def _Release(self, sink):
    # Releasing a queuing sink is a noop
    if (isinstance(sink, QueuingMessageSink) or
        isinstance(sink, FailingMessageSink)):
      self._varz.queue_size(len(self._waiters))
      return

    do_close = False
    # This sink is already shutting down
    if self.state == ChannelState.Closed:
      self._current_size -= 1
    # One of the underlying sinks failed, shut down
    elif sink.state == ChannelState.Closed:
      self._current_size -= 1
      self.Close()
    # There are some waiters queued, reuse this sink to process another request.
    elif any(self._waiters):
      gevent.spawn(self._ProcessQueue, sink)
    # We're below the min-size specified, cache this sink
    elif self._current_size <= self._min_size:
      self._cache.append(sink)
    # We're above the min-size, close the sink.
    else:
      self._current_size -= 1
      do_close = True

    self._varz.size(self._current_size)
    if do_close:
      self._DiscardSink(sink)
dataset/ETHPy150Open steveniemitz/scales/scales/pool/watermark.py/WatermarkPoolSink._Release/original