Dev/master

.gitignore

@@ -11,3 +11,7 @@ build
 dist
 .coverage*
 .vscode
+
+daisy_logs/
+.pytest_cache/
+.mypy_cache/

benchmarks/test_dep_graph.py → benchmarks/benchmark_dep_graph.py

@@ -47,7 +47,7 @@
 @pytest.mark.parametrize("test", ["iterate", "init"])
 @pytest.mark.parametrize("size", ["small", "medium", "large"])
 @pytest.mark.parametrize("block_gen", ["lazy", "enumerated"])
-def test_dep_graph(benchmark, test, size, block_gen):
+def benchmark_dep_graph(benchmark, test, size, block_gen):
     task = tasks[size]
     lazy = block_gen == "lazy"
     iterate = test == "iterate"

daisy/dependency_graph.py

@@ -410,7 +410,7 @@ def compute_level_conflicts(self) -> List[List[Coordinate]]:
 
             # get conflicts to previous level
             if prev_level_offset is not None and self.read_write_conflict:
-                conflict_offsets = self.get_conflict_offsets(
+                conflict_offsets: List[Coordinate] = self.get_conflict_offsets(
                     level_offset, prev_level_offset, self._level_stride
                 )
             else:

daisy/scheduler.py

@@ -7,7 +7,7 @@
 from .task import Task
 from .block import BlockStatus
 
-from typing import List
+from typing import Any, Dict, List
 import collections
 import logging
 
@@ -41,9 +41,11 @@ def __init__(self, tasks: List[Task], count_all_orphans=True):
             self.dependency_graph.downstream, self.dependency_graph.upstream
         )
 
-        self.task_map = {}
-        self.task_states = collections.defaultdict(TaskState)
-        self.task_queues = collections.defaultdict(ProcessingQueue)
+        self.task_map: Dict[Any, Task] = {}
+        self.task_states: Dict[Any, TaskState] = collections.defaultdict(TaskState)
+        self.task_queues: Dict[Any, ProcessingQueue] = collections.defaultdict(
+            ProcessingQueue
+        )
 
         # root tasks is a mapping from task_id -> (num_roots, root_generator)
         roots = self.dependency_graph.roots()

daisy/server.py

@@ -1,3 +1,4 @@
+from time import time
 from .block import BlockStatus
 from .block_bookkeeper import BlockBookkeeper
 from .context import Context
@@ -69,12 +70,15 @@ def run_blockwise(self, tasks, scheduler=None):
         return True if self.all_done else False
 
     def _event_loop(self):
-
+        last_time = time()
         while not self.stop_event.is_set():
-
+            current_time = time()
             self._handle_client_messages()
             self._check_for_lost_blocks()
             self.worker_pools.check_worker_health()
+            if current_time - last_time > 1:
+                self._check_all_tasks_completed()
+                last_time = current_time
 
     def _get_client_message(self):
 

daisy/tcp/io_looper.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import threading
+from typing import Dict
 import tornado.ioloop
 
 logger = logging.getLogger(__name__)
@@ -18,8 +19,8 @@ class IOLooper:
             thread per process.
     """
 
-    threads = {}
-    ioloops = {}
+    threads: Dict[int, threading.Thread] = {}
+    ioloops: Dict[int, tornado.ioloop.IOLoop] = {}
 
     @staticmethod
     def clear():

docs/conf.py

@@ -19,6 +19,9 @@
 
 # -- Project information -----------------------------------------------------
 
+from typing import Dict
+
+
 project = "daisy"
 copyright = "2019, Jan Funke, Tri Nguyen, Carolin Malin-Mayor, Arlo Sheridan, Philipp Hanslovsky, Chris Barnes"
 author = "Jan Funke, Tri Nguyen, Carolin Malin-Mayor, Arlo Sheridan, Philipp Hanslovsky, Chris Barnes"
@@ -109,7 +112,7 @@
 
 # -- Options for LaTeX output ------------------------------------------------
 
-latex_elements = {
+latex_elements: Dict[str, str] = {
     # The paper size ('letterpaper' or 'a4paper').
     #
     # 'papersize': 'letterpaper',

pyproject.toml

@@ -39,3 +39,12 @@ dev = [
 
 [tool.black]
 target_version = ['py37', 'py38', 'py39', 'py310']
+
+# # module specific overrides
+[[tool.mypy.overrides]]
+module = [
+    "funlib.*",
+    "tqdm.*",
+    "pkg_resources.*",
+]
+ignore_missing_imports = true