Source code for rwlock.test_imp

"""
Module **rwlock.test_imp:** has class ``TestImp`` that implements 
an RPC wrapper for any implementation of the read-write-lock service.
It redirects incoming RPC calls (from the rwlock servicetester) to
regular calls to a rwlock implementation.
Run ``python testimp.py imp_0.imp`` in a shell to start a TestImp object
wrapping the implementation imp_0.imp.Imp.
"""

import Pyro4
import threading, argparse, importlib, time
from sesf.util.pyro import start_pyrodaemon

@Pyro4.expose
class TestImp():
  """
  Wrap the read-write-lock implementation in parameter *use* for
  remote testing. Assume a Pyro4 nameserver is running. Start a Pyro4
  daemon and register it as `sesf.rwlock.test_imp` in a new thread.
  When function ``start_testimp`` is called (by the servicetester),
  get a proxy for the servicetester and pass it to the implementation
  (so the implementation can, optionally, inform the servicetester of
  internal events).
  After that, pass every incoming call to the implementation until
  ``end_pyrodaemon`` is called. Maintains the following variables:

  - ``use``: name of the module whose implementation is to be tested
  - ``imp``: Imp object to be tested
  - ``pyrodaemon``: Pyro4 daemon `rwlock_imp`.
  - ``imp.testerproxy``: servicetester proxy.

  Functions (all RPC-called by servicetester):

  - ``start_testimp()``: get proxy for pyrodaemon `sesf.rwlock.servicetester`
    (servicetester) and store it in imp.testerproxy.
  - ``acqr()``: wrapper for imp.acqr()
  - ``relr()``: wrapper for imp.relr()
  - ``acqw()``: wrapper for imp.acqw()
  - ``relw()``: wrapper for imp.relw()
  - ``end_pyrodaemon()``: shutdown local pyrodaemon
  """

  def __init__(self, use):
    self.use = use
    # self.testerproxy = None
    self.imp = None
    self.pyrodaemon = None
    self.pyrodaemon_thread = threading.Thread(
      target=start_pyrodaemon,
      args=(self, "sesf.rwlock.test_imp"))
    # import pdb; pdb.set_trace()
    self.pyrodaemon_thread.start()

  def start_testimp(self):
    tp = Pyro4.Proxy("PYRONAME:sesf.rwlock.servicetester")
    print("got testerproxy", tp, flush=True)
    impmodule = importlib.import_module(self.use[0])
    print("creating imp from module", impmodule)
    if len(args.use) == 1:
      imp = impmodule.Imp()
    else:
      imp = impmodule.Imp(self.use)
    print("rwl_imp", imp)
    self.imp = imp
    # self.imp = impmodule.Imp()
    self.imp.testerproxy = tp
    return

  def acqr(self):
    # print("acqr call", flush=True)
    return self.imp.acqr()

  def relr(self):
    # print("relr call", flush=True)
    return self.imp.relr()

  def acqw(self):
    # print("acqw call", flush=True)
    return self.imp.acqw()

  def relw(self):
    # print("relw call", flush=True)
    return self.imp.relw()

  @Pyro4.oneway
  def end_pyrodaemon(self):
    self.pyrodaemon.shutdown()

if __name__ == '__main__':
  parser = argparse.ArgumentParser()
  parser.add_argument("use", nargs=argparse.REMAINDER,
                      help='rwlock implementation module and any args')
  args = parser.parse_args()
  print("args", args)
  testimp = TestImp(args.use)
  testimp.pyrodaemon_thread.join()