text: """
  Socket server based on a worker thread pool. Doesn't use select.
  
  Uses a single worker thread per client connection.
  
  Pyro - Python Remote Objects.  Copyright by Irmen de Jong (irmen@razorvine.net).
  """
  
  import socket
  import logging
  import sys
  import time
  import threading
  import os
  from . import config, socketutil, errors
  
  log = logging.getLogger("Pyro5.threadpoolserver")
  _client_disconnect_lock = threading.Lock()
  
  
  class ClientConnectionJob(object):
      """
      Takes care of a single client connection and all requests
      that may arrive during its life span.
      """
  
      def __init__(self, clientSocket, clientAddr, daemon):
          self.csock = socketutil.SocketConnection(clientSocket)
          self.caddr = clientAddr
          self.daemon = daemon
  
      def __call__(self):
          if self.handleConnection():
              try:
                  while True:
                      try:
                          self.daemon.handleRequest(self.csock)
                      except (socket.error, errors.ConnectionClosedError):
                          # client went away.
                          log.debug("disconnected %s", self.caddr)
                          break
                      except errors.SecurityError:
                          log.debug("security error on client %s", self.caddr)
                          break
                      except errors.TimeoutError as x:
                          # for timeout errors we're not really interested in detailed traceback info
                          log.warning("error during handleRequest: %s" % x)
                          break
                      except:
                          # other errors log a warning, break this loop and close the client connection
                          ex_t, ex_v, ex_tb = sys.exc_info()
                          tb = errors.format_traceback(ex_t, ex_v, ex_tb)
                          msg = "error during handleRequest: %s; %s" % (ex_v, "".join(tb))
                          log.warning(msg)
                          break
              finally:
                  with _client_disconnect_lock:
                      try:
                          self.daemon._clientDisconnect(self.csock)
                      except Exception as x:
                          log.warning("Error in clientDisconnect: " + str(x))
                  self.csock.close()
  
      def handleConnection(self):
          # connection handshake
          try:
              if self.daemon._handshake(self.csock):
                  return True
              self.csock.close()
          except:
              ex_t, ex_v, ex_tb = sys.exc_info()
              tb = errors.format_traceback(ex_t, ex_v, ex_tb)
              log.warning("error during connect/handshake: %s; %s", ex_v, "\n".join(tb))
              self.csock.close()
          return False
  
      def denyConnection(self, reason):
          log.warning("client connection was denied: " + reason)
          # return failed handshake
          self.daemon._handshake(self.csock, denied_reason=reason)
          self.csock.close()
  
  
  class Housekeeper(threading.Thread):
      def __init__(self, daemon):
          super(Housekeeper, self).__init__(name="housekeeper")
          self.pyroDaemon = daemon
          self.stop = threading.Event()
          self.daemon = True
          self.waittime = min(config.POLLTIMEOUT or 0, max(config.COMMTIMEOUT or 0, 5))
  
      def run(self):
          while True:
              if self.stop.wait(self.waittime):
                  break
              self.pyroDaemon._housekeeping()
  
  
  class SocketServer_Threadpool(object):
      """transport server for socket connections, worker thread pool version."""
  
      def __init__(self):
          self.daemon = self.sock = self._socketaddr = self.locationStr = self.pool = None
          self.shutting_down = False
          self.housekeeper = None
  
      def init(self, daemon, host, port, unixsocket=None):
          log.info("starting thread pool socketserver")
          self.daemon = daemon
          self.sock = None
          bind_location = unixsocket if unixsocket else (host, port)
          if config.SSL:
              sslContext = socketutil.get_ssl_context(servercert=config.SSL_SERVERCERT,
                                                      serverkey=config.SSL_SERVERKEY,
                                                      keypassword=config.SSL_SERVERKEYPASSWD,
                                                      cacerts=config.SSL_CACERTS)
              log.info("using SSL,  cert=%s  key=%s  cacerts=%s", config.SSL_SERVERCERT, config.SSL_SERVERKEY, config.SSL_CACERTS)
          else:
              sslContext = None
              log.info("not using SSL")
          self.sock = socketutil.create_socket(bind=bind_location,
                                               reuseaddr=config.SOCK_REUSE,
                                               timeout=config.COMMTIMEOUT,
                                               noinherit=True,
                                               nodelay=config.SOCK_NODELAY,
                                               sslContext=sslContext)
          self._socketaddr = self.sock.getsockname()
          if not unixsocket and self._socketaddr[0].startswith("127."):
              if host is None or host.lower() != "localhost" and not host.startswith("127."):
                  log.warning("weird DNS setup: %s resolves to localhost (127.x.x.x)", host)
          if unixsocket:
              self.locationStr = "./u:" + unixsocket
          else:
              host = host or self._socketaddr[0]
              port = port or self._socketaddr[1]
              if ":" in host:  # ipv6
                  self.locationStr = "[%s]:%d" % (host, port)
              else:
                  self.locationStr = "%s:%d" % (host, port)
          self.pool = Pool()
          self.housekeeper = Housekeeper(daemon)
          self.housekeeper.start()
  
      def __del__(self):
          if self.sock is not None:
              self.sock.close()
              self.sock = None
          if self.pool is not None:
              self.pool.close()
              self.pool = None
          if self.housekeeper:
              self.housekeeper.stop.set()
              self.housekeeper.join()
              self.housekeeper = None
  
      def __repr__(self):
          return "<%s on %s; %d workers>" % (self.__class__.__name__, self.locationStr, self.pool.num_workers())
  
      def loop(self, loopCondition=lambda: True):
          log.debug("threadpool server requestloop")
          while (self.sock is not None) and not self.shutting_down and loopCondition():
              try:
                  self.events([self.sock])
              except (socket.error, OSError) as x:
                  if not loopCondition():
                      # swallow the socket error if loop terminates anyway
                      # this can occur if we are asked to shutdown, socket can be invalid then
                      break
                  # socket errors may not lead to a server abort, so we log it and continue
                  err = getattr(x, "errno", x.args[0])
                  log.warning("socket error '%s' with errno=%d, shouldn't happen", x, err)
                  continue
              except KeyboardInterrupt:
                  log.debug("stopping on break signal")
                  break
  
      def combine_loop(self, server):
          raise TypeError("You can't use the loop combiner on the threadpool server type")
  
      def events(self, eventsockets):
          """used for external event loops: handle events that occur on one of the sockets of this server"""
          # we only react on events on our own server socket.
          # all other (client) sockets are owned by their individual threads.
          assert self.sock in eventsockets
          try:
              csock, caddr = self.sock.accept()
              if self.shutting_down:
                  csock.close()
                  return
              if hasattr(csock, "getpeercert"):
                  log.debug("connected %s - SSL", caddr)
              else:
                  log.debug("connected %s - unencrypted", caddr)
              if config.COMMTIMEOUT:
                  csock.settimeout(config.COMMTIMEOUT)
              job = ClientConnectionJob(csock, caddr, self.daemon)
              try:
                  self.pool.process(job)
              except NoFreeWorkersError:
                  job.denyConnection("no free workers, increase server threadpool size")
          except socket.timeout:
              pass  # just continue the loop on a timeout on accept
  
      def shutdown(self):
          self.shutting_down = True
          self.wakeup()
          time.sleep(0.05)
          self.close()
          self.sock = None
  
      def close(self):
          if self.housekeeper:
              self.housekeeper.stop.set()
              self.housekeeper.join()
              self.housekeeper = None
          if self.sock:
              sockname = None
              try:
                  sockname = self.sock.getsockname()
              except (socket.error, OSError):
                  pass
              try:
                  self.sock.close()
                  if type(sockname) is str:
                      # it was a Unix domain socket, remove it from the filesystem
                      if os.path.exists(sockname):
                          os.remove(sockname)
              except Exception:
                  pass
              self.sock = None
          self.pool.close()
  
      @property
      def sockets(self):
          # the server socket is all we care about, all client sockets are running in their own threads
          return [self.sock]
  
      @property
      def selector(self):
          raise TypeError("threadpool server doesn't have multiplexing selector")
  
      def wakeup(self):
          socketutil.interrupt_socket(self._socketaddr)
  
  
  class PoolError(Exception):
      pass
  
  
  class NoFreeWorkersError(PoolError):
      pass
  
  
  class Worker(threading.Thread):
      def __init__(self, pool):
          super(Worker, self).__init__()
          self.daemon = True
          self.name = "Pyro-Worker-%d" % id(self)
          self.job_available = threading.Event()
          self.job = None
          self.pool = pool
  
      def process(self, job):
          self.job = job
          self.job_available.set()
  
      def run(self):
          while True:
              self.job_available.wait()
              self.job_available.clear()
              if self.job is None:
                  break
              try:
                  self.job()
              except Exception as x:
                  log.exception("unhandled exception from job in worker thread %s: %s", self.name, x)
              self.job = None
              self.pool.notify_done(self)
          self.pool = None
  
  
  class Pool(object):
      """
      A job processing pool that is using a pool of worker threads.
      The amount of worker threads in the pool is configurable and scales between min/max size.
      """
      def __init__(self):
          if config.THREADPOOL_SIZE < 1 or config.THREADPOOL_SIZE_MIN < 1:
              raise ValueError("threadpool sizes must be greater than zero")
          if config.THREADPOOL_SIZE_MIN > config.THREADPOOL_SIZE:
              raise ValueError("minimum threadpool size must be less than or equal to max size")
          self.idle = set()
          self.busy = set()
          self.closed = False
          for _ in range(config.THREADPOOL_SIZE_MIN):
              worker = Worker(self)
              self.idle.add(worker)
              worker.start()
          log.debug("worker pool created with initial size %d", self.num_workers())
          self.count_lock = threading.Lock()
  
      def __enter__(self):
          return self
  
      def __exit__(self, exc_type, exc_val, exc_tb):
          self.close()
  
      def close(self):
          if not self.closed:
              log.debug("closing down")
              for w in list(self.busy):
                  w.process(None)
              for w in list(self.idle):
                  w.process(None)
              self.closed = True
              time.sleep(0.1)
              idle, self.idle = self.idle, set()
              busy, self.busy = self.busy, set()
              # check if the threads that are joined are not the current thread,
              # otherwise Python 2.x crashes with "cannot join current thread".
              current_thread = threading.current_thread()
              while idle:
                  p = idle.pop()
                  if p is not current_thread:
                      p.join(timeout=0.1)
              while busy:
                  p = busy.pop()
                  if p is not current_thread:
                      p.join(timeout=0.1)
  
      def __repr__(self):
          return "<%s.%s at 0x%x; %d busy workers; %d idle workers>" % \
                 (self.__class__.__module__, self.__class__.__name__, id(self), len(self.busy), len(self.idle))
  
      def num_workers(self):
          return len(self.busy) + len(self.idle)
  
      def process(self, job):
          if self.closed:
              raise PoolError("job queue is closed")
          if self.idle:
              worker = self.idle.pop()
          elif self.num_workers() < config.THREADPOOL_SIZE:
              worker = Worker(self)
              worker.start()
          else:
              raise NoFreeWorkersError("no free workers available, increase thread pool size")
          self.busy.add(worker)
          worker.process(job)
          log.debug("worker counts: %d busy, %d idle", len(self.busy), len(self.idle))
  
      def notify_done(self, worker):
          if worker in self.busy:
              self.busy.remove(worker)
          if self.closed:
              worker.process(None)
              return
          if len(self.idle) >= config.THREADPOOL_SIZE_MIN:
              worker.process(None)
          else:
              self.idle.add(worker)
          log.debug("worker counts: %d busy, %d idle", len(self.busy), len(self.idle))
  