-
-
Notifications
You must be signed in to change notification settings - Fork 50.5k
Expand file tree
/
Copy pathsleep_sort.py
More file actions
83 lines (67 loc) · 2.23 KB
/
sleep_sort.py
File metadata and controls
83 lines (67 loc) · 2.23 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
from typing import List
import threading
import time
def sleep_sort(numbers: List[int], simulate: bool = True, scale: float = 0.01) -> None:
"""
Perform Sleep Sort on the given list of integers.
Sorts the list numbers in place using the Sleep Sort algorithm.
Behavior:
- Destructive: modifies the list numbers.
- Only accepts integers (positive, zero, or negative).
- Default simulate=True runs instantly by simulating timed wake-ups (safe for tests/CI).
- If simulate=False, the function spawns one thread per element and uses time.sleep;
this mode causes real waiting time proportional to element values.
- scale (seconds per unit) applies only when simulate=False.
Examples
--------
>>> nums = [3, 1, 2]
>>> sleep_sort(nums)
>>> nums
[1, 2, 3]
>>> nums = [0, 0, 1]
>>> sleep_sort(nums)
>>> nums
[0, 0, 1]
>>> nums = [-2, 1, 0]
>>> sleep_sort(nums)
>>> nums
[-2, 0, 1]
>>> sleep_sort([1.5, 2])
Traceback (most recent call last):
...
TypeError: integers only please
"""
if not numbers:
return
if any(not isinstance(x, int) for x in numbers):
raise TypeError("integers only please")
min_val = min(numbers)
offset = -min_val if min_val < 0 else 0
if simulate:
# Simulated wake-up: bucket by wake time (value + offset), preserve order
buckets = {}
for idx, val in enumerate(numbers):
wake = val + offset
buckets.setdefault(wake, []).append((idx, val))
result: List[int] = []
for wake in sorted(buckets.keys()):
for _, val in buckets[wake]:
result.append(val)
numbers[:] = result
return
# Real threaded mode: causes actual delays proportional to element values
results: List[int] = []
lock = threading.Lock()
def worker(value: int) -> None:
time.sleep((value + offset) * scale)
with lock:
results.append(value)
threads: List[threading.Thread] = []
for val in numbers:
t = threading.Thread(target=worker, args=(val,))
t.daemon = True
t.start()
threads.append(t)
for t in threads:
t.join()
numbers[:] = results