Repository: ManiacalLabs/BiblioPixel
Pull request #782 (2018-05-03 14:06:07)

Thread vevamp; ArtNet control
Because this branch got too large, I split off a bunch of uncontroversial bug fix or cleanup commits out at the start.

The last two are more interesting.

The second-to-last one, "Rework Runnable and threads for better shutdown" is about as ugly as I have written on this project.  :-/  

I needed to change the threading model somewhat to deal with some of the issues I recently documented and there was no neat way to do it gradually.

I did test everything really thoroughly.

It's all worth it,. though, for the last one: "New ArtNet control".  The code for the ArtNet control is [tiny and very clear](https://github.com/rec/BiblioPixel/blob/artnet/bibliopixel/control/artnet.py).

---

I haven't yet tested the ArtNet control - that last commit - on Adam2's ArtNet installation, although I have really a nice integration test that took a lot of work to really make sing.  

I'm going to test the hardware in parallel with the review of this branch, or if the branch gets finished before then, I'll pull out that last commit into a separate branch and pull all the other parts.



adammhaile: +1


Patch:
diff --git a/bibliopixel/animation/animation.py b/bibliopixel/animation/animation.py
index 6826380..947efcd 100644
--- a/bibliopixel/animation/animation.py
+++ b/bibliopixel/animation/animation.py
@@ -52,7 +52,7 @@ class Animation(object):
 
     @color_list.setter
     def color_list(self, cl):
-        self.layout.color_list[:] = cl
+        self.layout.color_list = cl
 
     @property
     def completed(self):
@@ -97,6 +97,9 @@ class Animation(object):
     def stop(self):
         self.threading.stop_event.set()
 
+    def join(self, timeout=None):
+        self.threading.join(timeout)
+
     def run_all_frames(self):
         for i in self.generate_frames():
             pass
diff --git a/bibliopixel/animation/collection.py b/bibliopixel/animation/collection.py
index aa2daba..c518974 100644
--- a/bibliopixel/animation/collection.py
+++ b/bibliopixel/animation/collection.py
@@ -1,4 +1,4 @@
-import os, traceback
+import os
 from .. project import aliases, construct, load, project
 from . import animation
 from .. util import json, log
diff --git a/bibliopixel/control/action.py b/bibliopixel/control/action.py
index fc0ef17..78158bb 100644
--- a/bibliopixel/control/action.py
+++ b/bibliopixel/control/action.py
@@ -21,7 +21,7 @@ class Action(Receiver):
             if len(values) == 1:
                 values = [self.ops(values[0])]
             else:
-                # TODO:They specified ops, but we can't use it.
+                # TODO: They specified ops, but we can't use it.
                 # Should we warn here?  Can we use the ops somehow?
                 pass
         return self.address.receive(values)
diff --git a/bibliopixel/control/artnet.py b/bibliopixel/control/artnet.py
new file mode 100644
index 0000000..f937d55
--- /dev/null
+++ b/bibliopixel/control/artnet.py
@@ -0,0 +1,25 @@
+import collections, copy, queue
+from . import control
+from .. util import artnet_message, log, server_cache, udp
+
+QUEUE_TIMEOUT = 0.1
+
+
+class ArtNet(control.ExtractedControl):
+    def __init__(self, *args, ip_address, port=artnet_message.UDP_PORT, **kwds):
+        super().__init__(*args, **kwds)
+        self.address = ip_address, port
+
+    def _convert(self, msg):
+        msg = artnet_message.bytes_to_message(msg)
+        assert any(i for i in msg.data)
+        msg = collections.OrderedDict((
+            ('type', 'dmx'),
+            ('net', msg.net),
+            ('subUni', msg.subUni),
+            ('data', msg.data)))
+
+        return super()._convert(msg)
+
+    def _make_thread(self):
+        return udp.Receiver(self.address, receive=self.receive)
diff --git a/bibliopixel/control/control.py b/bibliopixel/control/control.py
index 4f98fe3..194e4be 100644
--- a/bibliopixel/control/control.py
+++ b/bibliopixel/control/control.py
@@ -3,13 +3,14 @@ from . extractor import Extractor
 from .. project import construct, importer
 from .. util.log_errors import LogErrors
 from .. util import flatten, log, json
+from .. util.threads import runnable
 from . routing import ActionList, Routing
 
 
-class Control:
+class Control(runnable.Runnable):
     DEFAULT = {'datatype': ActionList}
 
-    def __init__(self, routing=None, default=None, errors=16,
+    def __init__(self, routing=None, default=None, errors='raise',
                  python_path='bibliopixel.control', verbose=False,
                  pre_routing=None):
         """
@@ -27,31 +28,26 @@ class Control:
         default = dict(self.DEFAULT, **(default or {}))
         self.pre_routing = ActionList(pre_routing)
         self.routing = Routing(routing or {}, default or {}, python_path)
-        self.running = False
 
     def set_root(self, root):
         self.pre_routing.set_root(root)
         self.routing.set_root(root)
 
     def start(self):
+        super().start()
         if self.verbose:
             log.info('Starting %s', self)
 
-        self.running = True
         self.thread = self._make_thread()
         self.thread.start()
 
-    def cleanup(self):
-        self.stop()
+    def join(self, timeout=None):
+        self.thread.join(timeout)
 
     def stop(self):
-        self.running = False
-
-    def loop(self):
-        for msg in self.messages():
-            self.receive(msg)
-            if not self.running:
-                return
+        super().stop()
+        stop_thread = getattr(self.thread, 'stop', None)
+        stop_thread and stop_thread()
 
     def _receive(self, msg):
         """
@@ -70,7 +66,8 @@ class Control:
         if receiver:
             receiver.receive(msg)
             if self.verbose:
-                log.info('Routed message %s to %s', str_msg, receiver)
+                log.info('Routed message %s (%s) to %s', str_msg, msg,
+                         repr(receiver))
 
     def _convert(self, msg):
         """
@@ -78,19 +75,11 @@ class Control:
         """
         raise NotImplementedError
 
-    def messages(self):
-        """Should yield a sequence of messages from the input source."""
-        raise NotImplementedError
-
     def _make_thread(self):
         """
-        Returns the thread that run the loop for this control source.
-
-        If the underlying source has its own thread or process, override this
-        method.  The object returned needs to have two methods:  ``start()``
-        and ``stop()``.
+        Returns a new thread to run the loop for this control source.
         """
-        return threading.Thread(target=self.loop, daemon=True)
+        raise NotImplementedError
 
     def _msg_to_str(self, msg):
         if msg is None:
@@ -101,10 +90,29 @@ class Control:
         return bool(self.routing or self.pre_routing)
 
 
+class ControlLoop:
+    """Mixin class for looping controls"""
+    def _receive_all_messages(self):
+        for msg in self.messages():
+            self.receive(msg)
+            if not self.running:
+                return
+
+    def messages(self):
+        """Should yield a sequence of messages from the input source."""
+        raise NotImplementedError
+
+    def _make_thread(self):
+        return threading.Thread(target=self._receive_all_messages, daemon=True)
+
+
 class ExtractedControl(Control):
     EXTRACTOR = {}
 
     def __init__(self, extractor=None, **kwds):
         super().__init__(**kwds)
         extractor = dict(self.EXTRACTOR, **(extractor or {}))
-        self._convert = Extractor(**extractor).extract
+        self.extractor = Extractor(**extractor)
+
+    def _convert(self, msg):
+        return self.extractor.extract(msg)
diff --git a/bibliopixel/control/extractor.py b/bibliopixel/control/extractor.py
index dff537d..3090602 100644
--- a/bibliopixel/control/extractor.py
+++ b/bibliopixel/control/extractor.py
@@ -49,16 +49,19 @@ class Extractor:
             return m and {k: to_set(v) for k, v in m.items()}
 
         self.accept, self.reject = make_match(accept), make_match(reject)
-        omit = to_set(omit)
+        self.omit = to_set(omit)
         if auto_omit and self.accept:
-            omit.update(k for k, v in self.accept.items() if len(v) == 1)
+            self.omit.update(k for k, v in self.accept.items() if len(v) == 1)
 
         self.normalizers = normalizers or {}
-        self.keys_by_type = {}
-        for k, v in (keys_by_type or {}).items():
-            if isinstance(v, str):
-                v = [v]
-            self.keys_by_type[k] = tuple(i for i in v if i not in omit)
+        if keys_by_type is None:
+            self.keys_by_type = None
+        else:
+            self.keys_by_type = {}
+            for k, v in keys_by_type.items():
+                if isinstance(v, str):
+                    v = [v]
+                self.keys_by_type[k] = tuple(i for i in v if i not in self.omit)
 
     def extract(self, msg):
         """Yield an ordered dictionary if msg['type'] is in keys_by_type."""
@@ -69,9 +72,19 @@ class Extractor:
             normalizer = self.normalizers.get(key, lambda x: x)
             return normalizer(v)
 
+        def odict(keys):
+            return collections.OrderedDict((k, normal(k)) for k in keys)
+
         def match(m):
             return (msg.get(k) in v for k, v in m.items()) if m else ()
 
-        keys = self.keys_by_type.get(msg.get('type'))
-        if keys and all(match(self.accept)) and not any(match(self.reject)):
-            return collections.OrderedDict((k, normal(k)) for k in keys)
+        accept = all(match(self.accept))
+        reject = any(match(self.reject))
+
+        if reject or not accept:
+            keys = ()
+        elif self.keys_by_type is None:
+            keys = [k for k in msg.keys() if k not in self.omit]
+        else:
+            keys = self.keys_by_type.get(msg.get('type'))
+        return odict(keys)
diff --git a/bibliopixel/control/keyboard.py b/bibliopixel/control/keyboard.py
index 5afb68a..790c4d1 100644
--- a/bibliopixel/control/keyboard.py
+++ b/bibliopixel/control/keyboard.py
@@ -1,4 +1,4 @@
-import getpass, platform, sys
+import getpass, platform, sys, threading
 from .. util import log
 from . control import ExtractedControl
 
@@ -19,6 +19,20 @@ Please install the pynput library with
 """
 
 
+try:
+    import pynput
+
+except ImportError:
+    pynput = Listener = None
+
+else:
+    class Listener(pynput.keyboard.Listener):
+        def join(self, timeout=None):
+            # join() on pynput.keyboard.Listener waits on a queue...
+            self._queue.put(None)
+            return super().join(timeout)
+
+
 def keyname(key):
     return getattr(key, 'name', None) or getattr(key, 'char')
 
@@ -42,14 +56,12 @@ class Keyboard(ExtractedControl):
         self.receive({'type': 'release', 'key': key})
 
     def _make_thread(self):
-        try:
-            import pynput
-        except Exception as e:
-            e.args = (INSTALL_ERROR,) + e.args
-            raise
+        if not pynput:
+            raise ValueError(INSTALL_ERROR)
 
         if platform.platform().startswith('Darwin'):
             if getpass.getuser() != 'root':
                 log.warning(DARWIN_ROOT_WARNING, sys.argv[0])
 
-        return pynput.keyboard.Listener(self._press, self._release)
+        log.info('Starting to listen for keyboard input')
+        return Listener(self._press, self._release)
diff --git a/bibliopixel/control/midi.py b/bibliopixel/control/midi.py
index a8c7047..bd7dc10 100644
--- a/bibliopixel/control/midi.py
+++ b/bibliopixel/control/midi.py
@@ -21,7 +21,7 @@ MIDO_BACKEND_ERROR = """You have the wrong rtmidi installed.  Try
 """
 
 
-class Midi(control.ExtractedControl):
+class Midi(control.ExtractedControl, control.ControlLoop):
     EXTRACTOR = {
         # There are more mido message types that we haven't used yet.
         'keys_by_type': {
@@ -51,9 +51,12 @@ class Midi(control.ExtractedControl):
     }
 
     def __init__(self, use_note_off=False, **kwds):
-        super().__init__(**kwds)
+        control.ExtractedControl.__init__(self, **kwds)
         self.use_note_off = use_note_off
 
+    def _make_thread(self):
+        return control.ControlLoop._make_thread(self)
+
     def messages(self):
         if not mido:
             raise ValueError(MIDO_ERROR)
@@ -69,7 +72,9 @@ class Midi(control.ExtractedControl):
             log.error('control.midi: no MIDI ports found')
             return
 
-        log.info('Starting to listen for MIDI')
+        port_names = ', '.join('"%s"' % p.name for p in ports)
+        log.info('Starting to listen for MIDI on port%s %s',
+                 '' if len(ports) == 1 else 's', port_names)
         for port, msg in mido.ports.MultiPort(ports, yield_ports=True):
             mdict = dict(vars(msg), port=port)
             if self.use_note_off or msg.type != 'note_off':
diff --git a/bibliopixel/drivers/artnet.py b/bibliopixel/drivers/artnet.py
index a89a459..4ade0bb 100644
--- a/bibliopixel/drivers/artnet.py
+++ b/bibliopixel/drivers/artnet.py
@@ -6,24 +6,27 @@ from .. util import artnet_message, log, server_cache, udp
 class ArtNet(ServerDriver):
     SERVER_CLASS = udp.Sender
 
-    def __init__(self, *args, ip_address, **kwds):
+    def __init__(self, *args, ip_address, port=artnet_message.UDP_PORT,
+                 filter_dupes=True, **kwds):
         """
         :param dict channel_map: maps DMX channels to positions in
             the color_list
         """
         self.msg = artnet_message.dmx_message()
+        self.last_message = None
 
-        address = ip_address, artnet_message.UDP_PORT
+        address = ip_address, port
         super().__init__(*args, address=address, **kwds)
+        self.filter_dupes = filter_dupes
 
     def _make_buffer(self):
         return self.msg.data
 
     def _send_packet(self):
-        # Regrettably, we need a copy here, because we can't be sure that the
-        # next _render won't come before this message has been sent.
-        # This is avoidable but not easily...
-        self.server.send(copy.copy(self.msg))
+        msg = bytes(self.msg)
+        if not (self.filter_dupes and msg == self.last_message):
+            self.server.send(msg)
+            self.last_message = msg
 
     def _on_positions(self):
         pass
diff --git a/bibliopixel/drivers/driver_base.py b/bibliopixel/drivers/driver_base.py
index 3bf7ca7..c93608b 100644
--- a/bibliopixel/drivers/driver_base.py
+++ b/bibliopixel/drivers/driver_base.py
@@ -98,7 +98,6 @@ class DriverBase(object):
         Called right before this driver will run.  This is the place
         to do things like start threads, not in the constructor.
         """
-        pass
 
     def stop(self):
         """
@@ -109,7 +108,11 @@ class DriverBase(object):
         """
         Called to shut this driver down, and stop all threads and processes.
         """
-        pass
+
+    def join(self, timeout=None):
+        """
+        Called to join threads.
+        """
 
     def bufByteCount(self):
         """
@@ -129,21 +132,18 @@ class DriverBase(object):
         one display which all need to display the next frame at exactly the same
         time.
         """
-        pass
 
     def _compute_packet(self):
         """Compute the packet from the colors and position.
 
         Eventually, this will run on the compute thread.
         """
-        pass
 
     def _send_packet(self):
         """Send the packet to the driver.
 
         Eventually, this will run on an I/O thread.
         """
-        pass
 
     def update_colors(self):
         """Apply any corrections to the current color list
diff --git a/bibliopixel/layout/layout.py b/bibliopixel/layout/layout.py
index 06486af..9781395 100644
--- a/bibliopixel/layout/layout.py
+++ b/bibliopixel/layout/layout.py
@@ -75,9 +75,15 @@ class Layout(object):
             d.start()
 
     def stop(self):
+        self.threading.stop()
         for d in self.drivers:
             d.stop()
 
+    def join(self, timeout=None):
+        self.threading.join(timeout)
+        for d in self.drivers:
+            d.join(timeout)
+
     def cleanup_drivers(self):
         for d in self.drivers:
             d.cleanup()
diff --git a/bibliopixel/main/monitor.py b/bibliopixel/main/monitor.py
index 4525208..cafe9f9 100644
--- a/bibliopixel/main/monitor.py
+++ b/bibliopixel/main/monitor.py
@@ -17,8 +17,8 @@ def run(args):
 
     control_object = tc(pre_routing='()')
     control_object.set_root(log.printer)
-    control_object.start()
-    control_object.thread.join()
+    with control_object.joiner(stop=False):
+        pass
 
 
 def set_parser(parser):
diff --git a/bibliopixel/project/project.py b/bibliopixel/project/project.py
index 6da5768..3f36c63 100644
--- a/bibliopixel/project/project.py
+++ b/bibliopixel/project/project.py
@@ -1,4 +1,4 @@
-import queue
+import queue, weakref
 from . import (
     attributes, construct, cleanup, defaults, event_queue, load, recurse)
 from .. util import exception, json, log
@@ -67,13 +67,16 @@ class Project:
         self.PROJECTS_RUNNING.add(self)
         self.layout.start()
         for c in self.controls:
-            c.start(self)
+            c.set_root(self)
+            c.start()
         self.animation.start()
 
     def stop(self):
-        log.debug('Project %s stop called', self)
-        if self in self.PROJECTS_RUNNING:
+        try:
             self.PROJECTS_RUNNING.remove(self)
+        except KeyError:
+            return
+        log.debug('Project %s stop called', self)
 
         self.animation.stop()
         for c in self.controls:
@@ -86,9 +89,22 @@ class Project:
             c.cleanup()
         self.layout.cleanup_drivers()
 
+    def join(self, timeout=None):
+        self.animation.join(timeout)
+
+        # TODO: This is a hack.  We only need to do this because we don't
+        # know if we exited through stop(), or naturally through running out of
+        # the thread.
+        self.stop()
+
+        for c in self.controls:
+            c.join(timeout)
+        self.layout.join(timeout)
+
     def run(self):
         try:
             self.start()
+            self.join()
         finally:
             self.cleanup()
         log.debug('Project %s finishes run()', self)
@@ -97,8 +113,9 @@ class Project:
     def stop_all():
         for p in list(Project.PROJECTS_RUNNING):
             p.stop()
+        Project.PROJECTS_RUNNING.clear()
 
-    PROJECTS_RUNNING = set()
+    PROJECTS_RUNNING = weakref.WeakSet()
 
 
 def project(*descs, root_file=None):
diff --git a/bibliopixel/util/threads/animation_threading.py b/bibliopixel/util/threads/animation_threading.py
index 3e43930..36ece69 100644
--- a/bibliopixel/util/threads/animation_threading.py
+++ b/bibliopixel/util/threads/animation_threading.py
@@ -2,7 +2,7 @@ import threading, time
 from .. import log
 
 
-class AnimationThreading(object):
+class AnimationThreading:
     """
     AnimationThreading handles threading - and eventually multiprocessing - for
     Animation.
@@ -15,6 +15,12 @@ class AnimationThreading(object):
         self.thread = None
         self.frame_overrun = False
 
+    def join(self, timeout=None):
+        self.thread and self.thread.join(timeout)
+
+    def is_alive(self):
+        return self.thread and self.thread.is_alive()
+
     def cleanup(self, wait=False):
         # if current thread is animation thread this was called
         # by the context manager and thread is therefore already stopped
@@ -23,9 +29,6 @@ class AnimationThreading(object):
             if self.thread and wait:
                 self.thread.join()
 
-    def stopped(self):
-        return not (self.thread and self.thread.isAlive())
-
     def target(self):
         is_main = threading.current_thread() is threading.main_thread()
         log.debug('Animation starts on %s thread', 'main' if is_main else 'new')
@@ -72,3 +75,7 @@ class AnimationThreading(object):
             self.target()
 
         self.thread and self.runner.main and self.thread.join()
+
+    def stopped(self):
+        # DEPRECATED
+        return not self.is_alive()
diff --git a/bibliopixel/util/threads/runnable.py b/bibliopixel/util/threads/runnable.py
index 492c872..c9426a3 100644
--- a/bibliopixel/util/threads/runnable.py
+++ b/bibliopixel/util/threads/runnable.py
@@ -1,5 +1,5 @@
 from .. import log
-import contextlib, functools, threading, traceback, queue
+import contextlib, functools, threading, time, traceback, queue
 
 
 class Runnable:
@@ -7,19 +7,48 @@ class Runnable:
     Base class for all objects that contain threads - including threads
     created and started by bp code, and threads started by external libraries.
 
-    You can use a Runnable in two ways:  you can call the methods start, stop,
-    and join just like for a thread, or you can use it as a context manager:
+    There are three possible thread categories for a Runnable:
+
+    1. Runs on the master thread - M
+    2. A new thread created by us - N
+    3. A new external thread created by some third-party code - X
+
+    Case X is tricky because we would like our code to be called at the start of
+    the new thread, and again at the end of that thread, but we can't.
+
+    Lifecycle - what a Runnable does, and what threads it could be called on
+
+    * construction: M
+    * start: M
+    * on_start: M
+      * called on the master thread after any new thread has started up
+    * run: MN
+    * callbacks: MNX
+    * join: M
+    * cleanup: M(N?)
+
+    TODO: right now we run all our cleanups on the new thread, if there is a new
+    thread, otherwise on the master thread.  Should we move to doing all the
+    cleanups on the master thread?
+
+    The way to use a Runnable is like a context manager:
 
     with some_runnable() as runnable:
-         do_some_things(runnable)
+         add_some_callbacks(runnable)
+         more_stuff_that_runs_on_start()
+
+         # Depending on the thread category, the `Runnable` isn't guaranteed to
+         # actually "go off" until the end of this block.
 
-         # At the end of the block, runnable.stop() is called and then
-         # the thread is joined
+    We're going to call the code inside the context manager `on_start`
 
     """
 
+    MASTER, NEW, EXTERNAL = 'M', 'N', 'X'
+    category = NEW
+
     timeout = 0.1
-    running = True
+    running = False
 
     def start(self):
         self.running = True
@@ -27,34 +56,24 @@ class Runnable:
     def stop(self):
         self.running = False
 
-    def join(self):
-        """Join this thread, or timeout after self.timeout seconds"""
-        pass
-
     def cleanup(self):
         pass
 
-    def is_alive(self):
-        return self.running
-
-    def __enter__(self):
-        self.start()
-        return self
+    def join(self, timeout=None):
+        """Join this thread, or timeout in `timeout` seconds"""
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.stop()
-        try:
-            self.join()
-        except Exception as e:
-            log.error('Unable to join thread %s: %s', self, e)
+    def wait(self, timeout=None):
+        """Wait until this runnable has started, or time out."""
 
+    def is_alive(self):
+        return self.running
 
-class RunnableTarget(Runnable):
     def run(self):
         try:
             self.target()
         except:
-            log.error('Exception at %s: \n%s', str(self), traceback.format_tb())
+            log.error('Exception at %s: \n%s',
+                      str(self), traceback.format_exc())
         finally:
             self.cleanup()
 
@@ -62,50 +81,53 @@ class RunnableTarget(Runnable):
         """The target code that is actually executed in the run method"""
         pass
 
+    @contextlib.contextmanager
+    def joiner(self, stop=True):
+        self.start()
+
+        try:
+            yield self
+        finally:
+            stop and self.stop()
+
+        while self.is_alive():
+            try:
+                self.join(self.timeout)
+            except Exception as e:
+                log.error('Unable to join thread %s: %s', self, e)
+                return
 
-class RunnableThread(threading.Thread, RunnableTarget):
+
+class RunnableThread(threading.Thread, Runnable):
     def __init__(self, daemon=True, **kwds):
         super().__init__(daemon=daemon, **kwds)
+        self.event = threading.Event()
 
     def start(self):
-        RunnableTarget.start(self)
+        Runnable.start(self)
         threading.Thread.start(self)
 
-    def run(self):
-        RunnableTarget.run(self)
+    def wait(self, timeout=None):
+        self.event.wait(timeout)
 
-    def join(self):
-        threading.Thread.join(self, self.timeout)
+    def run(self):
+        self.event.set()
+        Runnable.run(self)
 
 
 class Loop(RunnableThread):
-    def __init__(self, loop=None, **kwds):
+    def __init__(self, loop_once=None, **kwds):
         super().__init__(**kwds)
-        self.loop = loop or self.loop
+        self.loop_once = loop_once or self.loop_once
 
     def target(self):
         while self.running:
-            self.loop()
+            self.loop_once()
 
-    def loop(self):
+    def loop_once(self):
         raise NotImplementedError
 
 
-class Iterable(RunnableThread):
-    def __init__(self, iterator=None, **kwds):
-        super().__init__(**kwds)
-        self.iterator = iterator or self.iterator
-
-    def target(self):
-        for i in self.iterator():
-            self.receiver(i)
-            if not self.running:
-                break
-
-    def iterator(self):
-        pass
-
-
 class RunnableCollection(Runnable):
     def __init__(self, runnables):
         self.runnables = tuple(runnables)
@@ -118,9 +140,13 @@ class RunnableCollection(Runnable):
         for r in self.runnables:
             r.stop()
 
-    def join(self):
+    def join(self, timeout=None):
+        for r in self.runnables:
+            r.join(timeout)
+
+    def wait(self, timeout=None):
         for r in self.runnables:
-            r.join()
+            r.wait(timeout)
 
     def cleanup(self):
         for r in self.runnables:
@@ -137,7 +163,7 @@ class QueueHandler(Loop):
         self.queue = queue.Queue()
         self.send = send or self.send
 
-    def loop(self):
+    def loop_once(self):
         try:
             msg = self.queue.get(timeout=self.timeout)
         except queue.Empty:
diff --git a/bibliopixel/util/threads/task_thread.py b/bibliopixel/util/threads/task_thread.py
index 42960e6..3e2d4b3 100644
--- a/bibliopixel/util/threads/task_thread.py
+++ b/bibliopixel/util/threads/task_thread.py
@@ -25,5 +25,5 @@ class TaskThread(runnable.Loop):
     def produce(self):
         self.producer_task.run(self.consumer_task)
 
-    def loop(self):
+    def loop_once(self):
         self.consumer_task.run(self.producer_task)
diff --git a/bibliopixel/util/threads/update_threading.py b/bibliopixel/util/threads/update_threading.py
index c5a4053..43ee14c 100644
--- a/bibliopixel/util/threads/update_threading.py
+++ b/bibliopixel/util/threads/update_threading.py
@@ -26,7 +26,7 @@ class UpdateDriverThread(runnable.Loop):
     def sending(self):
         return self._wait.isSet()
 
-    def loop(self):
+    def loop_once(self):
         self._wait.wait()
         self._updating.clear()
         self._driver.update_colors()
@@ -59,7 +59,7 @@ class UpdateThread(runnable.Loop):
         self._reading.clear()
         self._wait.set()
 
-    def loop(self):
+    def loop_once(self):
         self._wait.wait()
         self._updated.wait()
 
@@ -71,7 +71,7 @@ class UpdateThread(runnable.Loop):
         self._reading.set()
 
 
-class NoThreading(object):
+class NoThreading(runnable.Runnable):
     """
     """
 
@@ -106,6 +106,12 @@ class UseThreading(NoThreading):
         while all([d._thread.sending() for d in self.layout.drivers]):
             time.sleep(0.000001)
 
+    def join(self, timeout=None):
+        self.update_thread.join(timeout)
+
+    def stop(self):
+        self.update_thread.stop()
+
 
 def UpdateThreading(enable, layout):
     """
diff --git a/bibliopixel/util/udp.py b/bibliopixel/util/udp.py
index ed77f49..98cc3d0 100644
--- a/bibliopixel/util/udp.py
+++ b/bibliopixel/util/udp.py
@@ -5,6 +5,7 @@ separate thread.
 
 import queue, socket
 from .. util.threads import runnable
+from .. util import log
 
 
 class Sender(runnable.Runnable):
@@ -62,24 +63,32 @@ class Receiver(runnable.Loop):
         self.address = address
         self.bufsize = bufsize
         self.receive = receive or self.receive
-
-    def run(self):
         self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.socket.bind(self.address)
-        self.socket.settimeout(self.timeout)
 
-        super().run()
+    def __str__(self):
+        return 'udp.Receiver(%s, %s)' % (self.address[0], hex(self.address[1]))
 
-    def loop(self):
+    def loop_once(self):
         try:
             data, addr = self.socket.recvfrom(self.bufsize)
-        except socket.timeout:
+        except OSError as e:
+            if e.errno != 9:
+                raise
+            if self.running:
+                log.error('Someone else closed the socket')
+                super().stop()
             return
+
         if data:
             self.receive(data)
 
-    def receive(self, msg):
-        pass
+    def stop(self):
+        super().stop()
+        try:
+            self.socket.close()
+        except Exception as e:
+            log.error('Exception in socket.close: %s', e)
 
 
 class QueuedReceiver(Receiver):
@@ -87,9 +96,9 @@ class QueuedReceiver(Receiver):
     Receive UDP messages in a thread and put them on a queue.
     """
 
-    def run(self):
+    def __init__(self, *args, **kwds):
         self.queue = queue.Queue()
-        super().run()
+        super().__init__(*args, **kwds)
 
     def receive(self, msg):
         self.queue.put(msg)
diff --git a/projects/01-matrix.yml b/projects/01-matrix.yml
index 9944e0d..85d097e 100644
--- a/projects/01-matrix.yml
+++ b/projects/01-matrix.yml
@@ -10,6 +10,7 @@ layout:
 run:
     fps: 30
     seconds: 14
+    threaded: true  # This is the only place we test threaded animations.
 
 aliases:
     bpa: BiblioPixelAnimations.matrix


Test patch:
diff --git a/test/bibliopixel/control/artnet_test.py b/test/bibliopixel/control/artnet_test.py
new file mode 100644
index 0000000..c635da9
--- /dev/null
+++ b/test/bibliopixel/control/artnet_test.py
@@ -0,0 +1,119 @@
+import os, time, queue, unittest
+from bibliopixel.control import artnet
+from bibliopixel.util import artnet_message, udp, json
+from .. project import make
+from .. import mark_tests
+from test.bibliopixel.util import udp_test
+
+
+class ArtNetControlTest(unittest.TestCase):
+    @mark_tests.fails_in_travis
+    def test_artnet_control(self):
+        project = make_project(ARTNET_CONTROL)
+        results = []
+        project.test_callback = results.append
+        project.controls[0].set_root(project)
+        sender = udp.QueuedSender(SEND_ADDRESS)
+
+        with project.controls[0].joiner(), sender.joiner():
+            sender.send(OUTGOING_MESSAGE)
+
+        self.assertTrue(results)
+        self.assertEquals(results[0], TEST_DATA)
+
+    @mark_tests.fails_in_travis
+    def test_artnet_integration(self):
+        sender = udp.QueuedSender(SEND_ADDRESS)
+        receiver = udp.QueuedReceiver(RECEIVE_ADDRESS)
+        project = make_project(ARTNET_PROJECT)
+
+        with sender.joiner(), receiver.joiner():
+            project.start()
+            time.sleep(0.01)
+            sender.queue.put(OUTGOING_MESSAGE)
+            project.join()
+
+        result = []
+        while True:
+            try:
+                result.append(receiver.queue.get(timeout=0.1))
+            except queue.Empty:
+                break
+
+        msgs = [artnet_message.bytes_to_message(i) for i in result]
+        data = [m.data for m in msgs]
+        self.assertEqual(len(msgs), 3)
+        self.assertTrue(not any(data[0]))
+        self.assertTrue(not any(data[2]))
+        actual = bytearray(data[1])
+
+        # We expect to lose the last two bytes because we have uneven sizes.
+        # TODO: perhaps we should round UP instead of DOWN?
+        self.assertEqual(actual[:-2], TEST_DATA[:-2])
+        self.assertEqual(actual[-2], 0)
+        self.assertEqual(actual[-1], 0)
+
+
+def make_project(datafile):
+    data = json.loads(datafile, '.yml')
+    return make.make_project(data)
+
+
+ROOT_DIR = os.path.dirname(__file__)
+TEST_DATA = bytearray(range(256)) + bytearray(range(256))
+OUTGOING_MESSAGE = artnet_message.dmx_message(data=TEST_DATA)
+PIXEL_COUNT = len(TEST_DATA) // 3
+
+IP_ADDRESS = '127.0.0.1'
+ARTNET_PORT = artnet_message.UDP_PORT
+ALTERNATE_PORT = ARTNET_PORT + 1
+SEND_ADDRESS = IP_ADDRESS, ARTNET_PORT
+RECEIVE_ADDRESS = IP_ADDRESS, ALTERNATE_PORT
+
+
+ARTNET_CONTROL = """
+shape: 1
+
+animation: animation
+
+controls:
+  typename: artnet
+  ip_address: {IP_ADDRESS}
+  pre_routing: ".test_callback()"
+  extractor:
+    omit: [type, net, subUni]
+  verbose: True
+
+run:
+  threaded: true
+""".format(**globals())
+
+
+ARTNET_PROJECT = """
+shape: {PIXEL_COUNT}
+
+animation: animation
+
+controls:
+  typename: artnet
+  ip_address: {IP_ADDRESS}
+  port: {ARTNET_PORT}
+
+  extractor:
+    omit: [net, subUni]
+
+  routing:
+    dmx: animation.color_list
+
+  verbose: True
+
+driver:
+  typename: artnet
+  ip_address: {IP_ADDRESS}
+  port: {ALTERNATE_PORT}
+
+run:
+  fps: 10
+  max_steps: 3
+  threaded: true
+""".format(**globals())
diff --git a/test/bibliopixel/control/extractor_test.py b/test/bibliopixel/control/extractor_test.py
index 5f5e8bf..7ccd90b 100644
--- a/test/bibliopixel/control/extractor_test.py
+++ b/test/bibliopixel/control/extractor_test.py
@@ -46,7 +46,7 @@ class ExtractorTest(unittest.TestCase):
     def test_accept(self):
         accept = {'channel': 2, 'type': 'control_change', 'control': 2}
         for msg in C3, C3_OFF, BC3, MOD, OTHER:
-            self.run_test(msg, None, accept=accept)
+            self.run_test(msg, collections.OrderedDict(), accept=accept)
 
         self.run_test(
             BC,
diff --git a/test/bibliopixel/control/midi_test.py b/test/bibliopixel/control/midi_test.py
index 2b80750..40d5e12 100644
--- a/test/bibliopixel/control/midi_test.py
+++ b/test/bibliopixel/control/midi_test.py
@@ -1,4 +1,4 @@
-import fractions, unittest
+import fractions, time, unittest
 from bibliopixel.control import midi
 from argparse import Namespace
 from test.bibliopixel import patch
@@ -24,7 +24,11 @@ class FakeMido:
         self.ports.MultiPort = multi_port
 
     def get_input_names(self):
-        return ['one', 'two']
+        class Port(str):
+            def name(self):
+                return self
+
+        return [Port('one'), Port('two')]
 
     def open_input(self, x):
         return x
@@ -48,8 +52,9 @@ class MidiTest(unittest.TestCase):
             root = Root()
             m = midi.Midi(routing=routing or self.routing, **kwds)
             m.set_root(root)
-            m.start()
-            m.thread.join()
+            with m.joiner(stop=True):
+                time.sleep(0.1)
+
             self.assertEqual(vars(root), expected)
 
     def test_one(self):
diff --git a/test/bibliopixel/drivers/artnet_test.py b/test/bibliopixel/drivers/artnet_test.py
index 6fbf0bc..c32f8d2 100644
--- a/test/bibliopixel/drivers/artnet_test.py
+++ b/test/bibliopixel/drivers/artnet_test.py
@@ -26,7 +26,7 @@ PROJECT = {
 
 class DMXMessageTest(unittest.TestCase):
     @mark_tests.long_test
-    @mark_tests.travis_test
+    @mark_tests.fails_in_travis
     def test_blackout(self):
         results = []
         with udp_test.receive_udp(ADDRESS, results, MAX_STEPS + 1):
diff --git a/test/bibliopixel/mark_tests.py b/test/bibliopixel/mark_tests.py
index 816923e..29f742d 100644
--- a/test/bibliopixel/mark_tests.py
+++ b/test/bibliopixel/mark_tests.py
@@ -10,8 +10,8 @@ with @long_test will not be run.
 
 SKIP_LONG_TESTS = os.getenv('SKIP_LONG_TESTS', '').lower().startswith('t')
 RUN_ALLPIXEL_TESTS = os.getenv('RUN_ALLPIXEL_TESTS', '').lower().startswith('t')
-TRAVIS_TESTS = os.getenv('TRAVIS', '').lower().startswith('t')
+FAILS_IN_TRAVIS = os.getenv('TRAVIS', '').lower().startswith('t')
 
 long_test = skipIf(SKIP_LONG_TESTS, 'Long tests skipped.')
 allpixel_test = skipIf(not RUN_ALLPIXEL_TESTS, 'AllPixel tests skipped.')
-travis_test = skipIf(TRAVIS_TESTS, 'Travis-flakey tests skipped')
+fails_in_travis = skipIf(FAILS_IN_TRAVIS, 'Travis-flakey tests skipped')
diff --git a/test/bibliopixel/util/udp_test.py b/test/bibliopixel/util/udp_test.py
index 0bdac84..90aec67 100644
--- a/test/bibliopixel/util/udp_test.py
+++ b/test/bibliopixel/util/udp_test.py
@@ -10,9 +10,8 @@ TIMEOUT = 0.1
 @contextlib.contextmanager
 def receive_udp(address, results, expected):
     receiver = udp.QueuedReceiver(address)
-    receiver.start()
-
-    yield
+    with receiver.joiner():
+        yield
 
     results.extend(receiver.queue.get() for i in range(expected))
     try:
@@ -24,7 +23,7 @@ def receive_udp(address, results, expected):
 
 
 class UDPTest(unittest.TestCase):
-    @mark_tests.travis_test
+    @mark_tests.fails_in_travis
     def test_full(self):
         messages = [s.encode() for s in ('foo', '', 'bar', 'baz', '', 'bing')]
         expected = [s for s in messages if s]
@@ -36,5 +35,6 @@ class UDPTest(unittest.TestCase):
             sender.start()
             for m in messages:
                 sender.send(m)
-
         self.assertEquals(actual, expected)
+        # TODO: if I don't have something here, this hangs!
+        time.sleep(0.001)