(self, info: DaskClusterInfo, workdir: Path)
| 71 | |
| 72 | class DaskEnvironment(Environment, EnvStateManager): |
| 73 | def __init__(self, info: DaskClusterInfo, workdir: Path): |
| 74 | super().__init__() |
| 75 | self.info = info |
| 76 | self._workdir = workdir.resolve() |
| 77 | self.cluster = ClusterHelper(info.cluster_info, workdir=self.workdir) |
| 78 | self.client: Optional[Client] = None |
| 79 | self.dask_dir = ensure_directory(self.workdir / "dask") |
| 80 | |
| 81 | self.nodes = self.info.cluster_info.node_list.resolve() |
| 82 | sanity_check_nodes(self.nodes) |
| 83 | |
| 84 | worker_nodes = self.nodes if isinstance(self.info.cluster_info.node_list, Local) else self.nodes[1:] |
| 85 | if not worker_nodes: |
| 86 | raise Exception("No worker nodes are available") |
| 87 | |
| 88 | self.worker_count = sum(w.processes for w in self.info.workers) |
| 89 | assert self.worker_count > 0 |
| 90 | self.worker_assignment = assign_workers(self.info.workers, worker_nodes) |
| 91 | logging.debug(f"Worker assignment: {self.worker_assignment}") |
| 92 | |
| 93 | @property |
| 94 | def workdir(self) -> Path: |
nothing calls this directly
no test coverage detected