We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix external dynamic deps
Since running tasks communicate with worker via a queue, all dynamic dependencies that they yield must be serialized and then deserialized back. This doesn't work if a task has `run = NotImplemented`, since there was a specific check for that in Register for unclear reason.

This PR adds a test case to reproduce the issue and fixes it by removing the check.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/task_register.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 """
18 Define the centralized register of all :class:`~luigi.task.Task` classes.
19 """
20 
21 import abc
22 try:
23     from collections import OrderedDict
24 except ImportError:
25     from ordereddict import OrderedDict
26 
27 from luigi import six
28 import logging
29 logger = logging.getLogger('luigi-interface')
30 
31 
32 class TaskClassException(Exception):
33     pass
34 
35 
36 class Register(abc.ABCMeta):
37     """
38     The Metaclass of :py:class:`Task`.
39 
40     Acts as a global registry of Tasks with the following properties:
41 
42     1. Cache instances of objects so that eg. ``X(1, 2, 3)`` always returns the
43        same object.
44     2. Keep track of all subclasses of :py:class:`Task` and expose them.
45     """
46     __instance_cache = {}
47     _default_namespace = None
48     _reg = []
49     AMBIGUOUS_CLASS = object()  # Placeholder denoting an error
50     """If this value is returned by :py:meth:`__get_reg` then there is an
51     ambiguous task name (two :py:class:`Task` have the same name). This denotes
52     an error."""
53 
54     def __new__(metacls, classname, bases, classdict):
55         """
56         Custom class creation for namespacing.
57 
58         Also register all subclasses.
59 
60         Set the task namespace to whatever the currently declared namespace is.
61         """
62         if "task_namespace" not in classdict:
63             classdict["task_namespace"] = metacls._default_namespace
64 
65         cls = super(Register, metacls).__new__(metacls, classname, bases, classdict)
66         metacls._reg.append(cls)
67 
68         return cls
69 
70     def __call__(cls, *args, **kwargs):
71         """
72         Custom class instantiation utilizing instance cache.
73 
74         If a Task has already been instantiated with the same parameters,
75         the previous instance is returned to reduce number of object instances.
76         """
77         def instantiate():
78             return super(Register, cls).__call__(*args, **kwargs)
79 
80         h = cls.__instance_cache
81 
82         if h is None:  # disabled
83             return instantiate()
84 
85         params = cls.get_params()
86         param_values = cls.get_param_values(params, args, kwargs)
87 
88         k = (cls, tuple(param_values))
89 
90         try:
91             hash(k)
92         except TypeError:
93             logger.debug("Not all parameter values are hashable so instance isn't coming from the cache")
94             return instantiate()  # unhashable types in parameters
95 
96         if k not in h:
97             h[k] = instantiate()
98 
99         return h[k]
100 
101     @classmethod
102     def clear_instance_cache(cls):
103         """
104         Clear/Reset the instance cache.
105         """
106         cls.__instance_cache = {}
107 
108     @classmethod
109     def disable_instance_cache(cls):
110         """
111         Disables the instance cache.
112         """
113         cls.__instance_cache = None
114 
115     @property
116     def task_family(cls):
117         """
118         The task family for the given class.
119 
120         If ``cls.task_namespace is None`` then it's the name of the class.
121         Otherwise, ``<task_namespace>.`` is prefixed to the class name.
122         """
123         if cls.task_namespace is None:
124             return cls.__name__
125         else:
126             return "%s.%s" % (cls.task_namespace, cls.__name__)
127 
128     @classmethod
129     def __get_reg(cls):
130         """Return all of the registered classes.
131 
132         :return:  an ``collections.OrderedDict`` of task_family -> class
133         """
134         # We have to do this on-demand in case task names have changed later
135         # We return this in a topologically sorted list of inheritance: this is useful in some cases (#822)
136         reg = OrderedDict()
137         for cls in cls._reg:
138             if cls.run == NotImplemented:
139                 continue
140             name = cls.task_family
141 
142             if name in reg and reg[name] != cls and \
143                     reg[name] != cls.AMBIGUOUS_CLASS and \
144                     not issubclass(cls, reg[name]):
145                 # Registering two different classes - this means we can't instantiate them by name
146                 # The only exception is if one class is a subclass of the other. In that case, we
147                 # instantiate the most-derived class (this fixes some issues with decorator wrappers).
148                 reg[name] = cls.AMBIGUOUS_CLASS
149             else:
150                 reg[name] = cls
151 
152         return reg
153 
154     @classmethod
155     def task_names(cls):
156         """
157         List of task names as strings
158         """
159         return sorted(cls.__get_reg().keys())
160 
161     @classmethod
162     def tasks_str(cls):
163         """
164         Human-readable register contents dump.
165         """
166         return ','.join(cls.task_names())
167 
168     @classmethod
169     def get_task_cls(cls, name):
170         """
171         Returns an unambiguous class or raises an exception.
172         """
173         task_cls = cls.__get_reg().get(name)
174         if not task_cls:
175             raise TaskClassException('Task %r not found. Candidates are: %s' % (name, cls.tasks_str()))
176 
177         if task_cls == cls.AMBIGUOUS_CLASS:
178             raise TaskClassException('Task %r is ambiguous' % name)
179         return task_cls
180 
181     @classmethod
182     def get_all_params(cls):
183         """
184         Compiles and returns all parameters for all :py:class:`Task`.
185 
186         :return: a generator of tuples (TODO: we should make this more elegant)
187         """
188         for task_name, task_cls in six.iteritems(cls.__get_reg()):
189             if task_cls == cls.AMBIGUOUS_CLASS:
190                 continue
191             for param_name, param_obj in task_cls.get_params():
192                 yield task_name, (not task_cls.use_cmdline_section), param_name, param_obj
193 
194 
195 def load_task(module, task_name, params_str):
196     """
197     Imports task dynamically given a module and a task name.
198     """
199     if module is not None:
200         __import__(module)
201     task_cls = Register.get_task_cls(task_name)
202     return task_cls.from_str_params(params_str)
203
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/luigi/task_register.py b/luigi/task_register.py
--- a/luigi/task_register.py
+++ b/luigi/task_register.py
@@ -135,8 +135,6 @@
         # We return this in a topologically sorted list of inheritance: this is useful in some cases (#822)
         reg = OrderedDict()
         for cls in cls._reg:
-            if cls.run == NotImplemented:
-                continue
             name = cls.task_family
 
             if name in reg and reg[name] != cls and \

{"golden_diff": "diff --git a/luigi/task_register.py b/luigi/task_register.py\n--- a/luigi/task_register.py\n+++ b/luigi/task_register.py\n@@ -135,8 +135,6 @@\n         # We return this in a topologically sorted list of inheritance: this is useful in some cases (#822)\n         reg = OrderedDict()\n         for cls in cls._reg:\n-            if cls.run == NotImplemented:\n-                continue\n             name = cls.task_family\n \n             if name in reg and reg[name] != cls and \\\n", "issue": "Fix external dynamic deps\nSince running tasks communicate with worker via a queue, all dynamic dependencies that they yield must be serialized and then deserialized back. This doesn't work if a task has `run = NotImplemented`, since there was a specific check for that in Register for unclear reason.\n\nThis PR adds a test case to reproduce the issue and fixes it by removing the check.\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nDefine the centralized register of all :class:`~luigi.task.Task` classes.\n\"\"\"\n\nimport abc\ntry:\n    from collections import OrderedDict\nexcept ImportError:\n    from ordereddict import OrderedDict\n\nfrom luigi import six\nimport logging\nlogger = logging.getLogger('luigi-interface')\n\n\nclass TaskClassException(Exception):\n    pass\n\n\nclass Register(abc.ABCMeta):\n    \"\"\"\n    The Metaclass of :py:class:`Task`.\n\n    Acts as a global registry of Tasks with the following properties:\n\n    1. Cache instances of objects so that eg. ``X(1, 2, 3)`` always returns the\n       same object.\n    2. Keep track of all subclasses of :py:class:`Task` and expose them.\n    \"\"\"\n    __instance_cache = {}\n    _default_namespace = None\n    _reg = []\n    AMBIGUOUS_CLASS = object()  # Placeholder denoting an error\n    \"\"\"If this value is returned by :py:meth:`__get_reg` then there is an\n    ambiguous task name (two :py:class:`Task` have the same name). This denotes\n    an error.\"\"\"\n\n    def __new__(metacls, classname, bases, classdict):\n        \"\"\"\n        Custom class creation for namespacing.\n\n        Also register all subclasses.\n\n        Set the task namespace to whatever the currently declared namespace is.\n        \"\"\"\n        if \"task_namespace\" not in classdict:\n            classdict[\"task_namespace\"] = metacls._default_namespace\n\n        cls = super(Register, metacls).__new__(metacls, classname, bases, classdict)\n        metacls._reg.append(cls)\n\n        return cls\n\n    def __call__(cls, *args, **kwargs):\n        \"\"\"\n        Custom class instantiation utilizing instance cache.\n\n        If a Task has already been instantiated with the same parameters,\n        the previous instance is returned to reduce number of object instances.\n        \"\"\"\n        def instantiate():\n            return super(Register, cls).__call__(*args, **kwargs)\n\n        h = cls.__instance_cache\n\n        if h is None:  # disabled\n            return instantiate()\n\n        params = cls.get_params()\n        param_values = cls.get_param_values(params, args, kwargs)\n\n        k = (cls, tuple(param_values))\n\n        try:\n            hash(k)\n        except TypeError:\n            logger.debug(\"Not all parameter values are hashable so instance isn't coming from the cache\")\n            return instantiate()  # unhashable types in parameters\n\n        if k not in h:\n            h[k] = instantiate()\n\n        return h[k]\n\n    @classmethod\n    def clear_instance_cache(cls):\n        \"\"\"\n        Clear/Reset the instance cache.\n        \"\"\"\n        cls.__instance_cache = {}\n\n    @classmethod\n    def disable_instance_cache(cls):\n        \"\"\"\n        Disables the instance cache.\n        \"\"\"\n        cls.__instance_cache = None\n\n    @property\n    def task_family(cls):\n        \"\"\"\n        The task family for the given class.\n\n        If ``cls.task_namespace is None`` then it's the name of the class.\n        Otherwise, ``<task_namespace>.`` is prefixed to the class name.\n        \"\"\"\n        if cls.task_namespace is None:\n            return cls.__name__\n        else:\n            return \"%s.%s\" % (cls.task_namespace, cls.__name__)\n\n    @classmethod\n    def __get_reg(cls):\n        \"\"\"Return all of the registered classes.\n\n        :return:  an ``collections.OrderedDict`` of task_family -> class\n        \"\"\"\n        # We have to do this on-demand in case task names have changed later\n        # We return this in a topologically sorted list of inheritance: this is useful in some cases (#822)\n        reg = OrderedDict()\n        for cls in cls._reg:\n            if cls.run == NotImplemented:\n                continue\n            name = cls.task_family\n\n            if name in reg and reg[name] != cls and \\\n                    reg[name] != cls.AMBIGUOUS_CLASS and \\\n                    not issubclass(cls, reg[name]):\n                # Registering two different classes - this means we can't instantiate them by name\n                # The only exception is if one class is a subclass of the other. In that case, we\n                # instantiate the most-derived class (this fixes some issues with decorator wrappers).\n                reg[name] = cls.AMBIGUOUS_CLASS\n            else:\n                reg[name] = cls\n\n        return reg\n\n    @classmethod\n    def task_names(cls):\n        \"\"\"\n        List of task names as strings\n        \"\"\"\n        return sorted(cls.__get_reg().keys())\n\n    @classmethod\n    def tasks_str(cls):\n        \"\"\"\n        Human-readable register contents dump.\n        \"\"\"\n        return ','.join(cls.task_names())\n\n    @classmethod\n    def get_task_cls(cls, name):\n        \"\"\"\n        Returns an unambiguous class or raises an exception.\n        \"\"\"\n        task_cls = cls.__get_reg().get(name)\n        if not task_cls:\n            raise TaskClassException('Task %r not found. Candidates are: %s' % (name, cls.tasks_str()))\n\n        if task_cls == cls.AMBIGUOUS_CLASS:\n            raise TaskClassException('Task %r is ambiguous' % name)\n        return task_cls\n\n    @classmethod\n    def get_all_params(cls):\n        \"\"\"\n        Compiles and returns all parameters for all :py:class:`Task`.\n\n        :return: a generator of tuples (TODO: we should make this more elegant)\n        \"\"\"\n        for task_name, task_cls in six.iteritems(cls.__get_reg()):\n            if task_cls == cls.AMBIGUOUS_CLASS:\n                continue\n            for param_name, param_obj in task_cls.get_params():\n                yield task_name, (not task_cls.use_cmdline_section), param_name, param_obj\n\n\ndef load_task(module, task_name, params_str):\n    \"\"\"\n    Imports task dynamically given a module and a task name.\n    \"\"\"\n    if module is not None:\n        __import__(module)\n    task_cls = Register.get_task_cls(task_name)\n    return task_cls.from_str_params(params_str)\n", "path": "luigi/task_register.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nDefine the centralized register of all :class:`~luigi.task.Task` classes.\n\"\"\"\n\nimport abc\ntry:\n    from collections import OrderedDict\nexcept ImportError:\n    from ordereddict import OrderedDict\n\nfrom luigi import six\nimport logging\nlogger = logging.getLogger('luigi-interface')\n\n\nclass TaskClassException(Exception):\n    pass\n\n\nclass Register(abc.ABCMeta):\n    \"\"\"\n    The Metaclass of :py:class:`Task`.\n\n    Acts as a global registry of Tasks with the following properties:\n\n    1. Cache instances of objects so that eg. ``X(1, 2, 3)`` always returns the\n       same object.\n    2. Keep track of all subclasses of :py:class:`Task` and expose them.\n    \"\"\"\n    __instance_cache = {}\n    _default_namespace = None\n    _reg = []\n    AMBIGUOUS_CLASS = object()  # Placeholder denoting an error\n    \"\"\"If this value is returned by :py:meth:`__get_reg` then there is an\n    ambiguous task name (two :py:class:`Task` have the same name). This denotes\n    an error.\"\"\"\n\n    def __new__(metacls, classname, bases, classdict):\n        \"\"\"\n        Custom class creation for namespacing.\n\n        Also register all subclasses.\n\n        Set the task namespace to whatever the currently declared namespace is.\n        \"\"\"\n        if \"task_namespace\" not in classdict:\n            classdict[\"task_namespace\"] = metacls._default_namespace\n\n        cls = super(Register, metacls).__new__(metacls, classname, bases, classdict)\n        metacls._reg.append(cls)\n\n        return cls\n\n    def __call__(cls, *args, **kwargs):\n        \"\"\"\n        Custom class instantiation utilizing instance cache.\n\n        If a Task has already been instantiated with the same parameters,\n        the previous instance is returned to reduce number of object instances.\n        \"\"\"\n        def instantiate():\n            return super(Register, cls).__call__(*args, **kwargs)\n\n        h = cls.__instance_cache\n\n        if h is None:  # disabled\n            return instantiate()\n\n        params = cls.get_params()\n        param_values = cls.get_param_values(params, args, kwargs)\n\n        k = (cls, tuple(param_values))\n\n        try:\n            hash(k)\n        except TypeError:\n            logger.debug(\"Not all parameter values are hashable so instance isn't coming from the cache\")\n            return instantiate()  # unhashable types in parameters\n\n        if k not in h:\n            h[k] = instantiate()\n\n        return h[k]\n\n    @classmethod\n    def clear_instance_cache(cls):\n        \"\"\"\n        Clear/Reset the instance cache.\n        \"\"\"\n        cls.__instance_cache = {}\n\n    @classmethod\n    def disable_instance_cache(cls):\n        \"\"\"\n        Disables the instance cache.\n        \"\"\"\n        cls.__instance_cache = None\n\n    @property\n    def task_family(cls):\n        \"\"\"\n        The task family for the given class.\n\n        If ``cls.task_namespace is None`` then it's the name of the class.\n        Otherwise, ``<task_namespace>.`` is prefixed to the class name.\n        \"\"\"\n        if cls.task_namespace is None:\n            return cls.__name__\n        else:\n            return \"%s.%s\" % (cls.task_namespace, cls.__name__)\n\n    @classmethod\n    def __get_reg(cls):\n        \"\"\"Return all of the registered classes.\n\n        :return:  an ``collections.OrderedDict`` of task_family -> class\n        \"\"\"\n        # We have to do this on-demand in case task names have changed later\n        # We return this in a topologically sorted list of inheritance: this is useful in some cases (#822)\n        reg = OrderedDict()\n        for cls in cls._reg:\n            name = cls.task_family\n\n            if name in reg and reg[name] != cls and \\\n                    reg[name] != cls.AMBIGUOUS_CLASS and \\\n                    not issubclass(cls, reg[name]):\n                # Registering two different classes - this means we can't instantiate them by name\n                # The only exception is if one class is a subclass of the other. In that case, we\n                # instantiate the most-derived class (this fixes some issues with decorator wrappers).\n                reg[name] = cls.AMBIGUOUS_CLASS\n            else:\n                reg[name] = cls\n\n        return reg\n\n    @classmethod\n    def task_names(cls):\n        \"\"\"\n        List of task names as strings\n        \"\"\"\n        return sorted(cls.__get_reg().keys())\n\n    @classmethod\n    def tasks_str(cls):\n        \"\"\"\n        Human-readable register contents dump.\n        \"\"\"\n        return ','.join(cls.task_names())\n\n    @classmethod\n    def get_task_cls(cls, name):\n        \"\"\"\n        Returns an unambiguous class or raises an exception.\n        \"\"\"\n        task_cls = cls.__get_reg().get(name)\n        if not task_cls:\n            raise TaskClassException('Task %r not found. Candidates are: %s' % (name, cls.tasks_str()))\n\n        if task_cls == cls.AMBIGUOUS_CLASS:\n            raise TaskClassException('Task %r is ambiguous' % name)\n        return task_cls\n\n    @classmethod\n    def get_all_params(cls):\n        \"\"\"\n        Compiles and returns all parameters for all :py:class:`Task`.\n\n        :return: a generator of tuples (TODO: we should make this more elegant)\n        \"\"\"\n        for task_name, task_cls in six.iteritems(cls.__get_reg()):\n            if task_cls == cls.AMBIGUOUS_CLASS:\n                continue\n            for param_name, param_obj in task_cls.get_params():\n                yield task_name, (not task_cls.use_cmdline_section), param_name, param_obj\n\n\ndef load_task(module, task_name, params_str):\n    \"\"\"\n    Imports task dynamically given a module and a task name.\n    \"\"\"\n    if module is not None:\n        __import__(module)\n    task_cls = Register.get_task_cls(task_name)\n    return task_cls.from_str_params(params_str)\n", "path": "luigi/task_register.py"}]}