Методы очереди для запуска на объекте различными потоками в Python

Question

Допустим, у меня есть объект, определение класса которого выглядит следующим образом:

class Command:
  foo = 5
  def run(self, bar):
    time.sleep(1)
    self.foo = bar
    return self.foo

Если этот класс создается один раз, но разные потоки обращаются к его методу run (через HTTP-запрос, обрабатываемый отдельно), передавая разные аргументы, каков наилучший способ поставить их в очередь? Можно ли это сделать в самом определении класса?

Я использую сервер RPC XML (отдельный класс). Для простоты мы можем сказать, что у него есть один экземпляр класса Command, созданный как переменная класса. Когда Command.run () обрабатывается двумя отдельными потоками, как я могу убедиться, что один метод run () завершен до запуска следующего?

Я мог бы сделать что-то вроде:

  while self.busy:
    time.sleep(1)
    self.busy = true
    ...
    self.busy = false
    return self.foo

но это, насколько я знаю, не отдает приоритет самому старому запросу.

Я понимаю, как звучит все это упражнение, поскольку я мог просто синхронно запускать сервер XML-RPC. Но, если коротко, есть несколько экземпляров Command, и я не хочу блокировать запросы для одного, потому что другой занят.

Надеюсь, это имеет больше смысла.

Спасибо.

Answer 1

Вот относительно простой подход (игнорирует исключения, доступ к атрибутам, специальные методы и т. Д.):

import Queue
import threading

def serialize(q):
  """runs a serializer on queue q: put [-1]*4 on q to terminate."""
  while True:
    # get output-queue for result, a callable, its args and kwds
    out_q, tocall, args, kwds = q.get()
    if out_q == -1:
      return
    result = tocall(*args, **kwds)
    out_q.put(result)

class WrapCall(object):
  """Wraps a callable to serialize calls to it."""

  def __init__(self, inq, ouq, tocall):
    self.inq = inq
    self.ouq = ouq
    self.tocall = tocall

  def __call__(self, *a, **k):
    self.inq.put((self.ouq, self.tocall, a, k))
    if self.ouq is None:
      return None
    return self.ouq.get()

class WrapObj(object):
  """Wraps any object to serialize all calls to its methods."""

  def __init__(self, obj):
    self._o = obj
    self._q = Queue.Queue()
    t = threading.Thread(target=serialize, args=(self._q,))
    t.setDaemon(True)
    t.start()
    self.t = t

  def __getattr__(self, n):
    """Wraps methods of self.w into an appropriate WrapCall instance."""
    towrap = getattr(self._o, n)
    if not callable(towrap):
      raise TypeError('Cannot wrap noncallable attribute %r (type: %s)'
                       % (n, type(towrap)))
    q = Queue.Queue()
    return WrapCall(self._q, q, towrap)

  def WrapperWait(self):
    """Return only when self.t has served all pending requests."""
    q = Queue.Queue()
    w = WrapCall(self.__q, q, lambda: None)
    return w()

С этим «сериализатором» вы можете сделать

myobj = WrapObj(Command())

и теперь все вызовы (не специальные) методов myobj сериализуются потокобезопасным способом.

В вашем конкретном случае, когда есть только один метод объекта, вы могли бы еще немного упростить это, но это уже упрощенная версия того, что я писал и часто использую (поддержка получения и установки атрибутов, специальные методы, и т. д., слишком сложный, чтобы того стоить, полная версия поддерживает перехват и повторный вызов исключений, возникающих в методах объекта-оболочки, оптимизацию вызовов, результаты или исключения которых вас не интересуют, и несколько дополнительных настроек, не сериализация атрибутов и специальные методы).

Answer 2

Это зависит от того, как вы планируете потреблять foo. Самое простое - использовать модуль Python queue для синхронизации доставки значений потребителю, но это предполагает, что потребитель хочет получить каждое значение. Возможно, вам придется быть более конкретным, чтобы получить лучший ответ.