def call(self, proc_name, args=[], kwargs={}, result='async', timeout=None):  #{
        """
        Call the remote method with *args and **kwargs.

        Parameters
        ----------
        proc_name : <str>   name of the remote procedure to call
        args      : <tuple> positional arguments of the procedure
        kwargs    : <dict>  keyword arguments of the procedure
        ignore    : <bool>  whether to ignore result or wait for it
        timeout   : <float> | None
            Number of seconds to wait for a reply.
            RPCTimeoutError is set as the future result in case of timeout.
            Set to None, 0 or a negative number to disable.

        Returns
        -------
        <Future> if result is 'async'
        None     if result is 'ignore'

        If remote call fails:
        - sets <RemoteRPCError> into the <Future> if result is 'async'
        """
        assert result in ('async', 'ignore'), \
            'expected any of "async", "ignore" -- got %r' % result

        if not (timeout is None or isinstance(timeout, (int, float))):
            raise TypeError("timeout param: <float> or None expected, got %r" % timeout)

        ignore = result == 'ignore'
        req_id, msg_list = self._build_request(proc_name, args, kwargs, ignore)
        self.socket.send_multipart(msg_list)

        if ignore:
            return None

        # The following logic assumes that the reply won't come back too
        # quickly, otherwise the callbacks won't be in place in time. It should
        # be fine as this code should run very fast. This approach improves
        # latency we send the request ASAP.
        def _abort_request():
            future_tout = self._futures.pop(req_id, None)
            if future_tout:
                future, _ = future_tout
                tout_msg  = "Request %s timed out after %s sec" % (req_id, timeout)
                #self.logger.debug(tout_msg)
                future.set_exception(RPCTimeoutError(tout_msg))

        timeout = timeout or 0

        if timeout > 0:
            tout_cb = DelayedCallback(_abort_request, int(timeout*1000), self.ioloop)
            tout_cb.start()
        else:
            tout_cb = None

        future = Future()
        self._futures[req_id] = (future, tout_cb)

        return future

def send_checkup():

    def checkup_timeout():
        global srv, responding, timeout

        timeout.stop()
        if not responding:
            # we've timed out, restart
            # TODO: provide config var for how many times to attempt start before exiting
            print('{0} not responding, attempting start.'.format(MODULE))
            srv = launch_service()

    def recv_checkup(msg):
        global responding
        responding = True

    # access globals
    global timeout, checkup, responding, loop

    # listen for ping back
    checkup.on_recv(recv_checkup)

    # do what's needed to rescue on timeout
    responding = False
    timeout = DelayedCallback(checkup_timeout, CHECKUP_TIMEOUT, io_loop=loop)
    timeout.start()

    # send ping
    checkup.send('You alive?')

def send_checkup():

    def checkup_timeout():
        global service, responding, timeout

        timeout.stop()
        if not responding:
            # we've timed out, restart
            print('Mongrel2 not responding, attempting restart.')
            
            # since control port isn't responding, do a dirty kill just in case
            kill_mongrel_with_pid(M2_PID_PATH)
            # start her back up
            # TODO: add configurable delay here before starting again
            start_mongrel()

    def recv_response(msg):
        global responding
        responding = True

    # access globals
    global timeout, control_port, responding

    # listen for ping back
    control_port.on_recv(recv_response)

    # do what's needed to rescue on timeout
    responding = False
    timeout = DelayedCallback(checkup_timeout, CHECKUP_TIMEOUT, io_loop=loop)
    timeout.start()

    # send status request
    control_port.send(tnetstrings.dump(['status', {'what': 'net'}]))

def stop(signum, frame):
    global loop, checkup_periodic, control_port, stop_timeout

    def stop_timeout():
        print('Terminate request timed out, mongrel2 might be orphaned.')
        kill_mongrel_with_pid(M2_PID_PATH)
        shutdown()

    def terminate_resp(msg):
        print('Mongrel2 control port confirmed SIGTERM sent.')
        shutdown()

    def shutdown():
        print('Shutting down.')
        remove_hosts(HOSTS)
        control_port.close()
        loop.stop()

    print('\nStopping.')

    # make sure checkup doesn't happen during termination
    checkup_periodic.stop()

    # register terminate response callback
    control_port.on_recv(terminate_resp)

    # prepare timeout
    stop_timeout = DelayedCallback(stop_timeout, STOP_TIMEOUT, io_loop=loop)
    stop_timeout.start()

    # send terminate request
    control_port.send(tnetstrings.dump(['terminate', {}]))

    def call(self, method, callback, errback, timeout, *args, **kwargs):
        """Call the remote method with *args and **kwargs.

        Parameters
        ----------
        method : str
            The name of the remote method to call.
        callback : callable
            The callable to call upon success or None. The result of the RPC
            call is passed as the single argument to the callback:
            `callback(result)`.
        errback : callable
            The callable to call upon a remote exception or None, The
            signature of this method is `errback(ename, evalue, tb)` where
            the arguments are passed as strings.
        timeout : int
            The number of milliseconds to wait before aborting the request.
            When a request is aborted, the errback will be called with an
            RPCTimeoutError. Set to 0 or a negative number to use an infinite
            timeout.
        args : tuple
            The tuple of arguments to pass as `*args` to the RPC method.
        kwargs : dict
            The dict of arguments to pass as `**kwargs` to the RPC method.
        """
        if not isinstance(timeout, int):
            raise TypeError("int expected, got %r" % timeout)
        if not (callback is None or callable(callback)):
            raise TypeError("callable or None expected, got %r" % callback)
        if not (errback is None or callable(errback)):
            raise TypeError("callable or None expected, got %r" % errback)

        msg_id, msg_list = self._build_request(method, args, kwargs)
        self.stream.send_multipart(msg_list)

        # The following logic assumes that the reply won't come back too
        # quickly, otherwise the callbacks won't be in place in time. It should
        # be fine as this code should run very fast. This approach improves
        # latency we send the request ASAP.
        def _abort_request():
            cb_eb_dc = self._callbacks.pop(msg_id, None)
            if cb_eb_dc is not None:
                eb = cb_eb_dc[1]
                if eb is not None:
                    try:
                        raise RPCTimeoutError()
                    except:
                        etype, evalue, tb = sys.exc_info()
                        eb(etype.__name__, evalue, traceback.format_exc(tb))
        if timeout > 0:
            dc = DelayedCallback(_abort_request, timeout, self.loop)
            dc.start()
        else:
            dc = None

        self._callbacks[msg_id] = (callback, errback, dc)

def restart_service():
    global srv

    def restart_checkup():
        global checkup_periodic
        checkup_periodic.start()

    global loop
    checkup_restart = DelayedCallback(restart_checkup, 
                                      PAUSE_AFTER_RESTART, 
                                      io_loop=loop)
    service = launch_service()
    checkup_restart.start()

    def __init__(self, broker, service, io_loop=None):
        """Create and setup an MDP worker.
           @param broker A string containing the broker's URL
           @param service A string containing the service name
           @param io_loop An existing I/O loop object. If None, the default will be used.
        """
        self.service=service
        self._broker = broker

        self.ctx = zmq.Context()
        sock = self.ctx.socket(zmq.DEALER)
        ZMQStream.__init__(self, sock, io_loop)
        # last watchdog timer tick
        self.watchdog = 0
        # connection callback one-shot
        self._conncb = DelayedCallback(self.send_ready, 3000, self.io_loop)
        # heartbeat callback..runs continuous when connected
        self._hbcb = PeriodicCallback(self.send_heartbeat, 2000, self.io_loop)
        # number of connection attempts
        self._conn_attempt = 0
        # waiting to connect state
        self._waiting_to_connect = True
        # have we been disconnected? (flags a reconnect attempt)
        self.disconnected = False

        # connect the socket and send a READY when the io_loop starts
        self.connect(self._broker)
        self._conncb.start()

    def _start_timeout(self, timeout):
        """Helper for starting the timeout.

        :param timeout:  the time to wait in milliseconds.
        :type timeout:   int
        """
        self._tmo = DelayedCallback(self._on_timeout, timeout)
        self._tmo.start()
        return

def check_for_change():

    global checkup, loop, checksums, responding, checkup_periodic, KEY

    curr_sums = check(path, watch)
    changed, deleted = get_diff(curr_sums, checksums)
    
    if len(changed) or len(deleted):
        checksums = curr_sums
        print('restarting {0}.'.format(MODULE))
        checkup_periodic.stop()
        command.send(json.dumps({
            'key': KEY,
            'command': 'die'
        }))
        delay = DelayedCallback(restart_service, 
                                PAUSE_BEFORE_RESTART + 300, 
                                io_loop=loop)
        delay.start()

    def __init__(self, context, device_id, address):

        self.context = context
        self.address = address
        self.device_id = device_id

        self.socket = None

        # delay = seconds_till_next('hour', duration=1) + 10  # 10 second buffer
        delay = 120  # 2 minutes after startup
        self.updater = DelayedCallback(self.update, delay * 1000)
        self.updater.start()
        return

 def _tick(self):
     """Method called every HB_INTERVAL milliseconds.
     """
     self.curr_liveness -= 1
     self.send_hb()
     if self.curr_liveness >= 0:
         return
     # ouch, connection seems to be dead
     self.on_log_event('broker.timeout', 'Connection to broker timeouted, disconnecting')
     self.shutdown(False)
     # try to recreate it
     self._delayed_cb = DelayedCallback(self._create_stream, 5000)
     self._delayed_cb.start()
     return

    def send_request(self, request, args, kwargs, handler, timeout):
        """Send a request to the service."""
        req = {}
        req['method'] = request.method
        req['uri'] = request.uri
        req['version'] = request.version
        req['headers'] = dict(request.headers)
        body = request.body
        req['remote_ip'] = request.remote_ip
        req['protocol'] = request.protocol
        req['host'] = request.host
        req['files'] = request.files
        req['arguments'] = request.arguments
        req['args'] = args
        req['kwargs'] = kwargs

        msg_id = bytes(uuid.uuid4())
        msg_list = [b'|', msg_id, jsonapi.dumps(req)]
        if body:
            msg_list.append(body)
        logging.debug('Sending request: %r', msg_list)
        self.stream.send_multipart(msg_list)

        if timeout > 0:
            def _handle_timeout():
                handler.send_error(504) # Gateway timeout
                try:
                    self._callbacks.pop(msg_id)
                except KeyError:
                    logging.error('Unexpected error removing callbacks')
            dc = DelayedCallback(_handle_timeout, timeout, self.loop)
            dc.start()
        else:
            dc = None
        self._callbacks[msg_id] = (handler, dc)
        return msg_id

    def _tick(self):
        """Method called every HB_INTERVAL milliseconds.
        """
        self.curr_liveness -= 1
##         print '%.3f tick - %d' % (time.time(), self.curr_liveness)
        self.send_hb()
        if self.curr_liveness >= 0:
            return
        print '%.3f lost connection' % time.time()
        # ouch, connection seems to be dead
        self.shutdown()
        # try to recreate it
        self._delayed_cb = DelayedCallback(self._create_stream, 5000)
        self._delayed_cb.start()
        return

 def _tick(self):
     """Method called every HB_INTERVAL milliseconds.
     """
     if DEBUG:
         print("MQREP > _tick")
     self.curr_liveness -= 1
     if DEBUG:
         print('MQREP > _tick - {0} tick = {1}'.format(time.time(), self.curr_liveness))
     self.send_hb()
     if self.curr_liveness >= 0:
         return
     if DEBUG:
         print('MQREP > _tick - {0} lost connection'.format(time.time()))
     # ouch, connection seems to be dead
     self.shutdown()
     # try to recreate it
     self._delayed_cb = DelayedCallback(self._create_stream, self.HB_INTERVAL)
     self._delayed_cb.start()
     return

 def test_01_simple_01(self):
     """Test MDPWorker simple req/reply.
     """
     self._start_broker()
     time.sleep(0.2)
     worker = MyWorker(self.context, self.endpoint, self.service)
     sender = DelayedCallback(self.send_req, 500)
     stopper = DelayedCallback(self.stop_test, 2500)
     sender.start()
     stopper.start()
     IOLoop.instance().start()
     worker.shutdown()
     self._stop_broker()
     return

    def _restart(self):

        self.shutdown()

        self.socket = self.context.socket(zmq.REQ)
        self.socket.setsockopt(zmq.LINGER, 0)
        self.socket.connect(self.address)

        self.can_send = True
        self.connection_attempts = 0
        self.device_state = 'unknown'
        self.ticker = PeriodicCallback(self.heartbeat,
                                       DeviceConnection.HB_INTERVAL)
        self.ticker.start()

        # delay = seconds_till_next('hour', duration=1) + 2  # 2 second buffer
        delay = 60  # send update one minute after startup
        self.updater = DelayedCallback(self.update, delay * 1000)
        self.updater.start()
        return

    def update(self):

        self.updater.stop()

        if self.device_state == 'connected':
            logger.info("Sending update for device")
            update = DeviceUpdate(self.device_id, 'farm_monitor')
            update.create()
            reply = self.send([pickle.dumps(update)])

            if reply:
                message = pickle.loads(reply[1])
                logger.debug("Update response from {0} :{1}".format(message.source, message.reply))

        # delay = seconds_till_next('hour', duration=1) + 2  # 2 second buffer
        delay = 60 * 60  # send next update in 60 minutes
        self.updater = DelayedCallback(self.update, delay * 1000)
        self.updater.start()

        return

    def update(self):
        self.updater.stop()

        if internal_dealer.check_connection():
            logger.info("Sending update for all grainbins")
            socket = self.context.socket(zmq.REQ)
            socket.setsockopt(zmq.LINGER, 0)
            socket.connect(self.address)

            update = GrainbinUpdate(self.device_id, 'farm_monitor')
            update.create()
            message = [pickle.dumps(update)]

            reply = mdp_request(socket, 'grainbin', message, 5)
            if reply:
                message = pickle.loads(reply[1])
                logger.debug("reply from farm monitor for update: {0}".format(message.reply))

        # delay = seconds_till_next('hour', duration=1) + 10  # 10 second buffer
        delay = 60 * 60  # send next update in one hour
        self.updater = DelayedCallback(self.update, delay * 1000)
        self.updater.start()
        return

class MqAsyncReq(object):

    """Class for the MDP client side.

    Thin asynchronous encapsulation of a zmq.REQ socket.
    Provides a :func:`request` method with optional timeout.

    Objects of this class are ment to be integrated into the
    asynchronous IOLoop of pyzmq.

    :param context:  the ZeroMQ context to create the socket in.
    :type context:   zmq.Context
    :param endpoint: the enpoint to connect to.
    :type endpoint:  str
    :param service:  the service the client should use
    :type service:   str
    """

    _proto_version = b'MDPC01'

    def __init__(self, context, service):
        """Initialize the MDPClient.
        """
        if ("domogik.common.configloader" in sys.modules):
            cfg = Loader('mq').load()
            confi = dict(cfg[1])
            self.endpoint = "tcp://{0}:{1}".format(config['ip'], config['req_rep_port'])
        else:
            ip = Parameter.objects.get(key='mq-ip')
            port = Parameter.objects.get(key='mq-req_rep_port')
            self.endpoint = "tcp://{0}:{1}".format(ip.value, port.value)
        socket = ZmqSocket(context, zmq.REQ)
        ioloop = IOLoop.instance()
        self.service = service
        self.stream = ZMQStream(socket, ioloop)
        self.stream.on_recv(self._on_message)
        self.can_send = True
        self._proto_prefix = [ PROTO_VERSION, service]
        self._tmo = None
        self.timed_out = False
        socket.connect(self.endpoint)
        return

    def shutdown(self):
        """Method to deactivate the client connection completely.

        Will delete the stream and the underlying socket.

        .. warning:: The instance MUST not be used after :func:`shutdown` has been called.

        :rtype: None
        """
        if not self.stream:
            return
        self.stream.socket.setsockopt(zmq.LINGER, 0)
        self.stream.socket.close()
        self.stream.close()
        self.stream = None
        return

    def request(self, msg, timeout=None):
        """Send the given message.

        :param msg:     message parts to send.
        :type msg:      list of str
        :param timeout: time to wait in milliseconds.
        :type timeout:  int
        
        :rtype None:
        """
        if not self.can_send:
            raise InvalidStateError()
        if type(msg) in (bytes, str):
            msg = [msg]
        # prepare full message
        to_send = self._proto_prefix[:]
        to_send.extend(msg)
        self.stream.send_multipart(to_send)
        self.can_send = False
        if timeout:
            self._start_timeout(timeout)
        return

    def _on_timeout(self):
        """Helper called after timeout.
        """
        self.timed_out = True
        self._tmo = None
        self.on_timeout()
        return

    def _start_timeout(self, timeout):
        """Helper for starting the timeout.

        :param timeout:  the time to wait in milliseconds.
        :type timeout:   int
        """
        self._tmo = DelayedCallback(self._on_timeout, timeout)
        self._tmo.start()
        return
#.........这里部分代码省略.........

class Worker(object):

    """Class for the MDP worker side.

    Thin encapsulation of a zmq.DEALER socket.
    Provides a send method with optional timeout parameter.

    Will use a timeout to indicate a broker failure.
    """
    max_forks = 10

    ipc = 'ipc:///tmp/zmq-rpc-'+str(uuid4())
    HB_INTERVAL = 1000  # in milliseconds
    HB_LIVENESS = 3    # HBs to miss before connection counts as dead

    def __init__(self, context, endpoint, service, multicasts=()):
        """Initialize the MDPWorker.

        :param context:    is the zmq context to create the socket from
        :type context:     zmq.Context
        :param service:    service name - you can put hostname here
        :type service:     str
        :param multicasts: list of groups to subscribe
        :type multicasts:  list
        """
        self.context = context
        self.endpoint = endpoint
        self.service = service.encode('utf-8')  # convert to byte-string - required in python 3
        self.multicasts = [m.encode('utf-8') for m in multicasts]  # convert to byte-string
        self.stream = None
        self._tmo = None
        self.need_handshake = True
        self.ticker = None
        self._delayed_cb = None
        self._create_stream()
        self.forks = []
        self.curr_liveness = self.HB_LIVENESS

        socket = self.context.socket(zmq.ROUTER)
        socket.bind(self.ipc)
        self.stream_w = ZMQStream(socket)
        self.stream_w.on_recv(self._on_fork_response)
        self.reply_socket = None
        return

    def _create_stream(self):
        """Helper to create the socket and the stream.
        """
        self.on_log_event('broker.connect', 'Trying to connect do broker')
        socket = self.context.socket(zmq.DEALER)
        ioloop = IOLoop.instance()
        self.stream = ZMQStream(socket, ioloop)
        self.stream.on_recv(self._on_message)
        self.stream.socket.setsockopt(zmq.LINGER, 0)
        self.stream.connect(self.endpoint)
        self.ticker = PeriodicCallback(self._tick, self.HB_INTERVAL)
        self._send_ready()
        for m in self.multicasts:
            self._register_worker_to_multicast(m)
        self.ticker.start()
        return

    def _tick(self):
        """Method called every HB_INTERVAL milliseconds.
        """
        self.curr_liveness -= 1
        self.send_hb()
        if self.curr_liveness >= 0:
            return
        # ouch, connection seems to be dead
        self.on_log_event('broker.timeout', 'Connection to broker timeouted, disconnecting')
        self.shutdown(False)
        # try to recreate it
        self._delayed_cb = DelayedCallback(self._create_stream, 5000)
        self._delayed_cb.start()
        return

    def send_hb(self):
        """Construct and send HB message to broker.
        """
        msg = [b'', MDP_WORKER_VERSION, b'\x05']
        self.stream.send_multipart(msg)
        return

    def shutdown(self, final=True):
        """Method to deactivate the worker connection completely.

        Will delete the stream and the underlying socket.

        :param final: if shutdown is final and we want to close all sockets
        :type final:  bool
        """

        if self.ticker:
            self.ticker.stop()
            self.ticker = None
        if not self.stream:
            return

        self.stream.on_recv(None)
#.........这里部分代码省略.........