gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Tpu vote 1.7 (#20187) (#20494)

Browse Source 
* Add separate vote processing tpu port

* Add feature to send to tpu vote port

* Add vote rejecting sigverify mode

* use packet.meta.is_simple_vote_tx in place of deserialization

* consolidate code that identifies vote tx atcommon path for cpu and gpu

* new key for feature set

* banking forward tpu vote

* add tpu vote port to dockerfile and other review changes

* Simplify thread id compare

* fix a test; updated cluster_info ABI change

Co-authored-by: Tao Zhu <tao@solana.com>

Co-authored-by: sakridge <sakridge@gmail.com>
This commit is contained in:
Tao Zhu
2021-10-07 04:38:23 -05:00
committed by GitHub
parent 2a91b34656
commit 177a375479
20 changed files with 435 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
gossip/src/cluster_info.rs
View File

@ -254,7 +254,7 @@ pub fn make_accounts_hashes_message(
pub(crate) type Ping = ping_pong::Ping<[u8; GOSSIP_PING_TOKEN_SIZE]>;
// TODO These messages should go through the gpu pipeline for spam filtering
#[frozen_abi(digest = "skH6cJ1MHfUyJKVaTfRRDV9y29HjuRpfReFpxXMWNky")]
#[frozen_abi(digest = "4VqzaZbxQkeTgo916HVoLtaWoM8bbGaQZy6Qgw7K9kLf")]
#[derive(Serialize, Deserialize, Debug, AbiEnumVisitor, AbiExample)]
#[allow(clippy::large_enum_variant)]
pub(crate) enum Protocol {
@ -765,7 +765,7 @@ impl ClusterInfo {
};
let ip_addr = node.gossip.ip();
Some(format!(
"{:15} {:2}| {:5} | {:44} |{:^9}| {:5}| {:5}| {:5}| {:5}| {:5}| {:5}| {:5}| {}\n",
"{:15} {:2}| {:5} | {:44} |{:^9}| {:5}| {:5}| {:5}| {:5}| {:5}| {:5}| {:5}| {:5}| {}\n",
if ContactInfo::is_valid_address(&node.gossip, &self.socket_addr_space) {
ip_addr.to_string()
} else {
@ -780,6 +780,7 @@ impl ClusterInfo {
"-".to_string()
},
addr_to_string(&ip_addr, &node.gossip),
addr_to_string(&ip_addr, &node.tpu_vote),
addr_to_string(&ip_addr, &node.tpu),
addr_to_string(&ip_addr, &node.tpu_forwards),
addr_to_string(&ip_addr, &node.tvu),
@ -794,9 +795,9 @@ impl ClusterInfo {
format!(
"IP Address |Age(ms)| Node identifier \
| Version |Gossip| TPU |TPUfwd| TVU |TVUfwd|Repair|ServeR|ShredVer\n\
| Version |Gossip|TPUvote| TPU |TPUfwd| TVU |TVUfwd|Repair|ServeR|ShredVer\n\
------------------+-------+----------------------------------------------+---------+\
------+------+------+------+------+------+------+--------\n\
------+------+-------+------+------+------+------+------+--------\n\
{}\
Nodes: {}{}{}",
nodes.join(""),
@ -2667,6 +2668,7 @@ pub struct Sockets {
pub tvu_forwards: Vec<UdpSocket>,
pub tpu: Vec<UdpSocket>,
pub tpu_forwards: Vec<UdpSocket>,
pub tpu_vote: Vec<UdpSocket>,
pub broadcast: Vec<UdpSocket>,
pub repair: UdpSocket,
pub retransmit_sockets: Vec<UdpSocket>,
@ -2693,6 +2695,7 @@ impl Node {
let tvu = UdpSocket::bind("127.0.0.1:0").unwrap();
let tvu_forwards = UdpSocket::bind("127.0.0.1:0").unwrap();
let tpu_forwards = UdpSocket::bind("127.0.0.1:0").unwrap();
let tpu_vote = UdpSocket::bind("127.0.0.1:0").unwrap();
let repair = UdpSocket::bind("127.0.0.1:0").unwrap();
let rpc_port = find_available_port_in_range(bind_ip_addr, (1024, 65535)).unwrap();
let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), rpc_port);
@ -2703,7 +2706,6 @@ impl Node {
let broadcast = vec![UdpSocket::bind("0.0.0.0:0").unwrap()];
let retransmit_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
let serve_repair = UdpSocket::bind("127.0.0.1:0").unwrap();
let unused = UdpSocket::bind("0.0.0.0:0").unwrap();
let info = ContactInfo {
id: *pubkey,
gossip: gossip_addr,
@ -2712,7 +2714,7 @@ impl Node {
repair: repair.local_addr().unwrap(),
tpu: tpu.local_addr().unwrap(),
tpu_forwards: tpu_forwards.local_addr().unwrap(),
unused: unused.local_addr().unwrap(),
tpu_vote: tpu_vote.local_addr().unwrap(),
rpc: rpc_addr,
rpc_pubsub: rpc_pubsub_addr,
serve_repair: serve_repair.local_addr().unwrap(),
@ -2728,6 +2730,7 @@ impl Node {
tvu_forwards: vec![tvu_forwards],
tpu: vec![tpu],
tpu_forwards: vec![tpu_forwards],
tpu_vote: vec![tpu_vote],
broadcast,
repair,
retransmit_sockets: vec![retransmit_socket],
@ -2768,6 +2771,7 @@ impl Node {
let (tvu_forwards_port, tvu_forwards) = Self::bind(bind_ip_addr, port_range);
let (tpu_port, tpu) = Self::bind(bind_ip_addr, port_range);
let (tpu_forwards_port, tpu_forwards) = Self::bind(bind_ip_addr, port_range);
let (tpu_vote_port, tpu_vote) = Self::bind(bind_ip_addr, port_range);
let (_, retransmit_socket) = Self::bind(bind_ip_addr, port_range);
let (repair_port, repair) = Self::bind(bind_ip_addr, port_range);
let (serve_repair_port, serve_repair) = Self::bind(bind_ip_addr, port_range);
@ -2784,7 +2788,7 @@ impl Node {
repair: SocketAddr::new(gossip_addr.ip(), repair_port),
tpu: SocketAddr::new(gossip_addr.ip(), tpu_port),
tpu_forwards: SocketAddr::new(gossip_addr.ip(), tpu_forwards_port),
unused: socketaddr_any!(),
tpu_vote: SocketAddr::new(gossip_addr.ip(), tpu_vote_port),
rpc: SocketAddr::new(gossip_addr.ip(), rpc_port),
rpc_pubsub: SocketAddr::new(gossip_addr.ip(), rpc_pubsub_port),
serve_repair: SocketAddr::new(gossip_addr.ip(), serve_repair_port),
@ -2802,6 +2806,7 @@ impl Node {
tvu_forwards: vec![tvu_forwards],
tpu: vec![tpu],
tpu_forwards: vec![tpu_forwards],
tpu_vote: vec![tpu_vote],
broadcast: vec![broadcast],
repair,
retransmit_sockets: vec![retransmit_socket],
@ -2831,6 +2836,9 @@ impl Node {
let (tpu_forwards_port, tpu_forwards_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("tpu_forwards multi_bind");
let (tpu_vote_port, tpu_vote_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 1).expect("tpu_vote multi_bind");
let (_, retransmit_sockets) =
multi_bind_in_range(bind_ip_addr, port_range, 8).expect("retransmit multi_bind");
@ -2848,7 +2856,7 @@ impl Node {
repair: SocketAddr::new(gossip_addr.ip(), repair_port),
tpu: SocketAddr::new(gossip_addr.ip(), tpu_port),
tpu_forwards: SocketAddr::new(gossip_addr.ip(), tpu_forwards_port),
unused: socketaddr_any!(),
tpu_vote: SocketAddr::new(gossip_addr.ip(), tpu_vote_port),
rpc: socketaddr_any!(),
rpc_pubsub: socketaddr_any!(),
serve_repair: SocketAddr::new(gossip_addr.ip(), serve_repair_port),
@ -2865,6 +2873,7 @@ impl Node {
tvu_forwards: tvu_forwards_sockets,
tpu: tpu_sockets,
tpu_forwards: tpu_forwards_sockets,
tpu_vote: tpu_vote_sockets,
broadcast,
repair,
retransmit_sockets,

20
gossip/src/contact_info.rs
View File

@ -28,7 +28,7 @@ pub struct ContactInfo {
/// address to forward unprocessed transactions to
pub tpu_forwards: SocketAddr,
/// address to which to send bank state requests
pub unused: SocketAddr,
pub tpu_vote: SocketAddr,
/// address to which to send JSON-RPC requests
pub rpc: SocketAddr,
/// websocket for JSON-RPC push notifications
@ -76,7 +76,7 @@ impl Default for ContactInfo {
repair: socketaddr_any!(),
tpu: socketaddr_any!(),
tpu_forwards: socketaddr_any!(),
unused: socketaddr_any!(),
tpu_vote: socketaddr_any!(),
rpc: socketaddr_any!(),
rpc_pubsub: socketaddr_any!(),
serve_repair: socketaddr_any!(),
@ -96,7 +96,7 @@ impl ContactInfo {
repair: socketaddr!("127.0.0.1:1237"),
tpu: socketaddr!("127.0.0.1:1238"),
tpu_forwards: socketaddr!("127.0.0.1:1239"),
unused: socketaddr!("127.0.0.1:1240"),
tpu_vote: socketaddr!("127.0.0.1:1240"),
rpc: socketaddr!("127.0.0.1:1241"),
rpc_pubsub: socketaddr!("127.0.0.1:1242"),
serve_repair: socketaddr!("127.0.0.1:1243"),
@ -126,7 +126,7 @@ impl ContactInfo {
repair: addr,
tpu: addr,
tpu_forwards: addr,
unused: addr,
tpu_vote: addr,
rpc: addr,
rpc_pubsub: addr,
serve_repair: addr,
@ -152,6 +152,7 @@ impl ContactInfo {
let rpc = SocketAddr::new(bind_addr.ip(), rpc_port::DEFAULT_RPC_PORT);
let rpc_pubsub = SocketAddr::new(bind_addr.ip(), rpc_port::DEFAULT_RPC_PUBSUB_PORT);
let serve_repair = next_port(bind_addr, 6);
let tpu_vote = next_port(bind_addr, 7);
Self {
id: *pubkey,
gossip,
@ -160,7 +161,7 @@ impl ContactInfo {
repair,
tpu,
tpu_forwards,
unused: "0.0.0.0:0".parse().unwrap(),
tpu_vote,
rpc,
rpc_pubsub,
serve_repair,
@ -262,7 +263,7 @@ mod tests {
assert!(ci.rpc.ip().is_unspecified());
assert!(ci.rpc_pubsub.ip().is_unspecified());
assert!(ci.tpu.ip().is_unspecified());
assert!(ci.unused.ip().is_unspecified());
assert!(ci.tpu_vote.ip().is_unspecified());
assert!(ci.serve_repair.ip().is_unspecified());
}
#[test]
@ -274,7 +275,7 @@ mod tests {
assert!(ci.rpc.ip().is_multicast());
assert!(ci.rpc_pubsub.ip().is_multicast());
assert!(ci.tpu.ip().is_multicast());
assert!(ci.unused.ip().is_multicast());
assert!(ci.tpu_vote.ip().is_multicast());
assert!(ci.serve_repair.ip().is_multicast());
}
#[test]
@ -287,7 +288,7 @@ mod tests {
assert!(ci.rpc.ip().is_unspecified());
assert!(ci.rpc_pubsub.ip().is_unspecified());
assert!(ci.tpu.ip().is_unspecified());
assert!(ci.unused.ip().is_unspecified());
assert!(ci.tpu_vote.ip().is_unspecified());
assert!(ci.serve_repair.ip().is_unspecified());
}
#[test]
@ -295,12 +296,12 @@ mod tests {
let addr = socketaddr!("127.0.0.1:10");
let ci = ContactInfo::new_with_socketaddr(&addr);
assert_eq!(ci.tpu, addr);
assert_eq!(ci.tpu_vote.port(), 17);
assert_eq!(ci.gossip.port(), 11);
assert_eq!(ci.tvu.port(), 12);
assert_eq!(ci.tpu_forwards.port(), 13);
assert_eq!(ci.rpc.port(), rpc_port::DEFAULT_RPC_PORT);
assert_eq!(ci.rpc_pubsub.port(), rpc_port::DEFAULT_RPC_PUBSUB_PORT);
assert!(ci.unused.ip().is_unspecified());
assert_eq!(ci.serve_repair.port(), 16);
}
@ -327,6 +328,7 @@ mod tests {
assert_eq!(d1.tvu_forwards, socketaddr!("127.0.0.1:1238"));
assert_eq!(d1.repair, socketaddr!("127.0.0.1:1239"));
assert_eq!(d1.serve_repair, socketaddr!("127.0.0.1:1240"));
assert_eq!(d1.tpu_vote, socketaddr!("127.0.0.1:1241"));
}
#[test]