Merge branch 'jsonrpc' into develop

docs/ARCHITECTURE.rst

@@ -36,7 +36,7 @@ Not started until the Block Processor has caught up with bitcoind.
 Daemon
 ------
 
-Encapsulates the RPC wire protcol with bitcoind for the whole server.
+Encapsulates the RPC wire protocol with bitcoind for the whole server.
 Transparently handles temporary bitcoind connection errors, and fails
 over if necessary.
 

docs/ENVIRONMENT.rst

@@ -205,7 +205,8 @@ below are low and encourage you to raise them.
   An integer number of seconds defaulting to 600.  Sessions with no
   activity for longer than this are disconnected.  Properly
   functioning Electrum clients by default will send pings roughly
-  every 60 seconds.
+  every 60 seconds, and servers doing peer discovery roughly every 300
+  seconds.
 
 IRC
 ---

electrumx_rpc.py

@@ -16,44 +16,60 @@ import json
 from functools import partial
 from os import environ
 
-from lib.jsonrpc import JSONRPC
+from lib.jsonrpc import JSONSession, JSONRPCv2
 from server.controller import Controller
 
 
-class RPCClient(JSONRPC):
+class RPCClient(JSONSession):
 
     def __init__(self):
-        super().__init__()
+        super().__init__(version=JSONRPCv2)
-        self.queue = asyncio.Queue()
+        self.max_send = 0
-        self.max_send = 1000000
+        self.max_buffer_size = 5*10**6
+        self.event = asyncio.Event()
 
-    def enqueue_request(self, request):
+    def have_pending_items(self):
-        self.queue.put_nowait(request)
+        self.event.set()
 
-    async def send_and_wait(self, method, params, timeout=None):
+    async def wait_for_response(self):
-        # Raise incoming buffer size - presumably connection is trusted
+        await self.event.wait()
-        self.max_buffer_size = 5000000
+        await self.process_pending_items()
-        if params:
-            params = [params]
-        self.send_request(method, method, params)
 
-        future = asyncio.ensure_future(self.queue.get())
+    def send_rpc_request(self, method, params):
-        for f in asyncio.as_completed([future], timeout=timeout):
+        handler = partial(self.handle_response, method)
-            try:
+        self.send_request(handler, method, params)
-                request = await f
+
-            except asyncio.TimeoutError:
+    def handle_response(self, method, result, error):
-                future.cancel()
+        if method in ('groups', 'sessions') and not error:
-                print('request timed out after {}s'.format(timeout))
+            if method == 'groups':
+                lines = Controller.groups_text_lines(result)
             else:
-                await request.process(self)
+                lines = Controller.sessions_text_lines(result)
-
+            for line in lines:
-    async def handle_response(self, result, error, method):
-        if result and method in ('groups', 'sessions'):
-            for line in Controller.text_lines(method, result):
                 print(line)
+        elif error:
+            print('error: {} (code {:d})'
+                  .format(error['message'], error['code']))
         else:
-            value = {'error': error} if error else result
+            print(json.dumps(result, indent=4, sort_keys=True))
-            print(json.dumps(value, indent=4, sort_keys=True))
+
+
+def rpc_send_and_wait(port, method, params, timeout=15):
+    loop = asyncio.get_event_loop()
+    coro = loop.create_connection(RPCClient, 'localhost', port)
+    try:
+        transport, rpc_client = loop.run_until_complete(coro)
+        rpc_client.send_rpc_request(method, params)
+        try:
+            coro = rpc_client.wait_for_response()
+            loop.run_until_complete(asyncio.wait_for(coro, timeout))
+        except asyncio.TimeoutError:
+            print('request timed out after {}s'.format(timeout))
+    except OSError:
+        print('cannot connect - is ElectrumX catching up, not running, or '
+              'is {:d} the wrong RPC port?'.format(port))
+    finally:
+        loop.close()
 
 
 def main():
@@ -67,20 +83,17 @@ def main():
                         help='params to send')
     args = parser.parse_args()
 
-    if args.port is None:
+    port = args.port
-        args.port = int(environ.get('RPC_PORT', 8000))
+    if port is None:
+        port = int(environ.get('RPC_PORT', 8000))
 
-    loop = asyncio.get_event_loop()
+    # Get the RPC request.
-    coro = loop.create_connection(RPCClient, 'localhost', args.port)
+    method = args.command[0]
-    try:
+    params = args.param
-        transport, protocol = loop.run_until_complete(coro)
+    if method in ('log', 'disconnect'):
-        coro = protocol.send_and_wait(args.command[0], args.param, timeout=15)
+        params = [params]
-        loop.run_until_complete(coro)
+
-    except OSError:
+    rpc_send_and_wait(port, method, params)
-        print('error connecting - is ElectrumX catching up or not running?')
-    finally:
-        loop.stop()
-        loop.close()
 
 
 if __name__ == '__main__':

server/controller.py

@@ -18,14 +18,14 @@ from functools import partial
 
 import pylru
 
-from lib.jsonrpc import JSONRPC, RPCError, RequestBase
+from lib.jsonrpc import JSONRPC, RPCError
 from lib.hash import sha256, double_sha256, hash_to_str, hex_str_to_hash
 import lib.util as util
 from server.block_processor import BlockProcessor
 from server.daemon import Daemon, DaemonError
-from server.session import LocalRPC, ElectrumX
-from server.peers import PeerManager
 from server.mempool import MemPool
+from server.peers import PeerManager
+from server.session import LocalRPC, ElectrumX
 from server.version import VERSION
 
 
@@ -39,16 +39,6 @@ class Controller(util.LoggedClass):
     BANDS = 5
     CATCHING_UP, LISTENING, PAUSED, SHUTTING_DOWN = range(4)
 
-    class NotificationRequest(RequestBase):
-        def __init__(self, height, touched):
-            super().__init__(1)
-            self.height = height
-            self.touched = touched
-
-        async def process(self, session):
-            self.remaining = 0
-            await session.notify(self.height, self.touched)
-
     def __init__(self, env):
         super().__init__()
         # Set this event to cleanly shutdown
@@ -56,7 +46,7 @@ class Controller(util.LoggedClass):
         self.loop = asyncio.get_event_loop()
         self.executor = ThreadPoolExecutor()
         self.loop.set_default_executor(self.executor)
-        self.start = time.time()
+        self.start_time = time.time()
         self.coin = env.coin
         self.daemon = Daemon(env.coin.daemon_urls(env.daemon_url))
         self.bp = BlockProcessor(env, self.daemon)
@@ -141,9 +131,9 @@ class Controller(util.LoggedClass):
         if isinstance(session, LocalRPC):
             return 0
         gid = self.sessions[session]
-        group_bandwidth = sum(s.bandwidth_used for s in self.groups[gid])
+        group_bw = sum(session.bw_used for session in self.groups[gid])
-        return 1 + (bisect_left(self.bands, session.bandwidth_used)
+        return 1 + (bisect_left(self.bands, session.bw_used)
-                    + bisect_left(self.bands, group_bandwidth)) // 2
+                    + bisect_left(self.bands, group_bw)) // 2
 
     def is_deprioritized(self, session):
         return self.session_priority(session) > self.BANDS
@@ -166,6 +156,15 @@ class Controller(util.LoggedClass):
                     and self.state == self.PAUSED):
                 await self.start_external_servers()
 
+            # Periodically log sessions
+            if self.env.log_sessions and time.time() > self.next_log_sessions:
+                if self.next_log_sessions:
+                    data = self.session_data(for_log=True)
+                    for line in Controller.sessions_text_lines(data):
+                        self.logger.info(line)
+                    self.logger.info(json.dumps(self.server_summary()))
+                self.next_log_sessions = time.time() + self.env.log_sessions
+
             await asyncio.sleep(1)
 
     def enqueue_session(self, session):
@@ -195,7 +194,10 @@ class Controller(util.LoggedClass):
         while True:
             priority_, id_, session = await self.queue.get()
             if session in self.sessions:
-                await session.serve_requests()
+                await session.process_pending_items()
+                # Re-enqueue the session if stuff is left
+                if session.items:
+                    self.enqueue_session(session)
 
     def initiate_shutdown(self):
         '''Call this function to start the shutdown process.'''
@@ -265,8 +267,8 @@ class Controller(util.LoggedClass):
 
     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)
+        protocol_factory = partial(protocol_class, self, kind)
-        server = self.loop.create_server(protocol, *args, **kw_args)
+        server = self.loop.create_server(protocol_factory, *args, **kw_args)
 
         host, port = args[:2]
         try:
@@ -329,17 +331,7 @@ class Controller(util.LoggedClass):
 
             for session in self.sessions:
                 if isinstance(session, ElectrumX):
-                    request = self.NotificationRequest(self.bp.db_height,
+                    await session.notify(self.bp.db_height, touched)
-                                                       touched)
-                    session.enqueue_request(request)
-            # Periodically log sessions
-            if self.env.log_sessions and time.time() > self.next_log_sessions:
-                if self.next_log_sessions:
-                    data = self.session_data(for_log=True)
-                    for line in Controller.sessions_text_lines(data):
-                        self.logger.info(line)
-                    self.logger.info(json.dumps(self.server_summary()))
-                self.next_log_sessions = time.time() + self.env.log_sessions
 
     def electrum_header(self, height):
         '''Return the binary header at the given height.'''
@@ -357,7 +349,7 @@ class Controller(util.LoggedClass):
         if now > self.next_stale_check:
             self.next_stale_check = now + 300
             self.clear_stale_sessions()
-        gid = int(session.start - self.start) // 900
+        gid = int(session.start_time - self.start_time) // 900
         self.groups[gid].append(session)
         self.sessions[session] = gid
         session.log_info('{} {}, {:,d} total'
@@ -381,12 +373,12 @@ class Controller(util.LoggedClass):
     def close_session(self, session):
         '''Close the session's transport and cancel its future.'''
         session.close_connection()
-        return 'disconnected {:d}'.format(session.id_)
+        return 'disconnected {:d}'.format(session.session_id)
 
     def toggle_logging(self, session):
         '''Toggle logging of the session.'''
         session.log_me = not session.log_me
-        return 'log {:d}: {}'.format(session.id_, session.log_me)
+        return 'log {:d}: {}'.format(session.session_id, session.log_me)
 
     def clear_stale_sessions(self, grace=15):
         '''Cut off sessions that haven't done anything for 10 minutes.  Force
@@ -400,17 +392,17 @@ class Controller(util.LoggedClass):
         stale = []
         for session in self.sessions:
             if session.is_closing():
-                if session.stop <= shutdown_cutoff:
+                if session.close_time <= shutdown_cutoff:
-                    session.transport.abort()
+                    session.abort()
             elif session.last_recv < stale_cutoff:
                 self.close_session(session)
-                stale.append(session.id_)
+                stale.append(session.session_id)
         if stale:
             self.logger.info('closing stale connections {}'.format(stale))
 
         # Consolidate small groups
         gids = [gid for gid, l in self.groups.items() if len(l) <= 4
-                and sum(session.bandwidth_used for session in l) < 10000]
+                and sum(session.bw_used for session in l) < 10000]
         if len(gids) > 1:
             sessions = sum([self.groups[gid] for gid in gids], [])
             new_gid = max(gids)
@@ -436,7 +428,7 @@ class Controller(util.LoggedClass):
             'paused': sum(s.pause for s in self.sessions),
             'pid': os.getpid(),
             'peers': self.peers.count(),
-            'requests': sum(s.requests_remaining() for s in self.sessions),
+            'requests': sum(s.count_pending_items() for s in self.sessions),
             'sessions': self.session_count(),
             'subs': self.sub_count(),
             'txs_sent': self.txs_sent,
@@ -445,13 +437,6 @@ class Controller(util.LoggedClass):
     def sub_count(self):
         return sum(s.sub_count() for s in self.sessions)
 
-    @staticmethod
-    def text_lines(method, data):
-        if method == 'sessions':
-            return Controller.sessions_text_lines(data)
-        else:
-            return Controller.groups_text_lines(data)
-
     @staticmethod
     def groups_text_lines(data):
         '''A generator returning lines for a list of groups.
@@ -482,8 +467,8 @@ class Controller(util.LoggedClass):
             sessions = self.groups[gid]
             result.append([gid,
                            len(sessions),
-                           sum(s.bandwidth_used for s in sessions),
+                           sum(s.bw_used for s in sessions),
-                           sum(s.requests_remaining() for s in sessions),
+                           sum(s.count_pending_items() for s in sessions),
                            sum(s.txs_sent for s in sessions),
                            sum(s.sub_count() for s in sessions),
                            sum(s.recv_count for s in sessions),
@@ -523,17 +508,17 @@ class Controller(util.LoggedClass):
     def session_data(self, for_log):
         '''Returned to the RPC 'sessions' call.'''
         now = time.time()
-        sessions = sorted(self.sessions, key=lambda s: s.start)
+        sessions = sorted(self.sessions, key=lambda s: s.start_time)
-        return [(session.id_,
+        return [(session.session_id,
                  session.flags(),
                  session.peername(for_log=for_log),
                  session.client,
-                 session.requests_remaining(),
+                 session.count_pending_items(),
                  session.txs_sent,
                  session.sub_count(),
                  session.recv_count, session.recv_size,
                  session.send_count, session.send_size,
-                 now - session.start)
+                 now - session.start_time)
                 for session in sessions]
 
     def lookup_session(self, session_id):
@@ -543,7 +528,7 @@ class Controller(util.LoggedClass):
             pass
         else:
             for session in self.sessions:
-                if session.id_ == session_id:
+                if session.session_id == session_id:
                     return session
         return None
 
@@ -562,42 +547,42 @@ class Controller(util.LoggedClass):
 
     # Local RPC command handlers
 
-    async def rpc_disconnect(self, session_ids):
+    def rpc_disconnect(self, session_ids):
         '''Disconnect sesssions.
 
         session_ids: array of session IDs
         '''
         return self.for_each_session(session_ids, self.close_session)
 
-    async def rpc_log(self, session_ids):
+    def rpc_log(self, session_ids):
         '''Toggle logging of sesssions.
 
         session_ids: array of session IDs
         '''
         return self.for_each_session(session_ids, self.toggle_logging)
 
-    async def rpc_stop(self):
+    def rpc_stop(self):
         '''Shut down the server cleanly.'''
         self.initiate_shutdown()
         return 'stopping'
 
-    async def rpc_getinfo(self):
+    def rpc_getinfo(self):
         '''Return summary information about the server process.'''
         return self.server_summary()
 
-    async def rpc_groups(self):
+    def rpc_groups(self):
         '''Return statistics about the session groups.'''
         return self.group_data()
 
-    async def rpc_sessions(self):
+    def rpc_sessions(self):
         '''Return statistics about connected sessions.'''
         return self.session_data(for_log=False)
 
-    async def rpc_peers(self):
+    def rpc_peers(self):
         '''Return a list of server peers, currently taken from IRC.'''
         return self.peers.peer_list()
 
-    async def rpc_reorg(self, count=3):
+    def rpc_reorg(self, count=3):
         '''Force a reorg of the given number of blocks.
 
         count: number of blocks to reorg (default 3)
@@ -779,14 +764,14 @@ class Controller(util.LoggedClass):
                  'height': utxo.height, 'value': utxo.value}
                 for utxo in sorted(await self.get_utxos(hashX))]
 
-    async def block_get_chunk(self, index):
+    def block_get_chunk(self, index):
         '''Return a chunk of block headers.
 
         index: the chunk index'''
         index = self.non_negative_integer(index)
         return self.get_chunk(index)
 
-    async def block_get_header(self, height):
+    def block_get_header(self, height):
         '''The deserialized header at a given height.
 
         height: the header's height'''
@@ -879,6 +864,6 @@ class Controller(util.LoggedClass):
 
         return banner
 
-    async def donation_address(self):
+    def donation_address(self):
         '''Return the donation address as a string, empty if there is none.'''
         return self.env.donation_address

server/peers.py

@@ -132,7 +132,7 @@ class PeerManager(util.LoggedClass):
     def peer_list(self):
         return self.irc_peers
 
-    async def subscribe(self):
+    def subscribe(self):
         '''Returns the server peers as a list of (ip, host, details) tuples.
 
         Despite the name this is not currently treated as a subscription.'''

server/session.py

@@ -9,39 +9,61 @@
 
 
 import asyncio
+import time
 import traceback
+from functools import partial
 
-from lib.jsonrpc import JSONRPC, RPCError
+from lib.jsonrpc import JSONSession, RPCError
 from server.daemon import DaemonError
 from server.version import VERSION
 
 
-class Session(JSONRPC):
+class SessionBase(JSONSession):
-    '''Base class of ElectrumX JSON session protocols.
+    '''Base class of ElectrumX JSON sessions.
 
     Each session runs its tasks in asynchronous parallelism with other
-    sessions.  To prevent some sessions blocking others, potentially
+    sessions.
-    long-running requests should yield.
     '''
 
-    def __init__(self, controller, bp, env, kind):
+    def __init__(self, controller, kind):
         super().__init__()
+        self.kind = kind  # 'RPC', 'TCP' etc.
         self.controller = controller
-        self.bp = bp
+        self.bp = controller.bp
-        self.env = env
+        self.env = controller.env
-        self.daemon = bp.daemon
+        self.daemon = self.bp.daemon
-        self.kind = kind
         self.client = 'unknown'
-        self.anon_logs = env.anon_logs
+        self.anon_logs = self.env.anon_logs
-        self.max_send = env.max_send
-        self.bandwidth_limit = env.bandwidth_limit
         self.last_delay = 0
         self.txs_sent = 0
         self.requests = []
+        self.start_time = time.time()
+        self.close_time = 0
+        self.bw_time = self.start_time
+        self.bw_interval = 3600
+        self.bw_used = 0
 
-    def is_closing(self):
+    def have_pending_items(self):
-        '''True if this session is closing.'''
+        '''Called each time the pending item queue goes from empty to having
-        return self.transport and self.transport.is_closing()
+        one item.'''
+        self.controller.enqueue_session(self)
+
+    def close_connection(self):
+        '''Call this to close the connection.'''
+        self.close_time = time.time()
+        super().close_connection()
+
+    def peername(self, *, for_log=True):
+        '''Return the peer name of this connection.'''
+        peer_info = self.peer_info()
+        if not peer_info:
+            return 'unknown'
+        if for_log and self.anon_logs:
+            return 'xx.xx.xx.xx:xx'
+        if ':' in peer_info[0]:
+            return '[{}]:{}'.format(peer_info[0], peer_info[1])
+        else:
+            return '{}:{}'.format(peer_info[0], peer_info[1])
 
     def flags(self):
         '''Status flags.'''
@@ -53,42 +75,6 @@ class Session(JSONRPC):
         status += str(self.controller.session_priority(self))
         return status
 
-    def requests_remaining(self):
-        return sum(request.remaining for request in self.requests)
-
-    def enqueue_request(self, request):
-        '''Add a request to the session's list.'''
-        self.requests.append(request)
-        if len(self.requests) == 1:
-            self.controller.enqueue_session(self)
-
-    async def serve_requests(self):
-        '''Serve requests in batches.'''
-        total = 0
-        errs = []
-        # Process 8 items at a time
-        for request in self.requests:
-            try:
-                initial = request.remaining
-                await request.process(self)
-                total += initial - request.remaining
-            except asyncio.CancelledError:
-                raise
-            except Exception:
-                # Should probably be considered a bug and fixed
-                self.log_error('error handling request {}'.format(request))
-                traceback.print_exc()
-                errs.append(request)
-            await asyncio.sleep(0)
-            if total >= 8:
-                break
-
-        # Remove completed requests and re-enqueue ourself if any remain.
-        self.requests = [req for req in self.requests
-                         if req.remaining and not req in errs]
-        if self.requests:
-            self.controller.enqueue_session(self)
-
     def connection_made(self, transport):
         '''Handle an incoming client connection.'''
         super().connection_made(transport)
@@ -96,27 +82,32 @@ class Session(JSONRPC):
 
     def connection_lost(self, exc):
         '''Handle client disconnection.'''
-        super().connection_lost(exc)
+        msg = ''
-        if (self.pause or self.controller.is_deprioritized(self)
+        if self.pause:
-                 or self.send_size >= 1024*1024 or self.error_count):
+            msg += ' whilst paused'
-            self.log_info('disconnected.  Sent {:,d} bytes in {:,d} messages '
+        if self.controller.is_deprioritized(self):
-                          '{:,d} errors'
+            msg += ' whilst deprioritized'
-                          .format(self.send_size, self.send_count,
+        if self.send_size >= 1024*1024:
-                                  self.error_count))
+            msg += ('.  Sent {:,d} bytes in {:,d} messages'
+                    .format(self.send_size, self.send_count))
+        if msg:
+            msg = 'disconnected' + msg
+            self.log_info(msg)
         self.controller.remove_session(self)
 
     def sub_count(self):
         return 0
 
 
-class ElectrumX(Session):
+class ElectrumX(SessionBase):
     '''A TCP server that handles incoming Electrum connections.'''
 
-    def __init__(self, *args):
+    def __init__(self, *args, **kwargs):
-        super().__init__(*args)
+        super().__init__(*args, **kwargs)
         self.subscribe_headers = False
         self.subscribe_height = False
         self.notified_height = None
+        self.max_send = self.env.max_send
         self.max_subs = self.env.max_session_subs
         self.hashX_subs = {}
         self.electrumx_handlers = {
@@ -124,7 +115,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,
+            'server.version': self.server_version,
         }
 
     def sub_count(self):
@@ -167,12 +158,12 @@ class ElectrumX(Session):
         '''Used as response to a headers subscription request.'''
         return self.controller.electrum_header(self.height())
 
-    async def headers_subscribe(self):
+    def headers_subscribe(self):
         '''Subscribe to get headers of new blocks.'''
         self.subscribe_headers = True
         return self.current_electrum_header()
 
-    async def numblocks_subscribe(self):
+    def numblocks_subscribe(self):
         '''Subscribe to get height of new blocks.'''
         self.subscribe_height = True
         return self.height()
@@ -190,7 +181,7 @@ class ElectrumX(Session):
         self.hashX_subs[hashX] = address
         return status
 
-    async def version(self, client_name=None, protocol_version=None):
+    def server_version(self, client_name=None, protocol_version=None):
         '''Returns the server version as a string.
 
         client_name: a string identifying the client
@@ -241,13 +232,13 @@ class ElectrumX(Session):
         return handler
 
 
-class LocalRPC(Session):
+class LocalRPC(SessionBase):
-    '''A local TCP RPC server for querying status.'''
+    '''A local TCP RPC server session.'''
 
-    def __init__(self, *args):
+    def __init__(self, *args, **kwargs):
-        super().__init__(*args)
+        super().__init__(*args, **kwargs)
         self.client = 'RPC'
-        self.max_send = 5000000
+        self.max_send = 0
 
     def request_handler(self, method):
         '''Return the async handler for the given request method.'''