def _discoro_proc():
    # coroutine
    """Server process receives computations and runs coroutines for it.
    """

    import os
    import shutil
    import traceback
    import sys
    import time

    try:
        import psutil
    except:
        psutil = None

    import asyncoro.disasyncoro as asyncoro
    from asyncoro import Coro
    from asyncoro.discoro import MinPulseInterval, MaxPulseInterval, \
        DiscoroNodeInfo, DiscoroNodeAvailInfo

    class Nonlocals(object):
        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)

    _discoro_coro = asyncoro.AsynCoro.cur_coro()
    _discoro_config = yield _discoro_coro.receive()
    assert _discoro_config['req'] == 'config'
    _discoro_timer_coro = _discoro_config['timer_coro']
    yield asyncoro.AsynCoro.instance().peer(_discoro_timer_coro.location)

    if _discoro_config['min_pulse_interval'] > 0:
        MinPulseInterval = _discoro_config['min_pulse_interval']
    if _discoro_config['max_pulse_interval'] > 0:
        MaxPulseInterval = _discoro_config['max_pulse_interval']

    _discoro_coro.register('discoro_server')
    _discoro_name = asyncoro.AsynCoro.instance().name
    asyncoro.AsynCoro.instance().dest_path = os.path.join('discoro', _discoro_name)
    _discoro_dest_path = asyncoro.AsynCoro.instance().dest_path
    _discoro_pid_path = os.path.join(_discoro_dest_path, '..', '%s.pid' % _discoro_name)
    _discoro_pid_path = os.path.normpath(_discoro_pid_path)
    # TODO: is file locking necessary?
    if os.path.exists(_discoro_pid_path):
        with open(_discoro_pid_path, 'r') as _discoro_req:
            _discoro_var = _discoro_req.read()
        _discoro_var = int(_discoro_var)
        if not _discoro_config['phoenix']:
            print('\n   Another discoronode seems to be running;\n'
                  '   make sure server with PID %d quit and remove "%s"\n' %
                  (_discoro_var, _discoro_pid_path))
            _discoro_var = os.getpid()

        import signal
        try:
            os.kill(_discoro_var, signal.SIGTERM)
        except:
            pass
        else:
            time.sleep(0.1)
            try:
                if os.waitpid(_discoro_var, os.WNOHANG)[0] != _discoro_var:
                    asyncoro.logger.warning('Killing process %d failed', _discoro_var)
            except:
                pass
        del signal
    if os.path.isdir(_discoro_dest_path):
        shutil.rmtree(_discoro_dest_path)
    os.makedirs(_discoro_dest_path)
    os.chdir(_discoro_dest_path)
    sys.path.insert(0, _discoro_dest_path)
    with open(_discoro_pid_path, 'w') as _discoro_var:
        _discoro_var.write('%s' % os.getpid())

    def _discoro_peer(peer, coro=None):
        yield asyncoro.AsynCoro.instance().peer(peer)

    for _discoro_var in _discoro_config.pop('peers', []):
        Coro(_discoro_peer, _discoro_var)
    del _discoro_peer

    for _discoro_var in ['req', 'phoenix', 'min_pulse_interval', 'max_pulse_interval']:
        del _discoro_config[_discoro_var]

    asyncoro.logger.info('discoro server "%s" started at %s; '
                         'computation files will be saved in "%s"',
                         _discoro_name, _discoro_coro.location, _discoro_dest_path)
    _discoro_req = _discoro_client = _discoro_auth = _discoro_msg = None
    _discoro_pulse_coro = _discoro_peer_status = _discoro_zombie_timeout = None
    _discoro_monitor_coro = _discoro_monitor_proc = _discoro_schedule_coro = None
    _discoro_computation = _discoro_func = _discoro_var = None
    _discoro_job_coros = set()
    _discoro_nonlocals = Nonlocals(busy_time=time.time())
    _discoro_globals = {}
    _discoro_locals = {}
    _discoro_modules = dict(sys.modules)
    _discoro_globals.update(globals())
    _discoro_locals.update(locals())

    def _discoro_peer_status(coro=None):
        coro.set_daemon()
        while 1:
            status = yield coro.receive()
            if isinstance(status, asyncoro.PeerStatus) and \
               status.status == asyncoro.PeerStatus.Offline and \
               _discoro_pulse_coro and _discoro_pulse_coro.location == status.location:
                asyncoro.logger.debug('scheduler at %s quit; closing computation %s',
                                      status.location, _discoro_computation._auth)
                _discoro_coro.send({'req': 'close', 'auth': _discoro_config['auth']})

    def _discoro_monitor_proc(coro=None):
        coro.set_daemon()
        while 1:
            msg = yield coro.receive(timeout=_discoro_zombie_timeout)
            if isinstance(msg, asyncoro.MonitorException):
                _discoro_nonlocals.busy_time = time.time()
                asyncoro.logger.debug('job %s done', msg.args[0])
                _discoro_job_coros.discard(msg.args[0])
            elif ((not _discoro_job_coros) and _discoro_zombie_timeout and
                  ((time.time() - _discoro_nonlocals.busy_time) > _discoro_zombie_timeout)):
                asyncoro.logger.debug('%s: zombie computation "%s"',
                                      coro.location, _discoro_computation._auth)
                _discoro_coro.send({'req': 'close', 'auth': _discoro_config['auth']})

    def _discoro_schedule_proc(client, job_coro, coro=None):
        if (yield client.deliver(job_coro, timeout=5)) == 1:
            Coro._asyncoro._add(job_coro)
        else:
            _discoro_job_coros.discard(job_coro)

    _discoro_monitor_coro = Coro(_discoro_monitor_proc)
    asyncoro.AsynCoro.instance().peer_status(Coro(_discoro_peer_status))

    while 1:
        _discoro_msg = yield _discoro_coro.receive()
        if not isinstance(_discoro_msg, dict):
            continue
        _discoro_req = _discoro_msg.get('req', None)

        if _discoro_req == 'run':
            _discoro_client = _discoro_msg.get('client', None)
            _discoro_auth = _discoro_msg.get('auth', None)
            _discoro_func = _discoro_msg.get('func', None)
            if not isinstance(_discoro_client, Coro) or not _discoro_computation or \
               _discoro_auth != _discoro_computation._auth:
                asyncoro.logger.warning('invalid run: %s', type(_discoro_func))
                if isinstance(_discoro_client, Coro):
                    _discoro_client.send(None)
                continue
            try:
                _discoro_func = asyncoro.unserialize(_discoro_func)
                if _discoro_func.code:
                    exec(_discoro_func.code) in globals()
                job_coro = Coro(globals()[_discoro_func.name],
                                *(_discoro_func.args), **(_discoro_func.kwargs))
            except:
                asyncoro.logger.debug('invalid computation to run')
                job_coro = (sys.exc_info()[0], getattr(_discoro_func, 'name', _discoro_func),
                            traceback.format_exc())
                _discoro_client.send(job_coro)
            else:
                asyncoro.logger.debug('job %s created', job_coro)
                _discoro_job_coros.add(job_coro)
                job_coro.notify(_discoro_monitor_coro)
                _discoro_var = _discoro_msg.get('notify', None)
                if isinstance(_discoro_var, Coro):
                    job_coro.notify(_discoro_var)
                if Coro._asyncoro._remove(job_coro) == 0:
                    Coro(_discoro_schedule_proc, _discoro_client, job_coro)
            _discoro_nonlocals.busy_time = time.time()
            del job_coro
        elif _discoro_req == 'setup':
            _discoro_client = _discoro_msg.get('client', None)
            _discoro_pulse_coro = _discoro_msg.get('pulse_coro', None)
            if not isinstance(_discoro_client, Coro) or not isinstance(_discoro_pulse_coro, Coro):
                continue
            if _discoro_computation is not None:
                asyncoro.logger.debug('invalid "setup" - busy')
                _discoro_client.send(-1)
                continue
            os.chdir(_discoro_dest_path)
            try:
                _discoro_computation = _discoro_msg['computation']
                exec('import asyncoro.disasyncoro as asyncoro') in globals()
                if _discoro_computation._code:
                    exec(_discoro_computation._code) in globals()
            except:
                _discoro_computation = None
                asyncoro.logger.warning('invalid computation')
                asyncoro.logger.debug(traceback.format_exc())
                _discoro_client.send(-1)
                continue
            if isinstance(_discoro_computation.pulse_interval, int) and \
               MinPulseInterval <= _discoro_computation.pulse_interval <= MaxPulseInterval:
                _discoro_computation.pulse_interval = _discoro_computation.pulse_interval
            else:
                _discoro_computation.pulse_interval = MinPulseInterval
            _discoro_timer_coro.send({'pulse_coro': _discoro_pulse_coro,
                                      'interval': _discoro_computation.pulse_interval,
                                      'disk_path': _discoro_dest_path})
            _discoro_nonlocals.busy_time = time.time()
            _discoro_zombie_timeout = _discoro_computation.zombie_period
            asyncoro.logger.debug('computation "%s" from %s with zombie period %s',
                                  _discoro_computation._auth, _discoro_msg['client'].location,
                                  _discoro_zombie_timeout)
            if not _discoro_zombie_timeout:
                _discoro_zombie_timeout = None
            _discoro_monitor_coro.send(None)
            _discoro_client.send(0)
        elif _discoro_req == 'close':
            _discoro_auth = _discoro_msg.get('auth', None)
            if not _discoro_computation or (_discoro_auth != _discoro_computation._auth and
                                            _discoro_auth != _discoro_config['auth']):
                continue
            asyncoro.logger.debug('%s deleting computation "%s"',
                                  _discoro_coro.location, _discoro_computation._auth)
            if _discoro_auth != _discoro_computation._auth and _discoro_pulse_coro:
                _discoro_pulse_coro.send({'status': 'ServerClosed',
                                          'location': _discoro_coro.location})
            for _discoro_var in _discoro_job_coros:
                _discoro_var.terminate()
            _discoro_job_coros.clear()

            for _discoro_var in list(globals()):
                if _discoro_var not in _discoro_globals:
                    globals().pop(_discoro_var, None)
            globals().update(_discoro_globals)

            for _discoro_var in sys.modules.keys():
                if _discoro_var not in _discoro_modules:
                    sys.modules.pop(_discoro_var, None)
            sys.modules.update(_discoro_modules)

            for _discoro_var in os.listdir(_discoro_dest_path):
                _discoro_var = os.path.join(_discoro_dest_path, _discoro_var)
                if os.path.isdir(_discoro_var) and not os.path.islink(_discoro_var):
                    shutil.rmtree(_discoro_var, ignore_errors=True)
                else:
                    os.remove(_discoro_var)
            if not os.path.isdir(_discoro_dest_path):
                try:
                    os.remove(_discoro_dest_path)
                except:
                    pass
                os.makedirs(_discoro_dest_path)
            if not os.path.isfile(_discoro_pid_path):
                try:
                    if os.path.islink(_discoro_pid_path):
                        os.remove(_discoro_pid_path)
                    else:
                        shutil.rmtree(_discoro_pid_path)
                    with open(_discoro_pid_path, 'w') as _discoro_var:
                        _discoro_var.write('%s' % os.getpid())
                except:
                    asyncoro.logger.warning('PID file "%s" is invalid', _discoro_pid_path)
            os.chdir(_discoro_dest_path)
            asyncoro.AsynCoro.instance().dest_path = _discoro_dest_path
            _discoro_computation = _discoro_client = _discoro_pulse_coro = None
            if _discoro_config['serve'] > 0:
                _discoro_config['serve'] -= 1
                if _discoro_config['serve'] == 0:
                    break
            _discoro_timer_coro.send({'pulse_coro': _discoro_pulse_coro, 'interval': None,
                                      'disk_path': ''})
            _discoro_zombie_timeout = None
        elif _discoro_req == 'node_info':
            if psutil:
                _discoro_var = DiscoroNodeAvailInfo(_discoro_coro.location.addr,
                                                    100.0 - psutil.cpu_percent(),
                                                    psutil.virtual_memory().available,
                                                    psutil.disk_usage(_discoro_dest_path).free,
                                                    100.0 - psutil.swap_memory().percent)
            else:
                _discoro_var = None
            # _discoro_name is host name followed by '-' and ID
            _discoro_var = DiscoroNodeInfo(_discoro_name[:_discoro_name.rfind('-')],
                                           _discoro_coro.location.addr, _discoro_var)
            _discoro_client = _discoro_msg.get('client', None)
            if isinstance(_discoro_client, Coro):
                _discoro_client.send(_discoro_var)
        elif _discoro_req == 'status':
            if _discoro_msg.get('auth', None) != _discoro_config['auth']:
                asyncoro.logger.debug('ignoring info: %s', _discoro_msg.get('auth'))
                continue
            if _discoro_pulse_coro:
                print('  Server "%s" @ %s running %d coroutines for %s' %
                      (_discoro_name, _discoro_coro.location, len(_discoro_job_coros),
                       _discoro_pulse_coro.location))
            else:
                print('  Server "%s" @ %s not used by any computation' %
                      (_discoro_name, _discoro_coro.location))
        elif _discoro_req == 'quit':
            if _discoro_msg.get('auth', None) != _discoro_config['auth']:
                asyncoro.logger.debug('ignoring quit: %s', _discoro_msg.get('auth'))
                continue
            if _discoro_pulse_coro:
                _discoro_pulse_coro.send({'status': 'ServerClosed',
                                          'location': _discoro_coro.location})
            break
        elif _discoro_req == 'terminate':
            if _discoro_msg.get('auth', None) != _discoro_config['auth']:
                asyncoro.logger.debug('ignoring terminate: %s', _discoro_msg.get('auth'))
                continue
            if _discoro_pulse_coro:
                _discoro_pulse_coro.send({'status': 'ServerTerminated',
                                          'location': _discoro_coro.location})
            if _discoro_computation:
                msg = {'req': 'close', 'auth': _discoro_computation._auth}
                _discoro_config['serve'] = 1
                _discoro_coro.send(msg)
            else:
                break
        else:
            asyncoro.logger.warning('invalid command "%s" ignored', _discoro_req)
            _discoro_client = _discoro_msg.get('client', None)
            if not isinstance(_discoro_client, Coro):
                continue
            _discoro_client.send(-1)

    # wait until all computations are done; process only 'close'
    while _discoro_job_coros:
        _discoro_msg = yield _discoro_coro.receive()
        if not isinstance(_discoro_msg, dict):
            continue
        _discoro_req = _discoro_msg.get('req', None)

        if _discoro_req == 'close':
            _discoro_auth = _discoro_msg.get('auth', None)
            if not _discoro_computation or _discoro_auth != _discoro_computation._auth:
                continue
            asyncoro.logger.debug('%s deleting computation "%s"',
                                  _discoro_coro.location, _discoro_computation._auth)

            for _discoro_var in list(globals()):
                if _discoro_var not in _discoro_globals:
                    globals().pop(_discoro_var, None)
            globals().update(_discoro_globals)

            break
        else:
            asyncoro.logger.warning('invalid command "%s" ignored', _discoro_req)
            _discoro_client = _discoro_msg.get('client', None)
            if not isinstance(_discoro_client, Coro):
                continue
            _discoro_client.send(-1)

    for _discoro_var in os.listdir(_discoro_dest_path):
        _discoro_var = os.path.join(_discoro_dest_path, _discoro_var)
        if os.path.isdir(_discoro_var) and not os.path.islink(_discoro_var):
            shutil.rmtree(_discoro_var, ignore_errors=True)
        else:
            os.remove(_discoro_var)
    if os.path.isfile(_discoro_pid_path):
        os.remove(_discoro_pid_path)
    _discoro_config['mp_queue'].put({'req': 'quit', 'auth': _discoro_config['auth']})
    asyncoro.logger.debug('discoro server "%s" @ %s terminated',
                          _discoro_name, _discoro_coro.location)
dataset/ETHPy150Open pgiri/asyncoro/py2/asyncoro/discoronode.py/_discoro_proc/original