Merge branch 'develop'

README.rst

@@ -137,6 +137,12 @@ version prior to the release of 1.0.
 ChangeLog
 =========
 
+Version 0.10.9
+--------------
+
+* restore client to sessions output
+* cleanup shutdown process; hopefully this resolves the log spew for good
+
 Version 0.10.8
 --------------
 
@@ -226,46 +232,6 @@ variables to use roughly the same amount of memory.
 For now this code should be considered experimental; if you want
 stability please stick with the 0.9 series.
 
-Version 0.9.23
---------------
-
-* Backport of the fix for issue `#94#` - stale references to old
-  sessions.  This would effectively memory and network handles.
-
-Version 0.9.22
---------------
-
-* documentation updates (ARCHITECTURE.rst, ENVIRONMENT.rst) only.
-
-Version 0.9.21
---------------
-
-* moved RELEASE-NOTES into this README
-* document the RPC interface in docs/RPC-INTERFACE.rst
-* clean up open DB handling, issue `#89`_
-
-Version 0.9.20
---------------
-
-* fix for IRC flood issue `#93`_
-
-Version 0.9.19
---------------
-
-* move sleep outside semaphore (issue `#88`_)
-
-Version 0.9.18
---------------
-
-* last release of 2016.  Just a couple of minor tweaks to logging.
-
-Version 0.9.17
---------------
-
-* have all the DBs use fsync on write; hopefully means DB won't corrupt in
-  case of a kernel panic (issue `#75`_)
-* replace $DONATION_ADDRESS in banner file
-
 
 **Neil Booth**  kyuupichan@gmail.com  https://github.com/kyuupichan
 
@@ -273,11 +239,7 @@ Version 0.9.17
 
 
 .. _#72: https://github.com/kyuupichan/electrumx/issues/72
-.. _#75: https://github.com/kyuupichan/electrumx/issues/75
-.. _#88: https://github.com/kyuupichan/electrumx/issues/88
-.. _#89: https://github.com/kyuupichan/electrumx/issues/89
 .. _#92: https://github.com/kyuupichan/electrumx/issues/92
-.. _#93: https://github.com/kyuupichan/electrumx/issues/93
 .. _#94: https://github.com/kyuupichan/electrumx/issues/94
 .. _#99: https://github.com/kyuupichan/electrumx/issues/99
 .. _#100: https://github.com/kyuupichan/electrumx/issues/100

server/block_processor.py

@@ -196,11 +196,14 @@ class BlockProcessor(server.db.DB):
             task = await self.task_queue.get()
             await task()
 
-    def shutdown(self):
+    def shutdown(self, executor):
+        '''Shutdown cleanly and flush to disk.'''
+        # First stut down the executor; it may be processing a block.
+        # Then we can flush anything remaining to disk.
+        executor.shutdown()
         if self.height != self.db_height:
             self.logger.info('flushing state to DB for a clean shutdown...')
             self.flush(True)
-            self.logger.info('shutdown complete')
 
     async def executor(self, func, *args, **kwargs):
         '''Run func taking args in the executor.'''

server/controller.py

@@ -95,7 +95,7 @@ class Controller(util.LoggedClass):
              'block.get_header block.get_chunk estimatefee relayfee '
              'transaction.get transaction.get_merkle utxo.get_address'),
             ('server',
-             'banner donation_address peers.subscribe version'),
+             'banner donation_address peers.subscribe'),
         ]
         self.electrumx_handlers = {'.'.join([prefix, suffix]):
                                    getattr(self, suffix.replace('.', '_'))
@@ -225,7 +225,13 @@ class Controller(util.LoggedClass):
 
         # Perform a clean shutdown when this event is signalled.
         await self.shutdown_event.wait()
-        self.logger.info('shutting down gracefully')
+
+        self.logger.info('shutting down')
+        await self.shutdown(futures)
+        self.logger.info('shutdown complete')
+
+    async def shutdown(self, futures):
+        '''Perform the shutdown sequence.'''
         self.state = self.SHUTTING_DOWN
 
         # Close servers and sessions
@@ -237,11 +243,12 @@ class Controller(util.LoggedClass):
         for future in futures:
             future.cancel()
 
-        await asyncio.wait(futures)
+        # Wait for all futures to finish
+        while any(not future.done() for future in futures):
+            await asyncio.sleep(1)
 
-        # Wait for the executor to finish anything it's doing
+        # Finally shut down the block processor and executor
-        self.executor.shutdown()
+        self.bp.shutdown(self.executor)
-        self.bp.shutdown()
 
     def close_servers(self, kinds):
         '''Close the servers of the given kinds (TCP etc.).'''
@@ -253,18 +260,6 @@ class Controller(util.LoggedClass):
             if server:
                 server.close()
 
-    async def wait_for_sessions(self, secs=30):
-        if not self.sessions:
-            return
-        self.logger.info('waiting up to {:d} seconds for socket cleanup'
-                         .format(secs))
-        limit = time.time() + secs
-        while self.sessions and time.time() < limit:
-            self.clear_stale_sessions(grace=secs//2)
-            await asyncio.sleep(2)
-            self.logger.info('{:,d} sessions remaining'
-                             .format(len(self.sessions)))
-
     async def start_server(self, kind, *args, **kw_args):
         protocol_class = LocalRPC if kind == 'RPC' else ElectrumX
         protocol = partial(protocol_class, self, self.bp, self.env, kind)
@@ -886,15 +881,3 @@ class Controller(util.LoggedClass):
 
         Despite the name this is not currently treated as a subscription.'''
         return list(self.irc.peers.values())
-
-    async def version(self, client_name=None, protocol_version=None):
-        '''Returns the server version as a string.
-
-        client_name: a string identifying the client
-        protocol_version: the protocol version spoken by the client
-        '''
-        if client_name:
-            self.client = str(client_name)[:15]
-        if protocol_version is not None:
-            self.protocol_version = protocol_version
-        return VERSION

server/session.py

@@ -13,6 +13,7 @@ import traceback
 
 from lib.jsonrpc import JSONRPC, RPCError
 from server.daemon import DaemonError
+from server.version import VERSION
 
 
 class Session(JSONRPC):
@@ -123,6 +124,7 @@ class ElectrumX(Session):
             'blockchain.headers.subscribe': self.headers_subscribe,
             'blockchain.numblocks.subscribe': self.numblocks_subscribe,
             'blockchain.transaction.broadcast': self.transaction_broadcast,
+            'server.version': self.version,
         }
 
     def sub_count(self):
@@ -191,6 +193,18 @@ class ElectrumX(Session):
         self.hashX_subs[hashX] = address
         return status
 
+    async def version(self, client_name=None, protocol_version=None):
+        '''Returns the server version as a string.
+
+        client_name: a string identifying the client
+        protocol_version: the protocol version spoken by the client
+        '''
+        if client_name:
+            self.client = str(client_name)[:15]
+        if protocol_version is not None:
+            self.protocol_version = protocol_version
+        return VERSION
+
     async def transaction_broadcast(self, raw_tx):
         '''Broadcast a raw transaction to the network.
 

server/version.py

@@ -1 +1 @@
-VERSION = "ElectrumX 0.10.8"
+VERSION = "ElectrumX 0.10.9"