util.py 文件源码

def __init__(self, path):
    import json
    self._path = path
    try:
      with open(path) as f:
        old_data = json.load(f)
    except json.JSONDecodeError:
      logger.warning('Could not decode config')
      old_data = {}
    except OSError:
      logger.debug('No config file')
      old_data = {}

    for i in range(10):
      try:
        self._f = open(path, 'w+')
        fcntl.flock(self._f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        self._locked = True
        break
      except BlockingIOError:
        import signal
        pid = old_data.get('pid')
        if pid:
          logger.info(f'Config file is locked (try {i}). Killing previous instance {pid}')
          os.kill(pid, signal.SIGTERM)
          time.sleep(.05)
        else:
          logger.error(f'Config file is locked and no pid to kill')
    assert self._locked