WIP mpp

Author: naumenkogs

lightning/src/routing/router.rs

@@ -22,7 +22,7 @@ use util::ser::{Writeable, Readable};
 use util::logger::Logger;
 
 use std::cmp;
-use std::collections::{HashMap,BinaryHeap};
+use std::collections::{HashMap, BinaryHeap, HashSet};
 use std::ops::Deref;
 use std::hash::{Hash, Hasher};
 
@@ -138,7 +138,7 @@ pub struct RouteHint {
 	pub htlc_maximum_msat: Option<u64>,
 }
 
-#[derive(Eq, PartialEq)]
+#[derive(Eq, PartialEq, Clone)]
 struct RouteGraphNode {
 	pubkey: PublicKey,
 	lowest_fee_to_peer_through_node: u64,
@@ -212,6 +212,7 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 
 	let mut targets = BinaryHeap::new(); //TODO: Do we care about switching to eg Fibbonaci heap?
 	let mut dist = HashMap::with_capacity(network.get_nodes().len());
+	let recommended_available_msat = final_value_msat * 4;
 
 	let mut first_hop_targets = HashMap::with_capacity(if first_hops.is_some() { first_hops.as_ref().unwrap().len() } else { 0 });
 	if let Some(hops) = first_hops {
@@ -236,6 +237,9 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 		}
 	}
 
+	// Track the use of channels to reduce their available capacity in the next MPP rounds.
+	let mut used_channels_with_amounts_msat = HashMap::new();
+
 	macro_rules! add_entry {
 		// Adds entry which goes from $src_node_id to $dest_node_id
 		// over the channel with id $chan_id with fees described in
@@ -249,12 +253,19 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 				{
 					let mut total_fee = $starting_fee_msat as u64;
 
-					let mut available_msat = $capacity_sats;
+					let mut available_capacity_msat = $capacity_sats;
 					if let Some(htlc_maximum_msat) = $directional_info.htlc_maximum_msat {
 						if let Some(capacity_sats) = $capacity_sats {
-							available_msat = Some(cmp::min(capacity_sats * 1000, htlc_maximum_msat));
+							available_capacity_msat = Some(cmp::min(capacity_sats * 1000, htlc_maximum_msat));
 						} else {
-							available_msat = Some(htlc_maximum_msat);
+							available_capacity_msat = Some(htlc_maximum_msat);
+						}
+					}
+
+					if let Some(max_available_capacity_msat) = available_capacity_msat {
+					   if let Some(used_amount_msat) = used_channels_with_amounts_msat.get(&$chan_id.clone()) {
+							// Take into account amounts used in previous mpp iterations.
+							available_capacity_msat = Some(max_available_capacity_msat - used_amount_msat)
 						}
 					}
 
@@ -279,6 +290,7 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 								cltv_expiry_delta: 0,
 							},
 							None,
+							$directional_info.fees,
 						)
 					});
 					if $src_node_id != *our_node_id {
@@ -308,13 +320,17 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 							fee_msat: new_fee, // This field is ignored on the last-hop anyway
 							cltv_expiry_delta: $directional_info.cltv_expiry_delta as u32,
 						};
-						old_entry.4 = available_msat;
+						old_entry.4 = available_capacity_msat;
+						old_entry.5 = $directional_info.fees;
 					}
 				}
 			}
 		};
 	}
 
+	// For MPP, just retry the same dest-to-source A*, but avoid some channels with lowest capacities.
+	let mut channels_to_avoid = HashSet::new();
+
 	macro_rules! add_entries_to_cheapest_to_target_node {
 		( $node: expr, $node_id: expr, $fee_to_target_msat: expr ) => {
 			if first_hops.is_some() {
@@ -332,6 +348,9 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 
 			if !features.requires_unknown_bits() {
 				for chan_id in $node.channels.iter() {
+					if channels_to_avoid.contains(chan_id) {
+						continue;
+					}
 					let chan = network.get_channels().get(chan_id).unwrap();
 					if !chan.features.requires_unknown_bits() {
 						if chan.node_one == *$node_id {
@@ -385,53 +404,160 @@ pub fn get_route<L: Deref>(our_node_id: &PublicKey, network: &NetworkGraph, targ
 		}
 	}
 
-	while let Some(RouteGraphNode { pubkey, lowest_fee_to_node, .. }) = targets.pop() {
-		if pubkey == *our_node_id {
-			let mut res = vec!(dist.remove(&our_node_id).unwrap().3);
-			loop {
-				if let Some(&(_, ref features)) = first_hop_targets.get(&res.last().unwrap().pubkey) {
-					res.last_mut().unwrap().node_features = features.to_context();
-				} else if let Some(node) = network.get_nodes().get(&res.last().unwrap().pubkey) {
-					if let Some(node_info) = node.announcement_info.as_ref() {
-						res.last_mut().unwrap().node_features = node_info.features.clone();
+	let mut payment_paths = Vec::new();
+	let mut tries_left = 10;
+	let mut found_amount_msat = 0;
+	// For every MPP start with the same pre-filled data.
+	let prefilled_targets = targets.clone();
+	let prefilled_dist = dist.clone();
+
+	'route_finding: loop {
+		targets = prefilled_targets.clone();
+		dist = prefilled_dist.clone();
+		'path_finding: while let Some(RouteGraphNode { pubkey, lowest_fee_to_node, .. }) = targets.pop() {
+			if pubkey == *our_node_id {
+				// TODO: currently pretending that always have blockchain/amount data for channel capacities.
+				// Should handle None better.
+				let mut new_entry = dist.remove(&our_node_id).unwrap();
+				let mut res = vec!(new_entry.3.clone());
+				let mut path_bottleneck_msat = None;
+				let mut cur_path_fees = Vec::new();
+				let mut smallest_channel_id = None;
+				loop {
+					if let Some(&(_, ref features)) = first_hop_targets.get(&res.last().unwrap().pubkey) {
+						res.last_mut().unwrap().node_features = features.to_context();
+					} else if let Some(node) = network.get_nodes().get(&res.last().unwrap().pubkey) {
+						if let Some(node_info) = node.announcement_info.as_ref() {
+							res.last_mut().unwrap().node_features = node_info.features.clone();
+						} else {
+							res.last_mut().unwrap().node_features = NodeFeatures::empty();
+						}
 					} else {
-						res.last_mut().unwrap().node_features = NodeFeatures::empty();
+						// We should be able to fill in features for everything except the last
+						// hop, if the last hop was provided via a BOLT 11 invoice (though we
+						// should be able to extend it further as BOLT 11 does have feature
+						// flags for the last hop node itself).
+						assert!(res.last().unwrap().pubkey == *target);
+					}
+
+
+					if let Some(current_hop_bottleneck_msat) = new_entry.4 {
+						if let Some(full_path_bottleneck_msat) = path_bottleneck_msat {
+							if current_hop_bottleneck_msat < full_path_bottleneck_msat {
+								path_bottleneck_msat = Some(current_hop_bottleneck_msat);
+								// Store smallest channel per path to avoid hitting it while looking
+								// for another path (per MPP).
+								smallest_channel_id = Some(res.last().unwrap().short_channel_id);
+							}
+						}
+					}
+
+					if let Some(bottleneck_amount_msat) = path_bottleneck_msat {
+						found_amount_msat += bottleneck_amount_msat;
 					}
-				} else {
-					// We should be able to fill in features for everything except the last
-					// hop, if the last hop was provided via a BOLT 11 invoice (though we
-					// should be able to extend it further as BOLT 11 does have feature
-					// flags for the last hop node itself).
-					assert!(res.last().unwrap().pubkey == *target);
+
+					if res.last().unwrap().pubkey == *target {
+						break;
+					}
+
+					new_entry = match dist.remove(&res.last().unwrap().pubkey) {
+						Some(hop) => hop,
+						None => return Err(LightningError{err: "Failed to find a non-fee-overflowing path to the given destination".to_owned(), action: ErrorAction::IgnoreError}),
+					};
+					res.last_mut().unwrap().fee_msat = new_entry.3.fee_msat;
+					res.last_mut().unwrap().cltv_expiry_delta = new_entry.3.cltv_expiry_delta;
+					res.push(new_entry.3);
+					cur_path_fees.push(new_entry.5);
 				}
-				if res.last().unwrap().pubkey == *target {
-					break;
+				res.last_mut().unwrap().fee_msat = final_value_msat;
+				res.last_mut().unwrap().cltv_expiry_delta = final_cltv;
+
+				// Recompute fee considering the bottleneck.
+				// Also remember used amounts per channels to avoid relying on them in the next MPP iterations.
+				let mut recomputed_fee_msat = 0;
+				let mut total_proportional_fee_millionths = 0;
+				let mut i = 0;
+				for fees in cur_path_fees.clone() {
+					let current_hop_amount_msat = path_bottleneck_msat.unwrap().checked_add(recomputed_fee_msat);
+					let proportional_fee_millions = current_hop_amount_msat.unwrap().checked_mul(fees.proportional_millionths as u64);
+					let mut hop_to_update = res.get_mut(i).unwrap();
+					if let Some(new_fee) = proportional_fee_millions.and_then(|part| {
+							(fees.base_msat as u64).checked_add(part / 1000000) }) {
+
+						recomputed_fee_msat += new_fee;
+						total_proportional_fee_millionths += fees.proportional_millionths;
+						hop_to_update.fee_msat = new_fee;
+					}
+
+					*used_channels_with_amounts_msat.entry(hop_to_update.short_channel_id).or_insert(0) += path_bottleneck_msat.unwrap();
+					i += 1;
 				}
 
-				let new_entry = match dist.remove(&res.last().unwrap().pubkey) {
-					Some(hop) => hop.3,
-					None => return Err(LightningError{err: "Failed to find a non-fee-overflowing path to the given destination".to_owned(), action: ErrorAction::IgnoreError}),
-				};
-				res.last_mut().unwrap().fee_msat = new_entry.fee_msat;
-				res.last_mut().unwrap().cltv_expiry_delta = new_entry.cltv_expiry_delta;
-				res.push(new_entry);
+				payment_paths.push((res, path_bottleneck_msat, total_proportional_fee_millionths, recomputed_fee_msat));
+				if let Some(chan_id_to_avoid) = smallest_channel_id {
+					channels_to_avoid.insert(chan_id_to_avoid);
+				}
+				break 'path_finding;
+			}
+
+			match network.get_nodes().get(&pubkey) {
+				None => {},
+				Some(node) => {
+					add_entries_to_cheapest_to_target_node!(node, &pubkey, lowest_fee_to_node);
+				},
 			}
-			res.last_mut().unwrap().fee_msat = final_value_msat;
-			res.last_mut().unwrap().cltv_expiry_delta = final_cltv;
-			let route = Route { paths: vec![res] };
-			log_trace!(logger, "Got route: {}", log_route!(route));
-			return Ok(route);
 		}
+		tries_left -= 1;
 
-		match network.get_nodes().get(&pubkey) {
-			None => {},
-			Some(node) => {
-				add_entries_to_cheapest_to_target_node!(node, &pubkey, lowest_fee_to_node);
-			},
+		if tries_left == 0 || found_amount_msat >= recommended_available_msat {
+			break 'route_finding;
 		}
 	}
 
-	Err(LightningError{err: "Failed to find a path to the given destination".to_owned(), action: ErrorAction::IgnoreError})
+	if payment_paths.len() == 0 {
+		return Err(LightningError{err: "Failed to find a path to the given destination".to_owned(), action: ErrorAction::IgnoreError});
+	}
+
+	// Sort by total fees and take the best paths.
+	payment_paths.sort_by_key(|k| k.3);
+	if payment_paths.len() > 50 {
+		payment_paths.resize(50, (Vec::new(), None, 0, 0));
+	}
+
+	let mut drawn_routes_with_fees = Vec::new();
+	for i in 0..10 {
+		let mut cur_route_with_fees = Vec::new();
+		let mut aggregate_amount_msat = 0;
+		let cur_payment_paths = &payment_paths[i..i+10]; // TODO: real random sampling to combine paths into random routes
+		let mut aggregate_fee_msat = 0;
+
+
+		for (path, path_bottleneck_msat, total_proportional_fee_millionths, fee_msat) in cur_payment_paths {
+			cur_route_with_fees.push((path, path_bottleneck_msat, total_proportional_fee_millionths, fee_msat));
+			aggregate_amount_msat += path_bottleneck_msat.unwrap();
+			aggregate_fee_msat += fee_msat;
+			if aggregate_amount_msat > final_value_msat {
+				break;
+			}
+		}
+		cur_route_with_fees.sort_by_key(|k| k.2); // sort by total proportional fee.
+		drawn_routes_with_fees.push((cur_route_with_fees, aggregate_fee_msat))
+	}
+
+
+	// TODO reduce amount in the last path (sorted by proportional fee) to not overshoot and adjust fees.
+
+	// Select the best route by total fee.
+	drawn_routes_with_fees.sort_by_key(|k| k.1);
+	let mut selected_paths = Vec::<Vec::<RouteHop>>::new();
+	for (path, amount, _, _) in &drawn_routes_with_fees.first().unwrap().0 {
+		selected_paths.push(path.to_vec());
+	}
+
+	// TODO add amount per path so that the caller knows how to distribute the amount being sent among MPPs.
+	let route = Route { paths: selected_paths };
+	log_trace!(logger, "Got route: {}", log_route!(route));
+	return Ok(route);
 }
 
 #[cfg(test)]