Repository: Duke-GCB/DukeDSClient

Issue:
download hangs after completing on windows
On windows after downloading larger files the terminal hangs after finishing downloading and verifying the data. This seems to be related to not closing the multiprocessing pool.
I found that closing the multiprocessing pool fixes this problem.
https://github.com/Duke-GCB/DukeDSClient/blob/869a4cfa4fbe4544e7e9463272951ec9979edcfb/ddsc/core/parallel.py#L162

Discussion:

Patch:
diff --git a/ddsc/core/download.py b/ddsc/core/download.py
index ca96cc6..c24935f 100644
--- a/ddsc/core/download.py
+++ b/ddsc/core/download.py
@@ -5,7 +5,7 @@ import math
 import requests
 from ddsc.core.localstore import HashUtil
 from ddsc.core.util import ProgressPrinter, plural_fmt
-from ddsc.core.parallel import TaskExecutor, TaskRunner
+from ddsc.core.parallel import TaskRunner
 from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
 from ddsc.core.remotestore import RemoteStore, ProjectFile, RemoteFileUrl
 from ddsc.core.retry import RetrySettings
@@ -296,7 +296,7 @@ class FileDownloader(object):
         task_runner.run()
 
     def _create_task_runner(self):
-        return TaskRunner(TaskExecutor(self.settings.config.download_workers))
+        return TaskRunner(self.settings.config.download_workers)
 
     def make_ranges(self, file_to_download):
         """
diff --git a/ddsc/core/parallel.py b/ddsc/core/parallel.py
index acafb3c..37d9616 100644
--- a/ddsc/core/parallel.py
+++ b/ddsc/core/parallel.py
@@ -90,13 +90,13 @@ class TaskRunner(object):
     """
     Runs a bunch of tasks in parallel with support for task waiting.
     """
-    def __init__(self, executor):
+    def __init__(self, num_workers):
         """
-        Setup runner to use executor to run it's tasks.
-        :param executor: TaskExecutor: actually executes tasks and returns their results
+        Setup runner to use num_workers to run it's tasks.
+        :param num_workers: int: number of workers to use when running tasks
         """
         self.waiting_task_list = WaitingTaskList()
-        self.executor = executor
+        self.num_workers = num_workers
         self.next_id = 1
 
     def _claim_next_id(self):
@@ -133,21 +133,24 @@ class TaskRunner(object):
         Blocks until all tasks have been completed.
         :return:
         """
+        executor = TaskExecutor(self.num_workers)
         for task in self.get_next_tasks(None):
-            self.executor.add_task(task, None)
-        while not self.executor.is_done():
-            done_task_and_result = self.executor.wait_for_tasks()
+            executor.add_task(task, None)
+        while not executor.is_done():
+            done_task_and_result = executor.wait_for_tasks()
             for task, task_result in done_task_and_result:
-                self._add_sub_tasks_to_executor(task, task_result)
+                self._add_sub_tasks_to_executor(executor, task, task_result)
+        executor.close()
 
-    def _add_sub_tasks_to_executor(self, parent_task, parent_task_result):
+    def _add_sub_tasks_to_executor(self, executor, parent_task, parent_task_result):
         """
         Add all subtasks for parent_task to the executor.
+        :param executor: TaskExecutor: executor to add tasks to
         :param parent_task: Task: task that has just finished
         :param parent_task_result: object: result of task that is finished
         """
         for sub_task in self.waiting_task_list.get_next_tasks(parent_task.id):
-            self.executor.add_task(sub_task, parent_task_result)
+            executor.add_task(sub_task, parent_task_result)
 
 
 class TaskExecutor(object):
@@ -262,6 +265,9 @@ class TaskExecutor(object):
                 self.pending_results.remove(pending_result)
         return task_and_results
 
+    def close(self):
+        self.pool.close()
+
 
 def execute_task_async(task_func, task_id, context):
     """
diff --git a/ddsc/core/projectuploader.py b/ddsc/core/projectuploader.py
index 6a5d42d..2e94b9e 100644
--- a/ddsc/core/projectuploader.py
+++ b/ddsc/core/projectuploader.py
@@ -1,7 +1,7 @@
 from ddsc.core.util import ProjectWalker, KindType
 from ddsc.core.ddsapi import DataServiceAuth, DataServiceApi
 from ddsc.core.fileuploader import FileUploader, FileUploadOperations, ParentData, ParallelChunkProcessor
-from ddsc.core.parallel import TaskExecutor, TaskRunner
+from ddsc.core.parallel import TaskRunner
 
 
 class UploadSettings(object):
@@ -100,7 +100,7 @@ class ProjectUploader(object):
         Setup to talk to the data service based on settings.
         :param settings: UploadSettings: settings to use for uploading.
         """
-        self.runner = TaskRunner(TaskExecutor(settings.config.upload_workers))
+        self.runner = TaskRunner(settings.config.upload_workers)
         self.settings = settings
         self.small_item_task_builder = SmallItemUploadTaskBuilder(self.settings, self.runner)
         self.small_files = []


Test patch:
diff --git a/ddsc/core/tests/test_download.py b/ddsc/core/tests/test_download.py
index bb08107..3e269dc 100644
--- a/ddsc/core/tests/test_download.py
+++ b/ddsc/core/tests/test_download.py
@@ -401,7 +401,8 @@ class TestFileToDownload(TestCase):
 
 class TestFileDownloader(TestCase):
     def setUp(self):
-        self.mock_config = Mock(download_bytes_per_chunk=1000)
+        self.num_download_workers = 5
+        self.mock_config = Mock(download_bytes_per_chunk=1000, download_workers=self.num_download_workers)
         self.mock_settings = Mock(dest_directory='/tmp/data2', config=self.mock_config)
         self.mock_file1 = Mock(path="data/file1.txt", size=200, local_path='/tmp/data2/data/file1.txt')
         self.mock_file1.get_remote_parent_path.return_value = 'data'
@@ -424,9 +425,8 @@ class TestFileDownloader(TestCase):
         downloader.download_files.assert_called_with()
 
     @patch('ddsc.core.download.TaskRunner')
-    @patch('ddsc.core.download.TaskExecutor')
     @patch('ddsc.core.download.os')
-    def test_make_local_directories(self, mock_os, mock_task_executor, mock_task_runner):
+    def test_make_local_directories(self, mock_os, mock_task_runner):
         mock_os.path.exists.return_value = True
         mock_os.path = os.path
         downloader = FileDownloader(self.mock_settings, self.mock_files_to_download)
@@ -434,9 +434,8 @@ class TestFileDownloader(TestCase):
         mock_os.makedirs.assert_called_with('/tmp/data2/data')
 
     @patch('ddsc.core.download.TaskRunner')
-    @patch('ddsc.core.download.TaskExecutor')
     @patch('ddsc.core.download.os')
-    def test_make_big_empty_files(self, mock_os, mock_task_executor, mock_task_runner):
+    def test_make_big_empty_files(self, mock_os, mock_task_runner):
         mock_os.path.exists.return_value = True
         mock_os.path = os.path
         downloader = FileDownloader(self.mock_settings, self.mock_files_to_download)
@@ -448,8 +447,7 @@ class TestFileDownloader(TestCase):
         fake_open.return_value.write.assert_called_with(b'\0')
 
     @patch('ddsc.core.download.TaskRunner')
-    @patch('ddsc.core.download.TaskExecutor')
-    def test_download_files(self, mock_task_executor, mock_task_runner):
+    def test_download_files(self, mock_task_runner):
         downloader = FileDownloader(self.mock_settings, self.mock_files_to_download)
         downloader.group_files_by_size = Mock()
         mock_small_file = Mock(size=100)
@@ -466,6 +464,7 @@ class TestFileDownloader(TestCase):
         downloader.download_files()
 
         # Download small file in one command, large file in two commands).
+        mock_task_runner.assert_called_with(self.num_download_workers)
         add_calls = mock_task_runner.return_value.add.call_args_list
         self.assertEqual(5, len(add_calls))
         command = add_calls[0][1]['command']
diff --git a/ddsc/core/tests/test_parallel.py b/ddsc/core/tests/test_parallel.py
index 4b7f704..8521615 100644
--- a/ddsc/core/tests/test_parallel.py
+++ b/ddsc/core/tests/test_parallel.py
@@ -98,8 +98,7 @@ class TestTaskRunner(TestCase):
     """
     def test_single_add(self):
         add_command = AddCommand(10, 30)
-        executor = TaskExecutor(10)
-        runner = TaskRunner(executor)
+        runner = TaskRunner(num_workers=10)
         runner.add(None, add_command)
         runner.run()
         self.assertEqual(add_command.parent_task_result, None)
@@ -109,8 +108,7 @@ class TestTaskRunner(TestCase):
     def test_two_adds_in_order(self):
         add_command = AddCommand(10, 30)
         add_command2 = AddCommand(4, 1)
-        executor = TaskExecutor(10)
-        runner = TaskRunner(executor)
+        runner = TaskRunner(num_workers=10)
         runner.add(None, add_command)
         runner.add(1, add_command2)
         runner.run()
@@ -122,8 +120,7 @@ class TestTaskRunner(TestCase):
     def test_two_adds_in_parallel(self):
         add_command = AddCommand(10, 30)
         add_command2 = AddCommand(4, 1)
-        executor = TaskExecutor(10)
-        runner = TaskRunner(executor)
+        runner = TaskRunner(num_workers=10)
         runner.add(None, add_command,)
         runner.add(None, add_command2)
         runner.run()
@@ -137,8 +134,7 @@ class TestTaskRunner(TestCase):
         add_command.send_message = 'ok'
         add_command2 = AddCommand(4, 1)
         add_command2.send_message = 'waiting'
-        executor = TaskExecutor(10)
-        runner = TaskRunner(executor)
+        runner = TaskRunner(num_workers=10)
         runner.add(None, add_command,)
         runner.add(None, add_command2)
         runner.run()
@@ -149,6 +145,15 @@ class TestTaskRunner(TestCase):
         self.assertEqual(add_command.on_message_data, ['ok'])
         self.assertEqual(add_command2.on_message_data, ['waiting'])
 
+    @patch('ddsc.core.parallel.TaskExecutor')
+    def test_run_closes_executor(self, mock_task_executor):
+        add_command = AddCommand(10, 30)
+        runner = TaskRunner(num_workers=10)
+        runner.add(None, add_command)
+        runner.run()
+        mock_task_executor.assert_called_with(10)
+        mock_task_executor.return_value.close.assert_called_with()
+
 
 class TestTaskExecutor(TestCase):
     @patch('ddsc.core.parallel.multiprocessing')
@@ -212,3 +217,11 @@ class TestTaskExecutor(TestCase):
         executor.wait_for_tasks()
         self.assertEqual(40, add_command.result)
         self.assertEqual(add_command.on_message_data, ['TEST', 'TEST2'])
+
+    @patch('ddsc.core.parallel.multiprocessing')
+    def test_close(self, mock_multiprocessing):
+        executor = TaskExecutor(2)
+        mock_multiprocessing.Pool.assert_called_with()
+        mock_multiprocessing.Pool.return_value.close.assert_not_called()
+        executor.close()
+        mock_multiprocessing.Pool.return_value.close.assert_called_with()
diff --git a/ddsc/core/tests/test_projectuploader.py b/ddsc/core/tests/test_projectuploader.py
index 19cc76a..7b35539 100644
--- a/ddsc/core/tests/test_projectuploader.py
+++ b/ddsc/core/tests/test_projectuploader.py
@@ -208,12 +208,12 @@ class TestCreateSmallFile(TestCase):
 class TestProjectUploader(TestCase):
     @patch('ddsc.core.projectuploader.ProjectWalker')
     @patch('ddsc.core.projectuploader.TaskRunner')
-    @patch('ddsc.core.projectuploader.TaskExecutor')
     @patch('ddsc.core.projectuploader.SmallItemUploadTaskBuilder')
-    def test_run_sorts_new_files_first(self, mock_small_task_builder, mock_task_executor, mock_task_runner,
-                                       mock_project_walker):
+    def test_run_sorts_new_files_first(self, mock_small_task_builder, mock_task_runner, mock_project_walker):
         settings = Mock()
         settings.config.upload_bytes_per_chunk = 100
+        num_upload_workers = 6
+        settings.config.upload_workers = num_upload_workers
         uploader = ProjectUploader(settings)
         uploader.process_large_file = Mock()
         small_file_existing = Mock(remote_id='abc123', size=1000)
@@ -234,6 +234,7 @@ class TestProjectUploader(TestCase):
 
         uploader.run(local_project)
 
+        mock_task_runner.assert_called_with(num_upload_workers)
         # small files should be sorted with new first
         uploader.small_item_task_builder.visit_file.assert_has_calls([
             call(small_file_new, None),
@@ -247,10 +248,8 @@ class TestProjectUploader(TestCase):
 
     @patch('ddsc.core.projectuploader.ProjectWalker')
     @patch('ddsc.core.projectuploader.TaskRunner')
-    @patch('ddsc.core.projectuploader.TaskExecutor')
     @patch('ddsc.core.projectuploader.SmallItemUploadTaskBuilder')
-    def test_run_with_large_files_hash_matching(self, mock_small_task_builder, mock_task_executor, mock_task_runner,
-                                                mock_project_walker):
+    def test_run_with_large_files_hash_matching(self, mock_small_task_builder, mock_task_runner, mock_project_walker):
         settings = Mock()
         settings.config.upload_bytes_per_chunk = 100
         uploader = ProjectUploader(settings)
@@ -276,11 +275,10 @@ class TestProjectUploader(TestCase):
         ])
 
     @patch('ddsc.core.projectuploader.TaskRunner')
-    @patch('ddsc.core.projectuploader.TaskExecutor')
     @patch('ddsc.core.projectuploader.SmallItemUploadTaskBuilder')
     @patch('ddsc.core.projectuploader.FileUploader')
     def test_process_large_file_sets_file_hash_and_alg(self, mock_file_uploader, mock_small_task_builder,
-                                                       mock_task_executor, mock_task_runner):
+                                                       mock_task_runner):
         uploader = ProjectUploader(Mock())
         local_file = Mock()
         parent = Mock()
@@ -295,11 +293,9 @@ class TestProjectUploader(TestCase):
         )
 
     @patch('ddsc.core.projectuploader.TaskRunner')
-    @patch('ddsc.core.projectuploader.TaskExecutor')
     @patch('ddsc.core.projectuploader.SmallItemUploadTaskBuilder')
     @patch('ddsc.core.projectuploader.FileUploader')
-    def test_upload_large_files__updates_watcher(self, mock_file_uploader, mock_small_task_builder,
-                                                 mock_task_executor, mock_task_runner):
+    def test_upload_large_files__updates_watcher(self, mock_file_uploader, mock_small_task_builder, mock_task_runner):
         local_file1 = Mock(size=1000)
         local_file1.hash_matches_remote.return_value = False
         local_file2 = Mock(size=2000)