Puts a list of data into the Ray object store in parallel using a thread pool. Args: data_list (List[Any]): A list of Python objects to be put into the Ray object store. max_workers (int, optional): The maximum number of worker threads to use.
(data_list: list[Any], max_workers: Optional[int] = None)
| 49 | |
| 50 | |
| 51 | def parallel_put(data_list: list[Any], max_workers: Optional[int] = None): |
| 52 | """ |
| 53 | Puts a list of data into the Ray object store in parallel using a thread pool. |
| 54 | |
| 55 | Args: |
| 56 | data_list (List[Any]): A list of Python objects to be put into the Ray object store. |
| 57 | max_workers (int, optional): The maximum number of worker threads to use. |
| 58 | Defaults to min(len(data_list), 16). |
| 59 | |
| 60 | Returns: |
| 61 | List[ray.ObjectRef]: A list of Ray object references corresponding to the input data_list, |
| 62 | maintaining the original order. |
| 63 | """ |
| 64 | assert len(data_list) > 0, "data_list must not be empty" |
| 65 | |
| 66 | def put_data(index, data): |
| 67 | return index, ray.put(data) |
| 68 | |
| 69 | if max_workers is None: |
| 70 | max_workers = min(len(data_list), 16) |
| 71 | |
| 72 | with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: |
| 73 | data_list_f = [executor.submit(put_data, i, data) for i, data in enumerate(data_list)] |
| 74 | res_lst = [] |
| 75 | for future in concurrent.futures.as_completed(data_list_f): |
| 76 | res_lst.append(future.result()) |
| 77 | |
| 78 | # reorder based on index |
| 79 | output = [None for _ in range(len(data_list))] |
| 80 | for res in res_lst: |
| 81 | index, data_ref = res |
| 82 | output[index] = data_ref |
| 83 | |
| 84 | return output |
| 85 | |
| 86 | |
| 87 | def get_event_loop(): |