Class for setting up GraphScope instance on hosts cluster
| 46 | |
| 47 | |
| 48 | class HostsClusterLauncher(Launcher): |
| 49 | """Class for setting up GraphScope instance on hosts cluster""" |
| 50 | |
| 51 | def __init__(self, config: Config): |
| 52 | self._config = copy.deepcopy(config) |
| 53 | self._proc = None |
| 54 | |
| 55 | port = self._config.coordinator.service_port |
| 56 | if not is_free_port(port): |
| 57 | port = get_free_port() |
| 58 | self._config.coordinator.service_port = port |
| 59 | self._coordinator_endpoint = f"{self._config.hosts_launcher.hosts[0]}:{port}" |
| 60 | |
| 61 | def poll(self): |
| 62 | if self._proc is not None: |
| 63 | return self._proc.poll() |
| 64 | return -1 |
| 65 | |
| 66 | def base64_encode(self, string): |
| 67 | return base64.b64encode(string.encode("utf-8")).decode("utf-8", errors="ignore") |
| 68 | |
| 69 | def _launch_coordinator(self): |
| 70 | cmd = [ |
| 71 | sys.executable, |
| 72 | "-m", |
| 73 | "gscoordinator", |
| 74 | "--config", |
| 75 | self.base64_encode(self._config.dumps_json()), |
| 76 | ] |
| 77 | |
| 78 | # logger.info("Initializing coordinator with command: %s", " ".join(cmd)) |
| 79 | |
| 80 | env = os.environ.copy() |
| 81 | env["PYTHONUNBUFFERED"] = "TRUE" |
| 82 | # add graphscope module to PYTHONPATH |
| 83 | graphscope_dir = os.path.join(os.path.dirname(graphscope.__file__), "..") |
| 84 | coordinator_dir = os.path.join(graphscope_dir, "..", "coordinator") |
| 85 | additional_path = graphscope_dir + os.pathsep + coordinator_dir |
| 86 | |
| 87 | if "PYTHONPATH" in env: |
| 88 | env["PYTHONPATH"] = additional_path + os.pathsep + env["PYTHONPATH"] |
| 89 | else: |
| 90 | env["PYTHONPATH"] = additional_path |
| 91 | |
| 92 | # Param `start_new_session=True` is for putting child process to a new process group |
| 93 | # so it won't get the signals from parent. |
| 94 | # In notebook environment, we need to accept the signal from kernel restarted/stopped. |
| 95 | process = subprocess.Popen( |
| 96 | cmd, |
| 97 | start_new_session=False if in_notebook() else True, |
| 98 | cwd=os.getcwd(), |
| 99 | env=env, |
| 100 | encoding="utf-8", |
| 101 | errors="replace", |
| 102 | stdin=subprocess.DEVNULL, |
| 103 | stdout=subprocess.PIPE, |
| 104 | stderr=subprocess.PIPE, |
| 105 | universal_newlines=True, |