-
Notifications
You must be signed in to change notification settings - Fork 269
Expand file tree
/
Copy paththreaded_ikram_celal_keskin.py
More file actions
51 lines (39 loc) · 1.56 KB
/
threaded_ikram_celal_keskin.py
File metadata and controls
51 lines (39 loc) · 1.56 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
import threading
def threaded(n:int) ->callable:
"""
A decorator that runs a function using multiple threads.
:param n: Number of threads to create
:type n: int
:return: A decorator function for threading
:rtype: function
"""
# if n < 1 or not isinstance(n, int):
# raise ValueError("Not Valid call of the decorator")
def decorator(func:callable) ->callable:
"""
An inner decorator function that wraps the original function with threads.
:param func: The original function to be run in threads
:type func: function
:return: A wrapper function that runs with threads
:rtype: function
"""
if not hasattr(func, "__call__"):
raise ValueError("Not Valid Function")
def wrapper(*args, **kwargs)->None:
"""
A wrapper function that runs the function with multiple threads.
:param args: Positional arguments to pass to the function
:param kwargs: Keyword arguments to pass to the function
"""
all_threads = []
try:
for _ in range(n):
one_thread = threading.Thread(target=func, args=args, kwargs=kwargs)
all_threads.append(one_thread)
one_thread.start()
for thread in all_threads:
thread.join()
except threading.ThreadError as te:
print(te)
return wrapper
return decorator