liz/veilid
liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

fix wasm and finish refactor

Browse Source
This commit is contained in:
John Smith 2022-01-04 14:25:32 -05:00
parent 2564d35cc1
commit 3035bc079f
16 changed files with 133 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
Cargo.lock generated
View File

@ -225,12 +225,6 @@ dependencies = [
"winapi",
]
[[package]]
name = "async-lock"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ab86ee898bb6d5d0118270acaa8536c59885231ca6f653b4c35a148f8ee6235"
[[package]]
name = "async-lock"
version = "2.4.0"
@ -276,7 +270,7 @@ dependencies = [
"async-channel",
"async-global-executor",
"async-io",
"async-lock 2.4.0",
"async-lock",
"async-process",
"crossbeam-utils",
"futures-channel",
@ -3749,7 +3743,7 @@ version = "0.1.0"
dependencies = [
"android_logger",
"anyhow",
"async-lock 0.1.0",
"async-lock",
"async-std",
"async-tls",
"async-tungstenite 0.16.0",

2
veilid-cli/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "veilid-cli"
version = "0.1.0"
authors = ["John Smith <jsmith@example.com>"]
edition = "2018"
edition = "2021"
build = "build.rs"
license = "LGPL-2.0-or-later OR MPL-2.0 OR (MIT AND BSD-3-Clause)"

8
veilid-core/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "veilid-core"
version = "0.1.0"
authors = ["John Smith <nobody@example.com>"]
edition = "2018"
edition = "2021"
build = "build.rs"
license = "LGPL-2.0-or-later OR MPL-2.0 OR (MIT AND BSD-3-Clause)"
@ -62,7 +62,7 @@ keyvaluedb-sqlite = { path = "../external/keyvaluedb/keyvaluedb-sqlite" }
data-encoding = { version = "^2" }
serde = { version = "^1", features = ["derive" ] }
serde_cbor = { version = "^0" }
async_executors = { version = "^0", features = [ "async_std" ]}
async_executors = { version = "^0", default-features = false, features = [ "async_std" ]}
socket2 = "^0"
bugsalot = "^0"
chrono = "^0"
@ -85,8 +85,8 @@ serde = { version = "^1", default-features = false, features = ["derive", "alloc
serde_cbor = { version = "^0", default-features = false, features = ["alloc"] }
getrandom = { version = "^0", features = ["js"] }
ws_stream_wasm = "^0"
async_executors = { version = "^0", features = [ "bindgen" ]}
async-lock = "^0"
async_executors = { version = "^0", default-features = false, features = [ "bindgen" ]}
async-lock = "^2"
# Configuration for WASM32 'web-sys' crate
[target.'cfg(target_arch = "wasm32")'.dependencies.web-sys]

17
veilid-core/src/connection_manager.rs
View File

@ -83,7 +83,7 @@ impl ConnectionManager {
// Returns a network connection if one already is established
pub async fn get_connection(
&self,
descriptor: &ConnectionDescriptor,
descriptor: ConnectionDescriptor,
) -> Option<NetworkConnection> {
let inner = self.arc.inner.lock().await;
inner.connection_table.get_connection(descriptor)
@ -92,7 +92,7 @@ impl ConnectionManager {
// Internal routine to register new connection atomically
async fn on_new_connection_internal(
&self,
mut inner: AsyncMutexGuard<'_, ConnectionManagerInner>,
inner: &mut ConnectionManagerInner,
conn: NetworkConnection,
) -> Result<(), String> {
let tx = inner
@ -116,8 +116,8 @@ impl ConnectionManager {
// either from incoming or outgoing connections. Registers connection in the connection table for later access
// and spawns a message processing loop for the connection
pub async fn on_new_connection(&self, conn: NetworkConnection) -> Result<(), String> {
let inner = self.arc.inner.lock().await;
self.on_new_connection_internal(inner, conn).await
let mut inner = self.arc.inner.lock().await;
self.on_new_connection_internal(&mut *inner, conn).await
}
pub async fn get_or_create_connection(
@ -134,16 +134,17 @@ impl ConnectionManager {
};
// If connection exists, then return it
let inner = self.arc.inner.lock().await;
let mut inner = self.arc.inner.lock().await;
if let Some(conn) = inner.connection_table.get_connection(&descriptor) {
if let Some(conn) = inner.connection_table.get_connection(descriptor) {
return Ok(conn);
}
// If not, attempt new connection
let conn = NetworkConnection::connect(local_addr, dial_info).await?;
self.on_new_connection_internal(inner, conn.clone()).await?;
self.on_new_connection_internal(&mut *inner, conn.clone())
.await?;
Ok(conn)
}
@ -178,7 +179,7 @@ impl ConnectionManager {
.lock()
.await
.connection_table
.remove_connection(&descriptor)
.remove_connection(descriptor)
{
log_net!(error e);
}

8
veilid-core/src/connection_table.rs
View File

@ -32,8 +32,8 @@ impl ConnectionTable {
Ok(())
}
pub fn get_connection(&self, descriptor: &ConnectionDescriptor) -> Option<NetworkConnection> {
self.conn_by_addr.get(descriptor).cloned()
pub fn get_connection(&self, descriptor: ConnectionDescriptor) -> Option<NetworkConnection> {
self.conn_by_addr.get(&descriptor).cloned()
}
pub fn connection_count(&self) -> usize {
@ -42,10 +42,10 @@ impl ConnectionTable {
pub fn remove_connection(
&mut self,
descriptor: &ConnectionDescriptor,
descriptor: ConnectionDescriptor,
) -> Result<NetworkConnection, String> {
self.conn_by_addr
.remove(descriptor)
.remove(&descriptor)
.ok_or_else(|| format!("Connection not in table: {:?}", descriptor))
}
}

30
veilid-core/src/intf/native/network/mod.rs
View File

@ -255,32 +255,33 @@ impl Network {
// This bypasses the connection table as it is not a 'node to node' connection.
pub async fn send_data_unbound_to_dial_info(
&self,
dial_info: &DialInfo,
dial_info: DialInfo,
data: Vec<u8>,
) -> Result<(), String> {
match &dial_info {
DialInfo::UDP(_) => {
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawUdpProtocolHandler::send_unbound_message(peer_socket_addr, data)
.await
.map_err(logthru_net!())
}
DialInfo::TCP(_) => {
ProtocolType::TCP => {
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::send_unbound_message(peer_socket_addr, data)
.await
.map_err(logthru_net!())
}
DialInfo::WS(_) => Err("WS protocol does not support unbound messages".to_owned())
.map_err(logthru_net!(error)),
DialInfo::WSS(_) => Err("WSS protocol does not support unbound messages".to_owned())
.map_err(logthru_net!(error)),
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::send_unbound_message(dial_info, data)
.await
.map_err(logthru_net!())
}
}
}
async fn send_data_to_existing_connection(
&self,
descriptor: &ConnectionDescriptor,
descriptor: ConnectionDescriptor,
data: Vec<u8>,
) -> Result<Option<Vec<u8>>, String> {
// Handle connectionless protocol
@ -317,7 +318,7 @@ impl Network {
// Send data directly to a dial info, possibly without knowing which node it is going to
pub async fn send_data_to_dial_info(
&self,
dial_info: &DialInfo,
dial_info: DialInfo,
data: Vec<u8>,
) -> Result<(), String> {
// Handle connectionless protocol
@ -334,10 +335,10 @@ impl Network {
}
// Handle connection-oriented protocols
let local_addr = self.get_preferred_local_address(dial_info);
let local_addr = self.get_preferred_local_address(&dial_info);
let conn = self
.connection_manager()
.get_or_create_connection(Some(local_addr), dial_info.clone())
.get_or_create_connection(Some(local_addr), dial_info)
.await?;
conn.send(data).await.map_err(logthru_net!(error))
@ -353,7 +354,7 @@ impl Network {
let data = if let Some(descriptor) = node_ref.last_connection() {
match self
.clone()
.send_data_to_existing_connection(&descriptor, data)
.send_data_to_existing_connection(descriptor, data)
.await?
{
None => {
@ -370,8 +371,7 @@ impl Network {
.best_dial_info()
.ok_or_else(|| "couldn't send data, no dial info or peer address".to_owned())?;
// Handle connectionless protocol
self.send_data_to_dial_info(&dial_info, data).await
self.send_data_to_dial_info(dial_info, data).await
}
/////////////////////////////////////////////////////////////////

2
veilid-core/src/intf/native/network/protocol/mod.rs
View File

@ -36,7 +36,7 @@ impl ProtocolNetworkConnection {
}
}
pub async fn send_unbound_message(dial_info: &DialInfo, data: Vec<u8>) -> Result<(), String> {
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> Result<(), String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
let peer_socket_addr = dial_info.to_socket_addr();

2
veilid-core/src/intf/native/network/protocol/ws.rs
View File

@ -290,7 +290,7 @@ impl WebsocketProtocolHandler {
}
}
pub async fn send_unbound_message(dial_info: &DialInfo, data: Vec<u8>) -> Result<(), String> {
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> Result<(), String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}

108
veilid-core/src/intf/wasm/network/mod.rs
View File

@ -1,8 +1,9 @@
mod protocol;
use crate::intf::*;
use crate::connection_manager::*;
use crate::network_manager::*;
use crate::routing_table::*;
use crate::intf::*;
use crate::*;
use protocol::ws::WebsocketProtocolHandler;
pub use protocol::*;
@ -42,11 +43,15 @@ impl Network {
}
}
fn connection_manager(&self) -> ConnectionManager {
self.inner.lock().network_manager.connection_manager()
}
/////////////////////////////////////////////////////////////////
async fn send_data_to_existing_connection(
&self,
descriptor: &ConnectionDescriptor,
descriptor: ConnectionDescriptor,
data: Vec<u8>,
) -> Result<Option<Vec<u8>>, String> {
match descriptor.protocol_type() {
@ -56,79 +61,70 @@ impl Network {
ProtocolType::TCP => {
return Err("no support for tcp protocol".to_owned()).map_err(logthru_net!(error))
}
ProtocolType::WS | ProtocolType::WSS => {
// find an existing connection in the connection table if one exists
let network_manager = self.inner.lock().network_manager.clone();
if let Some(entry) = network_manager
.connection_table()
.get_connection(&descriptor)
{
// connection exists, send over it
entry.conn.send(data).await.map_err(logthru_net!())?;
// Data was consumed
return Ok(None);
}
}
_ => {}
}
// Handle connection-oriented protocols
if let Some(conn) = self.connection_manager().get_connection(descriptor).await {
// connection exists, send over it
conn.send(data).await.map_err(logthru_net!())?;
// Data was consumed
Ok(None)
} else {
// Connection or didn't exist
// Pass the data back out so we don't own it any more
Ok(Some(data))
}
// connection or local socket didn't exist, we'll need to use dialinfo to create one
// Pass the data back out so we don't own it any more
Ok(Some(data))
}
pub async fn send_data_unbound_to_dial_info(
&self,
dial_info: &DialInfo,
dial_info: DialInfo,
data: Vec<u8>,
) -> Result<(), String> {
let network_manager = self.inner.lock().network_manager.clone();
match &dial_info {
DialInfo::UDP(_) => {
match dial_info.protocol_type() {
ProtocolType::UDP => {
return Err("no support for UDP protocol".to_owned()).map_err(logthru_net!(error))
}
DialInfo::TCP(_) => {
ProtocolType::TCP => {
return Err("no support for TCP protocol".to_owned()).map_err(logthru_net!(error))
}
DialInfo::WS(_) => Err("WS protocol does not support unbound messages".to_owned())
.map_err(logthru_net!(error)),
DialInfo::WSS(_) => Err("WSS protocol does not support unbound messages".to_owned())
.map_err(logthru_net!(error)),
ProtocolType::WS | ProtocolType::WSS => {
WebsocketProtocolHandler::send_unbound_message(dial_info, data)
.await
.map_err(logthru_net!())
}
}
}
pub async fn send_data_to_dial_info(
&self,
dial_info: &DialInfo,
dial_info: DialInfo,
data: Vec<u8>,
) -> Result<(), String> {
let network_manager = self.inner.lock().network_manager.clone();
if dial_info.protocol_type() == ProtocolType::UDP {
return Err("no support for UDP protocol".to_owned()).map_err(logthru_net!(error))
}
if dial_info.protocol_type() == ProtocolType::TCP {
return Err("no support for TCP protocol".to_owned()).map_err(logthru_net!(error))
}
let conn = match &dial_info {
DialInfo::UDP(_) => {
return Err("no support for UDP protocol".to_owned()).map_err(logthru_net!(error))
}
DialInfo::TCP(_) => {
return Err("no support for TCP protocol".to_owned()).map_err(logthru_net!(error))
}
DialInfo::WS(_) => WebsocketProtocolHandler::connect(network_manager, dial_info)
.await
.map_err(logthru_net!(error))?,
DialInfo::WSS(_) => WebsocketProtocolHandler::connect(network_manager, dial_info)
.await
.map_err(logthru_net!(error))?,
};
// Handle connection-oriented protocols
let conn = self
.connection_manager()
.get_or_create_connection(None, dial_info)
.await?;
conn.send(data).await.map_err(logthru_net!(error))
}
pub async fn send_data(&self, node_ref: NodeRef, data: Vec<u8>) -> Result<(), String> {
let dial_info = node_ref.best_dial_info();
let descriptor = node_ref.last_connection();
// First try to send data to the last socket we've seen this peer on
let di_data = if let Some(descriptor) = descriptor {
let data = if let Some(descriptor) = node_ref.last_connection() {
match self
.clone()
.send_data_to_existing_connection(&descriptor, data)
.send_data_to_existing_connection(descriptor, data)
.await?
{
None => {
@ -141,15 +137,11 @@ impl Network {
};
// If that fails, try to make a connection or reach out to the peer via its dial info
if let Some(di) = dial_info {
self.clone()
.send_data_to_dial_info(&di, di_data)
.await
.map_err(logthru_net!(error))
} else {
Err("couldn't send data, no dial info or peer address".to_owned())
.map_err(logthru_net!(error))
}
let dial_info = node_ref
.best_dial_info()
.ok_or_else(|| "couldn't send data, no dial info or peer address".to_owned())?;
self.send_data_to_dial_info(dial_info, data).await
}
/////////////////////////////////////////////////////////////////

6
veilid-core/src/intf/wasm/network/protocol/mod.rs
View File

@ -1,9 +1,9 @@
pub mod wrtc;
pub mod ws;
use crate::connection_manager::*;
use crate::veilid_api::ProtocolType;
use crate::network_connection::*;
use crate::xx::*;
use crate::*;
#[derive(Debug)]
pub enum ProtocolNetworkConnection {
@ -31,7 +31,7 @@ impl ProtocolNetworkConnection {
}
pub async fn send_unbound_message(
dial_info: &DialInfo,
dial_info: DialInfo,
data: Vec<u8>,
) -> Result<(), String> {
match dial_info.protocol_type() {

61
veilid-core/src/intf/wasm/network/protocol/ws.rs
View File

@ -1,10 +1,10 @@
use crate::intf::*;
use crate::network_manager::{NetworkManager, MAX_MESSAGE_SIZE};
use crate::network_connection::*;
use crate::network_manager::MAX_MESSAGE_SIZE;
use crate::*;
use alloc::fmt;
use futures_util::stream::StreamExt;
use web_sys::WebSocket;
use ws_stream_wasm::*;
use futures_util::{StreamExt, SinkExt};
struct WebsocketNetworkConnectionInner {
ws_meta: WsMeta,
@ -27,7 +27,7 @@ impl WebsocketNetworkConnection {
pub fn new(tls: bool, ws_meta: WsMeta, ws_stream: WsStream) -> Self {
Self {
tls,
inner: Arc::new(Mutex::new(WebsocketNetworkConnectionInner {
inner: Arc::new(AsyncMutex::new(WebsocketNetworkConnectionInner {
ws_meta,
ws_stream,
})),
@ -36,7 +36,7 @@ impl WebsocketNetworkConnection {
pub async fn close(&self) -> Result<(), String> {
let inner = self.inner.lock().await;
inner.ws_meta.close().await;
inner.ws_meta.close().await.map_err(map_to_string).map(drop)
}
pub async fn send(&self, message: Vec<u8>) -> Result<(), String> {
@ -77,46 +77,37 @@ pub struct WebsocketProtocolHandler {}
impl WebsocketProtocolHandler {
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: &DialInfo,
dial_info: DialInfo,
) -> Result<NetworkConnection, String> {
let url = dial_info
.request()
.ok_or_else(|| format!("missing url in websocket dialinfo: {:?}", dial_info))?;
let split_url = SplitUrl::from_str(&url)?;
let tls = match dial_info {
DialInfo::WS(ws) => {
if split_url.scheme.to_ascii_lowercase() != "ws" {
return Err(format!("wrong scheme for WS websocket url: {}", url));
}
false
}
DialInfo::WSS(wss) => {
if split_url.scheme.to_ascii_lowercase() != "wss" {
return Err(format!("wrong scheme for WSS websocket url: {}", url));
}
true
}
_ => {
return Err("wrong protocol for WebsocketProtocolHandler".to_owned())
.map_err(logthru_net!(error))
}
};
assert!(local_address.is_none());
let (_, wsio) = WsMeta::connect(url, None)
// Split dial info up
let (tls, scheme) = match &dial_info {
DialInfo::WS(_) => (false, "ws"),
DialInfo::WSS(_) => (true, "wss"),
_ => panic!("invalid dialinfo for WS/WSS protocol"),
};
let request = dial_info.request().unwrap();
let split_url = SplitUrl::from_str(&request)?;
if split_url.scheme != scheme {
return Err("invalid websocket url scheme".to_string());
}
let (wsmeta, wsio) = WsMeta::connect(request, None)
.await
.map_err(map_to_string)
.map_err(logthru_net!(error))?;
// Make our connection descriptor
let connection_descriptor = ConnectionDescriptor {
Ok(NetworkConnection::from_protocol(ConnectionDescriptor {
local: None,
remote: dial_info.to_peer_address(),
};
Ok(NetworkConnection::from_protocol(descriptor,ProtocolNetworkConnection::Ws(WebsocketNetworkConnection::new(tls, wsio))))
},ProtocolNetworkConnection::Ws(WebsocketNetworkConnection::new(tls, wsmeta, wsio))))
}
pub async fn send_unbound_message(dial_info: &DialInfo, data: Vec<u8>) -> Result<(), String> {
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> Result<(), String> {
if data.len() > MAX_MESSAGE_SIZE {
return Err("sending too large unbound WS message".to_owned());
}
@ -126,7 +117,7 @@ impl WebsocketProtocolHandler {
dial_info,
);
let conn = Self::connect(None, dial_info.clone())
let conn = Self::connect(None, dial_info)
.await
.map_err(|e| format!("failed to connect websocket for unbound message: {}", e))?;

2
veilid-core/src/network_manager.rs
View File

@ -405,7 +405,7 @@ impl NetworkManager {
// Called by the RPC handler when we want to issue an direct receipt
pub async fn send_direct_receipt<B: AsRef<[u8]>>(
&self,
dial_info: &DialInfo,
dial_info: DialInfo,
rcpt_data: B,
alternate_port: bool,
) -> Result<(), String> {

2
veilid-core/src/rpc_processor/mod.rs
View File

@ -941,7 +941,7 @@ impl RPCProcessor {
// Possibly from an alternate port
let network_manager = self.network_manager();
network_manager
.send_direct_receipt(&dial_info, rcpt_data, alternate_port)
.send_direct_receipt(dial_info.clone(), rcpt_data, alternate_port)
.await
.map_err(map_error_string!())
.map_err(

28
veilid-core/src/tests/common/test_connection_table.rs
View File

@ -59,37 +59,37 @@ pub async fn test_add_get_remove() {
assert_ne!(a4, c5.connection_descriptor());
assert_eq!(table.connection_count(), 0);
assert_eq!(table.get_connection(&a1), None);
assert_eq!(table.get_connection(a1), None);
table.add_connection(c1.clone()).unwrap();
assert_eq!(table.connection_count(), 1);
assert_err!(table.remove_connection(&a3));
assert_err!(table.remove_connection(&a4));
assert_err!(table.remove_connection(a3));
assert_err!(table.remove_connection(a4));
assert_eq!(table.connection_count(), 1);
assert_eq!(table.get_connection(&a1), Some(c1.clone()));
assert_eq!(table.get_connection(&a1), Some(c1.clone()));
assert_eq!(table.get_connection(a1), Some(c1.clone()));
assert_eq!(table.get_connection(a1), Some(c1.clone()));
assert_eq!(table.connection_count(), 1);
assert_err!(table.add_connection(c1.clone()));
assert_err!(table.add_connection(c2.clone()));
assert_eq!(table.connection_count(), 1);
assert_eq!(table.get_connection(&a1), Some(c1.clone()));
assert_eq!(table.get_connection(&a1), Some(c1.clone()));
assert_eq!(table.get_connection(a1), Some(c1.clone()));
assert_eq!(table.get_connection(a1), Some(c1.clone()));
assert_eq!(table.connection_count(), 1);
assert_eq!(table.remove_connection(&a2), Ok(c1.clone()));
assert_eq!(table.remove_connection(a2), Ok(c1.clone()));
assert_eq!(table.connection_count(), 0);
assert_err!(table.remove_connection(&a2));
assert_err!(table.remove_connection(a2));
assert_eq!(table.connection_count(), 0);
assert_eq!(table.get_connection(&a2), None);
assert_eq!(table.get_connection(&a1), None);
assert_eq!(table.get_connection(a2), None);
assert_eq!(table.get_connection(a1), None);
assert_eq!(table.connection_count(), 0);
table.add_connection(c1.clone()).unwrap();
assert_err!(table.add_connection(c2));
table.add_connection(c3.clone()).unwrap();
table.add_connection(c4.clone()).unwrap();
assert_eq!(table.connection_count(), 3);
assert_eq!(table.remove_connection(&a2), Ok(c1));
assert_eq!(table.remove_connection(&a3), Ok(c3));
assert_eq!(table.remove_connection(&a4), Ok(c4));
assert_eq!(table.remove_connection(a2), Ok(c1));
assert_eq!(table.remove_connection(a3), Ok(c3));
assert_eq!(table.remove_connection(a4), Ok(c4));
assert_eq!(table.connection_count(), 0);
}

1
veilid-core/src/xx/mod.rs
View File

@ -27,7 +27,6 @@ pub type SendPinBoxFutureLifetime<'a, T> = PinBox<dyn Future<Output = T> + Send
cfg_if! {
if #[cfg(target_arch = "wasm32")] {
extern crate alloc;
pub use alloc::string::String;
pub use alloc::vec::Vec;
pub use alloc::collections::btree_map::BTreeMap;

2
veilid-server/Cargo.toml
View File

@ -2,7 +2,7 @@
name = "veilid-server"
version = "0.1.0"
authors = ["John Smith <jsmith@example.com>"]
edition = "2018"
edition = "2021"
build = "build.rs"
license = "LGPL-2.0-or-later OR MPL-2.0 OR (MIT AND BSD-3-Clause)"