From d6c19e71a677d845bb850ba6665d57a582f7e5f6 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 14:54:30 +0800 Subject: [PATCH 01/14] Add typing and notes in pubsub --- libp2p/pubsub/floodsub.py | 6 ++- libp2p/pubsub/pubsub.py | 78 ++++++++++++++++++++++++++++++++------- 2 files changed, 69 insertions(+), 15 deletions(-) diff --git a/libp2p/pubsub/floodsub.py b/libp2p/pubsub/floodsub.py index 4b16020..949b929 100644 --- a/libp2p/pubsub/floodsub.py +++ b/libp2p/pubsub/floodsub.py @@ -1,3 +1,7 @@ +from libp2p.peer.id import ( + ID, +) + from .pb import rpc_pb2 from .pubsub_router_interface import IPubsubRouter @@ -42,7 +46,7 @@ class FloodSub(IPubsubRouter): :param rpc: rpc message """ - async def publish(self, sender_peer_id, rpc_message): + async def publish(self, from_peer: ID, pubsub_message: rpc_pb2.Message) -> None: """ Invoked to forward a new message that has been validated. This is where the "flooding" part of floodsub happens diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index f80ed97..7afdc2a 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -1,16 +1,62 @@ # pylint: disable=no-name-in-module import asyncio +import time +from typing import ( + Any, + Dict, + List, + Sequence, + Tuple, +) from lru import LRU +from libp2p.host.host_interface import ( + IHost, +) +from libp2p.peer.id import ( + ID, +) +from libp2p.network.stream.net_stream_interface import ( + INetStream, +) + from .pb import rpc_pb2 from .pubsub_notifee import PubsubNotifee +from .pubsub_router_interface import ( + IPubsubRouter, +) -class Pubsub(): +def get_msg_id(msg: rpc_pb2.Message) -> Tuple[bytes, bytes]: + # NOTE: `string(from, seqno)` in Go + return (msg.seqno, msg.from_id) + + +class Pubsub: # pylint: disable=too-many-instance-attributes, no-member - def __init__(self, host, router, my_id, cache_size=None): + host: IHost + my_id: ID + router: IPubsubRouter + peer_queue: asyncio.Queue + protocols: Sequence[str] + incoming_msgs_from_peers: asyncio.Queue() + outgoing_messages: asyncio.Queue() + seen_messages: LRU + my_topics: Dict[str, asyncio.Queue] + peer_topics: Dict[str, List[ID]] + # FIXME: Should be changed to `Dict[ID, INetStream]` + peers: Dict[str, INetStream] + # NOTE: Be sure it is increased atomically everytime. + counter: int # uint64 + + def __init__( + self, + host: IHost, + router: IPubsubRouter, + my_id: ID, + cache_size: int = None) -> None: """ Construct a new Pubsub object, which is responsible for handling all Pubsub-related messages and relaying messages as appropriate to the @@ -57,10 +103,12 @@ class Pubsub(): # Create peers map, which maps peer_id (as string) to stream (to a given peer) self.peers = {} + self.counter = time.time_ns() + # Call handle peer to keep waiting for updates to peer queue asyncio.ensure_future(self.handle_peer_queue()) - def get_hello_packet(self): + def get_hello_packet(self) -> bytes: """ Generate subscription message with all topics we are subscribed to only send hello packet if we have subscribed topics @@ -73,7 +121,7 @@ class Pubsub(): return packet.SerializeToString() - async def continuously_read_stream(self, stream): + async def continuously_read_stream(self, stream: INetStream) -> None: """ Read from input stream in an infinite loop. Process messages from other nodes @@ -120,7 +168,7 @@ class Pubsub(): # Force context switch await asyncio.sleep(0) - async def stream_handler(self, stream): + async def stream_handler(self, stream: INetStream) -> None: """ Stream handler for pubsub. Gets invoked whenever a new stream is created on one of the supported pubsub protocols. @@ -139,7 +187,7 @@ class Pubsub(): # Pass stream off to stream reader asyncio.ensure_future(self.continuously_read_stream(stream)) - async def handle_peer_queue(self): + async def handle_peer_queue(self) -> None: """ Continuously read from peer queue and each time a new peer is found, open a stream to the peer using a supported pubsub protocol @@ -170,7 +218,8 @@ class Pubsub(): # Force context switch await asyncio.sleep(0) - def handle_subscription(self, origin_id, sub_message): + # FIXME: `sub_message` can be further type hinted with mypy_protobuf + def handle_subscription(self, origin_id: ID, sub_message: Any) -> None: """ Handle an incoming subscription message from a peer. Update internal mapping to mark the peer as subscribed or unsubscribed to topics as @@ -189,7 +238,9 @@ class Pubsub(): if origin_id in self.peer_topics[sub_message.topicid]: self.peer_topics[sub_message.topicid].remove(origin_id) - async def handle_talk(self, publish_message): + # FIXME(mhchia): Change the function name? + # FIXME(mhchia): `publish_message` can be further type hinted with mypy_protobuf + async def handle_talk(self, publish_message: Any) -> None: """ Put incoming message from a peer onto my blocking queue :param talk: RPC.Message format @@ -203,7 +254,7 @@ class Pubsub(): # for each topic await self.my_topics[topic].put(publish_message) - async def subscribe(self, topic_id): + async def subscribe(self, topic_id: str) -> asyncio.Queue: """ Subscribe ourself to a topic :param topic_id: topic_id to subscribe to @@ -232,7 +283,7 @@ class Pubsub(): # Return the asyncio queue for messages on this topic return self.my_topics[topic_id] - async def unsubscribe(self, topic_id): + async def unsubscribe(self, topic_id: str) -> None: """ Unsubscribe ourself from a topic :param topic_id: topic_id to unsubscribe from @@ -257,15 +308,14 @@ class Pubsub(): # Tell router we are leaving this topic await self.router.leave(topic_id) - async def message_all_peers(self, rpc_msg): + # FIXME: `rpc_msg` can be further type hinted with mypy_protobuf + async def message_all_peers(self, rpc_msg: Any) -> None: """ Broadcast a message to peers :param raw_msg: raw contents of the message to broadcast """ # Broadcast message - for peer in self.peers: - stream = self.peers[peer] - + for _, stream in self.peers.items(): # Write message to stream await stream.write(rpc_msg) From 9497c3180f71fc6a5052937123bb8c46f8277141 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 15:54:30 +0800 Subject: [PATCH 02/14] Add tox - Put extras_require to setup.py - Add mypy --- mypy.ini | 12 ++++++++++++ setup.py | 21 +++++++++++++++++++++ tox.ini | 21 +++++++++++++++++++++ 3 files changed, 54 insertions(+) create mode 100644 mypy.ini create mode 100644 tox.ini diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..c062136 --- /dev/null +++ b/mypy.ini @@ -0,0 +1,12 @@ +[mypy] +warn_unused_ignores = True +ignore_missing_imports = True +strict_optional = False +check_untyped_defs = True +disallow_incomplete_defs = True +disallow_untyped_defs = True +disallow_any_generics = True +disallow_untyped_calls = True +warn_redundant_casts = True +warn_unused_configs = True +strict_equality = True diff --git a/setup.py b/setup.py index 4807b5b..a2b0aa2 100644 --- a/setup.py +++ b/setup.py @@ -7,6 +7,26 @@ classifiers = [ ] +# pylint: disable=invalid-name +extras_require = { + "test": [ + "codecov", + "pytest", + "pytest-cov", + "pytest-asyncio", + ], + "lint": [ + "pylint", + "mypy", + ], + "dev": [ + "tox", + ], +} + +extras_require["dev"] = extras_require["test"] + extras_require["lint"] + extras_require["dev"] + + setuptools.setup( name="libp2p", description="libp2p implementation written in python", @@ -26,6 +46,7 @@ setuptools.setup( "lru-dict>=1.1.6", "aio_timers>=0.0.1,<0.1.0", ], + extras_require=extras_require, packages=setuptools.find_packages(exclude=["tests", "tests.*"]), zip_safe=False, ) diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..d7cf5d7 --- /dev/null +++ b/tox.ini @@ -0,0 +1,21 @@ +# Reference: https://github.com/ethereum/py_ecc/blob/d0da74402210ea1503ef83b3c489d5b5eba7f7bf/tox.ini + +[tox] +envlist = + py37-test + lint + +[testenv] +deps = +extras = test +commands = + pytest --cov=./libp2p tests/ +basepython = + py37: python3.7 + +[testenv:lint] +basepython = python3 +extras = dev +commands = + pylint --rcfile={toxinidir}/.pylintrc libp2p tests + mypy -p p2pclient --config-file {toxinidir}/mypy.ini From ae4c135ae139c49ff7351de043d4ca5425c99002 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:05:33 +0800 Subject: [PATCH 03/14] Change travis CI config --- .travis.yml | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/.travis.yml b/.travis.yml index a5dfefb..c72ca2e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,19 +5,18 @@ matrix: - python: 3.7 dist: xenial sudo: true + after_success: + - codecov + - python: 3.7 + dist: xenial + sudo: true + env: TOXENV=lint install: - - pip install --upgrade pip - - pip install -r requirements_dev.txt - - python setup.py develop + - pip install --upgrade pip codecov script: - - pytest --cov=./libp2p tests/ - - pylint --rcfile=.pylintrc libp2p tests - -after_success: - - codecov + - tox notifications: slack: py-libp2p:RK0WVoQZhQXLgIKfHNPL1TR2 - From 859ec6e241c2de7dd9df9f562f9bf1edabb4e9c0 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:09:03 +0800 Subject: [PATCH 04/14] Add the missing env for py37-test --- .travis.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.travis.yml b/.travis.yml index c72ca2e..f8c6983 100644 --- a/.travis.yml +++ b/.travis.yml @@ -5,6 +5,7 @@ matrix: - python: 3.7 dist: xenial sudo: true + env: TOXENV=py37-test after_success: - codecov - python: 3.7 From e428897cc86ebfdade892d2f261b940a24ad9d45 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:11:05 +0800 Subject: [PATCH 05/14] Add the missing tox in CI --- .travis.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.travis.yml b/.travis.yml index f8c6983..78937e3 100644 --- a/.travis.yml +++ b/.travis.yml @@ -14,7 +14,7 @@ matrix: env: TOXENV=lint install: - - pip install --upgrade pip codecov + - pip install --upgrade pip codecov tox script: - tox From 8da4032c3bf789ae363526ae06f3d2715c856751 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:23:30 +0800 Subject: [PATCH 06/14] Let pylint not complain about FIXME, XXX --- .pylintrc | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.pylintrc b/.pylintrc index 7113484..4955221 100644 --- a/.pylintrc +++ b/.pylintrc @@ -197,8 +197,7 @@ spelling-store-unknown-words=no [MISCELLANEOUS] # List of note tags to take in consideration, separated by a comma. -notes=FIXME, - XXX +notes= [TYPECHECK] From d3a948be476ea2771615606d9c0e060106bf21f6 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:24:14 +0800 Subject: [PATCH 07/14] Fix error: Change params floodsub.publish back --- libp2p/pubsub/floodsub.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libp2p/pubsub/floodsub.py b/libp2p/pubsub/floodsub.py index 949b929..c056214 100644 --- a/libp2p/pubsub/floodsub.py +++ b/libp2p/pubsub/floodsub.py @@ -46,7 +46,7 @@ class FloodSub(IPubsubRouter): :param rpc: rpc message """ - async def publish(self, from_peer: ID, pubsub_message: rpc_pb2.Message) -> None: + async def publish(self, sender_peer_id: ID, rpc_message: rpc_pb2.Message) -> None: """ Invoked to forward a new message that has been validated. This is where the "flooding" part of floodsub happens From 1ae306ae8f0d07cd34b5c29d5537ff9d397652cc Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:34:55 +0800 Subject: [PATCH 08/14] Fix mypy command - Remove requirements_dev.txt - Add detailed versions --- requirements_dev.txt | 7 ------- setup.py | 14 +++++++------- tox.ini | 2 +- 3 files changed, 8 insertions(+), 15 deletions(-) delete mode 100644 requirements_dev.txt diff --git a/requirements_dev.txt b/requirements_dev.txt deleted file mode 100644 index e3d8c0e..0000000 --- a/requirements_dev.txt +++ /dev/null @@ -1,7 +0,0 @@ -pytest>=3.7 -codecov -pytest-cov -pytest-asyncio -pylint -grpcio -grpcio-tools diff --git a/setup.py b/setup.py index a2b0aa2..bc8cefb 100644 --- a/setup.py +++ b/setup.py @@ -10,17 +10,17 @@ classifiers = [ # pylint: disable=invalid-name extras_require = { "test": [ - "codecov", - "pytest", - "pytest-cov", - "pytest-asyncio", + "codecov>=2.0.15,<3.0.0", + "pytest>=4.6.3,<5.0.0", + "pytest-cov>=2.7.1,<3.0.0", + "pytest-asyncio>=0.10.0,<1.0.0", ], "lint": [ - "pylint", - "mypy", + "pylint>=2.3.1,<3.0.0", + "mypy>=0.701,<1.0", ], "dev": [ - "tox", + "tox>=3.13.2,<4.0.0", ], } diff --git a/tox.ini b/tox.ini index d7cf5d7..1fc2b12 100644 --- a/tox.ini +++ b/tox.ini @@ -18,4 +18,4 @@ basepython = python3 extras = dev commands = pylint --rcfile={toxinidir}/.pylintrc libp2p tests - mypy -p p2pclient --config-file {toxinidir}/mypy.ini + mypy -p libp2p --config-file {toxinidir}/mypy.ini From 529829b9f1f4e855d8fa60af98f3988353b2c34e Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 16:41:19 +0800 Subject: [PATCH 09/14] Move `codecov` to tox.ini --- .travis.yml | 5 ++--- tox.ini | 2 ++ 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/.travis.yml b/.travis.yml index 78937e3..92ab222 100644 --- a/.travis.yml +++ b/.travis.yml @@ -6,15 +6,14 @@ matrix: dist: xenial sudo: true env: TOXENV=py37-test - after_success: - - codecov - python: 3.7 dist: xenial sudo: true env: TOXENV=lint install: - - pip install --upgrade pip codecov tox + - pip install --upgrade pip + - pip install tox script: - tox diff --git a/tox.ini b/tox.ini index 1fc2b12..dd582b2 100644 --- a/tox.ini +++ b/tox.ini @@ -7,9 +7,11 @@ envlist = [testenv] deps = +passenv = CI TRAVIS TRAVIS_* extras = test commands = pytest --cov=./libp2p tests/ + codecov basepython = py37: python3.7 From 04b7df9fcf6e852cbe540665ab8c2f0afaff5ebb Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 18:00:57 +0800 Subject: [PATCH 10/14] Lint `examples` in tox --- examples/__init__.py | 0 examples/chat/__init__.py | 0 tox.ini | 4 ++-- 3 files changed, 2 insertions(+), 2 deletions(-) create mode 100644 examples/__init__.py create mode 100644 examples/chat/__init__.py diff --git a/examples/__init__.py b/examples/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/examples/chat/__init__.py b/examples/chat/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tox.ini b/tox.ini index dd582b2..10302ca 100644 --- a/tox.ini +++ b/tox.ini @@ -19,5 +19,5 @@ basepython = basepython = python3 extras = dev commands = - pylint --rcfile={toxinidir}/.pylintrc libp2p tests - mypy -p libp2p --config-file {toxinidir}/mypy.ini + pylint --rcfile={toxinidir}/.pylintrc libp2p examples tests + mypy -p libp2p -p examples --config-file {toxinidir}/mypy.ini From 381f5ddc3ad5f65bca5e901a661d1d13034839cc Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 18:43:49 +0800 Subject: [PATCH 11/14] Replace `click` with `argparse` --- examples/chat/chat.py | 49 ++++++++++++++++++++++++++++--------------- setup.py | 1 - 2 files changed, 32 insertions(+), 18 deletions(-) diff --git a/examples/chat/chat.py b/examples/chat/chat.py index 9c55687..0ac6e78 100755 --- a/examples/chat/chat.py +++ b/examples/chat/chat.py @@ -1,8 +1,8 @@ +import argparse import asyncio import sys import urllib.request -import click import multiaddr from libp2p import new_node @@ -52,9 +52,9 @@ async def run(port, destination): (int(port) + 1, external_ip, port, host.get_id().pretty())) print("\nWaiting for incoming connection\n\n") - else: # its the client - m = multiaddr.Multiaddr(destination) - info = info_from_p2p_addr(m) + else: # its the client + maddr = multiaddr.Multiaddr(destination) + info = info_from_p2p_addr(maddr) # Associate the peer with local ip address await host.connect(info) @@ -67,22 +67,37 @@ async def run(port, destination): print("Connected to peer %s" % info.addrs[0]) -@click.command() -@click.option('--port', '-p', help='source port number', default=8000) -@click.option('--destination', '-d', help="Destination multiaddr string") -@click.option('--help', is_flag=True, default=False, help='display help') -# @click.option('--debug', is_flag=True, default=False, help='Debug generates the same node ID on every execution') -def main(port, destination, help): - - if help: - print("This program demonstrates a simple p2p chat application using libp2p\n\n") - print("Usage: Run './chat -p ' where can be any port number.") - print("Now run './chat -p -d ' where is multiaddress of previous listener host.") - return +def main(): + description = """ + This program demonstrates a simple p2p chat application using libp2p. + To use it, first run 'python ./chat -p ', where is the port number. + Then, run another host with 'python ./chat -p -d ', + where is the multiaddress of the previous listener host. + """ + example_maddr = "/ip4/127.0.0.1/tcp/8000/p2p/QmQn4SwGkDZKkUEpBRBvTmheQycxAHJUNmVEnjA2v1qe8Q" + parser = argparse.ArgumentParser(description=description) + parser.add_argument( + "--debug", + action='store_true', + help='generate the same node ID on every execution', + ) + parser.add_argument( + "-p", + "--port", + default=8000, + help="source port number", + ) + parser.add_argument( + "-d", + "--destination", + type=str, + help=f"destination multiaddr string, e.g. {example_maddr}", + ) + args = parser.parse_args() loop = asyncio.get_event_loop() try: - asyncio.ensure_future(run(port, destination)) + asyncio.ensure_future(run(args.port, args.destination)) loop.run_forever() except KeyboardInterrupt: pass diff --git a/setup.py b/setup.py index bc8cefb..36d99a4 100644 --- a/setup.py +++ b/setup.py @@ -36,7 +36,6 @@ setuptools.setup( classifiers=classifiers, install_requires=[ "pycryptodome>=3.8.2,<4.0.0", - "click>=7.0,<8.0", "base58>=1.0.3,<2.0.0", "pymultihash>=0.8.2", "multiaddr>=0.0.8,<0.1.0", From ecf4e373da175f0e9ac26d79a3c79b33beedab26 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 18:49:51 +0800 Subject: [PATCH 12/14] Remove unused `aio_timers` --- setup.py | 1 - 1 file changed, 1 deletion(-) diff --git a/setup.py b/setup.py index 36d99a4..88c2147 100644 --- a/setup.py +++ b/setup.py @@ -43,7 +43,6 @@ setuptools.setup( "grpcio>=1.21.1,<2.0.0", "grpcio-tools>=1.21.1,<2.0.0", "lru-dict>=1.1.6", - "aio_timers>=0.0.1,<0.1.0", ], extras_require=extras_require, packages=setuptools.find_packages(exclude=["tests", "tests.*"]), From d64c7d6d56a5a16861c0a72fff029947c24e8098 Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 21:28:14 +0800 Subject: [PATCH 13/14] Add the missing type for `port` --- examples/chat/chat.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/examples/chat/chat.py b/examples/chat/chat.py index 0ac6e78..ea98598 100755 --- a/examples/chat/chat.py +++ b/examples/chat/chat.py @@ -23,6 +23,7 @@ async def read_data(stream): print("\x1b[32m %s\x1b[0m " % read_string, end="") +# FIXME(mhchia): Reconsider whether we should use a thread pool here. async def write_data(stream): loop = asyncio.get_event_loop() while True: @@ -85,6 +86,7 @@ def main(): "-p", "--port", default=8000, + type=int, help="source port number", ) parser.add_argument( From a97bac0a026c4f16608a73d3af88979f02b0dc4e Mon Sep 17 00:00:00 2001 From: mhchia Date: Wed, 24 Jul 2019 21:30:23 +0800 Subject: [PATCH 14/14] Remove sudo from .travis.yml --- .travis.yml | 2 -- 1 file changed, 2 deletions(-) diff --git a/.travis.yml b/.travis.yml index 92ab222..77b9e16 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,11 +4,9 @@ matrix: include: - python: 3.7 dist: xenial - sudo: true env: TOXENV=py37-test - python: 3.7 dist: xenial - sudo: true env: TOXENV=lint install: