diff --git a/substrate/Cargo.lock b/substrate/Cargo.lock
index f0da8b5390c6345a66dd3a1d52ada990adc19438..de2b174826e02f89887efd4b17258dea9722d9a2 100644
--- a/substrate/Cargo.lock
+++ b/substrate/Cargo.lock
@@ -398,6 +398,17 @@ dependencies = [
"constant_time_eq",
]
+[[package]]
+name = "blake2s_simd"
+version = "0.5.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab9e07352b829279624ceb7c64adb4f585dacdb81d35cafae81139ccd617cf44"
+dependencies = [
+ "arrayref",
+ "arrayvec 0.5.1",
+ "constant_time_eq",
+]
+
[[package]]
name = "block-buffer"
version = "0.7.3"
@@ -1227,7 +1238,7 @@ dependencies = [
"fixed-hash",
"impl-rlp",
"impl-serde 0.3.0",
- "tiny-keccak 2.0.2",
+ "tiny-keccak 2.0.1",
]
[[package]]
@@ -1791,7 +1802,6 @@ dependencies = [
"proc-macro-hack",
"proc-macro-nested",
"slab",
- "tokio-io",
]
[[package]]
@@ -1963,7 +1973,7 @@ dependencies = [
"indexmap",
"log",
"slab",
- "tokio 0.2.16",
+ "tokio 0.2.13",
"tokio-util",
]
@@ -2175,7 +2185,7 @@ dependencies = [
"net2",
"pin-project",
"time",
- "tokio 0.2.16",
+ "tokio 0.2.13",
"tower-service",
"want 0.3.0",
]
@@ -2193,7 +2203,7 @@ dependencies = [
"log",
"rustls 0.17.0",
"rustls-native-certs",
- "tokio 0.2.16",
+ "tokio 0.2.13",
"tokio-rustls",
"webpki",
]
@@ -2601,9 +2611,9 @@ checksum = "c7d73b3f436185384286bd8098d17ec07c9a7d2388a6599f824d8502b529702a"
[[package]]
name = "libp2p"
-version = "0.16.2"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bba17ee9cac4bb89de5812159877d9b4f0a993bf41697a5a875940cd1eb71f24"
+checksum = "8a261244b8d7ff58f5d62ffa33589eb1ba7733a1dfee0902ad9fdfe62ada7009"
dependencies = [
"bytes 0.5.4",
"futures 0.3.4",
@@ -2629,8 +2639,8 @@ dependencies = [
"libp2p-wasm-ext",
"libp2p-websocket",
"libp2p-yamux",
- "parity-multiaddr",
- "parity-multihash",
+ "multihash",
+ "parity-multiaddr 0.8.0",
"parking_lot 0.10.0",
"pin-project",
"smallvec 1.2.0",
@@ -2639,22 +2649,23 @@ dependencies = [
[[package]]
name = "libp2p-core"
-version = "0.16.0"
+version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b874594c4b29de1a29f27871feba8e6cd13aa54a8a1e8f8c7cf3dfac5ca287c"
+checksum = "1cfe1412f2afe1366a2661abd211bb1a27ee6a664d799071282f4fba997c6858"
dependencies = [
"asn1_der",
"bs58",
"ed25519-dalek",
+ "either",
"fnv",
"futures 0.3.4",
"futures-timer 3.0.2",
"lazy_static",
"libsecp256k1",
"log",
+ "multihash",
"multistream-select",
- "parity-multiaddr",
- "parity-multihash",
+ "parity-multiaddr 0.8.0",
"parking_lot 0.10.0",
"pin-project",
"prost",
@@ -2672,9 +2683,9 @@ dependencies = [
[[package]]
name = "libp2p-core-derive"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "96d472e9d522f588805c77801de10b957be84e10f019ca5f869fa1825b15ea9b"
+checksum = "a0eeb25d5f152a826eac57c7d1cc3de10d72dc4051e90fe4c0cd139f07a069a3"
dependencies = [
"quote 1.0.3",
"syn 1.0.17",
@@ -2682,9 +2693,9 @@ dependencies = [
[[package]]
name = "libp2p-deflate"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e25004d4d9837b44b22c5f1a69be1724a5168fef6cff1716b5176a972c3aa62"
+checksum = "136fcef31e3247f51946c3ebefb94d0798c4c8aae78bc59cb7431b220b5330cf"
dependencies = [
"flate2",
"futures 0.3.4",
@@ -2693,9 +2704,9 @@ dependencies = [
[[package]]
name = "libp2p-dns"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b99e552f9939b606eb4b59f7f64d9b01e3f96752f47e350fc3c5fc646ed3f649"
+checksum = "647178f8683bf868f7f14d5e5718dbdc2445b9f6b24ce99da96cecd7c5d2d1a6"
dependencies = [
"futures 0.3.4",
"libp2p-core",
@@ -2704,9 +2715,9 @@ dependencies = [
[[package]]
name = "libp2p-floodsub"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d3234f12e44f9a50351a9807b97fe7de11eb9ae4482370392ba10da6dc90722"
+checksum = "34c8dee172fd1630caf91a427d601d6a8d24c8cfcbcf7d5c09c9a1f3b4bbebc2"
dependencies = [
"cuckoofilter",
"fnv",
@@ -2721,9 +2732,9 @@ dependencies = [
[[package]]
name = "libp2p-gossipsub"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d46cb3e0841bd951cbf4feae56cdc081e6347836a644fb260c3ec554149b4006"
+checksum = "0042a2156fb6264bda9def93070e411dfaddf8c57c4b2d63634190d296458c76"
dependencies = [
"base64 0.11.0",
"byteorder 1.3.4",
@@ -2746,9 +2757,9 @@ dependencies = [
[[package]]
name = "libp2p-identify"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfeb935a9bd41263e4f3a24b988e9f4a044f3ae89ac284e83c17fe2f84e0d66b"
+checksum = "04efa011cda5232648b5aa50bd80be7ba0a695d682b01aa46b65e5be5ece0605"
dependencies = [
"futures 0.3.4",
"libp2p-core",
@@ -2762,9 +2773,9 @@ dependencies = [
[[package]]
name = "libp2p-kad"
-version = "0.16.2"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "464dc8412978d40f0286be72ed9ab5e0e1386a4a06e7f174526739b5c3c1f041"
+checksum = "97f4722d83af8fc0065cee7589a000b629961c1c11d90ba09f6685b3e123b9ae"
dependencies = [
"arrayvec 0.5.1",
"bytes 0.5.4",
@@ -2775,7 +2786,7 @@ dependencies = [
"libp2p-core",
"libp2p-swarm",
"log",
- "parity-multihash",
+ "multihash",
"prost",
"prost-build",
"rand 0.7.3",
@@ -2789,9 +2800,9 @@ dependencies = [
[[package]]
name = "libp2p-mdns"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "881fcfb360c2822db9f0e6bb6f89529621556ed9a8b038313414eda5107334de"
+checksum = "b752276b3bea2fca1c291f43cefc8082d8a639bb8f9052cf5adc6accfcd7b44e"
dependencies = [
"async-std",
"data-encoding",
@@ -2811,9 +2822,9 @@ dependencies = [
[[package]]
name = "libp2p-mplex"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8507b37ad0eed275efcde67a023c3d85af6c80768b193845b9288e848e1af95"
+checksum = "0f317db8c062beecde87a8765ca03784e6f1a55daa5b9868bf60ebf9b9a2b92f"
dependencies = [
"bytes 0.5.4",
"fnv",
@@ -2827,9 +2838,9 @@ dependencies = [
[[package]]
name = "libp2p-noise"
-version = "0.16.2"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b15a8a3d71f898beb6f854c8aae27aa1d198e0d1f2e49412261c2d90ef39675a"
+checksum = "98d3845f54288ff134dd78c131517bad8bc03965def6e6517efef03291d9b4d7"
dependencies = [
"curve25519-dalek",
"futures 0.3.4",
@@ -2848,9 +2859,9 @@ dependencies = [
[[package]]
name = "libp2p-ping"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33d22f2f228b3a828dca1cb8aa9fa331e0bc9c36510cb2c1916956e20dc85e8c"
+checksum = "aa1cb80ccbedb91d9b980aafc6bf39dc7e4616a7e37c631a4e6ef62629567a13"
dependencies = [
"futures 0.3.4",
"libp2p-core",
@@ -2863,9 +2874,9 @@ dependencies = [
[[package]]
name = "libp2p-plaintext"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56126a204d7b3382bac163143ff4125a14570b3ba76ba979103d1ae1abed1923"
+checksum = "da16d35e3990cc5dc22c8d7ea4a2aa1c18f518491bb29c0c3498fb9a2d8e486e"
dependencies = [
"bytes 0.5.4",
"futures 0.3.4",
@@ -2881,9 +2892,9 @@ dependencies = [
[[package]]
name = "libp2p-pnet"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b916938a8868f75180aeeffcc6a516a922d165e8fa2a90b57bad989d1ccbb57a"
+checksum = "45d11e8c6d83e294ef3d7ff3a9f5a7aa5aa0c39c2d4991f2905c23c438c84526"
dependencies = [
"futures 0.3.4",
"log",
@@ -2895,9 +2906,9 @@ dependencies = [
[[package]]
name = "libp2p-secio"
-version = "0.16.1"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1219e9ecb4945d7331a05f5ffe96a1f6e28051bfa1223d4c60353c251de0354e"
+checksum = "74130fa95effb780850ec790b7af777b893108d9b5983ab994b61d93d2eb0336"
dependencies = [
"aes-ctr",
"ctr",
@@ -2925,13 +2936,14 @@ dependencies = [
[[package]]
name = "libp2p-swarm"
-version = "0.16.1"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "275471e7c0e88ae004660866cd54f603bd8bd1f4caef541a27f50dd8640c4d4c"
+checksum = "a4ec53df8978a5d6d9dac243fb1e3adf004f8b8d28f72e6f2160df34d5f39564"
dependencies = [
"futures 0.3.4",
"libp2p-core",
"log",
+ "rand 0.7.3",
"smallvec 1.2.0",
"void",
"wasm-timer",
@@ -2939,9 +2951,9 @@ dependencies = [
[[package]]
name = "libp2p-tcp"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9e80ad4e3535345f3d666554ce347d3100453775611c05c60786bf9a1747a10"
+checksum = "e25c9d9c5448c189bba7ecdd1ca23800516281476e82810eff711ef04abaf9eb"
dependencies = [
"async-std",
"futures 0.3.4",
@@ -2954,9 +2966,9 @@ dependencies = [
[[package]]
name = "libp2p-uds"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "76d329564a43da9d0e055a5b938633c4a8ceab1f59cec133fbc4647917c07341"
+checksum = "d8dbcbe6567ea1b3c98ba4df5fd9d1b7c2bebbf50d46ceb0c2ce735c55af3f8d"
dependencies = [
"async-std",
"futures 0.3.4",
@@ -2966,9 +2978,9 @@ dependencies = [
[[package]]
name = "libp2p-wasm-ext"
-version = "0.16.2"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "923581c055bc4b8c5f42d4ce5ef43e52fe5216f1ea4bc26476cb8a966ce6220b"
+checksum = "076446cabb23b0d79d2375661d837a43cbda6719d88787f234e7661c33ef9554"
dependencies = [
"futures 0.3.4",
"js-sys",
@@ -2980,9 +2992,9 @@ dependencies = [
[[package]]
name = "libp2p-websocket"
-version = "0.16.0"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5351ca9eea122081c1c0f9323164d2918cac29b5a6bfe5054d4ba8ec9447cf42"
+checksum = "a0117ed6a6f60114c107c1232a0890a8fe997013c7e1920b6f0c811e05d2fae7"
dependencies = [
"async-tls",
"bytes 0.5.4",
@@ -3001,9 +3013,9 @@ dependencies = [
[[package]]
name = "libp2p-yamux"
-version = "0.16.2"
+version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9dac30de24ccde0e67f363d71a125c587bbe6589503f664947e9b084b68a34f1"
+checksum = "ee12c49426527908f81ffb6551b95f57149a8ea64f386bb7da3b123cdb9c01ba"
dependencies = [
"futures 0.3.4",
"libp2p-core",
@@ -3268,6 +3280,21 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0debeb9fcf88823ea64d64e4a815ab1643f33127d995978e099942ce38f25238"
+[[package]]
+name = "multihash"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47fbc227f7e2b1cb701f95404579ecb2668abbdd3c7ef7a6cbb3cc0d3b236869"
+dependencies = [
+ "blake2b_simd",
+ "blake2s_simd",
+ "digest",
+ "sha-1",
+ "sha2",
+ "sha3",
+ "unsigned-varint",
+]
+
[[package]]
name = "multimap"
version = "0.8.1"
@@ -3276,15 +3303,15 @@ checksum = "d8883adfde9756c1d30b0f519c9b8c502a94b41ac62f696453c37c7fc0a958ce"
[[package]]
name = "multistream-select"
-version = "0.7.0"
+version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f938ffe420493e77c8b6cbcc3f282283f68fc889c5dcbc8e51668d5f3a01ad94"
+checksum = "74cdcf7cfb3402881e15a1f95116cb033d69b33c83d481e1234777f5ef0c3d2c"
dependencies = [
"bytes 0.5.4",
- "futures 0.1.29",
+ "futures 0.3.4",
"log",
+ "pin-project",
"smallvec 1.2.0",
- "tokio-io",
"unsigned-varint",
]
@@ -4685,6 +4712,24 @@ dependencies = [
"url 2.1.1",
]
+[[package]]
+name = "parity-multiaddr"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4db35e222f783ef4e6661873f6c165c4eb7b65e0c408349818517d5705c2d7d3"
+dependencies = [
+ "arrayref",
+ "bs58",
+ "byteorder 1.3.4",
+ "data-encoding",
+ "multihash",
+ "percent-encoding 2.1.0",
+ "serde",
+ "static_assertions",
+ "unsigned-varint",
+ "url 2.1.1",
+]
+
[[package]]
name = "parity-multihash"
version = "0.2.3"
@@ -4827,9 +4872,9 @@ dependencies = [
[[package]]
name = "paste"
-version = "0.1.10"
+version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab4fb1930692d1b6a9cfabdde3d06ea0a7d186518e2f4d67660d8970e2fa647a"
+checksum = "092d791bf7847f70bbd49085489fba25fc2c193571752bff9e36e74e72403932"
dependencies = [
"paste-impl",
"proc-macro-hack",
@@ -4837,9 +4882,9 @@ dependencies = [
[[package]]
name = "paste-impl"
-version = "0.1.10"
+version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a62486e111e571b1e93b710b61e8f493c0013be39629b714cb166bdb06aa5a8a"
+checksum = "406c23fb4c45cc6f68a9bbabb8ec7bd6f8cfcbd17e9e8f72c2460282f8325729"
dependencies = [
"proc-macro-hack",
"proc-macro2",
@@ -5848,7 +5893,7 @@ dependencies = [
"substrate-prometheus-endpoint",
"tempfile",
"time",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -6107,7 +6152,7 @@ dependencies = [
"substrate-test-runtime-client",
"substrate-test-runtime-transaction-pool",
"tempfile",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -6285,7 +6330,7 @@ dependencies = [
"substrate-prometheus-endpoint",
"substrate-test-runtime-client",
"tempfile",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -6450,7 +6495,7 @@ dependencies = [
"sp-utils",
"substrate-test-runtime-client",
"threadpool",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -6566,7 +6611,7 @@ dependencies = [
"lazy_static",
"log",
"netstat2",
- "parity-multiaddr",
+ "parity-multiaddr 0.7.3",
"parity-scale-codec",
"parity-util-mem",
"parking_lot 0.10.0",
@@ -6866,18 +6911,18 @@ checksum = "f638d531eccd6e23b980caf34876660d38e265409d8e99b397ab71eb3612fad0"
[[package]]
name = "serde"
-version = "1.0.106"
+version = "1.0.105"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "36df6ac6412072f67cf767ebbde4133a5b2e88e76dc6187fa7104cd16f783399"
+checksum = "e707fbbf255b8fc8c3b99abb91e7257a622caeb20a9818cbadbeeede4e0932ff"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
-version = "1.0.106"
+version = "1.0.105"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e549e3abf4fb8621bd1609f11dfc9f5e50320802273b12f3811a67e6716ea6c"
+checksum = "ac5d00fc561ba2724df6758a17de23df5914f20e41cb00f94d5b7ae42fffaff8"
dependencies = [
"proc-macro2",
"quote 1.0.3",
@@ -6886,9 +6931,9 @@ dependencies = [
[[package]]
name = "serde_json"
-version = "1.0.51"
+version = "1.0.50"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da07b57ee2623368351e9a0488bb0b261322a15a6e0ae53e243cbdc0f4208da9"
+checksum = "78a7a12c167809363ec3bd7329fc0a3369056996de43c4b37ef3cd54a6ce4867"
dependencies = [
"itoa",
"ryu",
@@ -7329,7 +7374,7 @@ dependencies = [
"sp-storage",
"substrate-bip39",
"tiny-bip39",
- "tiny-keccak 2.0.2",
+ "tiny-keccak 2.0.1",
"twox-hash",
"wasmi",
"zeroize",
@@ -7901,7 +7946,7 @@ dependencies = [
"sc-rpc-api",
"serde",
"sp-storage",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -7937,7 +7982,7 @@ dependencies = [
"hyper 0.13.4",
"log",
"prometheus",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -8339,9 +8384,9 @@ dependencies = [
[[package]]
name = "tiny-keccak"
-version = "2.0.2"
+version = "2.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c9d3793400a45f954c52e73d068316d76b6f4e36977e3fcebb13a2721e80237"
+checksum = "2953ca5148619bc99695c1274cb54c5275bbb913c6adad87e72eaf8db9787f69"
dependencies = [
"crunchy",
]
@@ -8382,9 +8427,9 @@ dependencies = [
[[package]]
name = "tokio"
-version = "0.2.16"
+version = "0.2.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee5a0dd887e37d37390c13ff8ac830f992307fe30a1fff0ab8427af67211ba28"
+checksum = "0fa5e81d6bc4e67fe889d5783bd2a128ab2e0cfa487e0be16b6a8d177b101616"
dependencies = [
"bytes 0.5.4",
"fnv",
@@ -8516,7 +8561,7 @@ checksum = "4adb8b3e5f86b707f1b54e7c15b6de52617a823608ccda98a15d3a24222f265a"
dependencies = [
"futures-core",
"rustls 0.17.0",
- "tokio 0.2.16",
+ "tokio 0.2.13",
"webpki",
]
@@ -8628,7 +8673,7 @@ dependencies = [
"futures-sink",
"log",
"pin-project-lite",
- "tokio 0.2.16",
+ "tokio 0.2.13",
]
[[package]]
@@ -9165,18 +9210,18 @@ dependencies = [
[[package]]
name = "wast"
-version = "13.0.0"
+version = "12.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b20abd8b4a26f7e0d4dd5e357e90a3d555ec190e94472c9b2b27c5b9777f9ae"
+checksum = "0615ba420811bcda39cf80e8a1bd75997aec09222bda35165920a07ef15cc695"
dependencies = [
"leb128",
]
[[package]]
name = "wat"
-version = "1.0.14"
+version = "1.0.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "51a615830ee3e7200b505c441fec09aac2f114deae69df52f215cb828ba112c4"
+checksum = "095f615fbfcae695e3a4cea7d9f02f70561c81274c0142f45a12bf1e154d08bd"
dependencies = [
"wast",
]
diff --git a/substrate/Cargo.toml b/substrate/Cargo.toml
index b5829073853dc21cdf305b39de6564421cb63505..035ae7bc37184f3de33c44ced4f0c3aa899768bf 100644
--- a/substrate/Cargo.toml
+++ b/substrate/Cargo.toml
@@ -175,3 +175,4 @@ members = [
[profile.release]
# Substrate runtime requires unwinding.
panic = "unwind"
+
diff --git a/substrate/bin/utils/subkey/Cargo.toml b/substrate/bin/utils/subkey/Cargo.toml
index 9bf20146a990d8dd876495caa867812d331949a5..672f25275c74fbfbab2c1f0908cfbefef8afed42 100644
--- a/substrate/bin/utils/subkey/Cargo.toml
+++ b/substrate/bin/utils/subkey/Cargo.toml
@@ -29,7 +29,7 @@ derive_more = { version = "0.99.2" }
sc-rpc = { version = "2.0.0-alpha.5", path = "../../../client/rpc" }
jsonrpc-core-client = { version = "14.0.3", features = ["http"] }
hyper = "0.12.35"
-libp2p = "0.16.2"
+libp2p = "0.17.0"
serde_json = "1.0"
[features]
diff --git a/substrate/client/authority-discovery/Cargo.toml b/substrate/client/authority-discovery/Cargo.toml
index 3fe4de13e33b9daa1e37083dcfd9b53bb69ade84..7521101ae6f2a2f0e1df77bb25df381c45e3bf66 100644
--- a/substrate/client/authority-discovery/Cargo.toml
+++ b/substrate/client/authority-discovery/Cargo.toml
@@ -18,7 +18,7 @@ codec = { package = "parity-scale-codec", default-features = false, version = "1
derive_more = "0.99.2"
futures = "0.3.4"
futures-timer = "3.0.1"
-libp2p = { version = "0.16.2", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
+libp2p = { version = "0.17.0", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
log = "0.4.8"
prometheus-endpoint = { package = "substrate-prometheus-endpoint", path = "../../utils/prometheus", version = "0.8.0-alpha.5"}
prost = "0.6.1"
diff --git a/substrate/client/authority-discovery/src/lib.rs b/substrate/client/authority-discovery/src/lib.rs
index 176e8b2e81dcd2e53bb13ed906e54a78a68f85d3..45f05e1039a5bdc1c5e8eb5355117ff5a21719d4 100644
--- a/substrate/client/authority-discovery/src/lib.rs
+++ b/substrate/client/authority-discovery/src/lib.rs
@@ -300,7 +300,7 @@ where
.map_err(Error::EncodingProto)?;
self.network.put_value(
- hash_authority_id(key.1.as_ref())?,
+ hash_authority_id(key.1.as_ref()),
signed_addresses,
);
}
@@ -323,7 +323,7 @@ where
for authority_id in authorities.iter() {
self.network
- .get_value(&hash_authority_id(authority_id.as_ref())?);
+ .get_value(&hash_authority_id(authority_id.as_ref()));
}
Ok(())
@@ -408,8 +408,8 @@ where
self.addr_cache.retain_ids(&authorities);
authorities
.into_iter()
- .map(|id| hash_authority_id(id.as_ref()).map(|h| (h, id)))
- .collect::<Result<HashMap<_, _>>>()?
+ .map(|id| (hash_authority_id(id.as_ref()), id))
+ .collect::<HashMap<_, _>>()
};
// Check if the event origins from an authority in the current authority set.
@@ -586,10 +586,8 @@ where
}
}
-fn hash_authority_id(id: &[u8]) -> Result<libp2p::kad::record::Key> {
- libp2p::multihash::encode(libp2p::multihash::Hash::SHA2256, id)
- .map(|k| libp2p::kad::record::Key::new(&k))
- .map_err(Error::HashingAuthorityId)
+fn hash_authority_id(id: &[u8]) -> libp2p::kad::record::Key {
+ libp2p::kad::record::Key::new(&libp2p::multihash::Sha2_256::digest(id))
}
fn interval_at(start: Instant, duration: Duration) -> Interval {
diff --git a/substrate/client/authority-discovery/src/tests.rs b/substrate/client/authority-discovery/src/tests.rs
index d23836d6fac9cc3b0c677dafbae4da048159ddf6..358376e5db471f4aa218b9aa13ed700b7544ac95 100644
--- a/substrate/client/authority-discovery/src/tests.rs
+++ b/substrate/client/authority-discovery/src/tests.rs
@@ -304,7 +304,7 @@ fn handle_dht_events_with_value_found_should_call_set_priority_group() {
// Create sample dht event.
- let authority_id_1 = hash_authority_id(key_pair.public().as_ref()).unwrap();
+ let authority_id_1 = hash_authority_id(key_pair.public().as_ref());
let address_1: Multiaddr = "/ip6/2001:db8::".parse().unwrap();
let mut serialized_addresses = vec![];
diff --git a/substrate/client/network-gossip/Cargo.toml b/substrate/client/network-gossip/Cargo.toml
index ad2eb2ae0e391f31ce4551c29001f1a2d20ef922..c2887ed71925facd68b91fc2f08b2424181680d1 100644
--- a/substrate/client/network-gossip/Cargo.toml
+++ b/substrate/client/network-gossip/Cargo.toml
@@ -13,7 +13,7 @@ documentation = "https://docs.rs/sc-network-gossip"
[dependencies]
futures = "0.3.4"
futures-timer = "3.0.1"
-libp2p = { version = "0.16.2", default-features = false, features = ["libp2p-websocket"] }
+libp2p = { version = "0.17.0", default-features = false, features = ["websocket"] }
log = "0.4.8"
lru = "0.4.3"
sc-network = { version = "0.8.0-alpha.5", path = "../network" }
diff --git a/substrate/client/network/Cargo.toml b/substrate/client/network/Cargo.toml
index 8d67a15c354bdde4adfdee04a3b60293b55e86f7..5b4813e80a5fd57228dead65954f477eb6140f47 100644
--- a/substrate/client/network/Cargo.toml
+++ b/substrate/client/network/Cargo.toml
@@ -26,7 +26,6 @@ futures = "0.3.4"
futures_codec = "0.3.3"
futures-timer = "3.0.1"
wasm-timer = "0.2"
-libp2p = { version = "0.16.2", default-features = false, features = ["libp2p-websocket"] }
linked-hash-map = "0.5.2"
linked_hash_set = "0.1.3"
log = "0.4.8"
@@ -59,10 +58,16 @@ unsigned-varint = { version = "0.3.1", features = ["futures", "futures-codec"] }
void = "1.0.2"
zeroize = "1.0.0"
+[dependencies.libp2p]
+version = "0.17.0"
+default-features = false
+features = ["websocket", "kad", "mdns", "ping", "identify", "mplex", "yamux", "noise"]
+
[dev-dependencies]
async-std = "1.5"
assert_matches = "1.3"
env_logger = "0.7.0"
+libp2p = { version = "0.17.0", default-features = false, features = ["secio"] }
quickcheck = "0.9.0"
rand = "0.7.2"
sp-keyring = { version = "2.0.0-alpha.5", path = "../../primitives/keyring" }
diff --git a/substrate/client/network/src/debug_info.rs b/substrate/client/network/src/debug_info.rs
index 17fb622f7cd3cab44f3b17ab508d7f1a3b258882..e2803cde35a772e5e094919ef31b364af640c487 100644
--- a/substrate/client/network/src/debug_info.rs
+++ b/substrate/client/network/src/debug_info.rs
@@ -17,14 +17,15 @@
use fnv::FnvHashMap;
use futures::prelude::*;
use libp2p::Multiaddr;
-use libp2p::core::nodes::listeners::ListenerId;
+use libp2p::core::connection::{ConnectionId, ListenerId};
use libp2p::core::{ConnectedPoint, either::EitherOutput, PeerId, PublicKey};
use libp2p::swarm::{IntoProtocolsHandler, IntoProtocolsHandlerSelect, ProtocolsHandler};
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::identify::{Identify, IdentifyEvent, IdentifyInfo};
use libp2p::ping::{Ping, PingConfig, PingEvent, PingSuccess};
use log::{debug, trace, error};
-use std::error;
+use smallvec::SmallVec;
+use std::{error, io};
use std::collections::hash_map::Entry;
use std::pin::Pin;
use std::task::{Context, Poll};
@@ -56,14 +57,27 @@ struct NodeInfo {
/// When we will remove the entry about this node from the list, or `None` if we're connected
/// to the node.
info_expire: Option<Instant>,
- /// How we're connected to the node.
- endpoint: ConnectedPoint,
+ /// Non-empty list of connected endpoints, one per connection.
+ endpoints: SmallVec<[ConnectedPoint; crate::MAX_CONNECTIONS_PER_PEER]>,
/// Version reported by the remote, or `None` if unknown.
client_version: Option<String>,
/// Latest ping time with this node.
latest_ping: Option<Duration>,
}
+impl NodeInfo {
+ fn new(endpoint: ConnectedPoint) -> Self {
+ let mut endpoints = SmallVec::new();
+ endpoints.push(endpoint);
+ NodeInfo {
+ info_expire: None,
+ endpoints,
+ client_version: None,
+ latest_ping: None,
+ }
+ }
+}
+
impl DebugInfoBehaviour {
/// Builds a new `DebugInfoBehaviour`.
pub fn new(
@@ -121,9 +135,9 @@ impl DebugInfoBehaviour {
pub struct Node<'a>(&'a NodeInfo);
impl<'a> Node<'a> {
- /// Returns the endpoint we are connected to or were last connected to.
+ /// Returns the endpoint of an established connection to the peer.
pub fn endpoint(&self) -> &'a ConnectedPoint {
- &self.0.endpoint
+ &self.0.endpoints[0] // `endpoints` are non-empty by definition
}
/// Returns the latest version information we know of.
@@ -168,18 +182,17 @@ impl NetworkBehaviour for DebugInfoBehaviour {
list
}
- fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
- self.ping.inject_connected(peer_id.clone(), endpoint.clone());
- self.identify.inject_connected(peer_id.clone(), endpoint.clone());
+ fn inject_connected(&mut self, peer_id: &PeerId) {
+ self.ping.inject_connected(peer_id);
+ self.identify.inject_connected(peer_id);
+ }
- match self.nodes_info.entry(peer_id) {
+ fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.ping.inject_connection_established(peer_id, conn, endpoint);
+ self.identify.inject_connection_established(peer_id, conn, endpoint);
+ match self.nodes_info.entry(peer_id.clone()) {
Entry::Vacant(e) => {
- e.insert(NodeInfo {
- info_expire: None,
- endpoint,
- client_version: None,
- latest_ping: None,
- });
+ e.insert(NodeInfo::new(endpoint.clone()));
}
Entry::Occupied(e) => {
let e = e.into_mut();
@@ -188,14 +201,26 @@ impl NetworkBehaviour for DebugInfoBehaviour {
e.latest_ping = None;
}
e.info_expire = None;
- e.endpoint = endpoint;
+ e.endpoints.push(endpoint.clone());
}
}
}
- fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
- self.ping.inject_disconnected(peer_id, endpoint.clone());
- self.identify.inject_disconnected(peer_id, endpoint);
+ fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.ping.inject_connection_closed(peer_id, conn, endpoint);
+ self.identify.inject_connection_closed(peer_id, conn, endpoint);
+
+ if let Some(entry) = self.nodes_info.get_mut(peer_id) {
+ entry.endpoints.retain(|ep| ep != endpoint)
+ } else {
+ error!(target: "sub-libp2p",
+ "Unknown connection to {:?} closed: {:?}", peer_id, endpoint);
+ }
+ }
+
+ fn inject_disconnected(&mut self, peer_id: &PeerId) {
+ self.ping.inject_disconnected(peer_id);
+ self.identify.inject_disconnected(peer_id);
if let Some(entry) = self.nodes_info.get_mut(peer_id) {
entry.info_expire = Some(Instant::now() + CACHE_EXPIRE);
@@ -205,26 +230,15 @@ impl NetworkBehaviour for DebugInfoBehaviour {
}
}
- fn inject_node_event(
+ fn inject_event(
&mut self,
peer_id: PeerId,
+ connection: ConnectionId,
event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent
) {
match event {
- EitherOutput::First(event) => self.ping.inject_node_event(peer_id, event),
- EitherOutput::Second(event) => self.identify.inject_node_event(peer_id, event),
- }
- }
-
- fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
- self.ping.inject_replaced(peer_id.clone(), closed_endpoint.clone(), new_endpoint.clone());
- self.identify.inject_replaced(peer_id.clone(), closed_endpoint, new_endpoint.clone());
-
- if let Some(entry) = self.nodes_info.get_mut(&peer_id) {
- entry.endpoint = new_endpoint;
- } else {
- error!(target: "sub-libp2p",
- "Disconnected from node we were not connected to {:?}", peer_id);
+ EitherOutput::First(event) => self.ping.inject_event(peer_id, connection, event),
+ EitherOutput::Second(event) => self.identify.inject_event(peer_id, connection, event),
}
}
@@ -258,9 +272,9 @@ impl NetworkBehaviour for DebugInfoBehaviour {
self.identify.inject_listener_error(id, err);
}
- fn inject_listener_closed(&mut self, id: ListenerId) {
- self.ping.inject_listener_closed(id);
- self.identify.inject_listener_closed(id);
+ fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
+ self.ping.inject_listener_closed(id, reason);
+ self.identify.inject_listener_closed(id, reason);
}
fn poll(
@@ -283,11 +297,12 @@ impl NetworkBehaviour for DebugInfoBehaviour {
},
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
- Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }) =>
- return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
- Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) =>
- return Poll::Ready(NetworkBehaviourAction::SendEvent {
+ Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }) =>
+ return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
+ Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) =>
+ return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
peer_id,
+ handler,
event: EitherOutput::First(event)
}),
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) =>
@@ -312,11 +327,12 @@ impl NetworkBehaviour for DebugInfoBehaviour {
},
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
- Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }) =>
- return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
- Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) =>
- return Poll::Ready(NetworkBehaviourAction::SendEvent {
+ Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }) =>
+ return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
+ Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) =>
+ return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
peer_id,
+ handler,
event: EitherOutput::Second(event)
}),
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) =>
diff --git a/substrate/client/network/src/discovery.rs b/substrate/client/network/src/discovery.rs
index ed5016642be8b3e47e27588d2a86f3a01ea98614..a72c3cce65bd4a66a36bec3998920dc9f8578fc7 100644
--- a/substrate/client/network/src/discovery.rs
+++ b/substrate/client/network/src/discovery.rs
@@ -47,7 +47,7 @@
use futures::prelude::*;
use futures_timer::Delay;
-use libp2p::core::{nodes::listeners::ListenerId, ConnectedPoint, Multiaddr, PeerId, PublicKey};
+use libp2p::core::{connection::{ConnectionId, ListenerId}, ConnectedPoint, Multiaddr, PeerId, PublicKey};
use libp2p::swarm::{ProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use libp2p::kad::{Kademlia, KademliaEvent, Quorum, Record};
use libp2p::kad::GetClosestPeersError;
@@ -58,7 +58,7 @@ use libp2p::{swarm::toggle::Toggle};
use libp2p::mdns::{Mdns, MdnsEvent};
use libp2p::multiaddr::Protocol;
use log::{debug, info, trace, warn, error};
-use std::{cmp, collections::VecDeque, time::Duration};
+use std::{cmp, collections::VecDeque, io, time::Duration};
use std::task::{Context, Poll};
use sp_core::hexdisplay::HexDisplay;
@@ -149,6 +149,7 @@ impl DiscoveryBehaviour {
/// If we didn't know this address before, also generates a `Discovered` event.
pub fn add_known_address(&mut self, peer_id: PeerId, addr: Multiaddr) {
if self.user_defined.iter().all(|(p, a)| *p != peer_id && *a != addr) {
+ self.kademlia.add_address(&peer_id, addr.clone());
self.discoveries.push_back(peer_id.clone());
self.user_defined.push((peer_id, addr));
}
@@ -259,18 +260,22 @@ impl NetworkBehaviour for DiscoveryBehaviour {
list
}
- fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
+ fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
self.num_connections += 1;
- NetworkBehaviour::inject_connected(&mut self.kademlia, peer_id, endpoint)
+ NetworkBehaviour::inject_connection_established(&mut self.kademlia, peer_id, conn, endpoint)
}
- fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
+ fn inject_connected(&mut self, peer_id: &PeerId) {
+ NetworkBehaviour::inject_connected(&mut self.kademlia, peer_id)
+ }
+
+ fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
self.num_connections -= 1;
- NetworkBehaviour::inject_disconnected(&mut self.kademlia, peer_id, endpoint)
+ NetworkBehaviour::inject_connection_closed(&mut self.kademlia, peer_id, conn, endpoint)
}
- fn inject_replaced(&mut self, peer_id: PeerId, closed: ConnectedPoint, opened: ConnectedPoint) {
- NetworkBehaviour::inject_replaced(&mut self.kademlia, peer_id, closed, opened)
+ fn inject_disconnected(&mut self, peer_id: &PeerId) {
+ NetworkBehaviour::inject_disconnected(&mut self.kademlia, peer_id)
}
fn inject_addr_reach_failure(
@@ -282,12 +287,13 @@ impl NetworkBehaviour for DiscoveryBehaviour {
NetworkBehaviour::inject_addr_reach_failure(&mut self.kademlia, peer_id, addr, error)
}
- fn inject_node_event(
+ fn inject_event(
&mut self,
peer_id: PeerId,
+ connection: ConnectionId,
event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
- NetworkBehaviour::inject_node_event(&mut self.kademlia, peer_id, event)
+ NetworkBehaviour::inject_event(&mut self.kademlia, peer_id, connection, event)
}
fn inject_new_external_addr(&mut self, addr: &Multiaddr) {
@@ -315,9 +321,8 @@ impl NetworkBehaviour for DiscoveryBehaviour {
NetworkBehaviour::inject_listener_error(&mut self.kademlia, id, err);
}
- fn inject_listener_closed(&mut self, id: ListenerId) {
- error!(target: "sub-libp2p", "Libp2p listener {:?} closed", id);
- NetworkBehaviour::inject_listener_closed(&mut self.kademlia, id);
+ fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
+ NetworkBehaviour::inject_listener_closed(&mut self.kademlia, id, reason);
}
fn poll(
@@ -340,8 +345,9 @@ impl NetworkBehaviour for DiscoveryBehaviour {
while let Poll::Ready(_) = self.next_kad_random_query.poll_unpin(cx) {
let actually_started = if self.num_connections < self.discovery_only_if_under_num {
let random_peer_id = PeerId::random();
- debug!(target: "sub-libp2p", "Libp2p <= Starting random Kademlia request for \
- {:?}", random_peer_id);
+ debug!(target: "sub-libp2p",
+ "Libp2p <= Starting random Kademlia request for {:?}",
+ random_peer_id);
self.kademlia.get_closest_peers(random_peer_id);
true
@@ -451,10 +457,10 @@ impl NetworkBehaviour for DiscoveryBehaviour {
},
NetworkBehaviourAction::DialAddress { address } =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
- NetworkBehaviourAction::DialPeer { peer_id } =>
- return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
- NetworkBehaviourAction::SendEvent { peer_id, event } =>
- return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }),
+ NetworkBehaviourAction::DialPeer { peer_id, condition } =>
+ return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
+ NetworkBehaviourAction::NotifyHandler { peer_id, handler, event } =>
+ return Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }),
NetworkBehaviourAction::ReportObservedAddr { address } =>
return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }),
}
@@ -482,9 +488,9 @@ impl NetworkBehaviour for DiscoveryBehaviour {
},
NetworkBehaviourAction::DialAddress { address } =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
- NetworkBehaviourAction::DialPeer { peer_id } =>
- return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
- NetworkBehaviourAction::SendEvent { event, .. } =>
+ NetworkBehaviourAction::DialPeer { peer_id, condition } =>
+ return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
+ NetworkBehaviourAction::NotifyHandler { event, .. } =>
match event {}, // `event` is an enum with no variant
NetworkBehaviourAction::ReportObservedAddr { address } =>
return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }),
diff --git a/substrate/client/network/src/lib.rs b/substrate/client/network/src/lib.rs
index b425a7763b648fe87e234e56a992f954e2c8f369..d8afa1f1530ef8935df8a7177cd4d696d0c4adc2 100644
--- a/substrate/client/network/src/lib.rs
+++ b/substrate/client/network/src/lib.rs
@@ -255,3 +255,12 @@ pub use libp2p::{Multiaddr, PeerId};
pub use libp2p::multiaddr;
pub use sc_peerset::ReputationChange;
+
+/// The maximum allowed number of established connections per peer.
+///
+/// Typically, and by design of the network behaviours in this crate,
+/// there is a single established connection per peer. However, to
+/// avoid unnecessary and nondeterministic connection closure in
+/// case of (possibly repeated) simultaneous dialing attempts between
+/// two peers, the per-peer connection limit is not set to 1 but 2.
+const MAX_CONNECTIONS_PER_PEER: usize = 2;
diff --git a/substrate/client/network/src/protocol.rs b/substrate/client/network/src/protocol.rs
index bfe8226c8d61529d3817699731c3b4cd7fdbd1c3..365cbcbc5286f2efc643d3d563e06b2dc223b995 100644
--- a/substrate/client/network/src/protocol.rs
+++ b/substrate/client/network/src/protocol.rs
@@ -20,7 +20,7 @@ use bytes::{Bytes, BytesMut};
use futures::prelude::*;
use generic_proto::{GenericProto, GenericProtoOut};
use libp2p::{Multiaddr, PeerId};
-use libp2p::core::{ConnectedPoint, nodes::listeners::ListenerId};
+use libp2p::core::{ConnectedPoint, connection::{ConnectionId, ListenerId}};
use libp2p::swarm::{ProtocolsHandler, IntoProtocolsHandler};
use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use sp_core::{
@@ -48,7 +48,7 @@ use std::borrow::Cow;
use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::fmt::Write;
-use std::{cmp, num::NonZeroUsize, pin::Pin, task::Poll, time};
+use std::{cmp, io, num::NonZeroUsize, pin::Pin, task::Poll, time};
use log::{log, Level, trace, debug, warn, error};
use crate::chain::{Client, FinalityProofProvider};
use sc_client_api::{ChangesProof, StorageProof};
@@ -1830,20 +1830,29 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
self.behaviour.addresses_of_peer(peer_id)
}
- fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
- self.behaviour.inject_connected(peer_id, endpoint)
+ fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.behaviour.inject_connection_established(peer_id, conn, endpoint)
}
- fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
- self.behaviour.inject_disconnected(peer_id, endpoint)
+ fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.behaviour.inject_connection_closed(peer_id, conn, endpoint)
}
- fn inject_node_event(
+ fn inject_connected(&mut self, peer_id: &PeerId) {
+ self.behaviour.inject_connected(peer_id)
+ }
+
+ fn inject_disconnected(&mut self, peer_id: &PeerId) {
+ self.behaviour.inject_disconnected(peer_id)
+ }
+
+ fn inject_event(
&mut self,
peer_id: PeerId,
+ connection: ConnectionId,
event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent,
) {
- self.behaviour.inject_node_event(peer_id, event)
+ self.behaviour.inject_event(peer_id, connection, event)
}
fn poll(
@@ -1900,10 +1909,10 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
Poll::Ready(NetworkBehaviourAction::GenerateEvent(ev)) => ev,
Poll::Ready(NetworkBehaviourAction::DialAddress { address }) =>
return Poll::Ready(NetworkBehaviourAction::DialAddress { address }),
- Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }) =>
- return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id }),
- Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }) =>
- return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, event }),
+ Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }) =>
+ return Poll::Ready(NetworkBehaviourAction::DialPeer { peer_id, condition }),
+ Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }) =>
+ return Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, handler, event }),
Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }) =>
return Poll::Ready(NetworkBehaviourAction::ReportObservedAddr { address }),
};
@@ -1967,10 +1976,6 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
}
}
- fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
- self.behaviour.inject_replaced(peer_id, closed_endpoint, new_endpoint)
- }
-
fn inject_addr_reach_failure(
&mut self,
peer_id: Option<&PeerId>,
@@ -2000,8 +2005,8 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
self.behaviour.inject_listener_error(id, err);
}
- fn inject_listener_closed(&mut self, id: ListenerId) {
- self.behaviour.inject_listener_closed(id);
+ fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
+ self.behaviour.inject_listener_closed(id, reason);
}
}
diff --git a/substrate/client/network/src/protocol/block_requests.rs b/substrate/client/network/src/protocol/block_requests.rs
index 5a947c0b6b5854ce05e393ea563976bb85656592..6af5023d39fe6b20661a4883c0d8a47416091c4e 100644
--- a/substrate/client/network/src/protocol/block_requests.rs
+++ b/substrate/client/network/src/protocol/block_requests.rs
@@ -35,6 +35,7 @@ use libp2p::{
ConnectedPoint,
Multiaddr,
PeerId,
+ connection::ConnectionId,
upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated},
upgrade::{DeniedUpgrade, read_one, write_one}
},
@@ -43,6 +44,7 @@ use libp2p::{
NetworkBehaviour,
NetworkBehaviourAction,
OneShotHandler,
+ OneShotHandlerConfig,
PollParameters,
SubstreamProtocol
}
@@ -257,20 +259,27 @@ where
max_request_len: self.config.max_request_len,
protocol: self.config.protocol.clone(),
};
- OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout)
+ let mut cfg = OneShotHandlerConfig::default();
+ cfg.inactive_timeout = self.config.inactivity_timeout;
+ OneShotHandler::new(SubstreamProtocol::new(p), cfg)
}
fn addresses_of_peer(&mut self, _: &PeerId) -> Vec<Multiaddr> {
Vec::new()
}
- fn inject_connected(&mut self, _peer: PeerId, _info: ConnectedPoint) {
+ fn inject_connected(&mut self, _peer: &PeerId) {
}
- fn inject_disconnected(&mut self, _peer: &PeerId, _info: ConnectedPoint) {
+ fn inject_disconnected(&mut self, _peer: &PeerId) {
}
- fn inject_node_event(&mut self, peer: PeerId, Request(request, mut stream): Request<NegotiatedSubstream>) {
+ fn inject_event(
+ &mut self,
+ peer: PeerId,
+ connection: ConnectionId,
+ Request(request, mut stream): Request<NegotiatedSubstream>
+ ) {
match self.on_block_request(&peer, &request) {
Ok(res) => {
log::trace!("enqueueing block response for peer {} with {} blocks", peer, res.blocks.len());
diff --git a/substrate/client/network/src/protocol/generic_proto/behaviour.rs b/substrate/client/network/src/protocol/generic_proto/behaviour.rs
index c398f6df2d409e93cd774e8b2b2de6dcd048ae18..b38a97cb8f537501716f3923b6cb1501d9d4f5c8 100644
--- a/substrate/client/network/src/protocol/generic_proto/behaviour.rs
+++ b/substrate/client/network/src/protocol/generic_proto/behaviour.rs
@@ -21,8 +21,14 @@ use crate::protocol::generic_proto::upgrade::RegisteredProtocol;
use bytes::BytesMut;
use fnv::FnvHashMap;
use futures::prelude::*;
-use libp2p::core::{ConnectedPoint, Multiaddr, PeerId};
-use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
+use libp2p::core::{ConnectedPoint, Multiaddr, PeerId, connection::ConnectionId};
+use libp2p::swarm::{
+ DialPeerCondition,
+ NetworkBehaviour,
+ NetworkBehaviourAction,
+ NotifyHandler,
+ PollParameters
+};
use log::{debug, error, trace, warn};
use prometheus_endpoint::HistogramVec;
use rand::distributions::{Distribution as _, Uniform};
@@ -32,15 +38,16 @@ use std::{borrow::Cow, cmp, collections::hash_map::Entry};
use std::{error, mem, pin::Pin, str, time::Duration};
use wasm_timer::Instant;
-/// Network behaviour that handles opening substreams for custom protocols with other nodes.
+/// Network behaviour that handles opening substreams for custom protocols with other peers.
///
/// ## Legacy vs new protocol
///
/// The `GenericProto` behaves as following:
///
-/// - Whenever a connection is established, we open a single substream (called "legay protocol" in
-/// the source code). This substream name depends on the `protocol_id` and `versions` passed at
-/// initialization. If the remote refuses this substream, we close the connection.
+/// - Whenever a connection is established, we open a single substream (called "legacy protocol" in
+/// the source code) on that connection. This substream name depends on the `protocol_id` and
+/// `versions` passed at initialization. If the remote refuses this substream, we close the
+/// connection.
///
/// - For each registered protocol, we also open an additional substream for this protocol. If the
/// remote refuses this substream, then it's fine.
@@ -55,27 +62,51 @@ use wasm_timer::Instant;
///
/// - The libp2p swarm that opens new connections and reports disconnects.
/// - The connection handler (see `handler.rs`) that handles individual connections.
-/// - The peerset manager (PSM) that requests links to nodes to be established or broken.
+/// - The peerset manager (PSM) that requests links to peers to be established or broken.
/// - The external API, that requires knowledge of the links that have been established.
///
/// Each connection handler can be in four different states: Enabled+Open, Enabled+Closed,
/// Disabled+Open, or Disabled+Closed. The Enabled/Disabled component must be in sync with the
/// peerset manager. For example, if the peerset manager requires a disconnection, we disable the
-/// existing handler. The Open/Closed component must be in sync with the external API.
+/// connection handlers of that peer. The Open/Closed component must be in sync with the external
+/// API.
///
-/// However a connection handler only exists if we are actually connected to a node. What this
-/// means is that there are six possible states for each node: Disconnected, Dialing (trying to
-/// reach it), Enabled+Open, Enabled+Closed, Disabled+open, Disabled+Closed. Most notably, the
-/// Dialing state must correspond to a "link established" state in the peerset manager. In other
-/// words, the peerset manager doesn't differentiate whether we are dialing a node or connected
-/// to it.
+/// However, a connection handler for a peer only exists if we are actually connected to that peer.
+/// What this means is that there are six possible states for each peer: Disconnected, Dialing
+/// (trying to connect), Enabled+Open, Enabled+Closed, Disabled+Open, Disabled+Closed.
+/// Most notably, the Dialing state must correspond to a "link established" state in the peerset
+/// manager. In other words, the peerset manager doesn't differentiate whether we are dialing a
+/// peer or connected to it.
///
-/// Additionally, there also exists a "banning" system. If we fail to dial a node, we "ban" it for
-/// a few seconds. If the PSM requests a node that is in the "banned" state, then we delay the
-/// actual dialing attempt until after the ban expires, but the PSM will still consider the link
-/// to be established.
-/// Note that this "banning" system is not an actual ban. If a "banned" node tries to connect to
-/// us, we accept the connection. The "banning" system is only about delaying dialing attempts.
+/// There may be multiple connections to a peer. However, the status of a peer on
+/// the API of this behaviour and towards the peerset manager is aggregated in
+/// the following way:
+///
+/// 1. The enabled/disabled status is the same across all connections, as
+/// decided by the peerset manager.
+/// 2. `send_packet` and `write_notification` always send all data over
+/// the same connection to preserve the ordering provided by the transport,
+/// as long as that connection is open. If it closes, a second open
+/// connection may take over, if one exists, but that case should be no
+/// different than a single connection failing and being re-established
+/// in terms of potential reordering and dropped messages. Messages can
+/// be received on any connection.
+/// 3. The behaviour reports `GenericProtoOut::CustomProtocolOpen` when the
+/// first connection reports `NotifsHandlerOut::Open`.
+/// 4. The behaviour reports `GenericProtoOut::CustomProtocolClosed` when the
+/// last connection reports `NotifsHandlerOut::Closed`.
+///
+/// In this way, the number of actual established connections to the peer is
+/// an implementation detail of this behaviour. Note that, in practice and at
+/// the time of this writing, there may be at most two connections to a peer
+/// and only as a result of simultaneous dialing. However, the implementation
+/// accommodates for any number of connections.
+///
+/// Additionally, there also exists a "banning" system. If we fail to dial a peer, we "ban" it for
+/// a few seconds. If the PSM requests connecting to a peer that is currently "banned", the next
+/// dialing attempt is delayed until after the ban expires. However, the PSM will still consider
+/// the peer to be connected. This "ban" is thus not a ban in a strict sense: If a "banned" peer
+/// tries to connect, the connection is accepted. A ban only delays dialing attempts.
///
pub struct GenericProto {
/// Legacy protocol to open with peers. Never modified.
@@ -113,14 +144,14 @@ enum PeerState {
/// the state machine code.
Poisoned,
- /// The peer misbehaved. If the PSM wants us to connect to this node, we will add an artificial
+ /// The peer misbehaved. If the PSM wants us to connect to this peer, we will add an artificial
/// delay to the connection.
Banned {
- /// Until when the node is banned.
+ /// Until when the peer is banned.
until: Instant,
},
- /// The peerset requested that we connect to this peer. We are not connected to this node.
+ /// The peerset requested that we connect to this peer. We are currently not connected.
PendingRequest {
/// When to actually start dialing.
timer: futures_timer::Delay,
@@ -131,16 +162,13 @@ enum PeerState {
/// The peerset requested that we connect to this peer. We are currently dialing this peer.
Requested,
- /// We are connected to this peer but the peerset refused it. This peer can still perform
- /// Kademlia queries and such, but should get disconnected in a few seconds.
+ /// We are connected to this peer but the peerset refused it.
+ ///
+ /// We may still have ongoing traffic with that peer, but it should cease shortly.
Disabled {
- /// How we are connected to this peer.
- connected_point: ConnectedPoint,
- /// If true, we still have a custom protocol open with it. It will likely get closed in
- /// a short amount of time, but we need to keep the information in order to not have a
- /// state mismatch.
- open: bool,
- /// If `Some`, the node is banned until the given `Instant`.
+ /// The connections that are currently open for custom protocol traffic.
+ open: SmallVec<[ConnectionId; crate::MAX_CONNECTIONS_PER_PEER]>,
+ /// If `Some`, any dial attempts to this peer are delayed until the given `Instant`.
banned_until: Option<Instant>,
},
@@ -148,12 +176,8 @@ enum PeerState {
/// will be enabled when `timer` fires. This peer can still perform Kademlia queries and such,
/// but should get disconnected in a few seconds.
DisabledPendingEnable {
- /// How we are connected to this peer.
- connected_point: ConnectedPoint,
- /// If true, we still have a custom protocol open with it. It will likely get closed in
- /// a short amount of time, but we need to keep the information in order to not have a
- /// state mismatch.
- open: bool,
+ /// The connections that are currently open for custom protocol traffic.
+ open: SmallVec<[ConnectionId; crate::MAX_CONNECTIONS_PER_PEER]>,
/// When to enable this remote.
timer: futures_timer::Delay,
/// When the `timer` will trigger.
@@ -163,33 +187,41 @@ enum PeerState {
/// We are connected to this peer and the peerset has accepted it. The handler is in the
/// enabled state.
Enabled {
- /// How we are connected to this peer.
- connected_point: ConnectedPoint,
- /// If true, we have a custom protocol open with this peer.
- open: bool,
+ /// The connections that are currently open for custom protocol traffic.
+ open: SmallVec<[ConnectionId; crate::MAX_CONNECTIONS_PER_PEER]>,
},
- /// We are connected to this peer, and we sent an incoming message to the peerset. The handler
- /// is in initialization mode. We are waiting for the Accept or Reject from the peerset. There
- /// is a corresponding entry in `incoming`.
- Incoming {
- /// How we are connected to this peer.
- connected_point: ConnectedPoint,
- },
+ /// We received an incoming connection from this peer and forwarded that
+ /// connection request to the peerset. The connection handlers are waiting
+ /// for initialisation, i.e. to be enabled or disabled based on whether
+ /// the peerset accepts or rejects the peer.
+ Incoming,
}
impl PeerState {
- /// True if we have an open channel with that node.
+ /// True if there exists an established connection to tbe peer
+ /// that is open for custom protocol traffic.
fn is_open(&self) -> bool {
+ self.get_open().is_some()
+ }
+
+ /// Returns the connection ID of the first established connection
+ /// that is open for custom protocol traffic.
+ fn get_open(&self) -> Option<ConnectionId> {
match self {
- PeerState::Poisoned => false,
- PeerState::Banned { .. } => false,
- PeerState::PendingRequest { .. } => false,
- PeerState::Requested => false,
- PeerState::Disabled { open, .. } => *open,
- PeerState::DisabledPendingEnable { open, .. } => *open,
- PeerState::Enabled { open, .. } => *open,
- PeerState::Incoming { .. } => false,
+ PeerState::Disabled { open, .. } |
+ PeerState::DisabledPendingEnable { open, .. } |
+ PeerState::Enabled { open, .. } =>
+ if !open.is_empty() {
+ Some(open[0])
+ } else {
+ None
+ }
+ PeerState::Poisoned => None,
+ PeerState::Banned { .. } => None,
+ PeerState::PendingRequest { .. } => None,
+ PeerState::Requested => None,
+ PeerState::Incoming { .. } => None,
}
}
@@ -211,7 +243,7 @@ impl PeerState {
/// State of an "incoming" message sent to the peer set manager.
#[derive(Debug)]
struct IncomingPeer {
- /// Id of the node that is concerned.
+ /// Id of the remote peer of the incoming connection.
peer_id: PeerId,
/// If true, this "incoming" still corresponds to an actual connection. If false, then the
/// connection corresponding to it has been closed or replaced already.
@@ -225,10 +257,8 @@ struct IncomingPeer {
pub enum GenericProtoOut {
/// Opened a custom protocol with the remote.
CustomProtocolOpen {
- /// Id of the node we have opened a connection with.
+ /// Id of the peer we are connected to.
peer_id: PeerId,
- /// Endpoint used for this custom protocol.
- endpoint: ConnectedPoint,
},
/// Closed a custom protocol with the remote.
@@ -316,7 +346,7 @@ impl GenericProto {
self.peers.iter().filter(|(_, state)| state.is_open()).map(|(id, _)| id)
}
- /// Returns true if we have a channel open with this node.
+ /// Returns true if we have an open connection to the given peer.
pub fn is_open(&self, peer_id: &PeerId) -> bool {
self.peers.get(peer_id).map(|p| p.is_open()).unwrap_or(false)
}
@@ -327,8 +357,8 @@ impl GenericProto {
self.disconnect_peer_inner(peer_id, None);
}
- /// Inner implementation of `disconnect_peer`. If `ban` is `Some`, we ban the node for the
- /// specific duration.
+ /// Inner implementation of `disconnect_peer`. If `ban` is `Some`, we ban the peer
+ /// for the specific duration.
fn disconnect_peer_inner(&mut self, peer_id: &PeerId, ban: Option<Duration>) {
let mut entry = if let Entry::Occupied(entry) = self.peers.entry(peer_id.clone()) {
entry
@@ -344,7 +374,11 @@ impl GenericProto {
st @ PeerState::Banned { .. } => *entry.into_mut() = st,
// DisabledPendingEnable => Disabled.
- PeerState::DisabledPendingEnable { open, connected_point, timer_deadline, .. } => {
+ PeerState::DisabledPendingEnable {
+ open,
+ timer_deadline,
+ timer: _
+ } => {
debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
self.peerset.dropped(peer_id.clone());
let banned_until = Some(if let Some(ban) = ban {
@@ -352,24 +386,31 @@ impl GenericProto {
} else {
timer_deadline
});
- *entry.into_mut() = PeerState::Disabled { open, connected_point, banned_until }
+ *entry.into_mut() = PeerState::Disabled {
+ open,
+ banned_until
+ }
},
// Enabled => Disabled.
- PeerState::Enabled { open, connected_point } => {
+ PeerState::Enabled { open } => {
debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
self.peerset.dropped(peer_id.clone());
debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: peer_id.clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Disable,
});
let banned_until = ban.map(|dur| Instant::now() + dur);
- *entry.into_mut() = PeerState::Disabled { open, connected_point, banned_until }
+ *entry.into_mut() = PeerState::Disabled {
+ open,
+ banned_until
+ }
},
// Incoming => Disabled.
- PeerState::Incoming { connected_point, .. } => {
+ PeerState::Incoming => {
let inc = if let Some(inc) = self.incoming.iter_mut()
.find(|i| i.peer_id == *entry.key() && i.alive) {
inc
@@ -381,12 +422,16 @@ impl GenericProto {
inc.alive = false;
debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: peer_id.clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Disable,
});
let banned_until = ban.map(|dur| Instant::now() + dur);
- *entry.into_mut() = PeerState::Disabled { open: false, connected_point, banned_until }
+ *entry.into_mut() = PeerState::Disabled {
+ open: SmallVec::new(),
+ banned_until
+ }
},
PeerState::Poisoned =>
@@ -441,9 +486,15 @@ impl GenericProto {
message: impl Into<Vec<u8>>,
encoded_fallback_message: Vec<u8>,
) {
- if !self.is_open(target) {
- return;
- }
+ let conn = match self.peers.get(target).and_then(|p| p.get_open()) {
+ None => {
+ debug!(target: "sub-libp2p",
+ "Tried to sent notification to {:?} without an open channel.",
+ target);
+ return
+ },
+ Some(conn) => conn
+ };
trace!(
target: "sub-libp2p",
@@ -453,8 +504,9 @@ impl GenericProto {
);
trace!(target: "sub-libp2p", "Handler({:?}) <= Packet", target);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: target.clone(),
+ handler: NotifyHandler::One(conn),
event: NotifsHandlerIn::SendNotification {
message: message.into(),
encoded_fallback_message,
@@ -470,14 +522,21 @@ impl GenericProto {
/// Also note that even we have a valid open substream, it may in fact be already closed
/// without us knowing, in which case the packet will not be received.
pub fn send_packet(&mut self, target: &PeerId, message: Vec<u8>) {
- if !self.is_open(target) {
- return;
- }
+ let conn = match self.peers.get(target).and_then(|p| p.get_open()) {
+ None => {
+ debug!(target: "sub-libp2p",
+ "Tried to sent packet to {:?} without an open channel.",
+ target);
+ return
+ }
+ Some(conn) => conn
+ };
trace!(target: "sub-libp2p", "External API => Packet for {:?}", target);
trace!(target: "sub-libp2p", "Handler({:?}) <= Packet", target);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: target.clone(),
+ handler: NotifyHandler::One(conn),
event: NotifsHandlerIn::SendLegacy {
message,
}
@@ -489,7 +548,7 @@ impl GenericProto {
self.peerset.debug_info()
}
- /// Function that is called when the peerset wants us to connect to a node.
+ /// Function that is called when the peerset wants us to connect to a peer.
fn peerset_report_connect(&mut self, peer_id: PeerId) {
let mut occ_entry = match self.peers.entry(peer_id) {
Entry::Occupied(entry) => entry,
@@ -497,7 +556,10 @@ impl GenericProto {
// If there's no entry in `self.peers`, start dialing.
debug!(target: "sub-libp2p", "PSM => Connect({:?}): Starting to connect", entry.key());
debug!(target: "sub-libp2p", "Libp2p <= Dial {:?}", entry.key());
- self.events.push(NetworkBehaviourAction::DialPeer { peer_id: entry.key().clone() });
+ self.events.push(NetworkBehaviourAction::DialPeer {
+ peer_id: entry.key().clone(),
+ condition: DialPeerCondition::Disconnected
+ });
entry.insert(PeerState::Requested);
return;
}
@@ -518,36 +580,41 @@ impl GenericProto {
PeerState::Banned { .. } => {
debug!(target: "sub-libp2p", "PSM => Connect({:?}): Starting to connect", occ_entry.key());
debug!(target: "sub-libp2p", "Libp2p <= Dial {:?}", occ_entry.key());
- self.events.push(NetworkBehaviourAction::DialPeer { peer_id: occ_entry.key().clone() });
+ self.events.push(NetworkBehaviourAction::DialPeer {
+ peer_id: occ_entry.key().clone(),
+ condition: DialPeerCondition::Disconnected
+ });
*occ_entry.into_mut() = PeerState::Requested;
},
- PeerState::Disabled { open, ref connected_point, banned_until: Some(ref banned) }
- if *banned > now => {
- debug!(target: "sub-libp2p", "PSM => Connect({:?}): Has idle connection through \
- {:?} but node is banned until {:?}", occ_entry.key(), connected_point, banned);
+ PeerState::Disabled {
+ open,
+ banned_until: Some(ref banned)
+ } if *banned > now => {
+ debug!(target: "sub-libp2p", "PSM => Connect({:?}): But peer is banned until {:?}",
+ occ_entry.key(), banned);
*occ_entry.into_mut() = PeerState::DisabledPendingEnable {
- connected_point: connected_point.clone(),
open,
timer: futures_timer::Delay::new(banned.clone() - now),
timer_deadline: banned.clone(),
};
},
- PeerState::Disabled { open, connected_point, banned_until: _ } => {
- debug!(target: "sub-libp2p", "PSM => Connect({:?}): Enabling previously-idle \
- connection through {:?}", occ_entry.key(), connected_point);
+ PeerState::Disabled { open, banned_until: _ } => {
+ debug!(target: "sub-libp2p", "PSM => Connect({:?}): Enabling connections.",
+ occ_entry.key());
debug!(target: "sub-libp2p", "Handler({:?}) <= Enable", occ_entry.key());
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: occ_entry.key().clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Enable,
});
- *occ_entry.into_mut() = PeerState::Enabled { connected_point, open };
+ *occ_entry.into_mut() = PeerState::Enabled { open };
},
- PeerState::Incoming { connected_point, .. } => {
- debug!(target: "sub-libp2p", "PSM => Connect({:?}): Enabling incoming \
- connection through {:?}", occ_entry.key(), connected_point);
+ PeerState::Incoming => {
+ debug!(target: "sub-libp2p", "PSM => Connect({:?}): Enabling connections.",
+ occ_entry.key());
if let Some(inc) = self.incoming.iter_mut()
.find(|i| i.peer_id == *occ_entry.key() && i.alive) {
inc.alive = false;
@@ -556,26 +623,30 @@ impl GenericProto {
incoming for incoming peer")
}
debug!(target: "sub-libp2p", "Handler({:?}) <= Enable", occ_entry.key());
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: occ_entry.key().clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Enable,
});
- *occ_entry.into_mut() = PeerState::Enabled { connected_point, open: false };
+ *occ_entry.into_mut() = PeerState::Enabled { open: SmallVec::new() };
},
st @ PeerState::Enabled { .. } => {
- warn!(target: "sub-libp2p", "PSM => Connect({:?}): Already connected to this \
- peer", occ_entry.key());
+ warn!(target: "sub-libp2p",
+ "PSM => Connect({:?}): Already connected.",
+ occ_entry.key());
*occ_entry.into_mut() = st;
},
st @ PeerState::DisabledPendingEnable { .. } => {
- warn!(target: "sub-libp2p", "PSM => Connect({:?}): Already have an idle \
- connection to this peer and waiting to enable it", occ_entry.key());
+ warn!(target: "sub-libp2p",
+ "PSM => Connect({:?}): Already pending enabling.",
+ occ_entry.key());
*occ_entry.into_mut() = st;
},
st @ PeerState::Requested { .. } | st @ PeerState::PendingRequest { .. } => {
- warn!(target: "sub-libp2p", "PSM => Connect({:?}): Received a previous \
- request for that peer", occ_entry.key());
+ warn!(target: "sub-libp2p",
+ "PSM => Connect({:?}): Duplicate request.",
+ occ_entry.key());
*occ_entry.into_mut() = st;
},
@@ -584,55 +655,63 @@ impl GenericProto {
}
}
- /// Function that is called when the peerset wants us to disconnect from a node.
+ /// Function that is called when the peerset wants us to disconnect from a peer.
fn peerset_report_disconnect(&mut self, peer_id: PeerId) {
let mut entry = match self.peers.entry(peer_id) {
Entry::Occupied(entry) => entry,
Entry::Vacant(entry) => {
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Node already disabled", entry.key());
+ debug!(target: "sub-libp2p", "PSM => Drop({:?}): Already disabled.", entry.key());
return
}
};
match mem::replace(entry.get_mut(), PeerState::Poisoned) {
st @ PeerState::Disabled { .. } | st @ PeerState::Banned { .. } => {
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Node already disabled", entry.key());
+ debug!(target: "sub-libp2p", "PSM => Drop({:?}): Already disabled.", entry.key());
*entry.into_mut() = st;
},
- PeerState::DisabledPendingEnable { open, connected_point, timer_deadline, .. } => {
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Interrupting pending \
- enable", entry.key());
+ PeerState::DisabledPendingEnable {
+ open,
+ timer_deadline,
+ timer: _
+ } => {
+ debug!(target: "sub-libp2p",
+ "PSM => Drop({:?}): Interrupting pending enabling.",
+ entry.key());
*entry.into_mut() = PeerState::Disabled {
open,
- connected_point,
banned_until: Some(timer_deadline),
};
},
- PeerState::Enabled { open, connected_point } => {
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Disabling connection", entry.key());
+ PeerState::Enabled { open } => {
+ debug!(target: "sub-libp2p", "PSM => Drop({:?}): Disabling connections.", entry.key());
debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", entry.key());
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: entry.key().clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Disable,
});
- *entry.into_mut() = PeerState::Disabled { open, connected_point, banned_until: None }
+ *entry.into_mut() = PeerState::Disabled {
+ open,
+ banned_until: None
+ }
},
- st @ PeerState::Incoming { .. } => {
- error!(target: "sub-libp2p", "PSM => Drop({:?}): Was in incoming mode",
+ st @ PeerState::Incoming => {
+ error!(target: "sub-libp2p", "PSM => Drop({:?}): Not enabled (Incoming).",
entry.key());
*entry.into_mut() = st;
},
PeerState::Requested => {
// We don't cancel dialing. Libp2p doesn't expose that on purpose, as other
- // sub-systems (such as the discovery mechanism) may require dialing this node as
+ // sub-systems (such as the discovery mechanism) may require dialing this peer as
// well at the same time.
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Was not yet connected", entry.key());
+ debug!(target: "sub-libp2p", "PSM => Drop({:?}): Not yet connected.", entry.key());
entry.remove();
},
PeerState::PendingRequest { timer_deadline, .. } => {
- debug!(target: "sub-libp2p", "PSM => Drop({:?}): Was not yet connected", entry.key());
+ debug!(target: "sub-libp2p", "PSM => Drop({:?}): Not yet connected", entry.key());
*entry.into_mut() = PeerState::Banned { until: timer_deadline }
},
@@ -641,7 +720,8 @@ impl GenericProto {
}
}
- /// Function that is called when the peerset wants us to accept an incoming node.
+ /// Function that is called when the peerset wants us to accept a connection
+ /// request from a peer.
fn peerset_report_accept(&mut self, index: sc_peerset::IncomingIndex) {
let incoming = if let Some(pos) = self.incoming.iter().position(|i| i.incoming_id == index) {
self.incoming.remove(pos)
@@ -658,34 +738,25 @@ impl GenericProto {
return
}
- let state = if let Some(state) = self.peers.get_mut(&incoming.peer_id) {
- state
- } else {
- error!(target: "sub-libp2p", "State mismatch in libp2p: no entry in peers \
- corresponding to an alive incoming");
- return
- };
-
- let connected_point = if let PeerState::Incoming { connected_point } = state {
- connected_point.clone()
- } else {
- error!(target: "sub-libp2p", "State mismatch in libp2p: entry in peers corresponding \
- to an alive incoming is not in incoming state");
- return
- };
-
- debug!(target: "sub-libp2p", "PSM => Accept({:?}, {:?}): Enabling connection \
- through {:?}", index, incoming.peer_id, connected_point);
- debug!(target: "sub-libp2p", "Handler({:?}) <= Enable", incoming.peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
- peer_id: incoming.peer_id,
- event: NotifsHandlerIn::Enable,
- });
-
- *state = PeerState::Enabled { open: false, connected_point };
+ match self.peers.get_mut(&incoming.peer_id) {
+ Some(state @ PeerState::Incoming) => {
+ debug!(target: "sub-libp2p", "PSM => Accept({:?}, {:?}): Enabling connections.",
+ index, incoming.peer_id);
+ debug!(target: "sub-libp2p", "Handler({:?}) <= Enable", incoming.peer_id);
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
+ peer_id: incoming.peer_id,
+ handler: NotifyHandler::All,
+ event: NotifsHandlerIn::Enable,
+ });
+ *state = PeerState::Enabled { open: SmallVec::new() };
+ }
+ peer => error!(target: "sub-libp2p",
+ "State mismatch in libp2p: Expected alive incoming. Got {:?}.",
+ peer)
+ }
}
- /// Function that is called when the peerset wants us to reject an incoming node.
+ /// Function that is called when the peerset wants us to reject an incoming peer.
fn peerset_report_reject(&mut self, index: sc_peerset::IncomingIndex) {
let incoming = if let Some(pos) = self.incoming.iter().position(|i| i.incoming_id == index) {
self.incoming.remove(pos)
@@ -700,30 +771,25 @@ impl GenericProto {
return
}
- let state = if let Some(state) = self.peers.get_mut(&incoming.peer_id) {
- state
- } else {
- error!(target: "sub-libp2p", "State mismatch in libp2p: no entry in peers \
- corresponding to an alive incoming");
- return
- };
-
- let connected_point = if let PeerState::Incoming { connected_point } = state {
- connected_point.clone()
- } else {
- error!(target: "sub-libp2p", "State mismatch in libp2p: entry in peers corresponding \
- to an alive incoming is not in incoming state");
- return
- };
-
- debug!(target: "sub-libp2p", "PSM => Reject({:?}, {:?}): Rejecting connection through \
- {:?}", index, incoming.peer_id, connected_point);
- debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", incoming.peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
- peer_id: incoming.peer_id,
- event: NotifsHandlerIn::Disable,
- });
- *state = PeerState::Disabled { open: false, connected_point, banned_until: None };
+ match self.peers.get_mut(&incoming.peer_id) {
+ Some(state @ PeerState::Incoming) => {
+ debug!(target: "sub-libp2p", "PSM => Reject({:?}, {:?}): Rejecting connections.",
+ index, incoming.peer_id);
+ debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", incoming.peer_id);
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
+ peer_id: incoming.peer_id,
+ handler: NotifyHandler::All,
+ event: NotifsHandlerIn::Disable,
+ });
+ *state = PeerState::Disabled {
+ open: SmallVec::new(),
+ banned_until: None
+ };
+ }
+ peer => error!(target: "sub-libp2p",
+ "State mismatch in libp2p: Expected alive incoming. Got {:?}.",
+ peer)
+ }
}
}
@@ -743,26 +809,32 @@ impl NetworkBehaviour for GenericProto {
Vec::new()
}
- fn inject_connected(&mut self, peer_id: PeerId, connected_point: ConnectedPoint) {
- match (self.peers.entry(peer_id.clone()).or_insert(PeerState::Poisoned), connected_point) {
- (st @ &mut PeerState::Requested, connected_point) |
- (st @ &mut PeerState::PendingRequest { .. }, connected_point) => {
- debug!(target: "sub-libp2p", "Libp2p => Connected({:?}): Connection \
- requested by PSM (through {:?})", peer_id, connected_point
+ fn inject_connected(&mut self, _: &PeerId) {
+ }
+
+ fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ debug!(target: "sub-libp2p", "Libp2p => Connection ({:?},{:?}) to {} established.",
+ conn, endpoint, peer_id);
+ match (self.peers.entry(peer_id.clone()).or_insert(PeerState::Poisoned), endpoint) {
+ (st @ &mut PeerState::Requested, endpoint) |
+ (st @ &mut PeerState::PendingRequest { .. }, endpoint) => {
+ debug!(target: "sub-libp2p",
+ "Libp2p => Connected({}, {:?}): Connection was requested by PSM.",
+ peer_id, endpoint
);
- debug!(target: "sub-libp2p", "Handler({:?}) <= Enable", peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ *st = PeerState::Enabled { open: SmallVec::new() };
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: peer_id.clone(),
- event: NotifsHandlerIn::Enable,
+ handler: NotifyHandler::One(*conn),
+ event: NotifsHandlerIn::Enable
});
- *st = PeerState::Enabled { open: false, connected_point };
}
- // Note: it may seem weird that "Banned" nodes get treated as if there were absent.
+ // Note: it may seem weird that "Banned" peers get treated as if they were absent.
// This is because the word "Banned" means "temporarily prevent outgoing connections to
- // this node", and not "banned" in the sense that we would refuse the node altogether.
- (st @ &mut PeerState::Poisoned, connected_point @ ConnectedPoint::Listener { .. }) |
- (st @ &mut PeerState::Banned { .. }, connected_point @ ConnectedPoint::Listener { .. }) => {
+ // this peer", and not "banned" in the sense that we would refuse the peer altogether.
+ (st @ &mut PeerState::Poisoned, endpoint @ ConnectedPoint::Listener { .. }) |
+ (st @ &mut PeerState::Banned { .. }, endpoint @ ConnectedPoint::Listener { .. }) => {
let incoming_id = self.next_incoming_index.clone();
self.next_incoming_index.0 = match self.next_incoming_index.0.checked_add(1) {
Some(v) => v,
@@ -771,61 +843,79 @@ impl NetworkBehaviour for GenericProto {
return
}
};
- debug!(target: "sub-libp2p", "Libp2p => Connected({:?}): Incoming connection",
- peer_id);
- debug!(target: "sub-libp2p", "PSM <= Incoming({:?}, {:?}): Through {:?}",
- incoming_id, peer_id, connected_point);
+ debug!(target: "sub-libp2p", "Libp2p => Connected({}, {:?}): Incoming connection",
+ peer_id, endpoint);
+ debug!(target: "sub-libp2p", "PSM <= Incoming({}, {:?}).",
+ peer_id, incoming_id);
self.peerset.incoming(peer_id.clone(), incoming_id);
self.incoming.push(IncomingPeer {
peer_id: peer_id.clone(),
alive: true,
incoming_id,
});
- *st = PeerState::Incoming { connected_point };
+ *st = PeerState::Incoming { };
}
- (st @ &mut PeerState::Poisoned, connected_point) |
- (st @ &mut PeerState::Banned { .. }, connected_point) => {
+ (st @ &mut PeerState::Poisoned, endpoint) |
+ (st @ &mut PeerState::Banned { .. }, endpoint) => {
let banned_until = if let PeerState::Banned { until } = st {
Some(*until)
} else {
None
};
- debug!(target: "sub-libp2p", "Libp2p => Connected({:?}): Requested by something \
- else than PSM, disabling", peer_id);
- debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ debug!(target: "sub-libp2p",
+ "Libp2p => Connected({},{:?}): Not requested by PSM, disabling.",
+ peer_id, endpoint);
+ *st = PeerState::Disabled { open: SmallVec::new(), banned_until };
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: peer_id.clone(),
- event: NotifsHandlerIn::Disable,
+ handler: NotifyHandler::One(*conn),
+ event: NotifsHandlerIn::Disable
});
- *st = PeerState::Disabled { open: false, connected_point, banned_until };
}
- st => {
- // This is a serious bug either in this state machine or in libp2p.
- error!(target: "sub-libp2p", "Received inject_connected for \
- already-connected node; state is {:?}", st
- );
+ (PeerState::Incoming { .. }, _) => {
+ debug!(target: "sub-libp2p",
+ "Secondary connection {:?} to {} waiting for PSM decision.",
+ conn, peer_id);
+ },
+
+ (PeerState::Enabled { .. }, _) => {
+ debug!(target: "sub-libp2p", "Handler({},{:?}) <= Enable secondary connection",
+ peer_id, conn);
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
+ peer_id: peer_id.clone(),
+ handler: NotifyHandler::One(*conn),
+ event: NotifsHandlerIn::Enable
+ });
+ }
+
+ (PeerState::Disabled { .. }, _) | (PeerState::DisabledPendingEnable { .. }, _) => {
+ debug!(target: "sub-libp2p", "Handler({},{:?}) <= Disable secondary connection",
+ peer_id, conn);
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
+ peer_id: peer_id.clone(),
+ handler: NotifyHandler::One(*conn),
+ event: NotifsHandlerIn::Disable
+ });
}
}
}
- fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
- match self.peers.remove(peer_id) {
- None | Some(PeerState::Requested) | Some(PeerState::PendingRequest { .. }) |
- Some(PeerState::Banned { .. }) =>
- // This is a serious bug either in this state machine or in libp2p.
- error!(target: "sub-libp2p", "Received inject_disconnected for non-connected \
- node {:?}", peer_id),
-
- Some(PeerState::Disabled { open, banned_until, .. }) => {
- debug!(target: "sub-libp2p", "Libp2p => Disconnected({:?}): Was disabled \
- (through {:?})", peer_id, endpoint);
- if let Some(until) = banned_until {
- self.peers.insert(peer_id.clone(), PeerState::Banned { until });
- }
- if open {
- debug!(target: "sub-libp2p", "External API <= Closed({:?})", peer_id);
+ fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ debug!(target: "sub-libp2p", "Libp2p => Connection ({:?},{:?}) to {} closed.",
+ conn, endpoint, peer_id);
+ match self.peers.get_mut(peer_id) {
+ Some(PeerState::Disabled { open, .. }) |
+ Some(PeerState::DisabledPendingEnable { open, .. }) |
+ Some(PeerState::Enabled { open, .. }) => {
+ // Check if the "link" to the peer is already considered closed,
+ // i.e. there is no connection that is open for custom protocols,
+ // in which case `CustomProtocolClosed` was already emitted.
+ let closed = open.is_empty();
+ open.retain(|c| c != conn);
+ if !closed {
+ debug!(target: "sub-libp2p", "External API <= Closed({})", peer_id);
let event = GenericProtoOut::CustomProtocolClosed {
peer_id: peer_id.clone(),
reason: "Disconnected by libp2p".into(),
@@ -834,52 +924,52 @@ impl NetworkBehaviour for GenericProto {
self.events.push(NetworkBehaviourAction::GenerateEvent(event));
}
}
+ _ => {}
+ }
+ }
- Some(PeerState::DisabledPendingEnable { open, timer_deadline, .. }) => {
- debug!(target: "sub-libp2p", "Libp2p => Disconnected({:?}): Was disabled \
- (through {:?}) but pending enable", peer_id, endpoint);
- debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
- self.peerset.dropped(peer_id.clone());
- self.peers.insert(peer_id.clone(), PeerState::Banned { until: timer_deadline });
- if open {
- debug!(target: "sub-libp2p", "External API <= Closed({:?})", peer_id);
- let event = GenericProtoOut::CustomProtocolClosed {
- peer_id: peer_id.clone(),
- reason: "Disconnected by libp2p".into(),
- };
+ fn inject_disconnected(&mut self, peer_id: &PeerId) {
+ match self.peers.remove(peer_id) {
+ None | Some(PeerState::Requested) | Some(PeerState::PendingRequest { .. }) |
+ Some(PeerState::Banned { .. }) =>
+ // This is a serious bug either in this state machine or in libp2p.
+ error!(target: "sub-libp2p",
+ "`inject_disconnected` called for unknown peer {}",
+ peer_id),
- self.events.push(NetworkBehaviourAction::GenerateEvent(event));
+ Some(PeerState::Disabled { banned_until, .. }) => {
+ debug!(target: "sub-libp2p", "Libp2p => Disconnected({}): Was disabled.", peer_id);
+ if let Some(until) = banned_until {
+ self.peers.insert(peer_id.clone(), PeerState::Banned { until });
}
}
- Some(PeerState::Enabled { open, .. }) => {
- debug!(target: "sub-libp2p", "Libp2p => Disconnected({:?}): Was enabled \
- (through {:?})", peer_id, endpoint);
- debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", peer_id);
+ Some(PeerState::DisabledPendingEnable { timer_deadline, .. }) => {
+ debug!(target: "sub-libp2p",
+ "Libp2p => Disconnected({}): Was disabled but pending enable.",
+ peer_id);
+ debug!(target: "sub-libp2p", "PSM <= Dropped({})", peer_id);
self.peerset.dropped(peer_id.clone());
+ self.peers.insert(peer_id.clone(), PeerState::Banned { until: timer_deadline });
+ }
+ Some(PeerState::Enabled { .. }) => {
+ debug!(target: "sub-libp2p", "Libp2p => Disconnected({}): Was enabled.", peer_id);
+ debug!(target: "sub-libp2p", "PSM <= Dropped({})", peer_id);
+ self.peerset.dropped(peer_id.clone());
let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
self.peers.insert(peer_id.clone(), PeerState::Banned {
until: Instant::now() + Duration::from_secs(ban_dur)
});
-
- if open {
- debug!(target: "sub-libp2p", "External API <= Closed({:?})", peer_id);
- let event = GenericProtoOut::CustomProtocolClosed {
- peer_id: peer_id.clone(),
- reason: "Disconnected by libp2p".into(),
- };
-
- self.events.push(NetworkBehaviourAction::GenerateEvent(event));
- }
}
// In the incoming state, we don't report "Dropped". Instead we will just ignore the
// corresponding Accept/Reject.
- Some(PeerState::Incoming { .. }) => {
+ Some(PeerState::Incoming { }) => {
if let Some(state) = self.incoming.iter_mut().find(|i| i.peer_id == *peer_id) {
- debug!(target: "sub-libp2p", "Libp2p => Disconnected({:?}): Was in incoming \
- mode (id {:?}, through {:?})", peer_id, state.incoming_id, endpoint);
+ debug!(target: "sub-libp2p",
+ "Libp2p => Disconnected({}): Was in incoming mode with id {:?}.",
+ peer_id, state.incoming_id);
state.alive = false;
} else {
error!(target: "sub-libp2p", "State mismatch in libp2p: no entry in incoming \
@@ -888,7 +978,7 @@ impl NetworkBehaviour for GenericProto {
}
Some(PeerState::Poisoned) =>
- error!(target: "sub-libp2p", "State of {:?} is poisoned", peer_id),
+ error!(target: "sub-libp2p", "State of peer {} is poisoned", peer_id),
}
}
@@ -899,13 +989,13 @@ impl NetworkBehaviour for GenericProto {
fn inject_dial_failure(&mut self, peer_id: &PeerId) {
if let Entry::Occupied(mut entry) = self.peers.entry(peer_id.clone()) {
match mem::replace(entry.get_mut(), PeerState::Poisoned) {
- // The node is not in our list.
+ // The peer is not in our list.
st @ PeerState::Banned { .. } => {
trace!(target: "sub-libp2p", "Libp2p => Dial failure for {:?}", peer_id);
*entry.into_mut() = st;
},
- // "Basic" situation: we failed to reach a node that the peerset requested.
+ // "Basic" situation: we failed to reach a peer that the peerset requested.
PeerState::Requested | PeerState::PendingRequest { .. } => {
debug!(target: "sub-libp2p", "Libp2p => Dial failure for {:?}", peer_id);
*entry.into_mut() = PeerState::Banned {
@@ -915,7 +1005,7 @@ impl NetworkBehaviour for GenericProto {
self.peerset.dropped(peer_id.clone())
},
- // We can still get dial failures even if we are already connected to the node,
+ // We can still get dial failures even if we are already connected to the peer,
// as an extra diagnostic for an earlier attempt.
st @ PeerState::Disabled { .. } | st @ PeerState::Enabled { .. } |
st @ PeerState::DisabledPendingEnable { .. } | st @ PeerState::Incoming { .. } => {
@@ -928,92 +1018,130 @@ impl NetworkBehaviour for GenericProto {
}
} else {
- // The node is not in our list.
+ // The peer is not in our list.
trace!(target: "sub-libp2p", "Libp2p => Dial failure for {:?}", peer_id);
}
}
- fn inject_node_event(
+ fn inject_event(
&mut self,
source: PeerId,
+ connection: ConnectionId,
event: NotifsHandlerOut,
) {
match event {
- NotifsHandlerOut::Closed { reason } => {
- debug!(target: "sub-libp2p", "Handler({:?}) => Closed: {}", source, reason);
+ NotifsHandlerOut::Closed { endpoint, reason } => {
+ debug!(target: "sub-libp2p",
+ "Handler({:?}) => Endpoint {:?} closed for custom protocols: {}",
+ source, endpoint, reason);
let mut entry = if let Entry::Occupied(entry) = self.peers.entry(source.clone()) {
entry
} else {
- error!(target: "sub-libp2p", "State mismatch in the custom protos handler");
+ error!(target: "sub-libp2p", "Closed: State mismatch in the custom protos handler");
return
};
- debug!(target: "sub-libp2p", "External API <= Closed({:?})", source);
- let event = GenericProtoOut::CustomProtocolClosed {
- reason,
- peer_id: source.clone(),
- };
- self.events.push(NetworkBehaviourAction::GenerateEvent(event));
-
- match mem::replace(entry.get_mut(), PeerState::Poisoned) {
- PeerState::Enabled { open, connected_point } => {
- debug_assert!(open);
-
- debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
- self.peerset.dropped(source.clone());
+ let last = match mem::replace(entry.get_mut(), PeerState::Poisoned) {
+ PeerState::Enabled { mut open } => {
+ debug_assert!(open.iter().any(|c| c == &connection));
+ open.retain(|c| c != &connection);
debug!(target: "sub-libp2p", "Handler({:?}) <= Disable", source);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: source.clone(),
+ handler: NotifyHandler::One(connection),
event: NotifsHandlerIn::Disable,
});
+ let last = open.is_empty();
+
+ if last {
+ debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
+ self.peerset.dropped(source.clone());
+ *entry.into_mut() = PeerState::Disabled {
+ open,
+ banned_until: None
+ };
+ } else {
+ *entry.into_mut() = PeerState::Enabled { open };
+ }
+
+ last
+ },
+ PeerState::Disabled { mut open, banned_until } => {
+ debug_assert!(open.iter().any(|c| c == &connection));
+ open.retain(|c| c != &connection);
+ let last = open.is_empty();
*entry.into_mut() = PeerState::Disabled {
- open: false,
- connected_point,
- banned_until: None
+ open,
+ banned_until
};
+ last
},
- PeerState::Disabled { open, connected_point, banned_until } => {
- debug_assert!(open);
- *entry.into_mut() = PeerState::Disabled { open: false, connected_point, banned_until };
- },
- PeerState::DisabledPendingEnable { open, connected_point, timer, timer_deadline } => {
- debug_assert!(open);
+ PeerState::DisabledPendingEnable {
+ mut open,
+ timer,
+ timer_deadline
+ } => {
+ debug_assert!(open.iter().any(|c| c == &connection));
+ open.retain(|c| c != &connection);
+ let last = open.is_empty();
*entry.into_mut() = PeerState::DisabledPendingEnable {
- open: false,
- connected_point,
+ open,
timer,
timer_deadline
};
+ last
},
- _ => error!(target: "sub-libp2p", "State mismatch in the custom protos handler"),
+ state => {
+ error!(target: "sub-libp2p",
+ "Unexpected state in the custom protos handler: {:?}",
+ state);
+ return
+ }
+ };
+
+ if last {
+ debug!(target: "sub-libp2p", "External API <= Closed({:?})", source);
+ let event = GenericProtoOut::CustomProtocolClosed {
+ reason,
+ peer_id: source.clone(),
+ };
+ self.events.push(NetworkBehaviourAction::GenerateEvent(event));
+ } else {
+ debug!(target: "sub-libp2p", "Secondary connection closed custom protocol.");
}
}
- NotifsHandlerOut::Open => {
- debug!(target: "sub-libp2p", "Handler({:?}) => Open", source);
- let endpoint = match self.peers.get_mut(&source) {
- Some(PeerState::Enabled { ref mut open, ref connected_point }) |
- Some(PeerState::DisabledPendingEnable { ref mut open, ref connected_point, .. }) |
- Some(PeerState::Disabled { ref mut open, ref connected_point, .. }) if !*open => {
- *open = true;
- connected_point.clone()
+ NotifsHandlerOut::Open { endpoint } => {
+ debug!(target: "sub-libp2p",
+ "Handler({:?}) => Endpoint {:?} open for custom protocols.",
+ source, endpoint);
+
+ let first = match self.peers.get_mut(&source) {
+ Some(PeerState::Enabled { ref mut open, .. }) |
+ Some(PeerState::DisabledPendingEnable { ref mut open, .. }) |
+ Some(PeerState::Disabled { ref mut open, .. }) => {
+ let first = open.is_empty();
+ open.push(connection);
+ first
}
- _ => {
- error!(target: "sub-libp2p", "State mismatch in the custom protos handler");
+ state => {
+ error!(target: "sub-libp2p",
+ "Open: Unexpected state in the custom protos handler: {:?}",
+ state);
return
}
};
- debug!(target: "sub-libp2p", "External API <= Open({:?})", source);
- let event = GenericProtoOut::CustomProtocolOpen {
- peer_id: source,
- endpoint,
- };
-
- self.events.push(NetworkBehaviourAction::GenerateEvent(event));
+ if first {
+ debug!(target: "sub-libp2p", "External API <= Open({:?})", source);
+ let event = GenericProtoOut::CustomProtocolOpen { peer_id: source };
+ self.events.push(NetworkBehaviourAction::GenerateEvent(event));
+ } else {
+ debug!(target: "sub-libp2p", "Secondary connection opened custom protocol.");
+ }
}
NotifsHandlerOut::CustomMessage { message } => {
@@ -1065,11 +1193,12 @@ impl NetworkBehaviour for GenericProto {
}
NotifsHandlerOut::ProtocolError { error, .. } => {
- debug!(target: "sub-libp2p", "Handler({:?}) => Severe protocol error: {:?}",
+ warn!(target: "sub-libp2p",
+ "Handler({:?}) => Severe protocol error: {:?}",
source, error);
- // A severe protocol error happens when we detect a "bad" node, such as a node on
- // a different chain, or a node that doesn't speak the same protocol(s). We
- // decrease the node's reputation, hence lowering the chances we try this node
+ // A severe protocol error happens when we detect a "bad" peer, such as a per on
+ // a different chain, or a peer that doesn't speak the same protocol(s). We
+ // decrease the peer's reputation, hence lowering the chances we try this peer
// again in the short term.
self.peerset.report_peer(
source.clone(),
@@ -1123,27 +1252,34 @@ impl NetworkBehaviour for GenericProto {
}
debug!(target: "sub-libp2p", "Libp2p <= Dial {:?} now that ban has expired", peer_id);
- self.events.push(NetworkBehaviourAction::DialPeer { peer_id: peer_id.clone() });
+ self.events.push(NetworkBehaviourAction::DialPeer {
+ peer_id: peer_id.clone(),
+ condition: DialPeerCondition::Disconnected
+ });
*peer_state = PeerState::Requested;
}
- PeerState::DisabledPendingEnable { mut timer, connected_point, open, timer_deadline } => {
+ PeerState::DisabledPendingEnable {
+ mut timer,
+ open,
+ timer_deadline
+ } => {
if let Poll::Pending = Pin::new(&mut timer).poll(cx) {
*peer_state = PeerState::DisabledPendingEnable {
timer,
- connected_point,
open,
timer_deadline
};
continue;
}
- debug!(target: "sub-libp2p", "Handler({:?}) <= Enable now that ban has expired", peer_id);
- self.events.push(NetworkBehaviourAction::SendEvent {
+ debug!(target: "sub-libp2p", "Handler({:?}) <= Enable (ban expired)", peer_id);
+ self.events.push(NetworkBehaviourAction::NotifyHandler {
peer_id: peer_id.clone(),
+ handler: NotifyHandler::All,
event: NotifsHandlerIn::Enable,
});
- *peer_state = PeerState::Enabled { connected_point, open };
+ *peer_state = PeerState::Enabled { open };
}
st @ _ => *peer_state = st,
diff --git a/substrate/client/network/src/protocol/generic_proto/handler/group.rs b/substrate/client/network/src/protocol/generic_proto/handler/group.rs
index 21dc4091c0c4e70beec71c31f8263362ade80d3d..7e597f1be69f9d51b8f7982fc5bf1d4b38b9c765 100644
--- a/substrate/client/network/src/protocol/generic_proto/handler/group.rs
+++ b/substrate/client/network/src/protocol/generic_proto/handler/group.rs
@@ -143,7 +143,7 @@ impl IntoProtocolsHandler for NotifsHandlerProto {
}
/// Event that can be received by a `NotifsHandler`.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub enum NotifsHandlerIn {
/// The node should start using custom protocols.
Enable,
@@ -181,13 +181,18 @@ pub enum NotifsHandlerIn {
/// Event that can be emitted by a `NotifsHandler`.
#[derive(Debug)]
pub enum NotifsHandlerOut {
- /// Opened the substreams with the remote.
- Open,
+ /// The connection is open for custom protocols.
+ Open {
+ /// The endpoint of the connection that is open for custom protocols.
+ endpoint: ConnectedPoint,
+ },
- /// Closed the substreams with the remote.
+ /// The connection is closed for custom protocols.
Closed {
- /// Reason why the substream closed, for diagnostic purposes.
+ /// The reason for closing, for diagnostic purposes.
reason: Cow<'static, str>,
+ /// The endpoint of the connection that closed for custom protocols.
+ endpoint: ConnectedPoint,
},
/// Received a non-gossiping message on the legacy substream.
@@ -497,13 +502,13 @@ impl ProtocolsHandler for NotifsHandler {
protocol: protocol.map_upgrade(EitherUpgrade::B),
info: None,
}),
- ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolOpen { .. }) =>
+ ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolOpen { endpoint, .. }) =>
return Poll::Ready(ProtocolsHandlerEvent::Custom(
- NotifsHandlerOut::Open
+ NotifsHandlerOut::Open { endpoint }
)),
- ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolClosed { reason }) =>
+ ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomProtocolClosed { endpoint, reason }) =>
return Poll::Ready(ProtocolsHandlerEvent::Custom(
- NotifsHandlerOut::Closed { reason }
+ NotifsHandlerOut::Closed { endpoint, reason }
)),
ProtocolsHandlerEvent::Custom(LegacyProtoHandlerOut::CustomMessage { message }) =>
return Poll::Ready(ProtocolsHandlerEvent::Custom(
diff --git a/substrate/client/network/src/protocol/generic_proto/handler/legacy.rs b/substrate/client/network/src/protocol/generic_proto/handler/legacy.rs
index a2d2fc9246d1c79b761d5e5eb1ce375fba33b49a..bc84fd847c9c4ba44dbffcf5a7735c18c39f7776 100644
--- a/substrate/client/network/src/protocol/generic_proto/handler/legacy.rs
+++ b/substrate/client/network/src/protocol/generic_proto/handler/legacy.rs
@@ -40,9 +40,8 @@ use std::{pin::Pin, task::{Context, Poll}};
/// it is turned into a `LegacyProtoHandler`. It then handles all communications that are specific
/// to Substrate on that single connection.
///
-/// Note that there can be multiple instance of this struct simultaneously for same peer. However
-/// if that happens, only one main instance can communicate with the outer layers of the code. In
-/// other words, the outer layers of the code only ever see one handler.
+/// Note that there can be multiple instance of this struct simultaneously for same peer,
+/// if there are multiple established connections to the peer.
///
/// ## State of the handler
///
@@ -61,6 +60,7 @@ use std::{pin::Pin, task::{Context, Poll}};
/// these states. For example, if the handler reports a network misbehaviour, it will close the
/// substreams but it is the role of the user to send a `Disabled` event if it wants the connection
/// to close. Otherwise, the handler will try to reopen substreams.
+///
/// The handler starts in the "Initializing" state and must be transitionned to Enabled or Disabled
/// as soon as possible.
///
@@ -111,7 +111,7 @@ impl IntoProtocolsHandler for LegacyProtoHandlerProto {
fn into_handler(self, remote_peer_id: &PeerId, connected_point: &ConnectedPoint) -> Self::Handler {
LegacyProtoHandler {
protocol: self.protocol,
- endpoint: connected_point.to_endpoint(),
+ endpoint: connected_point.clone(),
remote_peer_id: remote_peer_id.clone(),
state: ProtocolState::Init {
substreams: SmallVec::new(),
@@ -136,7 +136,7 @@ pub struct LegacyProtoHandler {
/// Whether we are the connection dialer or listener. Used to determine who, between the local
/// node and the remote node, has priority.
- endpoint: Endpoint,
+ endpoint: ConnectedPoint,
/// Queue of events to send to the outside.
///
@@ -218,12 +218,16 @@ pub enum LegacyProtoHandlerOut {
CustomProtocolOpen {
/// Version of the protocol that has been opened.
version: u8,
+ /// The connected endpoint.
+ endpoint: ConnectedPoint,
},
/// Closed a custom protocol with the remote.
CustomProtocolClosed {
/// Reason why the substream closed, for diagnostic purposes.
reason: Cow<'static, str>,
+ /// The connected endpoint.
+ endpoint: ConnectedPoint,
},
/// Receives a message on a custom protocol substream.
@@ -272,7 +276,7 @@ impl LegacyProtoHandler {
ProtocolState::Init { substreams: incoming, .. } => {
if incoming.is_empty() {
- if let Endpoint::Dialer = self.endpoint {
+ if let ConnectedPoint::Dialer { .. } = self.endpoint {
self.events_queue.push(ProtocolsHandlerEvent::OutboundSubstreamRequest {
protocol: SubstreamProtocol::new(self.protocol.clone()),
info: (),
@@ -281,10 +285,10 @@ impl LegacyProtoHandler {
ProtocolState::Opening {
deadline: Delay::new(Duration::from_secs(60))
}
-
} else {
let event = LegacyProtoHandlerOut::CustomProtocolOpen {
- version: incoming[0].protocol_version()
+ version: incoming[0].protocol_version(),
+ endpoint: self.endpoint.clone()
};
self.events_queue.push(ProtocolsHandlerEvent::Custom(event));
ProtocolState::Normal {
@@ -404,6 +408,7 @@ impl LegacyProtoHandler {
if substreams.is_empty() {
let event = LegacyProtoHandlerOut::CustomProtocolClosed {
reason: "All substreams have been closed by the remote".into(),
+ endpoint: self.endpoint.clone()
};
self.state = ProtocolState::Disabled {
shutdown: shutdown.into_iter().collect(),
@@ -416,6 +421,7 @@ impl LegacyProtoHandler {
if substreams.is_empty() {
let event = LegacyProtoHandlerOut::CustomProtocolClosed {
reason: format!("Error on the last substream: {:?}", err).into(),
+ endpoint: self.endpoint.clone()
};
self.state = ProtocolState::Disabled {
shutdown: shutdown.into_iter().collect(),
@@ -479,7 +485,8 @@ impl LegacyProtoHandler {
ProtocolState::Opening { .. } => {
let event = LegacyProtoHandlerOut::CustomProtocolOpen {
- version: substream.protocol_version()
+ version: substream.protocol_version(),
+ endpoint: self.endpoint.clone()
};
self.events_queue.push(ProtocolsHandlerEvent::Custom(event));
ProtocolState::Normal {
diff --git a/substrate/client/network/src/protocol/generic_proto/handler/notif_in.rs b/substrate/client/network/src/protocol/generic_proto/handler/notif_in.rs
index 7558d1d361fd7cf445271b4351cd0c7c5fdd45d7..83923154bd6256de3bb4e983796807093f7019ad 100644
--- a/substrate/client/network/src/protocol/generic_proto/handler/notif_in.rs
+++ b/substrate/client/network/src/protocol/generic_proto/handler/notif_in.rs
@@ -72,7 +72,7 @@ pub struct NotifsInHandler {
}
/// Event that can be received by a `NotifsInHandler`.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub enum NotifsInHandlerIn {
/// Can be sent back as a response to an `OpenRequest`. Contains the status message to send
/// to the remote.
diff --git a/substrate/client/network/src/protocol/generic_proto/handler/notif_out.rs b/substrate/client/network/src/protocol/generic_proto/handler/notif_out.rs
index dd38826496e7a3f8289131183d27f7f2c235c230..b5d6cd61ada2a9e54bfa5873e0f7992ab1dfcfcb 100644
--- a/substrate/client/network/src/protocol/generic_proto/handler/notif_out.rs
+++ b/substrate/client/network/src/protocol/generic_proto/handler/notif_out.rs
@@ -33,7 +33,7 @@ use libp2p::swarm::{
SubstreamProtocol,
NegotiatedSubstream,
};
-use log::error;
+use log::{debug, warn, error};
use prometheus_endpoint::Histogram;
use smallvec::SmallVec;
use std::{borrow::Cow, fmt, mem, pin::Pin, task::{Context, Poll}, time::Duration};
@@ -280,7 +280,7 @@ impl ProtocolsHandler for NotifsOutHandler {
// be recovered. When in doubt, let's drop the existing substream and
// open a new one.
if sub.close().now_or_never().is_none() {
- log::warn!(
+ warn!(
target: "sub-libp2p",
"📞 Improperly closed outbound notifications substream"
);
@@ -293,16 +293,22 @@ impl ProtocolsHandler for NotifsOutHandler {
});
self.state = State::Opening { initial_message };
},
- State::Opening { .. } | State::Refused | State::Open { .. } =>
- error!("â˜Žï¸ Tried to enable notifications handler that was already enabled"),
- State::Poisoned => error!("â˜Žï¸ Notifications handler in a poisoned state"),
+ st @ State::Opening { .. } | st @ State::Refused | st @ State::Open { .. } => {
+ debug!(target: "sub-libp2p",
+ "Tried to enable notifications handler that was already enabled");
+ self.state = st;
+ }
+ State::Poisoned => error!("Notifications handler in a poisoned state"),
}
}
NotifsOutHandlerIn::Disable => {
match mem::replace(&mut self.state, State::Poisoned) {
- State::Disabled | State::DisabledOpen(_) | State::DisabledOpening =>
- error!("â˜Žï¸ Tried to disable notifications handler that was already disabled"),
+ st @ State::Disabled | st @ State::DisabledOpen(_) | st @ State::DisabledOpening => {
+ debug!(target: "sub-libp2p",
+ "Tried to disable notifications handler that was already disabled");
+ self.state = st;
+ }
State::Opening { .. } => self.state = State::DisabledOpening,
State::Refused => self.state = State::Disabled,
State::Open { substream, .. } => self.state = State::DisabledOpen(substream),
@@ -313,7 +319,7 @@ impl ProtocolsHandler for NotifsOutHandler {
NotifsOutHandlerIn::Send(msg) =>
if let State::Open { substream, .. } = &mut self.state {
if substream.push_message(msg).is_err() {
- log::warn!(
+ warn!(
target: "sub-libp2p",
"📞 Notifications queue with peer {} is full, dropped message (protocol: {:?})",
self.peer_id,
@@ -325,7 +331,7 @@ impl ProtocolsHandler for NotifsOutHandler {
}
} else {
// This is an API misuse.
- log::warn!(
+ warn!(
target: "sub-libp2p",
"📞 Tried to send a notification on a disabled handler"
);
diff --git a/substrate/client/network/src/protocol/generic_proto/tests.rs b/substrate/client/network/src/protocol/generic_proto/tests.rs
index 4548859ac420cd0ab52ed9c9c5cffd9167aef92b..1bc6e745f887685af41a6c9cfdfc05da84af7918 100644
--- a/substrate/client/network/src/protocol/generic_proto/tests.rs
+++ b/substrate/client/network/src/protocol/generic_proto/tests.rs
@@ -18,7 +18,7 @@
use futures::{prelude::*, ready};
use codec::{Encode, Decode};
-use libp2p::core::nodes::listeners::ListenerId;
+use libp2p::core::connection::{ConnectionId, ListenerId};
use libp2p::core::ConnectedPoint;
use libp2p::swarm::{Swarm, ProtocolsHandler, IntoProtocolsHandler};
use libp2p::swarm::{PollParameters, NetworkBehaviour, NetworkBehaviourAction};
@@ -148,20 +148,29 @@ impl NetworkBehaviour for CustomProtoWithAddr {
list
}
- fn inject_connected(&mut self, peer_id: PeerId, endpoint: ConnectedPoint) {
- self.inner.inject_connected(peer_id, endpoint)
+ fn inject_connected(&mut self, peer_id: &PeerId) {
+ self.inner.inject_connected(peer_id)
}
- fn inject_disconnected(&mut self, peer_id: &PeerId, endpoint: ConnectedPoint) {
- self.inner.inject_disconnected(peer_id, endpoint)
+ fn inject_disconnected(&mut self, peer_id: &PeerId) {
+ self.inner.inject_disconnected(peer_id)
}
- fn inject_node_event(
+ fn inject_connection_established(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.inner.inject_connection_established(peer_id, conn, endpoint)
+ }
+
+ fn inject_connection_closed(&mut self, peer_id: &PeerId, conn: &ConnectionId, endpoint: &ConnectedPoint) {
+ self.inner.inject_connection_closed(peer_id, conn, endpoint)
+ }
+
+ fn inject_event(
&mut self,
peer_id: PeerId,
+ connection: ConnectionId,
event: <<Self::ProtocolsHandler as IntoProtocolsHandler>::Handler as ProtocolsHandler>::OutEvent
) {
- self.inner.inject_node_event(peer_id, event)
+ self.inner.inject_event(peer_id, connection, event)
}
fn poll(
@@ -177,10 +186,6 @@ impl NetworkBehaviour for CustomProtoWithAddr {
self.inner.poll(cx, params)
}
- fn inject_replaced(&mut self, peer_id: PeerId, closed_endpoint: ConnectedPoint, new_endpoint: ConnectedPoint) {
- self.inner.inject_replaced(peer_id, closed_endpoint, new_endpoint)
- }
-
fn inject_addr_reach_failure(&mut self, peer_id: Option<&PeerId>, addr: &Multiaddr, error: &dyn std::error::Error) {
self.inner.inject_addr_reach_failure(peer_id, addr, error)
}
@@ -205,8 +210,8 @@ impl NetworkBehaviour for CustomProtoWithAddr {
self.inner.inject_listener_error(id, err);
}
- fn inject_listener_closed(&mut self, id: ListenerId) {
- self.inner.inject_listener_closed(id);
+ fn inject_listener_closed(&mut self, id: ListenerId, reason: Result<(), &io::Error>) {
+ self.inner.inject_listener_closed(id, reason);
}
}
diff --git a/substrate/client/network/src/protocol/light_client_handler.rs b/substrate/client/network/src/protocol/light_client_handler.rs
index 88a951914944ec04aaa7b2986836a6f2ae242506..85312b0803234e43dff213c4b1ade7f4f06a3c6c 100644
--- a/substrate/client/network/src/protocol/light_client_handler.rs
+++ b/substrate/client/network/src/protocol/light_client_handler.rs
@@ -37,6 +37,7 @@ use libp2p::{
ConnectedPoint,
Multiaddr,
PeerId,
+ connection::ConnectionId,
upgrade::{InboundUpgrade, ReadOneError, UpgradeInfo, Negotiated},
upgrade::{OutboundUpgrade, read_one, write_one}
},
@@ -44,9 +45,11 @@ use libp2p::{
NegotiatedSubstream,
NetworkBehaviour,
NetworkBehaviourAction,
+ NotifyHandler,
OneShotHandler,
+ OneShotHandlerConfig,
PollParameters,
- SubstreamProtocol
+ SubstreamProtocol,
}
};
use nohash_hasher::IntMap;
@@ -58,6 +61,7 @@ use sp_core::{
storage::{ChildInfo, StorageKey},
hexdisplay::HexDisplay,
};
+use smallvec::SmallVec;
use sp_blockchain::{Error as ClientError};
use sp_runtime::{
traits::{Block, Header, NumberFor, Zero},
@@ -237,25 +241,39 @@ struct RequestWrapper<B: Block, P> {
retries: usize,
/// The actual request.
request: Request<B>,
- /// Peer information, e.g. `PeerId`.
- peer: P
+ /// The peer to send the request to, e.g. `PeerId`.
+ peer: P,
+ /// The connection to use for sending the request.
+ connection: Option<ConnectionId>,
}
/// Information we have about some peer.
#[derive(Debug)]
struct PeerInfo<B: Block> {
- address: Multiaddr,
+ connections: SmallVec<[(ConnectionId, Multiaddr); crate::MAX_CONNECTIONS_PER_PEER]>,
best_block: Option<NumberFor<B>>,
status: PeerStatus,
}
+impl<B: Block> Default for PeerInfo<B> {
+ fn default() -> Self {
+ PeerInfo {
+ connections: SmallVec::new(),
+ best_block: None,
+ status: PeerStatus::Idle,
+ }
+ }
+}
+
+type RequestId = u64;
+
/// A peer is either idle or busy processing a request from us.
#[derive(Debug, Clone, PartialEq, Eq)]
enum PeerStatus {
/// The peer is available.
Idle,
/// We wait for the peer to return us a response for the given request ID.
- BusyWith(u64),
+ BusyWith(RequestId),
}
/// The light client handler behaviour.
@@ -273,9 +291,9 @@ pub struct LightClientHandler<B: Block> {
/// Pending (local) requests.
pending_requests: VecDeque<RequestWrapper<B, ()>>,
/// Requests on their way to remote peers.
- outstanding: IntMap<u64, RequestWrapper<B, PeerId>>,
+ outstanding: IntMap<RequestId, RequestWrapper<B, PeerId>>,
/// (Local) Request ID counter
- next_request_id: u64,
+ next_request_id: RequestId,
/// Handle to use for reporting misbehaviour of peers.
peerset: sc_peerset::PeersetHandle,
}
@@ -323,35 +341,18 @@ where
retries: retries(&req),
request: req,
peer: (), // we do not know the peer yet
+ connection: None,
};
self.pending_requests.push_back(rw);
Ok(())
}
- fn next_request_id(&mut self) -> u64 {
+ fn next_request_id(&mut self) -> RequestId {
let id = self.next_request_id;
self.next_request_id += 1;
id
}
- // Iterate over peers known to possess a certain block.
- fn idle_peers_with_block(&mut self, num: NumberFor<B>) -> impl Iterator<Item = PeerId> + '_ {
- self.peers.iter()
- .filter(move |(_, info)| {
- info.status == PeerStatus::Idle && info.best_block >= Some(num)
- })
- .map(|(peer, _)| peer.clone())
- }
-
- // Iterate over peers without a known block.
- fn idle_peers_with_unknown_block(&mut self) -> impl Iterator<Item = PeerId> + '_ {
- self.peers.iter()
- .filter(|(_, info)| {
- info.status == PeerStatus::Idle && info.best_block.is_none()
- })
- .map(|(peer, _)| peer.clone())
- }
-
/// Remove the given peer.
///
/// If we have a request to this peer in flight, we move it back to
@@ -364,12 +365,50 @@ where
retries: rw.retries,
request: rw.request,
peer: (), // need to find another peer
+ connection: None,
};
self.pending_requests.push_back(rw);
}
self.peers.remove(peer);
}
+ /// Prepares a request by selecting a suitable peer and connection to send it to.
+ ///
+ /// If there is currently no suitable peer for the request, the given request
+ /// is returned as `Err`.
+ fn prepare_request(&self, req: RequestWrapper<B, ()>)
+ -> Result<(PeerId, RequestWrapper<B, PeerId>), RequestWrapper<B, ()>>
+ {
+ let number = required_block(&req.request);
+
+ let mut peer = None;
+ for (peer_id, peer_info) in self.peers.iter() {
+ if peer_info.status == PeerStatus::Idle {
+ match peer_info.best_block {
+ Some(n) => if n >= number {
+ peer = Some((peer_id, peer_info));
+ break
+ },
+ None => peer = Some((peer_id, peer_info))
+ }
+ }
+ }
+
+ if let Some((peer_id, peer_info)) = peer {
+ let connection = peer_info.connections.iter().next().map(|(id, _)| *id);
+ let rw = RequestWrapper {
+ timestamp: req.timestamp,
+ retries: req.retries,
+ request: req.request,
+ peer: peer_id.clone(),
+ connection,
+ };
+ Ok((peer_id.clone(), rw))
+ } else {
+ Err(req)
+ }
+ }
+
/// Process a local request's response from remote.
///
/// If successful, this will give us the actual, checked data we should be
@@ -723,38 +762,68 @@ where
max_request_size: self.config.max_request_size,
protocol: self.config.light_protocol.clone(),
};
- OneShotHandler::new(SubstreamProtocol::new(p), self.config.inactivity_timeout)
+ let mut cfg = OneShotHandlerConfig::default();
+ cfg.inactive_timeout = self.config.inactivity_timeout;
+ OneShotHandler::new(SubstreamProtocol::new(p), cfg)
}
fn addresses_of_peer(&mut self, peer: &PeerId) -> Vec<Multiaddr> {
self.peers.get(peer)
- .map(|info| vec![info.address.clone()])
+ .map(|info| info.connections.iter().map(|(_, a)| a.clone()).collect())
.unwrap_or_default()
}
- fn inject_connected(&mut self, peer: PeerId, info: ConnectedPoint) {
+ fn inject_connected(&mut self, peer: &PeerId) {
+ }
+
+ fn inject_connection_established(&mut self, peer: &PeerId, conn: &ConnectionId, info: &ConnectedPoint) {
let peer_address = match info {
- ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
- ConnectedPoint::Dialer { address } => address
+ ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr.clone(),
+ ConnectedPoint::Dialer { address } => address.clone()
};
log::trace!("peer {} connected with address {}", peer, peer_address);
- let info = PeerInfo {
- address: peer_address,
- best_block: None,
- status: PeerStatus::Idle,
- };
-
- self.peers.insert(peer, info);
+ let entry = self.peers.entry(peer.clone()).or_default();
+ entry.connections.push((*conn, peer_address));
}
- fn inject_disconnected(&mut self, peer: &PeerId, _: ConnectedPoint) {
+ fn inject_disconnected(&mut self, peer: &PeerId) {
log::trace!("peer {} disconnected", peer);
self.remove_peer(peer)
}
- fn inject_node_event(&mut self, peer: PeerId, event: Event<NegotiatedSubstream>) {
+ fn inject_connection_closed(&mut self, peer: &PeerId, conn: &ConnectionId, info: &ConnectedPoint) {
+ let peer_address = match info {
+ ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
+ ConnectedPoint::Dialer { address } => address
+ };
+
+ log::trace!("connection to peer {} closed: {}", peer, peer_address);
+
+ if let Some(info) = self.peers.get_mut(peer) {
+ info.connections.retain(|(c, _)| c != conn)
+ }
+
+ // Add any outstanding requests on the closed connection back to the
+ // pending requests.
+ if let Some(id) = self.outstanding.iter()
+ .find(|(_, rw)| &rw.peer == peer && rw.connection == Some(*conn)) // (*)
+ .map(|(id, _)| *id)
+ {
+ let rw = self.outstanding.remove(&id).expect("by (*)");
+ let rw = RequestWrapper {
+ timestamp: rw.timestamp,
+ retries: rw.retries,
+ request: rw.request,
+ peer: (), // need to find another peer
+ connection: None,
+ };
+ self.pending_requests.push_back(rw);
+ }
+ }
+
+ fn inject_event(&mut self, peer: PeerId, conn: ConnectionId, event: Event<NegotiatedSubstream>) {
match event {
// An incoming request from remote has been received.
Event::Request(request, mut stream) => {
@@ -800,9 +869,10 @@ where
// A response to one of our own requests has been received.
Event::Response(id, response) => {
if let Some(request) = self.outstanding.remove(&id) {
- // We first just check if the response originates from the expected peer.
+ // We first just check if the response originates from the expected peer
+ // and connection.
if request.peer != peer {
- log::debug!("was expecting response from {} instead of {}", request.peer, peer);
+ log::debug!("Expected response from {} instead of {}.", request.peer, peer);
self.outstanding.insert(id, request);
self.remove_peer(&peer);
self.peerset.report_peer(peer, ReputationChange::new_fatal("response from unexpected peer"));
@@ -834,6 +904,7 @@ where
retries: request.retries,
request: request.request,
peer: (),
+ connection: None,
};
self.pending_requests.push_back(rw);
}
@@ -847,6 +918,7 @@ where
retries: request.retries - 1,
request: request.request,
peer: (),
+ connection: None,
};
self.pending_requests.push_back(rw)
} else {
@@ -886,54 +958,54 @@ where
request.timestamp = Instant::now();
request.retries -= 1
}
- let number = required_block(&request.request);
- let available_peer = {
- let p = self.idle_peers_with_block(number).next();
- if p.is_none() {
- self.idle_peers_with_unknown_block().next()
- } else {
- p
+
+
+ match self.prepare_request(request) {
+ Err(request) => {
+ self.pending_requests.push_front(request);
+ log::debug!("no peer available to send request to");
+ break
}
- };
- if let Some(peer) = available_peer {
- let buf = match serialize_request(&request.request) {
- Ok(b) => b,
- Err(e) => {
- log::debug!("failed to serialize request: {}", e);
- send_reply(Err(ClientError::RemoteFetchFailed), request.request);
- continue;
- }
- };
+ Ok((peer, request)) => {
+ let request_bytes = match serialize_request(&request.request) {
+ Ok(bytes) => bytes,
+ Err(error) => {
+ log::debug!("failed to serialize request: {}", error);
+ send_reply(Err(ClientError::RemoteFetchFailed), request.request);
+ continue
+ }
+ };
- let id = self.next_request_id();
- log::trace!("sending request {} to peer {}", id, peer);
- let protocol = OutboundProtocol {
- request: buf,
- request_id: id,
- expected: match request.request {
- Request::Body { .. } => ExpectedResponseTy::Block,
- _ => ExpectedResponseTy::Light,
- },
- max_response_size: self.config.max_response_size,
- protocol: match request.request {
- Request::Body { .. } => self.config.block_protocol.clone(),
- _ => self.config.light_protocol.clone(),
- },
- };
- self.peers.get_mut(&peer).map(|info| info.status = PeerStatus::BusyWith(id));
- let rw = RequestWrapper {
- timestamp: request.timestamp,
- retries: request.retries,
- request: request.request,
- peer: peer.clone(),
- };
- self.outstanding.insert(id, rw);
- return Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id: peer, event: protocol })
+ let (expected, protocol) = match request.request {
+ Request::Body { .. } =>
+ (ExpectedResponseTy::Block, self.config.block_protocol.clone()),
+ _ =>
+ (ExpectedResponseTy::Light, self.config.light_protocol.clone()),
+ };
- } else {
- self.pending_requests.push_front(request);
- log::debug!("no peer available to send request to");
- break
+ let peer_id = peer.clone();
+ let handler = request.connection.map_or(NotifyHandler::Any, NotifyHandler::One);
+
+ let request_id = self.next_request_id();
+ self.peers.get_mut(&peer).map(|p| p.status = PeerStatus::BusyWith(request_id));
+ self.outstanding.insert(request_id, request);
+
+ let event = OutboundProtocol {
+ request_id,
+ request: request_bytes,
+ expected,
+ max_response_size: self.config.max_response_size,
+ protocol,
+ };
+
+ log::trace!("sending request {} to peer {}", request_id, peer_id);
+
+ return Poll::Ready(NetworkBehaviourAction::NotifyHandler {
+ peer_id,
+ handler,
+ event,
+ })
+ }
}
}
@@ -959,6 +1031,7 @@ where
retries: rw.retries - 1,
request: rw.request,
peer: (),
+ connection: None,
};
self.pending_requests.push_back(rw)
}
@@ -1097,7 +1170,7 @@ pub enum Event<T> {
/// Incoming request from remote and substream to use for the response.
Request(api::v1::light::Request, T),
/// Incoming response from remote.
- Response(u64, Response),
+ Response(RequestId, Response),
}
/// Incoming response from remote.
@@ -1157,7 +1230,7 @@ pub struct OutboundProtocol {
/// The serialized protobuf request.
request: Vec<u8>,
/// Local identifier for the request. Used to associate it with a response.
- request_id: u64,
+ request_id: RequestId,
/// Kind of response expected for this request.
expected: ExpectedResponseTy,
/// The max. response length in bytes.
@@ -1244,6 +1317,7 @@ mod tests {
Multiaddr,
core::{
ConnectedPoint,
+ connection::ConnectionId,
identity,
muxing::{StreamMuxerBox, SubstreamRef},
transport::{Transport, boxed::Boxed, memory::MemoryTransport},
@@ -1457,10 +1531,12 @@ mod tests {
let pset = peerset();
let mut behaviour = make_behaviour(true, pset.1, make_config());
- behaviour.inject_connected(peer.clone(), empty_dialer());
+ behaviour.inject_connection_established(&peer, &ConnectionId::new(1), &empty_dialer());
+ behaviour.inject_connected(&peer);
assert_eq!(1, behaviour.peers.len());
- behaviour.inject_disconnected(&peer, empty_dialer());
+ behaviour.inject_connection_closed(&peer, &ConnectionId::new(1), &empty_dialer());
+ behaviour.inject_disconnected(&peer);
assert_eq!(0, behaviour.peers.len())
}
@@ -1471,8 +1547,10 @@ mod tests {
let pset = peerset();
let mut behaviour = make_behaviour(true, pset.1, make_config());
- behaviour.inject_connected(peer0.clone(), empty_dialer());
- behaviour.inject_connected(peer1.clone(), empty_dialer());
+ behaviour.inject_connection_established(&peer0, &ConnectionId::new(1), &empty_dialer());
+ behaviour.inject_connected(&peer0);
+ behaviour.inject_connection_established(&peer1, &ConnectionId::new(2), &empty_dialer());
+ behaviour.inject_connected(&peer1);
// We now know about two peers.
assert_eq!(HashSet::from_iter(&[peer0.clone(), peer1.clone()]), behaviour.peers.keys().collect::<HashSet<_>>());
@@ -1494,7 +1572,7 @@ mod tests {
assert_eq!(1, behaviour.pending_requests.len());
// The behaviour should now attempt to send the request.
- assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { peer_id, .. }) => {
+ assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { peer_id, .. }) => {
assert!(peer_id == peer0 || peer_id == peer1)
});
@@ -1534,7 +1612,9 @@ mod tests {
let mut behaviour = make_behaviour(false, pset.1, make_config());
// ^--- Making sure the response data check fails.
- behaviour.inject_connected(peer.clone(), empty_dialer());
+ let conn = ConnectionId::new(1);
+ behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+ behaviour.inject_connected(&peer);
assert_eq!(1, behaviour.peers.len());
let chan = oneshot::channel();
@@ -1562,7 +1642,7 @@ mod tests {
}
};
- behaviour.inject_node_event(peer.clone(), Event::Response(request_id, Response::Light(response)));
+ behaviour.inject_event(peer.clone(), conn, Event::Response(request_id, Response::Light(response)));
assert!(behaviour.peers.is_empty());
poll(&mut behaviour); // More progress
@@ -1578,7 +1658,9 @@ mod tests {
let pset = peerset();
let mut behaviour = make_behaviour(true, pset.1, make_config());
- behaviour.inject_connected(peer.clone(), empty_dialer());
+ let conn = ConnectionId::new(1);
+ behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+ behaviour.inject_connected(&peer);
assert_eq!(1, behaviour.peers.len());
assert_eq!(0, behaviour.pending_requests.len());
assert_eq!(0, behaviour.outstanding.len());
@@ -1591,7 +1673,7 @@ mod tests {
}
};
- behaviour.inject_node_event(peer.clone(), Event::Response(2347895932, Response::Light(response)));
+ behaviour.inject_event(peer.clone(), conn, Event::Response(2347895932, Response::Light(response)));
assert!(behaviour.peers.is_empty());
poll(&mut behaviour);
@@ -1605,7 +1687,9 @@ mod tests {
let pset = peerset();
let mut behaviour = make_behaviour(true, pset.1, make_config());
- behaviour.inject_connected(peer.clone(), empty_dialer());
+ let conn = ConnectionId::new(1);
+ behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+ behaviour.inject_connected(&peer);
assert_eq!(1, behaviour.peers.len());
let chan = oneshot::channel();
@@ -1633,7 +1717,7 @@ mod tests {
}
};
- behaviour.inject_node_event(peer.clone(), Event::Response(request_id, Response::Light(response)));
+ behaviour.inject_event(peer.clone(), conn, Event::Response(request_id, Response::Light(response)));
assert!(behaviour.peers.is_empty());
poll(&mut behaviour); // More progress
@@ -1653,10 +1737,18 @@ mod tests {
let mut behaviour = make_behaviour(false, pset.1, make_config());
// ^--- Making sure the response data check fails.
- behaviour.inject_connected(peer1.clone(), empty_dialer());
- behaviour.inject_connected(peer2.clone(), empty_dialer());
- behaviour.inject_connected(peer3.clone(), empty_dialer());
- behaviour.inject_connected(peer4.clone(), empty_dialer());
+ let conn1 = ConnectionId::new(1);
+ behaviour.inject_connection_established(&peer1, &conn1, &empty_dialer());
+ behaviour.inject_connected(&peer1);
+ let conn2 = ConnectionId::new(2);
+ behaviour.inject_connection_established(&peer2, &conn2, &empty_dialer());
+ behaviour.inject_connected(&peer2);
+ let conn3 = ConnectionId::new(3);
+ behaviour.inject_connection_established(&peer3, &conn3, &empty_dialer());
+ behaviour.inject_connected(&peer3);
+ let conn4 = ConnectionId::new(3);
+ behaviour.inject_connection_established(&peer4, &conn4, &empty_dialer());
+ behaviour.inject_connected(&peer4);
assert_eq!(4, behaviour.peers.len());
let mut chan = oneshot::channel();
@@ -1671,11 +1763,11 @@ mod tests {
assert_eq!(1, behaviour.pending_requests.len());
assert_eq!(0, behaviour.outstanding.len());
- assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+ assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
assert_eq!(0, behaviour.pending_requests.len());
assert_eq!(1, behaviour.outstanding.len());
- for _ in 0 .. 3 {
+ for i in 1 ..= 3 {
// Construct an invalid response
let request_id = *behaviour.outstanding.keys().next().unwrap();
let responding_peer = behaviour.outstanding.values().next().unwrap().peer.clone();
@@ -1685,8 +1777,9 @@ mod tests {
response: Some(api::v1::light::response::Response::RemoteCallResponse(r))
}
};
- behaviour.inject_node_event(responding_peer, Event::Response(request_id, Response::Light(response.clone())));
- assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+ let conn = ConnectionId::new(i);
+ behaviour.inject_event(responding_peer, conn, Event::Response(request_id, Response::Light(response.clone())));
+ assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
assert_matches!(chan.1.try_recv(), Ok(None))
}
// Final invalid response
@@ -1698,7 +1791,7 @@ mod tests {
response: Some(api::v1::light::response::Response::RemoteCallResponse(r)),
}
};
- behaviour.inject_node_event(responding_peer, Event::Response(request_id, Response::Light(response)));
+ behaviour.inject_event(responding_peer, conn4, Event::Response(request_id, Response::Light(response)));
assert_matches!(poll(&mut behaviour), Poll::Pending);
assert_matches!(chan.1.try_recv(), Ok(Some(Err(ClientError::RemoteFetchFailed))))
}
@@ -1708,7 +1801,9 @@ mod tests {
let pset = peerset();
let mut behaviour = make_behaviour(true, pset.1, make_config());
- behaviour.inject_connected(peer.clone(), empty_dialer());
+ let conn = ConnectionId::new(1);
+ behaviour.inject_connection_established(&peer, &conn, &empty_dialer());
+ behaviour.inject_connected(&peer);
assert_eq!(1, behaviour.peers.len());
let response = match request {
@@ -1757,12 +1852,12 @@ mod tests {
assert_eq!(1, behaviour.pending_requests.len());
assert_eq!(0, behaviour.outstanding.len());
- assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::SendEvent { .. }));
+ assert_matches!(poll(&mut behaviour), Poll::Ready(NetworkBehaviourAction::NotifyHandler { .. }));
assert_eq!(0, behaviour.pending_requests.len());
assert_eq!(1, behaviour.outstanding.len());
assert_eq!(1, *behaviour.outstanding.keys().next().unwrap());
- behaviour.inject_node_event(peer.clone(), Event::Response(1, Response::Light(response)));
+ behaviour.inject_event(peer.clone(), conn, Event::Response(1, Response::Light(response)));
poll(&mut behaviour);
diff --git a/substrate/client/network/src/service.rs b/substrate/client/network/src/service.rs
index ef2aa0aa233b6c8df176b5d0338b5d62a9e47583..fb33901dd09c3ed3a8b4121be38be26db4859e5f 100644
--- a/substrate/client/network/src/service.rs
+++ b/substrate/client/network/src/service.rs
@@ -37,9 +37,10 @@ use crate::{
transport, ReputationChange,
};
use futures::prelude::*;
-use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver};
+use libp2p::{PeerId, Multiaddr};
+use libp2p::core::{Executor, connection::PendingConnectionError};
+use libp2p::kad::record;
use libp2p::swarm::{NetworkBehaviour, SwarmBuilder, SwarmEvent};
-use libp2p::{kad::record, Multiaddr, PeerId};
use log::{error, info, trace, warn};
use parking_lot::Mutex;
use prometheus_endpoint::{
@@ -51,6 +52,7 @@ use sp_runtime::{
traits::{Block as BlockT, NumberFor},
ConsensusEngineId,
};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver};
use std::{
borrow::Cow,
collections::{HashMap, HashSet},
@@ -322,9 +324,16 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
};
transport::build_transport(local_identity, config_mem, config_wasm, flowctrl)
};
- let mut builder = SwarmBuilder::new(transport, behaviour, local_peer_id.clone());
+ let mut builder = SwarmBuilder::new(transport, behaviour, local_peer_id.clone())
+ .peer_connection_limit(crate::MAX_CONNECTIONS_PER_PEER);
if let Some(spawner) = params.executor {
- builder = builder.executor_fn(spawner);
+ struct SpawnImpl<F>(F);
+ impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for SpawnImpl<F> {
+ fn exec(&self, f: Pin<Box<dyn Future<Output = ()> + Send>>) {
+ (self.0)(f)
+ }
+ }
+ builder = builder.executor(Box::new(SpawnImpl(spawner)));
}
(builder.build(), bandwidth)
};
@@ -1038,13 +1047,13 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
metrics.update_with_network_event(&ev);
}
},
- Poll::Ready(SwarmEvent::Connected(peer_id)) => {
+ Poll::Ready(SwarmEvent::ConnectionEstablished { peer_id, .. }) => {
trace!(target: "sub-libp2p", "Libp2p => Connected({:?})", peer_id);
if let Some(metrics) = this.metrics.as_ref() {
metrics.connections.inc();
}
},
- Poll::Ready(SwarmEvent::Disconnected(peer_id)) => {
+ Poll::Ready(SwarmEvent::ConnectionClosed { peer_id, .. }) => {
trace!(target: "sub-libp2p", "Libp2p => Disconnected({:?})", peer_id);
if let Some(metrics) = this.metrics.as_ref() {
metrics.connections.dec();
@@ -1054,9 +1063,7 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
trace!(target: "sub-libp2p", "Libp2p => NewListenAddr({})", addr),
Poll::Ready(SwarmEvent::ExpiredListenAddr(addr)) =>
trace!(target: "sub-libp2p", "Libp2p => ExpiredListenAddr({})", addr),
- Poll::Ready(SwarmEvent::UnreachableAddr { peer_id, address, error }) => {
- let error = error.to_string();
-
+ Poll::Ready(SwarmEvent::UnreachableAddr { peer_id, address, error, .. }) => {
trace!(
target: "sub-libp2p", "Libp2p => Failed to reach {:?} through {:?}: {}",
peer_id,
@@ -1064,21 +1071,34 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
error,
);
- if let Some(peer_id) = peer_id {
- if this.boot_node_ids.contains(&peer_id)
- && error.contains("Peer ID mismatch")
- {
+ if this.boot_node_ids.contains(&peer_id) {
+ if let PendingConnectionError::InvalidPeerId = error {
error!(
- "💔 Connecting to bootnode with peer id `{}` and address `{}` failed \
- because it returned a different peer id!",
+ "💔 Invalid peer ID from bootnode, expected `{}` at address `{}`.",
peer_id,
address,
);
}
}
- },
- Poll::Ready(SwarmEvent::StartConnect(peer_id)) =>
- trace!(target: "sub-libp2p", "Libp2p => StartConnect({:?})", peer_id),
+ }
+ Poll::Ready(SwarmEvent::Dialing(peer_id)) =>
+ trace!(target: "sub-libp2p", "Libp2p => Dialing({:?})", peer_id),
+ Poll::Ready(SwarmEvent::IncomingConnection { local_addr, send_back_addr }) =>
+ trace!(target: "sub-libp2p", "Libp2p => IncomingConnection({},{}))",
+ local_addr, send_back_addr),
+ Poll::Ready(SwarmEvent::IncomingConnectionError { local_addr, send_back_addr, error }) =>
+ trace!(target: "sub-libp2p", "Libp2p => IncomingConnectionError({},{}): {}",
+ local_addr, send_back_addr, error),
+ Poll::Ready(SwarmEvent::BannedPeer { peer_id, endpoint }) =>
+ trace!(target: "sub-libp2p", "Libp2p => BannedPeer({}). Connected via {:?}.",
+ peer_id, endpoint),
+ Poll::Ready(SwarmEvent::UnknownPeerUnreachableAddr { address, error }) =>
+ trace!(target: "sub-libp2p", "Libp2p => UnknownPeerUnreachableAddr({}): {}",
+ address, error),
+ Poll::Ready(SwarmEvent::ListenerClosed { reason, addresses: _ }) =>
+ warn!(target: "sub-libp2p", "Libp2p => ListenerClosed: {:?}", reason),
+ Poll::Ready(SwarmEvent::ListenerError { error }) =>
+ trace!(target: "sub-libp2p", "Libp2p => ListenerError: {}", error),
};
}
diff --git a/substrate/client/network/test/Cargo.toml b/substrate/client/network/test/Cargo.toml
index 769e0faca9e5cd8a1055515f5c9abd66df22b915..90202008eb37fe99a1a98009bdce923ebe0dfc1c 100644
--- a/substrate/client/network/test/Cargo.toml
+++ b/substrate/client/network/test/Cargo.toml
@@ -16,7 +16,7 @@ parking_lot = "0.10.0"
futures = "0.3.4"
futures-timer = "3.0.1"
rand = "0.7.2"
-libp2p = { version = "0.16.2", default-features = false, features = ["libp2p-websocket"] }
+libp2p = { version = "0.17.0", default-features = false, features = ["libp2p-websocket"] }
sp-consensus = { version = "0.8.0-alpha.5", path = "../../../primitives/consensus/common" }
sc-client = { version = "0.8.0-alpha.5", path = "../../" }
sc-client-api = { version = "2.0.0-alpha.5", path = "../../api" }
diff --git a/substrate/client/peerset/Cargo.toml b/substrate/client/peerset/Cargo.toml
index 78d488a9899fc75ec68aad74e48c9017e8ae5de4..e026c6063a01fbd9e6716d0a3fe3b59cf23302eb 100644
--- a/substrate/client/peerset/Cargo.toml
+++ b/substrate/client/peerset/Cargo.toml
@@ -12,7 +12,7 @@ documentation = "https://docs.rs/sc-peerset"
[dependencies]
futures = "0.3.4"
-libp2p = { version = "0.16.2", default-features = false }
+libp2p = { version = "0.17.0", default-features = false }
sp-utils = { version = "2.0.0-alpha.5", path = "../../primitives/utils"}
log = "0.4.8"
serde_json = "1.0.41"
diff --git a/substrate/client/telemetry/Cargo.toml b/substrate/client/telemetry/Cargo.toml
index bb7e7e510372bc253ba6fa2130c6f0e1b7d4a31b..dcf218e9763fdb9cb39e7a7877881c7915a5d055 100644
--- a/substrate/client/telemetry/Cargo.toml
+++ b/substrate/client/telemetry/Cargo.toml
@@ -16,7 +16,7 @@ parking_lot = "0.10.0"
futures = "0.3.4"
futures-timer = "3.0.1"
wasm-timer = "0.2.0"
-libp2p = { version = "0.16.2", default-features = false, features = ["libp2p-websocket"] }
+libp2p = { version = "0.17.0", default-features = false, features = ["websocket", "wasm-ext", "tcp", "dns"] }
log = "0.4.8"
pin-project = "0.4.6"
rand = "0.7.2"
diff --git a/substrate/primitives/consensus/common/Cargo.toml b/substrate/primitives/consensus/common/Cargo.toml
index 112b9499a7485a289893965c644918b1451165a1..4734cde694261bd3d09c4eb688619f8d208c07eb 100644
--- a/substrate/primitives/consensus/common/Cargo.toml
+++ b/substrate/primitives/consensus/common/Cargo.toml
@@ -12,7 +12,7 @@ documentation = "https://docs.rs/sp-consensus/"
[dependencies]
derive_more = "0.99.2"
-libp2p = { version = "0.16.2", default-features = false }
+libp2p = { version = "0.17.0", default-features = false }
log = "0.4.8"
sp-core = { path= "../../core" , version = "2.0.0-alpha.5"}
sp-inherents = { version = "2.0.0-alpha.5", path = "../../inherents" }
diff --git a/substrate/utils/browser/Cargo.toml b/substrate/utils/browser/Cargo.toml
index 188f46bf197b831aaad001528befd4cc4b60f47c..5f6b18e00148c5e34e694b1d86113bb873b88461 100644
--- a/substrate/utils/browser/Cargo.toml
+++ b/substrate/utils/browser/Cargo.toml
@@ -12,7 +12,7 @@ repository = "https://github.com/paritytech/substrate/"
futures = "0.3"
futures01 = { package = "futures", version = "0.1.29" }
log = "0.4.8"
-libp2p-wasm-ext = { version = "0.16.2", features = ["websocket"] }
+libp2p-wasm-ext = { version = "0.17.0", features = ["websocket"] }
console_error_panic_hook = "0.1.6"
console_log = "0.1.2"
js-sys = "0.3.34"