WIP: gracefully shutdown asyncio dispatcherasyncio-dispatcher-fixes

4 files changed, 82 insertions, 52 deletions

diff --git a/pysnmp/carrier/asyncio/dispatch.py b/pysnmp/carrier/asyncio/dispatch.py
index cb4c0941..4227d100 100644
--- a/pysnmp/carrier/asyncio/dispatch.py
+++ b/pysnmp/carrier/asyncio/dispatch.py
@@ -51,13 +51,14 @@ class AsyncioDispatcher(AbstractTransportDispatcher):
         self.__transportCount = 0
         if 'timeout' in kwargs:
             self.setTimerResolution(kwargs['timeout'])
-        self.loopingcall = None
+        self._futureTimer = None
 
     @asyncio.coroutine
-    def handle_timeout(self):
-        while True:
-            yield asyncio.From(asyncio.sleep(self.getTimerResolution()))
-            self.handleTimerTick(loop.time())
+    def fireTimer(self):
+        yield asyncio.From(asyncio.sleep(self.getTimerResolution()))
+        self.handleTimerTick(loop.time())
+        if self._futureTimer:
+            self._futureTimer = asyncio.async(self.fireTimer())
 
     def runDispatcher(self, timeout=0.0):
         if not loop.is_running():
@@ -69,8 +70,8 @@ class AsyncioDispatcher(AbstractTransportDispatcher):
                 raise PySnmpError(';'.join(traceback.format_exception(*sys.exc_info())))
 
     def registerTransport(self, tDomain, transport):
-        if self.loopingcall is None and self.getTimerResolution() > 0:
-            self.loopingcall = asyncio.async(self.handle_timeout())
+        if not self._futureTimer and self.getTimerResolution() > 0:
+            self._futureTimer = asyncio.async(self.fireTimer())
         AbstractTransportDispatcher.registerTransport(
             self, tDomain, transport
         )
@@ -83,18 +84,20 @@ class AsyncioDispatcher(AbstractTransportDispatcher):
             self.__transportCount -= 1
 
         # The last transport has been removed, stop the timeout
-        if self.__transportCount == 0 and not self.loopingcall.done():
-            self.loopingcall.cancel()
-            self.loopingcall = None
+        if self.__transportCount == 0:
+            if self._futureTimer:
+                self._futureTimer.cancel()
+                self._futureTimer = None
 
 
 # Trollius or Tulip?
 if not hasattr(asyncio, "From"):
-    exec ("""\
+    exec("""\
 @asyncio.coroutine
-def handle_timeout(self):
-    while True:
-        yield from asyncio.sleep(self.getTimerResolution())
-        self.handleTimerTick(loop.time())
-AsyncioDispatcher.handle_timeout = handle_timeout\
+def fireTimer(self):
+    yield from asyncio.sleep(self.getTimerResolution())
+    self.handleTimerTick(loop.time())
+    if self._futureTimer:
+        self._futureTimer = asyncio.async(self.fireTimer())
+AsyncioDispatcher.fireTimer = fireTimer\
 """)
diff --git a/pysnmp/hlapi/asyncio/cmdgen.py b/pysnmp/hlapi/asyncio/cmdgen.py
index d9a8d4d6..ea65961f 100644
--- a/pysnmp/hlapi/asyncio/cmdgen.py
+++ b/pysnmp/hlapi/asyncio/cmdgen.py
@@ -44,7 +44,7 @@ try:
 except ImportError:
     import trollius as asyncio
 
-__all__ = ['getCmd', 'nextCmd', 'setCmd', 'bulkCmd', 'isEndOfMib']
+__all__ = ['getCmd', 'nextCmd', 'setCmd', 'bulkCmd', 'isEndOfMib', 'unconfigureCmdGen']
 
 vbProcessor = CommandGeneratorVarBinds()
 lcd = CommandGeneratorLcdConfigurator()
@@ -53,6 +53,21 @@ isEndOfMib = lambda x: not cmdgen.getNextVarBinds(x)[1]
 
 
 @asyncio.coroutine
+def unconfigureCmdGen(snmpEngine, authData=None):
+    """Remove LCD configuration entry.
+
+    If `authData` is not given, all currently configured LCD entries will be
+    removed.
+
+    Note
+    ----
+    Configuration entry removal may have a side effect of removing unused transport
+    and shutting down unused transport dispatcher.
+    """
+    lcd.unconfigure(snmpEngine, authData)
+
+
+@asyncio.coroutine
 def getCmd(snmpEngine, authData, transportTarget, contextData,
            *varBinds, **options):
     """Creates a generator to perform SNMP GET query.
diff --git a/pysnmp/hlapi/asyncio/ntforg.py b/pysnmp/hlapi/asyncio/ntforg.py
index 13ec63bd..0541684e 100644
--- a/pysnmp/hlapi/asyncio/ntforg.py
+++ b/pysnmp/hlapi/asyncio/ntforg.py
@@ -21,13 +21,28 @@ try:
 except ImportError:
     import trollius as asyncio
 
-__all__ = ['sendNotification']
+__all__ = ['sendNotification', 'unconfigureNtfOrg']
 
 vbProcessor = NotificationOriginatorVarBinds()
 lcd = NotificationOriginatorLcdConfigurator()
 
 
 @asyncio.coroutine
+def unconfigureNtfOrg(snmpEngine, authData=None):
+    """Remove LCD configuration entry.
+
+    If `authData` is not given, all currently configured LCD entries will be
+    removed.
+
+    Note
+    ----
+    Configuration entry removal may have a side effect of removing unused transport
+    and shutting down unused transport dispatcher.
+    """
+    lcd.unconfigure(snmpEngine, authData)
+
+
+@asyncio.coroutine
 def sendNotification(snmpEngine, authData, transportTarget, contextData,
                      notifyType, varBinds, **options):
     """Creates a generator to send SNMP notification.
diff --git a/pysnmp/hlapi/lcd.py b/pysnmp/hlapi/lcd.py
index 91042645..c276eb3b 100644
--- a/pysnmp/hlapi/lcd.py
+++ b/pysnmp/hlapi/lcd.py
@@ -37,7 +37,8 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
     def configure(self, snmpEngine, authData, transportTarget, *options):
         cache = self._getCache(snmpEngine)
         if isinstance(authData, CommunityData):
-            if authData.communityIndex not in cache['auth']:
+            authDataKey = authData.communityIndex
+            if authDataKey not in cache['auth']:
                 config.addV1System(
                     snmpEngine,
                     authData.communityIndex,
@@ -66,21 +67,26 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
         paramsKey = (authData.securityName,
                      authData.securityLevel,
                      authData.mpModel)
+
         if paramsKey in cache['parm']:
             paramsName, useCount = cache['parm'][paramsKey]
-            cache['parm'][paramsKey] = paramsName, useCount + 1
+            useCount.add(authDataKey)
         else:
             paramsName = 'p%s' % self.nextID()
             config.addTargetParams(
                 snmpEngine, paramsName,
                 authData.securityName, authData.securityLevel, authData.mpModel
             )
-            cache['parm'][paramsKey] = paramsName, 1
+            cache['parm'][paramsKey] = paramsName, set([authDataKey])
+
+        transportKey = (paramsName, transportTarget.transportDomain,
+                        transportTarget.transportAddr,
+                        transportTarget.tagList)
 
         if transportTarget.transportDomain in cache['tran']:
             transport, useCount = cache['tran'][transportTarget.transportDomain]
             transportTarget.verifyDispatcherCompatibility(snmpEngine)
-            cache['tran'][transportTarget.transportDomain] = transport, useCount + 1
+            useCount.add(transportKey)
         elif config.getTransport(snmpEngine, transportTarget.transportDomain):
             transportTarget.verifyDispatcherCompatibility(snmpEngine)
         else:
@@ -90,15 +96,10 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
                 transportTarget.transportDomain,
                 transport
             )
-            cache['tran'][transportTarget.transportDomain] = transport, 1
-
-        transportKey = (paramsName, transportTarget.transportDomain,
-                        transportTarget.transportAddr,
-                        transportTarget.tagList)
+            cache['tran'][transportTarget.transportDomain] = transport, set([transportKey])
 
         if transportKey in cache['addr']:
-            addrName, useCount = cache['addr'][transportKey]
-            cache['addr'][transportKey] = addrName, useCount + 1
+            addrName = cache['addr'][transportKey]
         else:
             addrName = 'a%s' % self.nextID()
             config.addTargetAddr(
@@ -110,7 +111,7 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
                 transportTarget.retries,
                 transportTarget.tagList
             )
-            cache['addr'][transportKey] = addrName, 1
+            cache['addr'][transportKey] = addrName
 
         return addrName, paramsName
 
@@ -152,12 +153,12 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
             paramsKey = (authDataX.securityName,
                          authDataX.securityLevel,
                          authDataX.mpModel)
+
             if paramsKey in cache['parm']:
                 paramsName, useCount = cache['parm'][paramsKey]
-                useCount -= 1
-                if useCount:
-                    cache['parm'][paramsKey] = paramsName, useCount
-                else:
+                if authDataKey in useCount:
+                    useCount.remove(authDataKey)
+                if not useCount:
                     del cache['parm'][paramsKey]
                     config.delTargetParams(
                         snmpEngine, paramsName
@@ -169,24 +170,20 @@ class CommandGeneratorLcdConfigurator(AbstractLcdConfigurator):
             addrKeys = [x for x in cache['addr'] if x[0] == paramsName]
 
             for addrKey in addrKeys:
-                addrName, useCount = cache['addr'][addrKey]
-                useCount -= 1
-                if useCount:
-                    cache['addr'][addrKey] = addrName, useCount
-                else:
-                    config.delTargetAddr(snmpEngine, addrName)
-
-                    addrNames.add(addrKey)
-
-                    if addrKey[1] in cache['tran']:
-                        transport, useCount = cache['tran'][addrKey[1]]
-                        if useCount > 1:
-                            useCount -= 1
-                            cache['tran'][addrKey[1]] = transport, useCount
-                        else:
-                            config.delTransport(snmpEngine, addrKey[1])
-                            transport.closeTransport()
-                            del cache['tran'][addrKey[1]]
+                addrName = cache['addr'][addrKey]
+
+                config.delTargetAddr(snmpEngine, addrName)
+
+                addrNames.add(addrKey)
+
+                if addrKey[1] in cache['tran']:
+                    transport, useCount = cache['tran'][addrKey[1]]
+                    if addrKey in useCount:
+                        useCount.remove(addrKey)
+                    if not useCount:
+                        config.delTransport(snmpEngine, addrKey[1])
+                        transport.closeTransport()
+                        del cache['tran'][addrKey[1]]
 
         return addrNames, paramsNames