weakref.WeakValueDictionary (at least in Python 3, at least at this moment

in time, at least on this platform) is sometimes overly aggressive about
garbage collecting, including some things that are not garbage. Replace it
with a regular dict, and have RPCIOStream.rpc_input manage both adding and
removing queues.

1 files changed, 11 insertions, 7 deletions

diff --git a/cryptech_muxd b/cryptech_muxd
index d2fd54e..32fa36e 100755
--- a/cryptech_muxd
+++ b/cryptech_muxd
@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 #
 # Copyright (c) 2016-2017, NORDUnet A/S All rights reserved.
+# Copyright: 2020-2021, The Commons Conservancy Cryptech Project
+# SPDX-License-Identifier: BSD-3-Clause
 #
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are
@@ -12,9 +14,9 @@
 #   notice, this list of conditions and the following disclaimer in the
 #   documentation and/or other materials provided with the distribution.
 #
-# - Neither the name of the NORDUnet nor the names of its contributors may
-#   be used to endorse or promote products derived from this software
-#   without specific prior written permission.
+# - Neither the name of the copyright holder nor the names of its
+#   contributors may be used to endorse or promote products derived from
+#   this software without specific prior written permission.
 #
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
@@ -42,7 +44,6 @@ import sys
 import time
 import struct
 import atexit
-import weakref
 import logging
 import argparse
 import logging.handlers
@@ -175,7 +176,7 @@ class RPCIOStream(SerialIOStream):
 
     def __init__(self, device):
         super(RPCIOStream, self).__init__(device)
-        self.queues = weakref.WeakValueDictionary()
+        self.queues = dict()
         self.rpc_input_lock = tornado.locks.Lock()
 
     @tornado.gen.coroutine
@@ -184,6 +185,8 @@ class RPCIOStream(SerialIOStream):
         logger.debug("RPC send: %s", colon_hex(query))
         if queue is not None:
             self.queues[handle] = queue
+        else:
+            del self.queues[handle]
         with (yield self.rpc_input_lock.acquire()):
             yield self.write(query)
         logger.debug("RPC sent")
@@ -246,8 +249,9 @@ class RPCServer(PFUnixServer):
                 reply = yield queue.get()
                 if reply is None:
                     raise QueuedStreamClosedError()
-                logger.debug("RPC socket write, handle 0x%x", handle)
-                yield stream.write(SLIP_END + reply)
+                reply = SLIP_END + reply
+                logger.debug("RPC socket write, handle 0x%x: %s", handle, colon_hex(reply))
+                yield stream.write(reply)
             except tornado.iostream.StreamClosedError:
                 logger.info("RPC closing %r, handle 0x%x", stream, handle)
                 stream.close()