Remove unnecessary channels

`alice::swap::run_until` will be called once the execution setup is
done. The steps before are directly handled by the event loop,
hence no channels are needed for said steps: connection established,
swap request/response & execution setup.
This commit is contained in:
Franck Royer 2021-02-08 16:53:05 +11:00
parent f5ca5faabf
commit 1b2be804ed
No known key found for this signature in database
GPG Key ID: A82ED75A8DFC50A4
3 changed files with 22 additions and 137 deletions

View File

@ -4,7 +4,7 @@ use crate::{
protocol::{ protocol::{
alice::{ alice::{
behaviour::{Behaviour, OutEvent}, behaviour::{Behaviour, OutEvent},
State0, State3, SwapResponse, TransferProof, State3, SwapResponse, TransferProof,
}, },
bob::{EncryptedSignature, SwapRequest}, bob::{EncryptedSignature, SwapRequest},
}, },
@ -17,7 +17,7 @@ use tokio::{
sync::mpsc::{Receiver, Sender}, sync::mpsc::{Receiver, Sender},
time::timeout, time::timeout,
}; };
use tracing::{error, trace}; use tracing::{debug, error, trace};
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct Channels<T> { pub struct Channels<T> {
@ -34,36 +34,12 @@ impl<T> Default for Channels<T> {
#[derive(Debug)] #[derive(Debug)]
pub struct EventLoopHandle { pub struct EventLoopHandle {
done_execution_setup: Receiver<State3>,
recv_encrypted_signature: Receiver<EncryptedSignature>, recv_encrypted_signature: Receiver<EncryptedSignature>,
recv_swap_request: Receiver<(SwapRequest, ResponseChannel<SwapResponse>)>,
conn_established: Receiver<PeerId>,
send_swap_response: Sender<(ResponseChannel<SwapResponse>, SwapResponse)>,
start_execution_setup: Sender<(PeerId, State0)>,
send_transfer_proof: Sender<(PeerId, TransferProof)>, send_transfer_proof: Sender<(PeerId, TransferProof)>,
recv_transfer_proof_ack: Receiver<()>, recv_transfer_proof_ack: Receiver<()>,
} }
impl EventLoopHandle { impl EventLoopHandle {
pub async fn recv_conn_established(&mut self) -> Result<PeerId> {
self.conn_established
.recv()
.await
.ok_or_else(|| anyhow!("Failed to receive connection established from Bob"))
}
pub async fn execution_setup(&mut self, bob_peer_id: PeerId, state0: State0) -> Result<State3> {
let _ = self
.start_execution_setup
.send((bob_peer_id, state0))
.await?;
self.done_execution_setup
.recv()
.await
.ok_or_else(|| anyhow!("Failed to setup execution with Bob"))
}
pub async fn recv_encrypted_signature(&mut self) -> Result<EncryptedSignature> { pub async fn recv_encrypted_signature(&mut self) -> Result<EncryptedSignature> {
self.recv_encrypted_signature self.recv_encrypted_signature
.recv() .recv()
@ -71,27 +47,6 @@ impl EventLoopHandle {
.ok_or_else(|| anyhow!("Failed to receive Bitcoin encrypted signature from Bob")) .ok_or_else(|| anyhow!("Failed to receive Bitcoin encrypted signature from Bob"))
} }
pub async fn recv_swap_request(
&mut self,
) -> Result<(SwapRequest, ResponseChannel<SwapResponse>)> {
self.recv_swap_request
.recv()
.await
.ok_or_else(|| anyhow!("Failed to receive amounts request from Bob"))
}
pub async fn send_swap_response(
&mut self,
channel: ResponseChannel<SwapResponse>,
swap_response: SwapResponse,
) -> Result<()> {
let _ = self
.send_swap_response
.send((channel, swap_response))
.await?;
Ok(())
}
pub async fn send_transfer_proof( pub async fn send_transfer_proof(
&mut self, &mut self,
bob: PeerId, bob: PeerId,
@ -120,12 +75,7 @@ impl EventLoopHandle {
#[allow(missing_debug_implementations)] #[allow(missing_debug_implementations)]
pub struct EventLoop { pub struct EventLoop {
swarm: libp2p::Swarm<Behaviour>, swarm: libp2p::Swarm<Behaviour>,
start_execution_setup: Receiver<(PeerId, State0)>,
done_execution_setup: Sender<State3>,
recv_encrypted_signature: Sender<EncryptedSignature>, recv_encrypted_signature: Sender<EncryptedSignature>,
recv_swap_request: Sender<(SwapRequest, ResponseChannel<SwapResponse>)>,
conn_established: Sender<PeerId>,
send_swap_response: Receiver<(ResponseChannel<SwapResponse>, SwapResponse)>,
send_transfer_proof: Receiver<(PeerId, TransferProof)>, send_transfer_proof: Receiver<(PeerId, TransferProof)>,
recv_transfer_proof_ack: Sender<()>, recv_transfer_proof_ack: Sender<()>,
} }
@ -148,34 +98,19 @@ impl EventLoop {
Swarm::listen_on(&mut swarm, listen.clone()) Swarm::listen_on(&mut swarm, listen.clone())
.with_context(|| format!("Address is not supported: {:#}", listen))?; .with_context(|| format!("Address is not supported: {:#}", listen))?;
let start_execution_setup = Channels::default();
let done_execution_setup = Channels::default();
let recv_encrypted_signature = Channels::default(); let recv_encrypted_signature = Channels::default();
let request = Channels::default();
let conn_established = Channels::default();
let send_swap_response = Channels::default();
let send_transfer_proof = Channels::default(); let send_transfer_proof = Channels::default();
let recv_transfer_proof_ack = Channels::default(); let recv_transfer_proof_ack = Channels::default();
let driver = EventLoop { let driver = EventLoop {
swarm, swarm,
start_execution_setup: start_execution_setup.receiver,
done_execution_setup: done_execution_setup.sender,
recv_encrypted_signature: recv_encrypted_signature.sender, recv_encrypted_signature: recv_encrypted_signature.sender,
recv_swap_request: request.sender,
conn_established: conn_established.sender,
send_swap_response: send_swap_response.receiver,
send_transfer_proof: send_transfer_proof.receiver, send_transfer_proof: send_transfer_proof.receiver,
recv_transfer_proof_ack: recv_transfer_proof_ack.sender, recv_transfer_proof_ack: recv_transfer_proof_ack.sender,
}; };
let handle = EventLoopHandle { let handle = EventLoopHandle {
start_execution_setup: start_execution_setup.sender,
done_execution_setup: done_execution_setup.receiver,
recv_encrypted_signature: recv_encrypted_signature.receiver, recv_encrypted_signature: recv_encrypted_signature.receiver,
recv_swap_request: request.receiver,
conn_established: conn_established.receiver,
send_swap_response: send_swap_response.sender,
send_transfer_proof: send_transfer_proof.sender, send_transfer_proof: send_transfer_proof.sender,
recv_transfer_proof_ack: recv_transfer_proof_ack.receiver, recv_transfer_proof_ack: recv_transfer_proof_ack.receiver,
}; };
@ -189,13 +124,13 @@ impl EventLoop {
swarm_event = self.swarm.next().fuse() => { swarm_event = self.swarm.next().fuse() => {
match swarm_event { match swarm_event {
OutEvent::ConnectionEstablished(alice) => { OutEvent::ConnectionEstablished(alice) => {
let _ = self.conn_established.send(alice).await; debug!("Connection Established with {}", alice);
} }
OutEvent::SwapRequest { msg, channel } => { OutEvent::SwapRequest { msg, channel } => {
let _ = self.recv_swap_request.send((msg, channel)).await; let _ = self.handle_swap_request(msg, channel).await;
} }
OutEvent::ExecutionSetupDone(state3) => { OutEvent::ExecutionSetupDone(state3) => {
let _ = self.done_execution_setup.send(*state3).await; let _ = self.handle_execution_setup_done(*state3).await;
} }
OutEvent::TransferProofAcknowledged => { OutEvent::TransferProofAcknowledged => {
trace!("Bob acknowledged transfer proof"); trace!("Bob acknowledged transfer proof");
@ -214,21 +149,6 @@ impl EventLoop {
} }
} }
}, },
swap_response = self.send_swap_response.recv().fuse() => {
if let Some((channel, swap_response)) = swap_response {
let _ = self
.swarm
.send_swap_response(channel, swap_response)
.map_err(|err|error!("Failed to send swap response: {:#}", err));
}
},
option = self.start_execution_setup.recv().fuse() => {
if let Some((bob_peer_id, state0)) = option {
let _ = self
.swarm
.start_execution_setup(bob_peer_id, state0);
}
},
transfer_proof = self.send_transfer_proof.recv().fuse() => { transfer_proof = self.send_transfer_proof.recv().fuse() => {
if let Some((bob_peer_id, msg)) = transfer_proof { if let Some((bob_peer_id, msg)) = transfer_proof {
self.swarm.send_transfer_proof(bob_peer_id, msg); self.swarm.send_transfer_proof(bob_peer_id, msg);
@ -237,4 +157,13 @@ impl EventLoop {
} }
} }
} }
async fn handle_swap_request(
&self,
_msg: SwapRequest,
_channel: ResponseChannel<SwapResponse>,
) {
}
async fn handle_execution_setup_done(&self, _state3: State3) {}
} }

View File

@ -12,7 +12,7 @@ use crate::{
monero::Transfer, monero::Transfer,
protocol::{ protocol::{
alice, alice,
alice::{event_loop::EventLoopHandle, SwapResponse, TransferProof}, alice::{event_loop::EventLoopHandle, TransferProof},
SwapAmounts, SwapAmounts,
}, },
}; };
@ -26,43 +26,7 @@ use libp2p::PeerId;
use sha2::Sha256; use sha2::Sha256;
use std::sync::Arc; use std::sync::Arc;
use tokio::time::timeout; use tokio::time::timeout;
use tracing::{info, trace}; use tracing::info;
pub async fn negotiate(
state0: alice::State0,
xmr_amount: monero::Amount,
event_loop_handle: &mut EventLoopHandle,
execution_params: ExecutionParams,
) -> Result<(PeerId, alice::State3)> {
trace!("Starting negotiate");
// todo: we can move this out, we dont need to timeout here
let bob_peer_id = timeout(
execution_params.bob_time_to_act,
event_loop_handle.recv_conn_established(),
)
.await
.context("Failed to receive dial connection from Bob")??;
let event = timeout(
execution_params.bob_time_to_act,
event_loop_handle.recv_swap_request(),
)
.await
.context("Failed to receive swap request from Bob")??;
event_loop_handle
.send_swap_response(event.1, SwapResponse { xmr_amount })
.await?;
let state3 = timeout(
execution_params.bob_time_to_act,
event_loop_handle.execution_setup(bob_peer_id, state0),
)
.await??;
Ok((bob_peer_id, state3))
}
// TODO(Franck): Use helper functions from xmr-btc instead of re-writing them // TODO(Franck): Use helper functions from xmr-btc instead of re-writing them
// here // here

View File

@ -17,10 +17,10 @@ use crate::{
event_loop::EventLoopHandle, event_loop::EventLoopHandle,
steps::{ steps::{
build_bitcoin_punish_transaction, build_bitcoin_redeem_transaction, build_bitcoin_punish_transaction, build_bitcoin_redeem_transaction,
extract_monero_private_key, lock_xmr, negotiate, extract_monero_private_key, lock_xmr, publish_bitcoin_punish_transaction,
publish_bitcoin_punish_transaction, publish_bitcoin_redeem_transaction, publish_bitcoin_redeem_transaction, publish_cancel_transaction,
publish_cancel_transaction, wait_for_bitcoin_encrypted_signature, wait_for_bitcoin_encrypted_signature, wait_for_bitcoin_refund,
wait_for_bitcoin_refund, wait_for_locked_bitcoin, wait_for_locked_bitcoin,
}, },
AliceState, AliceState,
}, },
@ -92,18 +92,10 @@ async fn run_until_internal(
} else { } else {
match state { match state {
AliceState::Started { amounts, state0 } => { AliceState::Started { amounts, state0 } => {
let (bob_peer_id, state3) = negotiate(
state0,
amounts.xmr,
&mut event_loop_handle,
execution_params,
)
.await?;
let state = AliceState::Negotiated { let state = AliceState::Negotiated {
bob_peer_id, bob_peer_id: todo!(),
amounts, amounts,
state3: Box::new(state3), state3: todo!(),
}; };
let db_state = (&state).into(); let db_state = (&state).into();