Cuprate/cuprate
Cuprate/cuprate
1
0
Fork 0
mirror of https://github.com/Cuprate/cuprate.git synced 2024-11-16 15:58:17 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

PId -> PeerId

Browse source
This commit is contained in:
Boog900 2024-08-22 00:22:23 +01:00
parent ca8ac91f22
commit d6122f34dd
No known key found for this signature in database
GPG key ID: 42AB1287CB0041C2
3 changed files with 31 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
p2p/dandelion-tower/src/pool/incoming_tx.rs
View file

@ -2,13 +2,13 @@
use crate::{State, TxState};
/// An incoming transaction that has gone through the preprocessing stage.
pub struct IncomingTx<Tx, TxId, PId> {
pub struct IncomingTx<Tx, TxId, PeerId> {
/// The transaction.
pub(crate) tx: Tx,
/// The transaction ID.
pub(crate) tx_id: TxId,
/// The routing state of the transaction.
pub(crate) routing_state: TxState<PId>,
pub(crate) routing_state: TxState<PeerId>,
}
/// An [`IncomingTx`] builder.
@ -17,18 +17,18 @@ pub struct IncomingTx<Tx, TxId, PId> {
///
/// - `RS`: routing state; a `bool` for if the routing state is set
/// - `DBS`: database state; a `bool` for if the state in the DB is set
pub struct IncomingTxBuilder<const RS: bool, const DBS: bool, Tx, TxId, PId> {
pub struct IncomingTxBuilder<const RS: bool, const DBS: bool, Tx, TxId, PeerId> {
/// The transaction.
tx: Tx,
/// The transaction ID.
tx_id: TxId,
/// The routing state of the transaction.
routing_state: Option<TxState<PId>>,
routing_state: Option<TxState<PeerId>>,
/// The state of this transaction in the DB.
state_in_db: Option<State>,
}
impl<Tx, TxId, PId> IncomingTxBuilder<false, false, Tx, TxId, PId> {
impl<Tx, TxId, PeerId> IncomingTxBuilder<false, false, Tx, TxId, PeerId> {
/// Creates a new [`IncomingTxBuilder`].
pub fn new(tx: Tx, tx_id: TxId) -> Self {
Self {
@ -40,14 +40,14 @@ impl<Tx, TxId, PId> IncomingTxBuilder<false, false, Tx, TxId, PId> {
}
}
impl<const DBS: bool, Tx, TxId, PId> IncomingTxBuilder<false, DBS, Tx, TxId, PId> {
impl<const DBS: bool, Tx, TxId, PeerId> IncomingTxBuilder<false, DBS, Tx, TxId, PeerId> {
/// Adds the routing state to the builder.
///
/// The routing state is the origin of this transaction from our perspective.
pub fn with_routing_state(
self,
state: TxState<PId>,
) -> IncomingTxBuilder<true, DBS, Tx, TxId, PId> {
state: TxState<PeerId>,
) -> IncomingTxBuilder<true, DBS, Tx, TxId, PeerId> {
IncomingTxBuilder {
tx: self.tx,
tx_id: self.tx_id,
@ -57,14 +57,14 @@ impl<const DBS: bool, Tx, TxId, PId> IncomingTxBuilder<false, DBS, Tx, TxId, PId
}
}
impl<const RS: bool, Tx, TxId, PId> IncomingTxBuilder<RS, false, Tx, TxId, PId> {
impl<const RS: bool, Tx, TxId, PeerId> IncomingTxBuilder<RS, false, Tx, TxId, PeerId> {
/// Adds the database state to the builder.
///
/// If the transaction is not in the DB already then the state should be [`None`].
pub fn with_state_in_db(
self,
state: Option<State>,
) -> IncomingTxBuilder<RS, true, Tx, TxId, PId> {
) -> IncomingTxBuilder<RS, true, Tx, TxId, PeerId> {
IncomingTxBuilder {
tx: self.tx,
tx_id: self.tx_id,
@ -74,12 +74,12 @@ impl<const RS: bool, Tx, TxId, PId> IncomingTxBuilder<RS, false, Tx, TxId, PId>
}
}
impl<Tx, TxId, PId> IncomingTxBuilder<true, true, Tx, TxId, PId> {
impl<Tx, TxId, PeerId> IncomingTxBuilder<true, true, Tx, TxId, PeerId> {
/// Builds the [`IncomingTx`].
///
/// If this returns [`None`] then the transaction does not need to be given to the dandelion pool
/// manager.
pub fn build(self) -> Option<IncomingTx<Tx, TxId, PId>> {
pub fn build(self) -> Option<IncomingTx<Tx, TxId, PeerId>> {
let routing_state = self.routing_state.unwrap();
if self.state_in_db == Some(State::Fluff) {

18
p2p/dandelion-tower/src/pool/manager.rs
View file

@ -28,16 +28,16 @@ pub struct DandelionPoolShutDown;
/// The dandelion++ pool manager.
///
/// See the [module docs](super) for more.
pub struct DandelionPoolManager<P, R, Tx, TxId, PId> {
pub struct DandelionPoolManager<P, R, Tx, TxId, PeerId> {
/// The dandelion++ router
pub(crate) dandelion_router: R,
/// The backing tx storage.
pub(crate) backing_pool: P,
/// The set of tasks that are running the future returned from `dandelion_router`.
pub(crate) routing_set: JoinSet<(TxId, Result<State, TxState<PId>>)>,
pub(crate) routing_set: JoinSet<(TxId, Result<State, TxState<PeerId>>)>,
/// The origin of stem transactions.
pub(crate) stem_origins: HashMap<TxId, HashSet<PId>>,
pub(crate) stem_origins: HashMap<TxId, HashSet<PeerId>>,
/// Current stem pool embargo timers.
pub(crate) embargo_timers: DelayQueue<TxId>,
@ -50,14 +50,14 @@ pub struct DandelionPoolManager<P, R, Tx, TxId, PId> {
pub(crate) _tx: PhantomData<Tx>,
}
impl<P, R, Tx, TxId, PId> DandelionPoolManager<P, R, Tx, TxId, PId>
impl<P, R, Tx, TxId, PeerId> DandelionPoolManager<P, R, Tx, TxId, PeerId>
where
Tx: Clone + Send,
TxId: Hash + Eq + Clone + Send + 'static,
PId: Hash + Eq + Clone + Send + 'static,
PeerId: Hash + Eq + Clone + Send + 'static,
P: Service<TxStoreRequest<TxId>, Response = TxStoreResponse<Tx>, Error = tower::BoxError>,
P::Future: Send + 'static,
R: Service<DandelionRouteReq<Tx, PId>, Response = State, Error = DandelionRouterError>,
R: Service<DandelionRouteReq<Tx, PeerId>, Response = State, Error = DandelionRouterError>,
R::Future: Send + 'static,
{
/// Adds a new embargo timer to the running timers, with a duration pulled from [`Self::embargo_dist`]
@ -79,7 +79,7 @@ where
&mut self,
tx: Tx,
tx_id: TxId,
from: Option<PId>,
from: Option<PeerId>,
) -> Result<(), tower::BoxError> {
if let Some(peer) = &from {
self.stem_origins
@ -126,7 +126,7 @@ where
async fn handle_incoming_tx(
&mut self,
tx: Tx,
tx_state: TxState<PId>,
tx_state: TxState<PeerId>,
tx_id: TxId,
) -> Result<(), tower::BoxError> {
match tx_state {
@ -225,7 +225,7 @@ where
/// Starts the [`DandelionPoolManager`].
pub(crate) async fn run(
mut self,
mut rx: mpsc::Receiver<(IncomingTx<Tx, TxId, PId>, oneshot::Sender<()>)>,
mut rx: mpsc::Receiver<(IncomingTx<Tx, TxId, PeerId>, oneshot::Sender<()>)>,
) {
tracing::debug!("Starting dandelion++ tx-pool, config: {:?}", self.config);

19
p2p/dandelion-tower/src/pool/mod.rs
View file

@ -60,21 +60,21 @@ pub use manager::DandelionPoolManager;
/// user to customise routing functionality.
/// - `backing_pool` is the backing transaction storage service
/// - `config` is [`DandelionConfig`].
pub fn start_dandelion_pool_manager<P, R, Tx, TxId, PId>(
pub fn start_dandelion_pool_manager<P, R, Tx, TxId, PeerId>(
buffer_size: usize,
dandelion_router: R,
backing_pool: P,
config: DandelionConfig,
) -> DandelionPoolService<Tx, TxId, PId>
) -> DandelionPoolService<Tx, TxId, PeerId>
where
Tx: Clone + Send + 'static,
TxId: Hash + Eq + Clone + Send + 'static,
PId: Hash + Eq + Clone + Send + 'static,
PeerId: Hash + Eq + Clone + Send + 'static,
P: Service<TxStoreRequest<TxId>, Response = TxStoreResponse<Tx>, Error = tower::BoxError>
+ Send
+ 'static,
P::Future: Send + 'static,
R: Service<DandelionRouteReq<Tx, PId>, Response = State, Error = DandelionRouterError>
R: Service<DandelionRouteReq<Tx, PeerId>, Response = State, Error = DandelionRouterError>
+ Send
+ 'static,
R::Future: Send + 'static,
@ -105,16 +105,17 @@ where
///
/// Used to send [`IncomingTx`]s to the [`DandelionPoolManager`]
#[derive(Clone)]
pub struct DandelionPoolService<Tx, TxId, PId> {
pub struct DandelionPoolService<Tx, TxId, PeerId> {
/// The channel to [`DandelionPoolManager`].
tx: PollSender<(IncomingTx<Tx, TxId, PId>, oneshot::Sender<()>)>,
tx: PollSender<(IncomingTx<Tx, TxId, PeerId>, oneshot::Sender<()>)>,
}
impl<Tx, TxId, PId> Service<IncomingTx<Tx, TxId, PId>> for DandelionPoolService<Tx, TxId, PId>
impl<Tx, TxId, PeerId> Service<IncomingTx<Tx, TxId, PeerId>>
for DandelionPoolService<Tx, TxId, PeerId>
where
Tx: Clone + Send,
TxId: Hash + Eq + Clone + Send + 'static,
PId: Hash + Eq + Clone + Send + 'static,
PeerId: Hash + Eq + Clone + Send + 'static,
{
type Response = ();
type Error = DandelionPoolShutDown;
@ -124,7 +125,7 @@ where
self.tx.poll_reserve(cx).map_err(|_| DandelionPoolShutDown)
}
fn call(&mut self, req: IncomingTx<Tx, TxId, PId>) -> Self::Future {
fn call(&mut self, req: IncomingTx<Tx, TxId, PeerId>) -> Self::Future {
// although the channel isn't sending anything we want to wait for the request to be handled before continuing.
let (tx, rx) = oneshot::channel();