Source code for util.service

def myprint(*args):
  """Print with flush on."""
  print(*args, flush=True)


def do_service_input(j, check_cond, do_update, args, cond):
  """Service input-part factory, used in service model and service tester.
  The arguments are:

  - *j* (int): node id for the input step.
  - *check_cond* (fn): CCI or RCI function from service.
  - *do_update* (fn): CU or RU function from service.
  - *args*: tuple of arguments supplied by implementation; (None,) if none.
  - *cond* (Condition): condition used in service model or service tester.
  """
  if args == (None,):
    args = ()
  with cond:
    rc_rval = check_cond(j, *args)
    # DEBUG: myprint("do_service_input: rc_rval", rc_rval, "*args", *args)
    if not rc_rval:
      raise Exception("input condition violated: args", args)
    do_update(j, *args)
    cond.notify_all()


def do_service_output(j, get_args, do_update, cond):
  """Service output-part factory, used in service model and service tester.
  The arguments are:

  - *j* (int): node id for the input step.
  - *get_args* (fn): CCO or RCO function from service.
  - *do_update* (fn): CU or RU function from service.
  - *cond* (Condition): condition used in service tester or service model.
  """
  with cond:
    while True:
      args = get_args(j)
      # (0,): cond not satisfiable
      # (1,x): cond satisfiable with args list x (may be empty)
      if args == (0,):
        cond.wait()
      else:
        break
    # DEBUG: myprint("do_service_output: args", args)
    # args == (1,x)
    args = args[1:]
    # args == (x)
    do_update(j, *args)
    cond.notify_all()
    # *args will be used in call|return to implementation
    return args