Close all the writing process and load/gather the data from all the nodes if main node or all_process is True.
(self)
| 379 | rendez_vous_lock.release() |
| 380 | |
| 381 | def _finalize(self): |
| 382 | """Close all the writing process and load/gather the data |
| 383 | from all the nodes if main node or all_process is True. |
| 384 | """ |
| 385 | if self.writer is not None: |
| 386 | self.writer.finalize() |
| 387 | self.writer = None |
| 388 | # release the locks of the processes > 0 so that process 0 can lock them to read + delete the data |
| 389 | if self.filelock is not None and self.process_id > 0: |
| 390 | self.filelock.release() |
| 391 | |
| 392 | if self.keep_in_memory: |
| 393 | # Read the predictions and references |
| 394 | reader = ArrowReader(path=self.data_dir, info=DatasetInfo(features=self.selected_feature_format)) |
| 395 | self.data = Dataset.from_buffer(self.buf_writer.getvalue()) |
| 396 | |
| 397 | elif self.process_id == 0: |
| 398 | # Let's acquire a lock on each node files to be sure they are finished writing |
| 399 | file_paths, filelocks = self._get_all_cache_files() |
| 400 | |
| 401 | # Read the predictions and references |
| 402 | try: |
| 403 | reader = ArrowReader(path="", info=DatasetInfo(features=self.selected_feature_format)) |
| 404 | self.data = Dataset(**reader.read_files([{"filename": f} for f in file_paths])) |
| 405 | except FileNotFoundError: |
| 406 | raise ValueError( |
| 407 | "Error in finalize: another evaluation module instance is already using the local cache file. " |
| 408 | "Please specify an experiment_id to avoid collision between distributed evaluation module instances." |
| 409 | ) from None |
| 410 | |
| 411 | # Store file paths and locks and we will release/delete them after the computation. |
| 412 | self.file_paths = file_paths |
| 413 | self.filelocks = filelocks |
| 414 | |
| 415 | def compute(self, *, predictions=None, references=None, **kwargs) -> Optional[dict]: |
| 416 | """Compute the evaluation module. |
no test coverage detected