(self)
| 145 | return result_line |
| 146 | |
| 147 | def multi_process_evaluation(self): |
| 148 | start_eval_time = time.perf_counter() |
| 149 | nr_devices = len(self.devices) |
| 150 | stride = int(np.ceil(self.ndata / nr_devices)) |
| 151 | |
| 152 | # start multi-process on multi-gpu |
| 153 | procs = [] |
| 154 | for d in range(nr_devices): |
| 155 | e_record = min((d + 1) * stride, self.ndata) |
| 156 | shred_list = list(range(d * stride, e_record)) |
| 157 | device = self.devices[d] |
| 158 | logger.info("GPU %s handle %d data." % (device, len(shred_list))) |
| 159 | |
| 160 | p = self.context.Process(target=self.worker, args=(shred_list, device)) |
| 161 | procs.append(p) |
| 162 | |
| 163 | for p in procs: |
| 164 | p.start() |
| 165 | |
| 166 | all_results = [] |
| 167 | for _ in tqdm(range(self.ndata)): |
| 168 | t = self.results_queue.get() |
| 169 | all_results.append(t) |
| 170 | if self.verbose: |
| 171 | self.compute_metric(all_results) |
| 172 | |
| 173 | for p in procs: |
| 174 | p.join() |
| 175 | |
| 176 | result_line = self.compute_metric(all_results) |
| 177 | logger.info("Evaluation Elapsed Time: %.2fs" % (time.perf_counter() - start_eval_time)) |
| 178 | return result_line |
| 179 | |
| 180 | def worker(self, shred_list, device): |
| 181 | start_load_time = time.time() |
no test coverage detected