Page MenuHomePhabricator
Files F13115460

D14947.diff
No OneTemporary
Actions

D14947.diff
View Options

diff --git a/electrum/electrumabc/interface.py b/electrum/electrumabc/interface.py
--- a/electrum/electrumabc/interface.py
+++ b/electrum/electrumabc/interface.py
@@ -23,6 +23,8 @@
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
+from __future__ import annotations
+
import os
import re
import socket
@@ -31,8 +33,8 @@
import threading
import time
import traceback
-from collections import namedtuple
-from typing import Optional, Tuple
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, NamedTuple, Optional, Tuple
import requests
from pathvalidate import sanitize_filename
@@ -40,20 +42,33 @@
from . import pem, util, x509
from .json_util import JSONSocketPipe
from .printerror import PrintError, is_verbose, print_error, print_msg
+from .simple_config import SimpleConfig
from .utils import Event
+if TYPE_CHECKING:
+ from queue import Queue
+
+ Request = Tuple[str, List[Any], int]
+
ca_path = requests.certs.where()
PING_INTERVAL = 300
-def Connection(server, queue, config_path, callback=None):
+def Connection(
+ server: str,
+ queue: Queue,
+ config_path: str,
+ callback: Optional[Callable[[TcpConnection], None]] = None,
+) -> TcpConnection:
"""Makes asynchronous connections to a remote electrum server.
Returns the running thread that is making the connection.
Once the thread has connected, it finishes, placing a tuple on the
queue of the form (server, socket), where socket is None if
connection failed.
+
+ server is a "<host>:<port>:<protocol>" string
"""
host, port, protocol = server.rsplit(":", 2)
if protocol not in "st":
@@ -70,13 +85,13 @@
def __init__(self, server, queue, config_path):
threading.Thread.__init__(self)
- self.config_path = config_path
- self.queue = queue
- self.server = server
- self.host, self.port, self.protocol = self.server.rsplit(":", 2)
- self.host = str(self.host)
- self.port = int(self.port)
- self.use_ssl = self.protocol == "s"
+ self.config_path: str = config_path
+ self.queue: Queue = queue
+ self.server: str = server
+ host, port, protocol = self.server.rsplit(":", 2)
+ self.host: str = host
+ self.port = int(port)
+ self.use_ssl = protocol == "s"
self.daemon = True
def diagnostic_name(self):
@@ -331,27 +346,39 @@
- Member variable server.
"""
- MODE_DEFAULT = "default"
- MODE_BACKWARD = "backward"
- MODE_BINARY = "binary"
- MODE_CATCH_UP = "catch_up"
- MODE_VERIFICATION = "verification"
-
- def __init__(self, server, socket, *, max_message_bytes=0, config=None):
+ class Mode(Enum):
+ DEFAULT = "default"
+ BACKWARD = "backward"
+ BINARY = "binary"
+ CATCH_UP = "catch_up"
+ VERIFICATION = "verification"
+
+ def __init__(
+ self,
+ server: str,
+ socket: ssl.SSLSocket,
+ *,
+ max_message_bytes: int = 0,
+ config: Optional[SimpleConfig] = None,
+ ):
self.server = server
self.config = config
- self.host, self.port, _ = server.rsplit(":", 2)
+ host, port, _ = server.rsplit(":", 2)
+ self.host: str = host
+ self.port: str = port
self.socket = socket
self.pipe = JSONSocketPipe(socket, max_message_bytes=max_message_bytes)
# Dump network messages. Set at runtime from the console.
self.debug = False
self.request_time = time.time()
- self.unsent_requests = []
- self.unanswered_requests = {}
+ self.unsent_requests: List[Request] = []
+ """[(method, params, id), ...]"""
+ self.unanswered_requests: Dict[int, Request] = {}
+ """{id: (method, params, id), ...}"""
self.last_send = time.time()
- self.mode = None
+ self.mode: Optional[Interface.Mode] = None
def __repr__(self):
return "<{}.{} {}>".format(__name__, type(self).__name__, self.format_address())
@@ -359,7 +386,7 @@
def format_address(self):
return "{}:{}".format(self.host, self.port)
- def set_mode(self, mode):
+ def set_mode(self, mode: Mode):
self.print_error("set_mode({})".format(mode))
self.mode = mode
@@ -380,26 +407,32 @@
except Exception:
pass
- def queue_request(self, *args): # method, params, _id
+ def queue_request(self, method: str, params: List[Any], id_: int):
"""Queue a request, later to be sent with send_requests when the
socket is available for writing."""
self.request_time = time.time()
- self.unsent_requests.append(args)
+ self.unsent_requests.append((method, params, id_))
+
+ class ReqThrottleParams(NamedTuple):
+ max: int
+ chunkSize: int
- ReqThrottleParams = namedtuple("ReqThrottleParams", "max chunkSize")
req_throttle_default = ReqThrottleParams(2000, 100)
@classmethod
- def get_req_throttle_params(cls, config):
+ def get_req_throttle_params(cls, config: Optional[SimpleConfig]):
tup = config and config.get("network_unanswered_requests_throttle")
if not isinstance(tup, (list, tuple)) or len(tup) != 2:
- tup = cls.req_throttle_default
+ return cls.req_throttle_default
tup = cls.ReqThrottleParams(*tup)
return tup
@classmethod
def set_req_throttle_params(
- cls, config, max_unanswered_requests=None, chunkSize=None
+ cls,
+ config: Optional[SimpleConfig],
+ max_unanswered_requests: Optional[int] = None,
+ chunkSize: Optional[int] = None,
):
if not config:
return
@@ -557,8 +590,6 @@
def test_certificates():
- from .simple_config import SimpleConfig
-
config = SimpleConfig()
mydir = os.path.join(config.path, "certs")
certs = os.listdir(mydir)
diff --git a/electrum/electrumabc/network.py b/electrum/electrumabc/network.py
--- a/electrum/electrumabc/network.py
+++ b/electrum/electrumabc/network.py
@@ -1160,7 +1160,7 @@
interface.blockchain = None
interface.tip_header = None
interface.tip = 0
- interface.set_mode(Interface.MODE_VERIFICATION)
+ interface.set_mode(Interface.Mode.VERIFICATION)
with self.interface_lock:
self.interfaces[server_key] = interface
@@ -1377,7 +1377,7 @@
)
initial_interface_mode = interface.mode
- if interface.mode == Interface.MODE_VERIFICATION:
+ if interface.mode == Interface.Mode.VERIFICATION:
if not was_verification_request:
interface.print_error(
"disconnecting unverified server for sending unrelated header chunk"
@@ -1434,7 +1434,7 @@
return
# This interface was verified above. Get it syncing.
- if initial_interface_mode == Interface.MODE_VERIFICATION:
+ if initial_interface_mode == Interface.Mode.VERIFICATION:
self._process_latest_tip(interface)
return
@@ -1449,7 +1449,7 @@
interface, request_base_height + actual_header_count, 2016
)
else:
- interface.set_mode(Interface.MODE_DEFAULT)
+ interface.set_mode(Interface.Mode.DEFAULT)
interface.print_error("catch up done", interface.blockchain.height())
interface.blockchain.catch_up = None
self.notify("blockchain_updated")
@@ -1525,10 +1525,10 @@
header = blockchain.deserialize_header(bytes.fromhex(hexheader), height)
# Is there a blockchain that already includes this header?
chain = blockchain.check_header(header)
- if interface.mode == Interface.MODE_BACKWARD:
+ if interface.mode == Interface.Mode.BACKWARD:
if chain:
interface.print_error("binary search")
- interface.set_mode(Interface.MODE_BINARY)
+ interface.set_mode(Interface.Mode.BINARY)
interface.blockchain = chain
interface.good = height
next_height = (interface.bad + interface.good) // 2
@@ -1556,7 +1556,7 @@
interface.tip - 2 * delta,
)
- elif interface.mode == Interface.MODE_BINARY:
+ elif interface.mode == Interface.Mode.BINARY:
if chain:
interface.good = height
interface.blockchain = chain
@@ -1586,7 +1586,7 @@
)
branch.write(b"", 0)
branch.save_header(interface.bad_header)
- interface.set_mode(Interface.MODE_CATCH_UP)
+ interface.set_mode(Interface.Mode.CATCH_UP)
interface.blockchain = branch
next_height = interface.bad + 1
interface.blockchain.catch_up = interface.server
@@ -1599,20 +1599,20 @@
self.blockchains[interface.bad] = b
interface.blockchain = b
interface.print_error("new chain", b.base_height)
- interface.set_mode(Interface.MODE_CATCH_UP)
+ interface.set_mode(Interface.Mode.CATCH_UP)
next_height = interface.bad + 1
interface.blockchain.catch_up = interface.server
else:
assert bh == interface.good
if interface.blockchain.catch_up is None and bh < interface.tip:
interface.print_error("catching up from %d" % (bh + 1))
- interface.set_mode(Interface.MODE_CATCH_UP)
+ interface.set_mode(Interface.Mode.CATCH_UP)
next_height = bh + 1
interface.blockchain.catch_up = interface.server
self.notify("blockchain_updated")
- elif interface.mode == Interface.MODE_CATCH_UP:
+ elif interface.mode == Interface.Mode.CATCH_UP:
can_connect = interface.blockchain.can_connect(header)
if can_connect:
interface.blockchain.save_header(header)
@@ -1620,7 +1620,7 @@
else:
# go back
interface.print_error("cannot connect", height)
- interface.set_mode(Interface.MODE_BACKWARD)
+ interface.set_mode(Interface.Mode.BACKWARD)
interface.bad = height
interface.bad_header = header
next_height = height - 1
@@ -1631,7 +1631,7 @@
interface.blockchain.catch_up = None
self.switch_lagging_interface()
self.notify("blockchain_updated")
- elif interface.mode == Interface.MODE_DEFAULT:
+ elif interface.mode == Interface.Mode.DEFAULT:
interface.print_error(
"ignored header {} received in default mode".format(height)
)
@@ -1640,14 +1640,14 @@
# If not finished, get the next header
if next_height:
if (
- interface.mode == Interface.MODE_CATCH_UP
+ interface.mode == Interface.Mode.CATCH_UP
and interface.tip > next_height
):
self.request_headers(interface, next_height, 2016)
else:
self.request_header(interface, next_height)
else:
- interface.set_mode(Interface.MODE_DEFAULT)
+ interface.set_mode(Interface.Mode.DEFAULT)
self.notify("blockchain_updated")
# refresh network dialog
self.notify("interfaces")
@@ -1814,7 +1814,7 @@
interface.tip_header = header
interface.tip = height
- if interface.mode == Interface.MODE_VERIFICATION:
+ if interface.mode == Interface.Mode.VERIFICATION:
# If the server has already had this requested, this will be a no-op.
self.request_initial_proof_and_headers(interface)
return
@@ -1822,7 +1822,7 @@
self._process_latest_tip(interface)
def _process_latest_tip(self, interface):
- if interface.mode != Interface.MODE_DEFAULT:
+ if interface.mode != Interface.Mode.DEFAULT:
return
header = interface.tip_header
@@ -1854,7 +1854,7 @@
tip, heights
)
)
- interface.set_mode(Interface.MODE_BACKWARD)
+ interface.set_mode(Interface.Mode.BACKWARD)
interface.bad = height
interface.bad_header = header
self.request_header(interface, min(tip, height - 1))
@@ -1865,7 +1865,7 @@
chain = self.blockchains[0]
if chain.catch_up is None:
chain.catch_up = interface
- interface.set_mode(Interface.MODE_CATCH_UP)
+ interface.set_mode(Interface.Mode.CATCH_UP)
interface.blockchain = chain
interface.print_error("switching to catchup mode", tip)
self.request_header(
@@ -1908,7 +1908,7 @@
else:
# We already have them verified, maybe we got disconnected.
interface.print_error("request_initial_proof_and_headers bypassed")
- interface.set_mode(Interface.MODE_DEFAULT)
+ interface.set_mode(Interface.Mode.DEFAULT)
self._process_latest_tip(interface)
def apply_successful_verification(
@@ -1932,7 +1932,7 @@
self.verified_checkpoint = True
interface.print_error("server was verified correctly")
- interface.set_mode(Interface.MODE_DEFAULT)
+ interface.set_mode(Interface.Mode.DEFAULT)
return True
def validate_checkpoint_result(

File Metadata

Mime Type
text/plain
Expires
Sat, Mar 1, 11:08 (19 h, 17 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
5187514
Default Alt Text
D14947.diff (13 KB)

Event Timeline