Add 'run_in_thread' to Scheduler constructor

Author: TheGreatCabbage

scheduler/ProcessTask.py

@@ -0,0 +1,50 @@
+#  MIT License
+#
+#  Copyright (c) 2020 Sam McCormack
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+from multiprocessing import Process, Queue
+
+from scheduler.Task import Task
+from scheduler.utils import terminate_tree
+
+
+class ProcessTask(Task):
+    def __init__(self, process: Process, queue: Queue, subtasks: int = 0):
+        super(ProcessTask, self).__init__(queue, subtasks)
+
+        self.process = process
+
+    def start(self) -> None:
+        """Starts the task."""
+        if not self.running and not self.finished:
+            self.process.start()
+            self.running = True
+
+    def terminate(self) -> None:
+        """
+        Terminates the task and all running sub-tasks.
+        """
+        try:
+            terminate_tree(self.process)
+            self.queue.close()
+        except:
+            pass
+        finally:
+            self.running = False

scheduler/Scheduler.py

@@ -23,14 +23,18 @@
 import asyncio
 import functools
 import multiprocessing
-import time
 import sys
+import threading
+import time
 from multiprocessing import cpu_count, Process, Queue
+from queue import Queue as MTQueue
 from typing import List, Callable, Type, Optional, Union, Any, Tuple, Iterable
 
 import psutil
 
+from scheduler.ProcessTask import ProcessTask
 from scheduler.Task import Task
+from scheduler.ThreadTask import ThreadTask
 from scheduler.utils import SchedulerException
 
 
@@ -54,7 +58,8 @@ def __init__(
         cpu_threshold: float = 95,
         cpu_update_interval: float = 5,
         shared_memory: bool = False,
-        shared_memory_threshold:int = 1e7,
+        shared_memory_threshold: int = 1e7,
+        run_in_thread: bool = False,
     ):
         """
         :param progress_callback: a function taking the number of finished tasks and the total number of tasks, which is
@@ -66,7 +71,15 @@ def __init__(
         :param cpu_update_interval: the time, in seconds, between consecutive CPU usage checks when `dynamic` is enabled
         :param shared_memory: whether to use shared memory if possible
         :param shared_memory_threshold: the minimum size of a Numpy array which will cause it to be transferred using shared memory if possible
+        :param run_in_thread: if True, a single task will be run in a thread instead of a process.
+        This reduces the overhead (caused by spawning processes instead of forking) on Windows/macOS systems
         """
+        self.run_in_thread = run_in_thread
+        if self.run_in_thread and shared_memory:
+            raise SchedulerException(
+                f"Shared memory cannot currently be used when 'run_in_thread' is True."
+            )
+
         self.dynamic = dynamic
         self.update_interval = update_interval
 
@@ -125,7 +138,7 @@ def add_process(self, process: Process, queue: Queue, subtasks: int = 0) -> None
                 "add() cannot be called on a Scheduler which has already been started."
             )
 
-        task = Task(process, queue, subtasks)
+        task = ProcessTask(process, queue, subtasks)
         self.tasks.append(task)
 
     def add(
@@ -150,13 +163,27 @@ def add(
                 "add_function() cannot be called on a scheduler which has already been started."
             )
 
-        queue = queue_type()
+        thread = self.run_in_thread and len(self.tasks) == 0
+
+        if thread:
+            queue = MTQueue()
+
+            _args = (queue, self.mgr, self.shared_memory_threshold) + args
+            _wrapper = functools.partial(wrapper, target)
+
+            task = ThreadTask(
+                threading.Thread(target=_wrapper, args=_args), queue, subtasks=subtasks
+            )
+        else:
+            queue = queue_type()
+
+            _args = (queue, self.mgr, self.shared_memory_threshold) + args
+            _wrapper = functools.partial(wrapper, target)
 
-        _args = (queue, self.mgr, self.shared_memory_threshold) + args
-        _wrapper = functools.partial(wrapper, target)
+            process = process_type(target=_wrapper, args=_args)
+            task = ProcessTask(process, queue, subtasks=subtasks)
 
-        process = process_type(target=_wrapper, args=_args)
-        self.tasks.append(Task(process, queue, subtasks=subtasks))
+        self.tasks.append(task)
 
     async def map(
         self,
@@ -463,7 +490,7 @@ def wrapper(
     queue: Queue,
     manager: Optional["SharedMemoryManager"],
     threshold: int,
-    *args: Any
+    *args: Any,
 ) -> None:
     """
     Wrapper which calls a function with its specified arguments and puts the output in a queue.

scheduler/Task.py

@@ -19,45 +19,30 @@
 #  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 #  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 #  SOFTWARE.
+from abc import ABC, abstractmethod
 
-from multiprocessing import Process, Queue
 
-from scheduler.utils import terminate_tree
-
-
-class Task:
+class Task(ABC):
     """
     A simple class containing a process and an associated queue. The queue should have been
     passed to the process, so that the process can put its output in the queue.
     """
 
-    def __init__(self, process: Process, queue: Queue, subtasks: int = 0):
-        self.process = process
+    def __init__(self, queue, subtasks: int):
         self.queue = queue
-
         self.running = False
         self.finished = False
 
         # The number of processes which will be spawned as part of this task.
         self.subtasks: int = subtasks
 
+    @abstractmethod
     def start(self) -> None:
-        """Starts the task."""
-        if not self.running and not self.finished:
-            self.process.start()
-            self.running = True
+        pass
 
+    @abstractmethod
     def terminate(self) -> None:
-        """
-        Terminates the task and all running sub-tasks.
-        """
-        try:
-            terminate_tree(self.process)
-            self.queue.close()
-        except:
-            pass
-        finally:
-            self.running = False
+        pass
 
     def total_tasks(self) -> int:
         """

scheduler/ThreadTask.py

@@ -0,0 +1,43 @@
+#  MIT License
+#
+#  Copyright (c) 2020 Sam McCormack
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+import warnings
+from queue import Queue
+from threading import Thread
+
+from scheduler.Task import Task
+
+
+class ThreadTask(Task):
+    def __init__(self, thread: Thread, queue: Queue, subtasks: int = 0):
+        super(ThreadTask, self).__init__(queue, subtasks)
+
+        self.thread = thread
+
+    def start(self) -> None:
+        if not self.running and not self.finished:
+            self.thread.start()
+            self.running = True
+
+    def terminate(self) -> None:
+        warnings.warn(f"Cannot terminate a thread task.", RuntimeWarning)
+
+        self.running = False