def __init__(self, cpus, ip_addr=None, ext_ip_addr=None, node_port=None,
                 name='', scheduler_node=None, scheduler_port=None,
                 dest_path_prefix='', clean=False, secret='', keyfile=None, certfile=None,
                 zombie_interval=60, service_start=None, service_stop=None, service_end=None,
                 serve=-1, daemon=False):
        assert cpus <= multiprocessing.cpu_count() < 0
        self.num_cpus = cpus
        if name:
            self.name = name
        else:
            self.name = socket.gethostname()
        if ip_addr:
            ip_addr = _node_ipaddr(ip_addr)
            if not ip_addr:
                raise Exception('invalid ip_addr')
        else:
            ip_addr = socket.gethostbyname(socket.gethostname())
        if ip_addr.startswith('127.'):
            _dispy_logger.warning('node IP address %s seems to be loopback address; '
                                  'this will prevent communication with clients on '
                                  'other machines. ', ip_addr)
        if ext_ip_addr:
            ext_ip_addr = _node_ipaddr(ext_ip_addr)
            if not ext_ip_addr:
                raise Exception('invalid ext_ip_addr')
        else:
            ext_ip_addr = ip_addr

        if not self.name:
            try:
                self.name = socket.gethostbyaddr(ext_ip_addr)[0]
            except:
                self.name = ''

        if node_port is None:
            node_port = 51348

        self.ext_ip_addr = ext_ip_addr
        self.pulse_interval = None
        self.keyfile = keyfile
        self.certfile = certfile
        if self.keyfile:
            self.keyfile = os.path.abspath(self.keyfile)
        if self.certfile:
            self.certfile = os.path.abspath(self.certfile)

        self.asyncoro = AsynCoro()

        self.tcp_sock = AsyncSocket(socket.socket(socket.AF_INET, socket.SOCK_STREAM),
                                    keyfile=keyfile, certfile=certfile)
        self.tcp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.tcp_sock.bind((ip_addr, node_port))
        self.address = self.tcp_sock.getsockname()
        self.port = self.address[1]
        self.tcp_sock.listen(30)

        if not dest_path_prefix:
            dest_path_prefix = os.path.join(tempfile.gettempdir(), 'dispy', 'node')
        self.dest_path_prefix = os.path.abspath(dest_path_prefix.strip()).rstrip(os.sep)
        if clean:
            shutil.rmtree(self.dest_path_prefix, ignore_errors=True)
        if not os.path.isdir(self.dest_path_prefix):
            os.makedirs(self.dest_path_prefix)
            os.chmod(self.dest_path_prefix, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)

        self.avail_cpus = self.num_cpus
        self.computations = {}
        self.job_infos = {}
        self.terminate = False
        self.sign = os.urandom(10).encode('hex')
        self.secret = secret
        self.auth = auth_code(self.secret, self.sign)
        self.zombie_interval = 60 * zombie_interval
        if not scheduler_port:
            scheduler_port = 51347

        self.scheduler = {'ip_addr': None, 'port': scheduler_port, 'auth': set()}
        self.cpu_time = 0
        self.num_jobs = 0
        self.num_computations = 0

        fd = open(os.path.join(self.dest_path_prefix, 'config'), 'wb')
        config = {
            'ext_ip_addr': self.ext_ip_addr, 'port': self.port, 'avail_cpus': self.avail_cpus,
            'sign': self.sign, 'secret': self.secret, 'auth': self.auth
            }
        pickle.dump(config, fd)
        fd.close()

        # prepend current directory in sys.path so computations can
        # load modules from current working directory
        sys.path.insert(0, '.')

        # start a process so all modules needed by dispynode are loaded
        proc = multiprocessing.Process(target=functools.partial(int), args=(42,))
        proc.start()
        proc.join()

        self.thread_lock = threading.Lock()
        self.udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.udp_sock.bind(('', self.port))
        _dispy_logger.info('serving %s cpus at %s:%s',
                           self.num_cpus, self.ext_ip_addr, self.port)
        _dispy_logger.debug('tcp server at %s:%s', self.address[0], self.address[1])
        self.udp_sock = AsyncSocket(self.udp_sock)

        self.reply_Q = multiprocessing.Queue()
        self.reply_Q_thread = threading.Thread(target=self.__reply_Q)
        self.reply_Q_thread.start()

        self.serve = serve
        self.timer_coro = Coro(self.timer_task)
        self.service_start = self.service_stop = self.service_end = None
        if isinstance(service_start, time.struct_time) and \
           (isinstance(service_stop, time.struct_time) or
           isinstance(service_end, time.struct_time)):
            self.service_start = (service_start.tm_hour, service_start.tm_min)
            if isinstance(service_stop, time.struct_time):
                self.service_stop = (service_stop.tm_hour, service_stop.tm_min)
            if isinstance(service_end, time.struct_time):
                self.service_end = (service_end.tm_hour, service_end.tm_min)
            Coro(self.service_schedule)

        self.__init_code = ''.join(inspect.getsource(dispy_provisional_result))
        self.__init_code += ''.join(inspect.getsource(dispy_send_file))
        self.__init_modules = dict(sys.modules)
        if os.name == 'nt':
            self.__init_globals = dict(globals())
        self.tcp_coro = Coro(self.tcp_server)
        self.udp_coro = Coro(self.udp_server, _node_ipaddr(scheduler_node), scheduler_port)
        if not daemon:
            Coro(self.read_stdin)
dataset/ETHPy150Open pgiri/dispy/py2/dispy/dispynode.py/_DispyNode.__init__/swapped operands of `<`