# Conrad Meyer <cemeyer@uw.edu>
# 0824410
# CSE 550 Problem Set 3
# Thu Nov 10 2011

import threading
import time
import zmq

class Server(threading.Thread):
  def __init__(self, zctx, endpoint, obj):
    threading.Thread.__init__(self)
    self.methods = {}
    for x in dir(obj):
      if not x.startswith("_"):
        self.methods[x] = getattr(obj, x)
    self.socket = zctx.socket(zmq.REP)
    self.socket.bind(endpoint)
    self.daemon = True
    self.start()

  def run(self):
    while True:
      req = self.socket.recv_pyobj()
      rep = None

      try:
        rep = self.methods[req[0]](*req[1])

      except Exception as e:
        print(e)
        pass

      self.socket.send_pyobj(rep)

class Client:
  def __init__(self, zctx, endpoint):
    self._socket = zctx.socket(zmq.REQ)
    self._socket.connect(endpoint)
    self._needs_recv = False

  def _unblock(self):
    try:
      self._socket.recv(zmq.NOBLOCK)
    except zmq.ZMQError as e:
      if e.errno == zmq.EAGAIN:
        return False
      raise
    return True

  def send_pyobj(self, obj):
    if self._needs_recv:
      if self._unblock():
        self._needs_recv = False

    if not self._needs_recv:
      self._socket.send_pyobj(obj)
      self._needs_recv = True

  def recv_pyobj(self):
    o = self._socket.recv_pyobj()
    self._needs_recv = False
    return o

  class _Call:
    def __init__(self, name, client):
      self.name = name
      self.client = client

    def __call__(self, *args):
      self.client.send_pyobj((self.name, args))
      return self.client.recv_pyobj()

  def __getattr__(self, name):
    return Client._Call(name, self)

  class _FutureCall:
    def __init__(self, name, client, args):
      self.client = client
      self.client.send_pyobj((name, args))

    def socket(self):
      return self.client._socket

    def force(self):
      return self.client.recv_pyobj()

  def async_call(self, name, args):
    return Client._FutureCall(name, self, args)

# Performs a given RPC against a set of peers; returns when a quorum of servers has
# replied, after the timeout interval (in seconds) has passed, whichever happens
# first.
#
# On success, returns (True, values), where values is a list of at least quorum
#   values.
# On timeout, returns (False, values).
def async_multicall(peers, timeout, name, args):
  futures = {}
  for peer in peers:
    future = peer.async_call(name, args)
    futures[future.socket()] = future

  starttime = time.time()
  results = []

  poller = zmq.Poller()
  for future_sock in futures.keys():
    poller.register(future_sock, zmq.POLLIN)

  # while len(results) < ((len(peers) // 2) + 1):  # quit on majority responded
  while len(results) < len(peers):  # quit when everyone responds
    elapsed = time.time() - starttime
    if elapsed >= timeout: break

    for (sock, status) in poller.poll(1000*(timeout - elapsed)):
      if status == zmq.POLLIN:
        poller.unregister(sock)
        results.append(futures[sock].force())

  return (len(results) >= ((len(peers) // 2) + 1), results)

@@ -119,16 +119,140 @@
 rt zmq%0A%0A
+# Turns an arbitrary python object into an RPC server by introspecting and%0A# ignoring private (i.e., %22_%22-prefixed) methods.%0A
 class Se
@@ -617,16 +617,37 @@
 e True:%0A
+      # Deserialize,%0A
       re
@@ -700,24 +700,40 @@
 %0A      try:%0A
+        # Call,%0A
         rep 
@@ -835,193 +835,795 @@
     
-self.socket.send_pyobj(rep)%0A%0Aclass Client:%0A  def __init__(self, zctx, endpoint):%0A    self._socket = zctx.socket(zmq.REQ)%0A    self._socket.connect(endpoint)%0A    self._needs_recv = False%0A
+# And serialize the response.%0A      self.socket.send_pyobj(rep)%0A%0A%0A# Clients are fairly dumb; they'll just serialize calls into them and send%0A# the bits on to the server end of things (de-serializing responses, of%0A# course).%0Aclass Client:%0A  def __init__(self, zctx, endpoint):%0A    self._socket = zctx.socket(zmq.REQ)%0A    self._socket.connect(endpoint)%0A    self._needs_recv = False%0A%0A  # We attempt to cleverly handle asynchronously available servers by%0A  # dropping backed-up requests to them, and occasionally checking to see%0A  # if they've come back. Obviously this doesn't work if the server totally%0A  # crashes because it'll never reply to a request it hasn't gotten; the%0A  # solution to this is to use PUB/SUB and message passing instead of REQ/REP.%0A  # 20-20 hindsight, motherfucker.
 %0A  d
@@ -1795,24 +1795,116 @@
 eturn True%0A%0A
+  # Wrap send/recv in order to do the clever things we talked about ten lines%0A  # previous.%0A
   def send_p
@@ -2201,16 +2201,34 @@
 turn o%0A%0A
+  # Syntax sugar.%0A
   class 
@@ -2509,16 +2509,133 @@
  self)%0A%0A
+  # A basic Future so that we can send out many RPCs simultaneously and check%0A  # for responses modulo some timeout.%0A
   class 
@@ -3136,16 +3136,182 @@
 # first.
+ (Actually, returns when *all* servers has replied, or the timeout has%0A# occurred. Waiting for quorum just seems stupid esp. with our RPC REQ/REP socket%0A# interface.)
 %0A#%0A# On 

comment-ify zmqrpc