StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #298 from NIC619/add_more_logging

Browse Source 
Add a little more loggings
This commit is contained in:
NIC Lin 2019-09-15 15:34:28 +08:00 committed by GitHub
commit 7483da762e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 119 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -24,6 +24,7 @@ wheels/
.installed.cfg
*.egg
MANIFEST
pip-wheel-metadata
# PyInstaller
# Usually these files are written by a python script from a template

24
libp2p/host/basic_host.py
View File

@ -21,7 +21,7 @@ from .host_interface import IHost
class BasicHost(IHost):
_network: INetwork
router: KadmeliaPeerRouter
_router: KadmeliaPeerRouter
peerstore: IPeerStore
# default options constructor
@ -56,7 +56,7 @@ class BasicHost(IHost):
def get_addrs(self) -> List[multiaddr.Multiaddr]:
"""
:return: all the multiaddr addresses this host is listening too
:return: all the multiaddr addresses this host is listening to
"""
p2p_part = multiaddr.Multiaddr("/p2p/{}".format(self.get_id().pretty()))
@ -68,23 +68,22 @@ class BasicHost(IHost):
def set_stream_handler(
self, protocol_id: TProtocol, stream_handler: StreamHandlerFn
) -> bool:
) -> None:
"""
set stream handler for host
set stream handler for given `protocol_id`
:param protocol_id: protocol id used on stream
:param stream_handler: a stream handler function
:return: true if successful
"""
return self._network.set_stream_handler(protocol_id, stream_handler)
self._network.set_stream_handler(protocol_id, stream_handler)
# protocol_id can be a list of protocol_ids
# stream will decide which protocol_id to run on
# `protocol_ids` can be a list of `protocol_id`
# stream will decide which `protocol_id` to run on
async def new_stream(
self, peer_id: ID, protocol_ids: Sequence[TProtocol]
) -> INetStream:
"""
:param peer_id: peer_id that host is connecting
:param protocol_id: protocol id that stream runs on
:param protocol_ids: available protocol ids to use for stream
:return: stream: new stream created
"""
return await self._network.new_stream(peer_id, protocol_ids)
@ -92,12 +91,11 @@ class BasicHost(IHost):
async def connect(self, peer_info: PeerInfo) -> None:
"""
connect ensures there is a connection between this host and the peer with
given peer_info.peer_id. connect will absorb the addresses in peer_info into its internal
given `peer_info.peer_id`. connect will absorb the addresses in peer_info into its internal
peerstore. If there is not an active connection, connect will issue a
dial, and block until a connection is open, or an error is
returned.
dial, and block until a connection is opened, or an error is returned.
:param peer_info: peer_info of the host we want to connect to
:param peer_info: peer_info of the peer we want to connect to
:type peer_info: peer.peerinfo.PeerInfo
"""
self.peerstore.add_addrs(peer_info.peer_id, peer_info.addrs, 10)

12
libp2p/host/host_interface.py
View File

@ -33,18 +33,17 @@ class IHost(ABC):
@abstractmethod
def get_addrs(self) -> List[multiaddr.Multiaddr]:
"""
:return: all the multiaddr addresses this host is listening too
:return: all the multiaddr addresses this host is listening to
"""
@abstractmethod
def set_stream_handler(
self, protocol_id: TProtocol, stream_handler: StreamHandlerFn
) -> bool:
) -> None:
"""
set stream handler for host
:param protocol_id: protocol id used on stream
:param stream_handler: a stream handler function
:return: true if successful
"""
# protocol_id can be a list of protocol_ids
@ -55,7 +54,7 @@ class IHost(ABC):
) -> INetStream:
"""
:param peer_id: peer_id that host is connecting
:param protocol_ids: protocol ids that stream can run on
:param protocol_ids: available protocol ids to use for stream
:return: stream: new stream created
"""
@ -65,10 +64,9 @@ class IHost(ABC):
connect ensures there is a connection between this host and the peer with
given peer_info.peer_id. connect will absorb the addresses in peer_info into its internal
peerstore. If there is not an active connection, connect will issue a
dial, and block until a connection is open, or an error is
returned.
dial, and block until a connection is opened, or an error is returned.
:param peer_info: peer_info of the host we want to connect to
:param peer_info: peer_info of the peer we want to connect to
:type peer_info: peer.peerinfo.PeerInfo
"""

3
libp2p/network/network_interface.py
View File

@ -40,11 +40,10 @@ class INetwork(ABC):
@abstractmethod
def set_stream_handler(
self, protocol_id: TProtocol, stream_handler: StreamHandlerFn
) -> bool:
) -> None:
"""
:param protocol_id: protocol id used on stream
:param stream_handler: a stream handler instance
:return: true if successful
"""
@abstractmethod

2
libp2p/network/stream/net_stream.py
View File

@ -34,7 +34,6 @@ class NetStream(INetStream):
def set_protocol(self, protocol_id: TProtocol) -> None:
"""
:param protocol_id: protocol id that stream runs on
:return: true if successful
"""
self.protocol_id = protocol_id
@ -64,7 +63,6 @@ class NetStream(INetStream):
async def close(self) -> None:
"""
close stream
:return: true if successful
"""
await self.muxed_stream.close()

3
libp2p/network/stream/net_stream_interface.py
View File

@ -16,10 +16,9 @@ class INetStream(ReadWriteCloser):
"""
@abstractmethod
def set_protocol(self, protocol_id: TProtocol) -> bool:
def set_protocol(self, protocol_id: TProtocol) -> None:
"""
:param protocol_id: protocol id that stream runs on
:return: true if successful
"""
@abstractmethod

51
libp2p/network/swarm.py
View File

@ -80,14 +80,12 @@ class Swarm(INetwork):
def set_stream_handler(
self, protocol_id: TProtocol, stream_handler: StreamHandlerFn
) -> bool:
) -> None:
"""
:param protocol_id: protocol id used on stream
:param stream_handler: a stream handler instance
:return: true if successful
"""
self.multiselect.add_handler(protocol_id, stream_handler)
return True
async def dial_peer(self, peer_id: ID) -> IMuxedConn:
"""
@ -128,11 +126,10 @@ class Swarm(INetwork):
try:
secured_conn = await self.upgrader.upgrade_security(raw_conn, peer_id, True)
except SecurityUpgradeFailure as error:
# TODO: Add logging to indicate the failure
error_msg = "fail to upgrade security for peer %s"
logger.debug(error_msg, peer_id)
await raw_conn.close()
raise SwarmException(
f"fail to upgrade the connection to a secured connection from {peer_id}"
) from error
raise SwarmException(error_msg % peer_id) from error
logger.debug("upgraded security for peer %s", peer_id)
@ -141,11 +138,10 @@ class Swarm(INetwork):
secured_conn, self.generic_protocol_handler, peer_id
)
except MuxerUpgradeFailure as error:
# TODO: Add logging to indicate the failure
error_msg = "fail to upgrade mux for peer %s"
logger.debug(error_msg, peer_id)
await secured_conn.close()
raise SwarmException(
f"fail to upgrade the connection to a muxed connection from {peer_id}"
) from error
raise SwarmException(error_msg % peer_id) from error
logger.debug("upgraded mux for peer %s", peer_id)
@ -168,6 +164,11 @@ class Swarm(INetwork):
:param protocol_id: protocol id
:return: net stream instance
"""
logger.debug(
"attempting to open a stream to peer %s, over one of the protocols %s",
peer_id,
protocol_ids,
)
muxed_conn = await self.dial_peer(peer_id)
@ -183,6 +184,12 @@ class Swarm(INetwork):
net_stream = NetStream(muxed_stream)
net_stream.set_protocol(selected_protocol)
logger.debug(
"successfully opened a stream to peer %s, over protocol %s",
peer_id,
selected_protocol,
)
# Call notifiers since event occurred
for notifee in self.notifees:
await notifee.opened_stream(self, net_stream)
@ -215,8 +222,6 @@ class Swarm(INetwork):
peer_addr = f"/ip4/{connection_info[0]}/tcp/{connection_info[1]}"
logger.debug("inbound connection at %s", peer_addr)
# logger.debug("inbound connection request", peer_id)
# Upgrade reader/write to a net_stream and pass \
# to appropriate stream handler (using multiaddr)
raw_conn = RawConnection(reader, writer, False)
# Per, https://discuss.libp2p.io/t/multistream-security/130, we first secure
@ -227,11 +232,10 @@ class Swarm(INetwork):
raw_conn, ID(b""), False
)
except SecurityUpgradeFailure as error:
# TODO: Add logging to indicate the failure
error_msg = "fail to upgrade security for peer at %s"
logger.debug(error_msg, peer_addr)
await raw_conn.close()
raise SwarmException(
"fail to upgrade the connection to a secured connection"
) from error
raise SwarmException(error_msg % peer_addr) from error
peer_id = secured_conn.get_remote_peer()
logger.debug("upgraded security for peer at %s", peer_addr)
@ -242,11 +246,10 @@ class Swarm(INetwork):
secured_conn, self.generic_protocol_handler, peer_id
)
except MuxerUpgradeFailure as error:
# TODO: Add logging to indicate the failure
error_msg = "fail to upgrade mux for peer %s"
logger.debug(error_msg, peer_id)
await secured_conn.close()
raise SwarmException(
f"fail to upgrade the connection to a muxed connection from {peer_id}"
) from error
raise SwarmException(error_msg % peer_id) from error
logger.debug("upgraded mux for peer %s", peer_id)
# Store muxed_conn with peer id
self.connections[peer_id] = muxed_conn
@ -269,7 +272,7 @@ class Swarm(INetwork):
return True
except IOError:
# Failed. Continue looping.
print("Failed to connect to: " + str(maddr))
logger.debug("fail to listen on: " + str(maddr))
# No maddr succeeded
return False
@ -301,6 +304,8 @@ class Swarm(INetwork):
*[connection.close() for connection in self.connections.values()]
)
logger.debug("swarm successfully closed")
async def close_peer(self, peer_id: ID) -> None:
if peer_id not in self.connections:
return
@ -308,6 +313,8 @@ class Swarm(INetwork):
del self.connections[peer_id]
await connection.close()
logger.debug("successfully close the connection to peer %s", peer_id)
def create_generic_protocol_handler(swarm: Swarm) -> GenericProtocolHandlerFn:
"""

18
libp2p/protocol_muxer/multiselect_client.py
View File

@ -38,24 +38,6 @@ class MultiselectClient(IMultiselectClient):
# Handshake succeeded if this point is reached
async def select_protocol_or_fail(
self, protocol: TProtocol, communicator: IMultiselectCommunicator
) -> TProtocol:
"""
Send message to multiselect selecting protocol
and fail if multiselect does not return same protocol
:param protocol: protocol to select
:param stream: stream to communicate with multiselect over
:return: selected protocol
"""
# Perform handshake to ensure multiselect protocol IDs match
await self.handshake(communicator)
# Try to select the given protocol
selected_protocol = await self.try_select(communicator, protocol)
return selected_protocol
async def select_one_of(
self, protocols: Sequence[TProtocol], communicator: IMultiselectCommunicator
) -> TProtocol:

23
libp2p/protocol_muxer/multiselect_client_interface.py
View File

@ -13,16 +13,12 @@ class IMultiselectClient(ABC):
module in order to select a protocol id to communicate over
"""
@abstractmethod
async def select_protocol_or_fail(
self, protocol: TProtocol, communicator: IMultiselectCommunicator
) -> TProtocol:
async def handshake(self, communicator: IMultiselectCommunicator) -> None:
"""
Send message to multiselect selecting protocol
and fail if multiselect does not return same protocol
:param protocol: protocol to select
Ensure that the client and multiselect
are both using the same multiselect protocol
:param stream: stream to communicate with multiselect over
:return: selected protocol
:raise Exception: multiselect protocol ID mismatch
"""
@abstractmethod
@ -37,3 +33,14 @@ class IMultiselectClient(ABC):
:param stream: stream to communicate with multiselect over
:return: selected protocol
"""
async def try_select(
self, communicator: IMultiselectCommunicator, protocol: TProtocol
) -> TProtocol:
"""
Try to select the given protocol or raise exception if fails
:param communicator: communicator to use to communicate with counterparty
:param protocol: protocol to select
:raise Exception: error in protocol selection
:return: selected protocol
"""

7
libp2p/pubsub/floodsub.py
View File

@ -1,3 +1,4 @@
import logging
from typing import Iterable, List, Sequence
from libp2p.peer.id import ID
@ -10,6 +11,9 @@ from .pubsub_router_interface import IPubsubRouter
PROTOCOL_ID = TProtocol("/floodsub/1.0.0")
logger = logging.getLogger("libp2p.pubsub.floodsub")
logger.setLevel(logging.DEBUG)
class FloodSub(IPubsubRouter):
@ -75,6 +79,9 @@ class FloodSub(IPubsubRouter):
origin=ID(pubsub_msg.from_id),
)
rpc_msg = rpc_pb2.RPC(publish=[pubsub_msg])
logger.debug("publishing message %s", pubsub_msg)
for peer_id in peers_gen:
stream = self.pubsub.peers[peer_id]
# FIXME: We should add a `WriteMsg` similar to write delimited messages.

25
libp2p/pubsub/gossipsub.py
View File

@ -1,5 +1,6 @@
from ast import literal_eval
import asyncio
import logging
import random
from typing import Any, Dict, Iterable, List, Sequence, Set
@ -15,6 +16,9 @@ from .pubsub_router_interface import IPubsubRouter
PROTOCOL_ID = TProtocol("/meshsub/1.0.0")
logger = logging.getLogger("libp2p.pubsub.gossipsub")
logger.setLevel(logging.DEBUG)
class GossipSub(IPubsubRouter):
@ -98,6 +102,8 @@ class GossipSub(IPubsubRouter):
"""
self.pubsub = pubsub
logger.debug("attached to pusub")
# Start heartbeat now that we have a pubsub instance
# TODO: Start after delay
asyncio.ensure_future(self.heartbeat())
@ -108,7 +114,7 @@ class GossipSub(IPubsubRouter):
:param peer_id: id of peer to add
:param protocol_id: router protocol the peer speaks, e.g., floodsub, gossipsub
"""
self.peers_to_protocol[peer_id] = protocol_id
logger.debug("adding peer %s with protocol %s", peer_id, protocol_id)
if protocol_id == PROTOCOL_ID:
self.peers_gossipsub.append(peer_id)
@ -121,19 +127,23 @@ class GossipSub(IPubsubRouter):
# in multistream-select, or wrong versions.
# TODO: Better handling
raise Exception(f"protocol is not supported: protocol_id={protocol_id}")
self.peers_to_protocol[peer_id] = protocol_id
def remove_peer(self, peer_id: ID) -> None:
"""
Notifies the router that a peer has been disconnected
:param peer_id: id of peer to remove
"""
del self.peers_to_protocol[peer_id]
logger.debug("removing peer %s", peer_id)
if peer_id in self.peers_gossipsub:
self.peers_gossipsub.remove(peer_id)
if peer_id in self.peers_gossipsub:
elif peer_id in self.peers_floodsub:
self.peers_floodsub.remove(peer_id)
if peer_id in self.peers_to_protocol:
del self.peers_to_protocol[peer_id]
async def handle_rpc(self, rpc: rpc_pb2.RPC, sender_peer_id: ID) -> None:
"""
Invoked to process control messages in the RPC envelope.
@ -169,6 +179,9 @@ class GossipSub(IPubsubRouter):
origin=ID(pubsub_msg.from_id),
)
rpc_msg = rpc_pb2.RPC(publish=[pubsub_msg])
logger.debug("publishing message %s", pubsub_msg)
for peer_id in peers_gen:
stream = self.pubsub.peers[peer_id]
# FIXME: We should add a `WriteMsg` similar to write delimited messages.
@ -227,6 +240,8 @@ class GossipSub(IPubsubRouter):
subscription announcement
:param topic: topic to join
"""
logger.debug("joining topic %s", topic)
if topic in self.mesh:
return
# Create mesh[topic] if it does not yet exist
@ -262,6 +277,8 @@ class GossipSub(IPubsubRouter):
It is invoked after the unsubscription announcement.
:param topic: topic to leave
"""
logger.debug("leaving topic %s", topic)
if topic not in self.mesh:
return
# Notify the peers in mesh[topic] with a PRUNE(topic) message
@ -269,7 +286,7 @@ class GossipSub(IPubsubRouter):
await self.emit_prune(topic, peer)
# Forget mesh[topic]
self.mesh.pop(topic, None)
del self.mesh[topic]
# Heartbeat
async def heartbeat(self) -> None:

30
libp2p/pubsub/pubsub.py
View File

@ -31,7 +31,8 @@ if TYPE_CHECKING:
from .pubsub_router_interface import IPubsubRouter # noqa: F401
log = logging.getLogger(__name__)
logger = logging.getLogger("libp2p.pubsub")
logger.setLevel(logging.DEBUG)
def get_msg_id(msg: rpc_pb2.Message) -> Tuple[bytes, bytes]:
@ -161,6 +162,9 @@ class Pubsub:
for msg in rpc_incoming.publish:
if not self._is_subscribed_to_msg(msg):
continue
logger.debug(
"received `publish` message %s from peer %s", msg, peer_id
)
asyncio.ensure_future(self.push_msg(msg_forwarder=peer_id, msg=msg))
if rpc_incoming.subscriptions:
@ -170,6 +174,11 @@ class Pubsub:
# to know that it is subscribed to the topic (doesn't
# need everyone to know)
for message in rpc_incoming.subscriptions:
logger.debug(
"received `subscriptions` message %s from peer %s",
message,
peer_id,
)
self.handle_subscription(peer_id, message)
# NOTE: Check if `rpc_incoming.control` is set through `HasField`.
@ -177,6 +186,7 @@ class Pubsub:
# Ref: https://developers.google.com/protocol-buffers/docs/reference/python-generated#singular-fields-proto2 # noqa: E501
if rpc_incoming.HasField("control"):
# Pass rpc to router so router could perform custom logic
logger.debug("received `control` message %s from peer %s", peer_id)
await self.router.handle_rpc(rpc_incoming, peer_id)
# Force context switch
@ -230,7 +240,13 @@ class Pubsub:
await stream.write(encode_varint_prefixed(hello.SerializeToString()))
# TODO: Check EOF of this stream.
# TODO: Check if the peer in black list.
try:
self.router.add_peer(peer_id, stream.get_protocol())
except Exception as error:
logger.debug("fail to add new peer %s, error %s", peer_id, error)
return
logger.debug("added new peer %s", peer_id)
async def handle_peer_queue(self) -> None:
"""
@ -291,6 +307,8 @@ class Pubsub:
:param topic_id: topic_id to subscribe to
"""
logger.debug("subscribing to topic %s", topic_id)
# Already subscribed
if topic_id in self.my_topics:
return self.my_topics[topic_id]
@ -319,6 +337,8 @@ class Pubsub:
:param topic_id: topic_id to unsubscribe from
"""
logger.debug("unsubscribing from topic %s", topic_id)
# Return if we already unsubscribed from the topic
if topic_id not in self.my_topics:
return
@ -366,6 +386,8 @@ class Pubsub:
await self.push_msg(self.host.get_id(), msg)
logger.debug("successfully published message %s", msg)
async def validate_msg(self, msg_forwarder: ID, msg: rpc_pb2.Message) -> None:
"""
Validate the received message
@ -401,6 +423,8 @@ class Pubsub:
:param msg_forwarder: the peer who forward us the message.
:param msg: the message we are going to push out.
"""
logger.debug("attempting to publish message %s", msg)
# TODO: Check if the `source` is in the blacklist. If yes, reject.
# TODO: Check if the `from` is in the blacklist. If yes, reject.
@ -415,14 +439,14 @@ class Pubsub:
# Validate the signature of the message
# FIXME: `signature_validator` is currently a stub.
if not signature_validator(msg.key, msg.SerializeToString()):
log.debug("Signature validation failed for msg: %s", msg)
logger.debug("Signature validation failed for msg: %s", msg)
return
# Validate the message with registered topic validators.
# If the validation failed, return(i.e., don't further process the message).
try:
await self.validate_msg(msg_forwarder, msg)
except ValidationError:
log.debug(
logger.debug(
"Topic validation failed: sender %s sent data %s under topic IDs: %s",
f"{base58.b58encode(msg.from_id).decode()}:{msg.seqno.hex()}",
msg.data.hex(),