Mercurial
view grok_interview/merge_sort.py @ 170:7fce234bfdca
Closing old fzf head
| author | MrJuneJune <me@mrjunejune.com> |
|---|---|
| date | Mon, 19 Jan 2026 17:35:39 -0800 |
| parents | 68fa88ac73fe |
| children |
line wrap: on
line source
from concurrent.futures import ProcessPoolExecutor import random from typing import List def merge_sort(lst: List[int]) -> List[int]: # no executor parameter! if len(lst) <= 1: return lst middle = len(lst) // 2 # These run in separate processes → no shared thread pool problem left = merge_sort(lst[:middle]) right = merge_sort(lst[middle:]) return merge(left, right) def merge(left: List[int], right: List[int]) -> List[int]: res = [] i = j = 0 while i < len(left) and j < len(right): if left[i] <= right[j]: res.append(left[i]) i += 1 else: res.append(right[j]) j += 1 res.extend(left[i:]) res.extend(right[j:]) return res RANGE = 10_000 cpu_num = 10 def main(): data = [random.randint(0, 100) for _ in range(RANGE)] sorted_data = [] ranges = [i for i in range(RANGE // cpu_num, RANGE, RANGE // 100)] with ProcessPoolExecutor() as executor: for pos_range in ranges: sorted_data.append(executor.submit(merge_sort, data[:pos_range]).result()) while len(sorted_data) > 1: new_sorted_data = [] for i in range(0, len(sorted_data), 2): if res = merge(sorted_data[i], sorted_data[i+1]) new_sorted_data.append(res) print(len(sorted_data)) if __name__ == "__main__": # important on Windows/macOS main()