Gossipsub Protocol

Cargo.toml

@@ -22,6 +22,7 @@ libp2p-mplex = { version = "0.3.0", path = "./muxers/mplex" }
 libp2p-identify = { version = "0.3.0", path = "./protocols/identify" }
 libp2p-kad = { version = "0.3.0", path = "./protocols/kad" }
 libp2p-floodsub = { version = "0.3.0", path = "./protocols/floodsub" }
+libp2p-gossipsub = { version = "0.1.0", path = "./protocols/gossipsub" }
 libp2p-ping = { version = "0.3.0", path = "./protocols/ping" }
 libp2p-plaintext = { version = "0.3.0", path = "./protocols/plaintext" }
 libp2p-ratelimit = { version = "0.3.0", path = "./transports/ratelimit" }
@@ -64,6 +65,7 @@ members = [
     "muxers/mplex",
     "muxers/yamux",
     "protocols/floodsub",
+    "protocols/gossipsub",
     "protocols/identify",
     "protocols/kad",
     "protocols/noise",

protocols/gossipsub/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "libp2p-gossipsub"
+version = "0.1.0"
+authors = ["Age Manning <age@agemanning.com>"]
+license = "MIT"
+
+[dependencies]
+libp2p-core = { path = "../../core" }
+libp2p-floodsub = { path = "../floodsub" }
+bs58 = "0.2.2"
+bytes = "0.4.11"
+byteorder = "1.3.1"
+cuckoofilter = "0.3.2"
+fnv = "1.0.6"
+futures = "0.1.25"
+protobuf = "2.3.0"
+rand = "0.6.5"
+smallvec = "0.6.8"
+tokio-codec = "0.1.1"
+tokio-io = "0.1.11"
+tokio-timer = "0.2.8"
+unsigned-varint = "0.2.1"
+log = "0.4.6"
+
+
+[dev-dependencies]
+libp2p = { path = "../../" }
+tokio = "0.1"
+tokio-stdin-stdout = "0.1"
+env_logger = "0.6.0"

protocols/gossipsub/examples/chat.rs

@@ -0,0 +1,88 @@
+extern crate env_logger;
+extern crate futures;
+extern crate libp2p;
+extern crate log;
+extern crate tokio;
+
+use env_logger::{Builder, Env};
+use futures::prelude::*;
+use libp2p::gossipsub::GossipsubEvent;
+use libp2p::{
+    gossipsub, secio,
+    tokio_codec::{FramedRead, LinesCodec},
+};
+use std::time::Duration;
+
+fn main() {
+    Builder::from_env(Env::default().default_filter_or("debug")).init();
+
+    // Create a random PeerId
+    let local_key = secio::SecioKeyPair::ed25519_generated().unwrap();
+    let local_peer_id = local_key.to_peer_id();
+    println!("Local peer id: {:?}", local_peer_id);
+
+    // Set up an encrypted TCP Transport over the Mplex and Yamux protocols
+    let transport = libp2p::build_development_transport(local_key);
+
+    // Create a Floodsub/Gossipsub topic
+    let topic = libp2p::floodsub::TopicBuilder::new("test").build();
+
+    // Create a Swarm to manage peers and events
+    let mut swarm = {
+        // set default parameters for gossipsub
+        // let gossipsub_config = gossipsub::GossipsubConfig::default();
+        // set custom gossipsub
+        let gossipsub_config = gossipsub::GossipsubConfigBuilder::new()
+            .heartbeat_interval(Duration::from_secs(10))
+            .build();
+        // build a gossipsub network behaviour
+        let mut gossipsub = gossipsub::Gossipsub::new(local_peer_id.clone(), gossipsub_config);
+        gossipsub.subscribe(topic.clone());
+        libp2p::Swarm::new(transport, gossipsub, local_peer_id)
+    };
+
+    // Listen on all interfaces and whatever port the OS assigns
+    let addr = libp2p::Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
+    println!("Listening on {:?}", addr);
+
+    // Reach out to another node if specified
+    if let Some(to_dial) = std::env::args().nth(1) {
+        let dialing = to_dial.clone();
+        match to_dial.parse() {
+            Ok(to_dial) => match libp2p::Swarm::dial_addr(&mut swarm, to_dial) {
+                Ok(_) => println!("Dialed {:?}", dialing),
+                Err(e) => println!("Dial {:?} failed: {:?}", dialing, e),
+            },
+            Err(err) => println!("Failed to parse address to dial: {:?}", err),
+        }
+    }
+
+    // Read full lines from stdin
+    let stdin = tokio_stdin_stdout::stdin(0);
+    let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
+
+    // Kick it off
+    tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
+        loop {
+            match framed_stdin.poll().expect("Error while polling stdin") {
+                Async::Ready(Some(line)) => swarm.publish(&topic, line.as_bytes()),
+                Async::Ready(None) => panic!("Stdin closed"),
+                Async::NotReady => break,
+            };
+        }
+
+        loop {
+            match swarm.poll().expect("Error while polling swarm") {
+                Async::Ready(Some(gossip_event)) => match gossip_event {
+                    GossipsubEvent::Message(message) => {
+                        println!("Got message: {:?}", String::from_utf8_lossy(&message.data))
+                    }
+                    _ => {}
+                },
+                Async::Ready(None) | Async::NotReady => break,
+            }
+        }
+
+        Ok(Async::NotReady)
+    }));
+}

protocols/gossipsub/regen_structs_proto.sh

@@ -0,0 +1,13 @@
+#!/bin/sh
+
+# This script regenerates the `src/rpc_proto.rs` file from `rpc.proto`.
+
+docker run --rm -v `pwd`:/usr/code:z -w /usr/code rust /bin/bash -c " \
+    apt-get update; \
+    apt-get install -y protobuf-compiler; \
+    cargo install --version 2.3.0 protobuf-codegen; \
+    protoc --rust_out . rpc.proto"
+
+sudo chown $USER:$USER *.rs
+
+mv -f rpc.rs ./src/rpc_proto.rs

protocols/gossipsub/rpc.proto

@@ -0,0 +1,75 @@
+syntax = "proto2";
+
+package gossipsub.pb;
+
+message RPC {
+	repeated SubOpts subscriptions = 1;
+	repeated Message publish = 2;
+
+	message SubOpts {
+		optional bool subscribe = 1; // subscribe or unsubscribe
+		optional string topicid = 2;
+	}
+
+	optional ControlMessage control = 3;
+}
+
+message Message {
+	optional bytes from = 1;
+	optional bytes data = 2;
+	optional bytes seqno = 3;
+	repeated string topicIDs = 4;
+}
+
+message ControlMessage {
+	repeated ControlIHave ihave = 1;
+	repeated ControlIWant iwant = 2;
+	repeated ControlGraft graft = 3;
+	repeated ControlPrune prune = 4;
+}
+
+message ControlIHave {
+	optional string topicID = 1;
+	repeated string messageIDs = 2;
+}
+
+message ControlIWant {
+	repeated string messageIDs = 1;
+}
+
+message ControlGraft {
+	optional string topicID = 1;
+}
+
+message ControlPrune {
+	optional string topicID = 1;
+}
+
+// topicID = hash(topicDescriptor); (not the topic.name)
+message TopicDescriptor {
+	optional string name = 1;
+	optional AuthOpts auth = 2;
+	optional EncOpts enc = 3;
+
+	message AuthOpts {
+		optional AuthMode mode = 1;
+		repeated bytes keys = 2; // root keys to trust
+
+		enum AuthMode {
+			NONE = 0; // no authentication, anyone can publish
+			KEY = 1; // only messages signed by keys in the topic descriptor are accepted
+			WOT = 2; // web of trust, certificates can allow publisher set to grow
+		}
+	}
+
+	message EncOpts {
+		optional EncMode mode = 1;
+		repeated bytes keyHashes = 2; // the hashes of the shared keys used (salted)
+
+		enum EncMode {
+			NONE = 0; // no encryption, anyone can read
+			SHAREDKEY = 1; // messages are encrypted with shared key
+			WOT = 2; // web of trust, certificates can allow publisher set to grow
+		}
+	}
+}

protocols/gossipsub/src/gossipsub_config.rs

@@ -0,0 +1,167 @@
+use std::time::Duration;
+
+/// Configuration parameters that define the performance of the gossipsub network.
+#[derive(Clone)]
+pub struct GossipsubConfig {
+    /// Overlay network parameters.
+    /// Number of heartbeats to keep in the `memcache`.
+    pub history_length: usize,
+    /// Number of past heartbeats to gossip about.
+    pub history_gossip: usize,
+
+    /// Target number of peers for the mesh network (D in the spec).
+    pub mesh_n: usize,
+    /// Minimum number of peers in mesh network before adding more (D_lo in the spec).
+    pub mesh_n_low: usize,
+    /// Maximum number of peers in mesh network before removing some (D_high in the spec).
+    pub mesh_n_high: usize,
+
+    /// Number of peers to emit gossip to during a heartbeat (D_lazy in the spec).
+    pub gossip_lazy: usize,
+
+    /// Initial delay in each heartbeat.
+    pub heartbeat_initial_delay: Duration,
+    /// Time between each heartbeat.
+    pub heartbeat_interval: Duration,
+    /// Time to live for fanout peers.
+    pub fanout_ttl: Duration,
+}
+
+impl Default for GossipsubConfig {
+    fn default() -> GossipsubConfig {
+        GossipsubConfig {
+            history_length: 5,
+            history_gossip: 3,
+            mesh_n: 6,
+            mesh_n_low: 4,
+            mesh_n_high: 12,
+            gossip_lazy: 6, // default to mesh_n
+            heartbeat_initial_delay: Duration::from_secs(5),
+            heartbeat_interval: Duration::from_secs(1),
+            fanout_ttl: Duration::from_secs(60),
+        }
+    }
+}
+
+pub struct GossipsubConfigBuilder {
+    history_length: usize,
+    /// Number of past heartbeats to gossip about.
+    history_gossip: usize,
+
+    /// Target number of peers for the mesh network (D in the spec).
+    mesh_n: usize,
+    /// Minimum number of peers in mesh network before adding more (D_lo in the spec).
+    mesh_n_low: usize,
+    /// Maximum number of peers in mesh network before removing some (D_high in the spec).
+    mesh_n_high: usize,
+
+    /// Number of peers to emit gossip to during a heartbeat (D_lazy in the spec).
+    gossip_lazy: usize,
+
+    /// Initial delay in each heartbeat.
+    heartbeat_initial_delay: Duration,
+    /// Time between each heartbeat.
+    heartbeat_interval: Duration,
+    /// Time to live for fanout peers.
+    fanout_ttl: Duration,
+}
+
+impl Default for GossipsubConfigBuilder {
+    fn default() -> GossipsubConfigBuilder {
+        GossipsubConfigBuilder {
+            history_length: 5,
+            history_gossip: 3,
+            mesh_n: 6,
+            mesh_n_low: 4,
+            mesh_n_high: 12,
+            gossip_lazy: 6, // default to mesh_n
+            heartbeat_initial_delay: Duration::from_secs(5),
+            heartbeat_interval: Duration::from_secs(1),
+            fanout_ttl: Duration::from_secs(60),
+        }
+    }
+}
+
+impl GossipsubConfigBuilder {
+    // set default values
+    pub fn new() -> GossipsubConfigBuilder {
+        GossipsubConfigBuilder::default()
+    }
+
+    pub fn history_length(&mut self, history_length: usize) -> &mut Self {
+        assert!(
+            history_length >= self.history_gossip,
+            "The history_length must be greater than or equal to the history_gossip length"
+        );
+        self.history_length = history_length;
+        self
+    }
+
+    pub fn history_gossip(&mut self, history_gossip: usize) -> &mut Self {
+        assert!(
+            self.history_length >= history_gossip,
+            "The history_length must be greater than or equal to the history_gossip length"
+        );
+        self.history_gossip = history_gossip;
+        self
+    }
+
+    pub fn mesh_n(&mut self, mesh_n: usize) -> &mut Self {
+        assert!(
+            self.mesh_n_low <= mesh_n && mesh_n <= self.mesh_n_high,
+            "The following equality doesn't hold mesh_n_low <= mesh_n <= mesh_n_high"
+        );
+        self.mesh_n = mesh_n;
+        self
+    }
+
+    pub fn mesh_n_low(&mut self, mesh_n_low: usize) -> &mut Self {
+        assert!(
+            mesh_n_low <= self.mesh_n && self.mesh_n <= self.mesh_n_high,
+            "The following equality doesn't hold mesh_n_low <= mesh_n <= mesh_n_high"
+        );
+        self.mesh_n_low = mesh_n_low;
+        self
+    }
+
+    pub fn mesh_n_high(&mut self, mesh_n_high: usize) -> &mut Self {
+        assert!(
+            self.mesh_n_low <= self.mesh_n && self.mesh_n <= mesh_n_high,
+            "The following equality doesn't hold mesh_n_low <= mesh_n <= mesh_n_high"
+        );
+        self.mesh_n_high = mesh_n_high;
+        self
+    }
+
+    pub fn gossip_lazy(&mut self, gossip_lazy: usize) -> &mut Self {
+        self.gossip_lazy = gossip_lazy;
+        self
+    }
+
+    pub fn heartbeat_initial_delay(&mut self, heartbeat_initial_delay: Duration) -> &mut Self {
+        self.heartbeat_initial_delay = heartbeat_initial_delay;
+        self
+    }
+    pub fn heartbeat_interval(&mut self, heartbeat_interval: Duration) -> &mut Self {
+        self.heartbeat_interval = heartbeat_interval;
+        self
+    }
+    pub fn fanout_ttl(&mut self, fanout_ttl: Duration) -> &mut Self {
+        self.fanout_ttl = fanout_ttl;
+        self
+    }
+
+    pub fn build(&self) -> GossipsubConfig {
+        GossipsubConfig {
+            history_length: self.history_length,
+            history_gossip: self.history_gossip,
+            mesh_n: self.mesh_n,
+            mesh_n_low: self.mesh_n_low,
+            mesh_n_high: self.mesh_n_high,
+            gossip_lazy: self.gossip_lazy,
+            heartbeat_initial_delay: self.heartbeat_initial_delay,
+            heartbeat_interval: self.heartbeat_interval,
+            fanout_ttl: self.fanout_ttl,
+        }
+    }
+}