From 55c23e1c5732c82e4af7e576b0e8e4ef2b73e885 Mon Sep 17 00:00:00 2001
From: David Wilson <dw@botanicus.net>
Date: Tue, 13 Feb 2018 18:49:03 +0545
Subject: [PATCH] issue #68: replace sets with lists

Fix a MyPy warning by only passing lists to select.select(). At least on
Python 2.x, select.select() was internally converting the sets to lists
anyway.

By the time lists become inefficient here, it is likely that
select.select() itself will also be inefficient, and need replaced with
.poll() or similar.

No discernible performance different when transferring django.db.models
to a local VM.
---
 mitogen/core.py | 26 ++++++++++++++++++--------
 1 file changed, 18 insertions(+), 8 deletions(-)

diff --git a/mitogen/core.py b/mitogen/core.py
index 2a0b9035..31b8e9a2 100644
--- a/mitogen/core.py
+++ b/mitogen/core.py
@@ -1015,8 +1015,8 @@ class Broker(object):
     def __init__(self):
         self._alive = True
         self._queue = Queue.Queue()
-        self._readers = set()
-        self._writers = set()
+        self._readers = []
+        self._writers = []
         self._waker = Waker(self)
         self.start_receive(self._waker)
         self._thread = threading.Thread(
@@ -1033,23 +1033,33 @@ class Broker(object):
             self._queue.put((func, args, kwargs))
             self._waker.wake()
 
+    def _list_discard(self, lst, value):
+        try:
+            lst.remove(value)
+        except ValueError:
+            pass
+
+    def _list_add(self, lst, value):
+        if value not in lst:
+            lst.append(value)
+
     def start_receive(self, stream):
         IOLOG.debug('%r.start_receive(%r)', self, stream)
         assert stream.receive_side and stream.receive_side.fd is not None
-        self.defer(self._readers.add, stream.receive_side)
+        self.defer(self._list_add, self._readers, stream.receive_side)
 
     def stop_receive(self, stream):
         IOLOG.debug('%r.stop_receive(%r)', self, stream)
-        self.defer(self._readers.discard, stream.receive_side)
+        self.defer(self._list_discard, self._readers, stream.receive_side)
 
     def start_transmit(self, stream):
         IOLOG.debug('%r.start_transmit(%r)', self, stream)
         assert stream.transmit_side and stream.transmit_side.fd is not None
-        self.defer(self._writers.add, stream.transmit_side)
+        self.defer(self._list_add, self._writers, stream.transmit_side)
 
     def stop_transmit(self, stream):
         IOLOG.debug('%r.stop_transmit(%r)', self, stream)
-        self.defer(self._writers.discard, stream.transmit_side)
+        self.defer(self._list_discard, self._writers, stream.transmit_side)
 
     def _call(self, stream, func):
         try:
@@ -1096,7 +1106,7 @@ class Broker(object):
             self._run_defer()
             fire(self, 'shutdown')
 
-            for side in self._readers | self._writers:
+            for side in set(self._readers).union(self._writers):
                 self._call(side.stream, side.stream.on_shutdown)
 
             deadline = time.time() + self.shutdown_timeout
@@ -1109,7 +1119,7 @@ class Broker(object):
                           'more child processes still connected to '
                           'our stdout/stderr pipes.', self)
 
-            for side in self._readers | self._writers:
+            for side in set(self._readers).union(self._writers):
                 LOG.error('_broker_main() force disconnecting %r', side)
                 side.stream.on_disconnect(self)
         except Exception: