Garbage collect task result queue when worker context exits (#2973)

* Close our link to the task result queue when a worker exits

* Don't exit worker context until worker has run in test

luigi/worker.py

@@ -606,6 +606,7 @@ def __exit__(self, type, value, traceback):
         for task in self._running_tasks.values():
             if task.is_alive():
                 task.terminate()
+        self._task_result_queue.close()
         return False  # Don't suppress exception
 
     def _generate_worker_info(self):

test/worker_external_task_test.py

@@ -185,9 +185,9 @@ def test_external_task_complete_but_missing_dep_at_runtime(self):
         # split up scheduling task and running to simulate runtime scenario
         with self._make_worker() as w:
             w.add(test_task)
-        # touch output so test_task should be considered complete at runtime
-        open(test_task.output_path, 'a').close()
-        success = w.run()
+            # touch output so test_task should be considered complete at runtime
+            open(test_task.output_path, 'a').close()
+            success = w.run()
 
         self.assertTrue(success)
         # upstream dependency output didn't exist at runtime