DaLaw2/VisioGrid
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimize Connection Channel structure.

Browse Source
This commit is contained in:
DaLaw2 2024-02-16 21:44:42 +08:00
parent aaefe354c7
commit cac0427ca6
20 changed files with 545 additions and 412 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -1,2 +1,3 @@
target/
.idea/
.idea/
Unimplemented.txt

48
Server/src/main.rs
View File

@ -1,23 +1,49 @@
#![allow(non_snake_case)]
use tokio::time::sleep;
use std::time::Duration;
use library::web::page::log;
use library::web::page::config;
use library::web::server::Server;
use library::web::page::inference;
use library::web::page::javascript;
use library::web::page::config;
use library::utils::config::Config;
use actix_web::{App, Error, HttpServer};
use library::utils::logger::{Logger, LogLevel};
use library::manager::file_manager::FileManager;
use library::manager::node_cluster::NodeCluster;
#[actix_web::main]
async fn main() -> Result<(), Error> {
Config::now().await;
FileManager::run().await;
HttpServer::new(|| {
App::new()
.service(config::initialize())
.service(inference::initialize())
.service(log::initialize())
.service(javascript::initialize())
})
.bind("127.0.0.1:8080")?
.run()
.await?;
NodeCluster::run().await;
Server::run().await;
let http_server = loop {
let config = Config::now().await;
let http_server = HttpServer::new(|| {
App::new()
.service(config::initialize())
.service(inference::initialize())
.service(log::initialize())
.service(javascript::initialize())
}).bind(format!("127.0.0.1:{}", config.http_server_bind_port));
match http_server {
Ok(http_server) => break http_server,
Err(err) => {
Logger::append_system_log(LogLevel::ERROR, format!("Http Server: Bind port failed.\nReason: {}", err)).await;
sleep(Duration::from_millis(config.internal_timestamp)).await;
continue;
},
}
};
Logger::append_system_log(LogLevel::INFO, "Http Server: Online.".to_string()).await;
match http_server.run().await {
Ok(_) => {},
Err(err) => Logger::append_system_log(LogLevel::ERROR, format!("Http Server: Internal server error.\nReason: {}", err)).await,
};
FileManager::terminate().await;
NodeCluster::terminate().await;
Server::terminate().await;
Ok(())
}

7
Unimplemented.txt
View File

@ -1,7 +0,0 @@
實踐SOLID原則
Logger使用全局 RwLock 可能會在高並發情況下成為瓶頸。可以使用異步非阻塞 I/O 或消息隊列
任務不管是完成還是失敗，都要清除資料(Repository)
加入對除了YOLO以外的支持
目前影片拆解最多30FPS、組合時固定30FPS

60
library/src/connection/connection_channel/control_channel.rs
View File

@ -1,58 +1,16 @@
use uuid::Uuid;
use tokio::sync::{mpsc, oneshot};
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::connection_channel::sender::Sender;
use crate::connection::socket::socket_stream::SocketStream;
use crate::connection::connection_channel::control_channel_receiver::Receiver;
use crate::connection::connection_channel::control_packet_channel::{ControlPacketChannel, PacketReceiver};
use crate::connection::connection_channel::control_channel_sender::ControlChannelSender;
use crate::connection::connection_channel::control_channel_receiver::ControlChannelReceiver;
pub struct ControlChannel {
node_id: Uuid,
sender: mpsc::UnboundedSender<Option<Box<dyn Packet + Send>>>,
stop_signal: Option<oneshot::Sender<()>>,
}
pub struct ControlChannel;
impl ControlChannel {
pub fn new(node_id: Uuid, socket: SocketStream) -> (Self, PacketReceiver) {
let (socket_sender, socket_receiver) = socket.into_split();
let (sender_tx, sender_rx) = mpsc::unbounded_channel();
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let (control_packet_channel_tx, control_packet_channel_rx) = ControlPacketChannel::split();
let mut sender = Sender::new(node_id, socket_sender, sender_rx);
let mut receiver = Receiver::new(node_id, socket_receiver, stop_signal_rx, control_packet_channel_tx);
tokio::spawn(async move {
sender.run().await;
});
tokio::spawn(async move {
receiver.run().await;
});
let control_channel = Self {
node_id,
sender: sender_tx,
stop_signal: Some(stop_signal_tx),
};
(control_channel, control_packet_channel_rx)
}
pub async fn disconnect(&mut self) {
match self.sender.send(None) {
Ok(_) => Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel: Destroyed Sender successfully.".to_string()).await,
Err(err) => Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Control Channel: Failed to destroy Sender.\nReason: {}.", err)).await,
}
match self.stop_signal.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel: Destroyed Receiver successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Control Channel: Failed to destroy Receiver.".to_string()).await,
}
}
pub async fn send<T: Packet + Send + 'static>(&mut self, packet: T) {
let packet: Box<dyn Packet + Send + 'static> = Box::new(packet);
if let Err(err) = self.sender.send(Some(packet)) {
Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Control Channel: Failed to send packet to client.\nReason: {}.", err)).await;
}
pub fn new(node_id: Uuid, socket: SocketStream) -> (ControlChannelSender, ControlChannelReceiver) {
let (socket_tx, socket_rx) = socket.into_split();
(
ControlChannelSender::new(node_id, socket_tx),
ControlChannelReceiver::new(node_id, socket_rx),
)
}
}

59
library/src/connection/connection_channel/control_channel_receive_thread.rs Normal file
View File

@ -0,0 +1,59 @@
use uuid::Uuid;
use tokio::select;
use tokio::sync::oneshot;
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::packet::definition::PacketType;
use crate::connection::socket::socket_stream::ReadHalf;
use crate::connection::connection_channel::control_channel_receiver::ReceiverTX;
pub struct ReceiveThread {
node_id: Uuid,
socket_rx: ReadHalf,
receiver_tx: ReceiverTX,
stop_signal_rx: oneshot::Receiver<()>,
}
impl ReceiveThread {
pub fn new(node_id: Uuid, socket_rx: ReadHalf, receiver_tx: ReceiverTX, stop_signal_rx: oneshot::Receiver<()>) -> Self {
Self {
node_id,
socket_rx,
receiver_tx,
stop_signal_rx,
}
}
pub async fn run(&mut self) {
loop {
select! {
biased;
packet = self.socket_rx.receive_packet() => {
match packet {
Ok(packet) => {
let packet_type = PacketType::parse_packet_type(&packet.clone_id_byte());
let result = match packet_type {
PacketType::ControlReplyPacket => self.receiver_tx.control_reply_packet.send(packet),
PacketType::NodeInformationPacket => self.receiver_tx.node_information_packet.send(packet),
PacketType::PerformancePacket => self.receiver_tx.performance_packet.send(packet),
_ => {
Logger::append_node_log(self.node_id, LogLevel::WARNING, "Receive Thread: Receive unknown packet.".to_string()).await;
Ok(())
},
};
if result.is_err() {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Receive Thread: Unable to submit packet to receiver.".to_string()).await;
break;
}
},
Err(_) => {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Receive Thread: Client disconnect.".to_string()).await;
break;
},
}
},
_ = &mut self.stop_signal_rx => break,
}
}
}
}

81
library/src/connection/connection_channel/control_channel_receiver.rs
View File

@ -1,57 +1,56 @@
use uuid::Uuid;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::packet::definition::PacketType;
use crate::connection::packet::base_packet::BasePacket;
use crate::connection::socket::socket_stream::ReadHalf;
use crate::connection::connection_channel::control_packet_channel::PacketSender;
use crate::connection::connection_channel::control_channel_receive_thread::ReceiveThread;
pub struct Receiver {
pub struct ControlChannelReceiver {
node_id: Uuid,
socket: ReadHalf,
stop_signal: oneshot::Receiver<()>,
control_packet_channel: PacketSender,
stop_signal_tx: Option<oneshot::Sender<()>>,
pub control_reply_packet: mpsc::UnboundedReceiver<BasePacket>,
pub node_information_packet: mpsc::UnboundedReceiver<BasePacket>,
pub performance_packet: mpsc::UnboundedReceiver<BasePacket>,
}
impl Receiver {
pub fn new(node_id: Uuid, socket: ReadHalf, stop_signal: oneshot::Receiver<()>, control_packet_channel: PacketSender) -> Self {
impl ControlChannelReceiver {
pub fn new(node_id: Uuid, socket_rx: ReadHalf) -> Self {
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let (control_reply_packet_tx, control_reply_packet_rx) = mpsc::unbounded_channel();
let (node_information_packet_tx, node_information_packet_rx) = mpsc::unbounded_channel();
let (performance_packet_tx, performance_packet_rx) = mpsc::unbounded_channel();
let receiver_tx = ReceiverTX {
control_reply_packet: control_reply_packet_tx,
node_information_packet: node_information_packet_tx,
performance_packet: performance_packet_tx,
};
let mut receive_thread = ReceiveThread::new(node_id, socket_rx, receiver_tx, stop_signal_rx);
tokio::spawn(async move {
receive_thread.run().await;
});
Self {
node_id,
socket,
stop_signal,
control_packet_channel,
stop_signal_tx: Some(stop_signal_tx),
control_reply_packet: control_reply_packet_rx,
node_information_packet: node_information_packet_rx,
performance_packet: performance_packet_rx,
}
}
pub async fn run(&mut self) {
loop {
tokio::select! {
packet = self.socket.receive_packet() => {
match packet {
Ok(packet) => {
let packet_type = PacketType::parse_packet_type(&packet.clone_id_byte());
let result = match packet_type {
PacketType::ControlReplyPacket => self.control_packet_channel.control_reply_packet.send(packet),
PacketType::NodeInformationPacket => self.control_packet_channel.node_information_packet.send(packet),
PacketType::PerformancePacket => self.control_packet_channel.performance_packet.send(packet),
_ => {
Logger::append_node_log(self.node_id, LogLevel::WARNING, "Control Channel Receiver: Receive unknown packet.".to_string()).await;
Ok(())
},
};
if result.is_err() {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel Receiver: Client disconnect.".to_string()).await;
break;
}
},
Err(_) => {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel Receiver: Client disconnect.".to_string()).await;
break;
},
}
},
_ = &mut self.stop_signal => break,
}
pub async fn disconnect(&mut self) {
match self.stop_signal_tx.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel: Destroyed Receiver successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Control Channel: Failed to destroy Receiver.".to_string()).await,
}
}
}
pub struct ReceiverTX {
pub control_reply_packet: mpsc::UnboundedSender<BasePacket>,
pub node_information_packet: mpsc::UnboundedSender<BasePacket>,
pub performance_packet: mpsc::UnboundedSender<BasePacket>,
}

48
library/src/connection/connection_channel/control_channel_sender.rs Normal file
View File

@ -0,0 +1,48 @@
use uuid::Uuid;
use tokio::sync::{mpsc, oneshot};
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::socket::socket_stream::WriteHalf;
use crate::connection::connection_channel::send_thread::SendThread;
pub type SenderTX = mpsc::UnboundedSender<Box<dyn Packet+Send>>;
pub type SenderRX = mpsc::UnboundedReceiver<Box<dyn Packet+Send>>;
pub struct ControlChannelSender {
node_id: Uuid,
sender_tx: SenderTX,
stop_signal_tx: Option<oneshot::Sender<()>>,
}
impl ControlChannelSender {
pub fn new(node_id: Uuid, socket_tx: WriteHalf) -> Self {
let (sender_tx, sender_rx) = mpsc::unbounded_channel();
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let mut send_thread = SendThread::new(node_id, socket_tx, sender_rx, stop_signal_rx);
tokio::spawn(async move {
send_thread.run().await;
});
Self {
node_id,
sender_tx,
stop_signal_tx: Some(stop_signal_tx),
}
}
pub async fn disconnect(&mut self) {
match self.stop_signal_tx.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Control Channel: Destroyed Sender successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Control Channel: Failed to destroy Sender.".to_string()).await,
}
}
pub async fn send<T: Packet + Send + 'static>(&mut self, packet: T) {
let packet: Box<dyn Packet + Send + 'static> = Box::new(packet);
if let Err(err) = self.sender_tx.send(packet) {
Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Control Channel: Unable to submit packet to Send Thread.\nReason: {}.", err)).await;
}
}
}

36
library/src/connection/connection_channel/control_packet_channel.rs
View File

@ -1,36 +0,0 @@
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use crate::connection::packet::base_packet::BasePacket;
pub struct ControlPacketChannel;
impl ControlPacketChannel {
pub fn split() -> (PacketSender, PacketReceiver) {
let (control_reply_packet_tx, control_reply_packet_rx) = mpsc::unbounded_channel();
let (node_information_packet_tx, node_information_packet_rx) = mpsc::unbounded_channel();
let (performance_packet_tx, performance_packet_rx) = mpsc::unbounded_channel();
(
PacketSender {
control_reply_packet: control_reply_packet_tx,
node_information_packet: node_information_packet_tx,
performance_packet: performance_packet_tx,
},
PacketReceiver {
control_reply_packet: control_reply_packet_rx,
node_information_packet: node_information_packet_rx,
performance_packet: performance_packet_rx,
},
)
}
}
pub struct PacketSender {
pub control_reply_packet: UnboundedSender<BasePacket>,
pub node_information_packet: UnboundedSender<BasePacket>,
pub performance_packet: UnboundedSender<BasePacket>,
}
pub struct PacketReceiver {
pub control_reply_packet: UnboundedReceiver<BasePacket>,
pub node_information_packet: UnboundedReceiver<BasePacket>,
pub performance_packet: UnboundedReceiver<BasePacket>,
}

60
library/src/connection/connection_channel/data_channel.rs
View File

@ -1,58 +1,16 @@
use uuid::Uuid;
use tokio::sync::{mpsc, oneshot};
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::connection_channel::sender::Sender;
use crate::connection::socket::socket_stream::SocketStream;
use crate::connection::connection_channel::data_channel_receiver::Receiver;
use crate::connection::connection_channel::data_packet_channel::{DataPacketChannel, PacketReceiver};
use crate::connection::connection_channel::data_channel_sender::DataChannelSender;
use crate::connection::connection_channel::data_channel_receiver::DataChannelReceiver;
pub struct DataChannel {
node_id: Uuid,
sender: mpsc::UnboundedSender<Option<Box<dyn Packet + Send>>>,
stop_signal: Option<oneshot::Sender<()>>,
}
pub struct DataChannel;
impl DataChannel {
pub fn new(node_id: Uuid, socket: SocketStream) -> (Self, PacketReceiver) {
let (sender_tx, sender_rx) = mpsc::unbounded_channel();
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let (socket_sender, socket_receiver) = socket.into_split();
let (data_packet_channel_tx, data_packet_channel_rx) = DataPacketChannel::split();
let mut sender = Sender::new(node_id, socket_sender, sender_rx);
let mut receiver = Receiver::new(node_id, socket_receiver, stop_signal_rx, data_packet_channel_tx);
tokio::spawn(async move {
sender.run().await;
});
tokio::spawn(async move {
receiver.run().await;
});
let data_channel = Self {
node_id,
sender: sender_tx,
stop_signal: Some(stop_signal_tx),
};
(data_channel, data_packet_channel_rx)
}
pub async fn disconnect(&mut self) {
match self.sender.send(None) {
Ok(_) => Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel: Destroyed Sender successfully.".to_string()).await,
Err(err) => Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Data Channel: Failed to destroy Sender.\nReason: {}.", err)).await,
}
match self.stop_signal.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel: Destroyed Receiver successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Data Channel: Failed to destroy Receiver.".to_string()).await,
}
}
pub async fn send<T: Packet + Send + 'static>(&mut self, packet: T) {
let packet: Box<dyn Packet + Send + 'static> = Box::new(packet);
if let Err(err) = self.sender.send(Some(packet)) {
Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Data Channel: Failed to send packet to client.\nReason: {}.", err)).await;
}
pub fn new(node_id: Uuid, socket: SocketStream) -> (DataChannelSender, DataChannelReceiver) {
let (socket_tx, socket_rx) = socket.into_split();
(
DataChannelSender::new(node_id, socket_tx),
DataChannelReceiver::new(node_id, socket_rx),
)
}
}

61
library/src/connection/connection_channel/data_channel_receive_thread.rs Normal file
View File

@ -0,0 +1,61 @@
use uuid::Uuid;
use tokio::select;
use tokio::sync::oneshot;
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::packet::definition::PacketType;
use crate::connection::socket::socket_stream::ReadHalf;
use crate::connection::connection_channel::data_channel_receiver::ReceiverTX;
pub struct ReceiveThread {
node_id: Uuid,
socket_rx: ReadHalf,
receiver_tx: ReceiverTX,
stop_signal_rx: oneshot::Receiver<()>,
}
impl ReceiveThread {
pub fn new(node_id: Uuid, socket_rx: ReadHalf, receiver_tx: ReceiverTX, stop_signal_rx: oneshot::Receiver<()>) -> Self {
Self {
node_id,
socket_rx,
receiver_tx,
stop_signal_rx,
}
}
pub async fn run(&mut self) {
loop {
select! {
biased;
packet = self.socket_rx.receive_packet() => {
match packet {
Ok(packet) => {
let packet_type = PacketType::parse_packet_type(&packet.clone_id_byte());
let result = match packet_type {
PacketType::AliveReplyPacket => self.receiver_tx.alive_reply_packet.send(packet),
PacketType::FileTransferReplyPacket => self.receiver_tx.file_transfer_reply_packet.send(packet),
PacketType::ResultPacket => self.receiver_tx.result_packet.send(packet),
PacketType::StillProcessReplyPacket => self.receiver_tx.still_process_reply_packet.send(packet),
PacketType::TaskInfoReplyPacket => self.receiver_tx.task_info_reply_packet.send(packet),
_ => {
Logger::append_node_log(self.node_id, LogLevel::WARNING, "Receive Thread: Receive unknown packet.".to_string()).await;
Ok(())
},
};
if result.is_err() {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Receive Thread: Unable to submit packet to receiver.".to_string()).await;
break;
}
},
Err(_) => {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Receive Thread: Client disconnect.".to_string()).await;
break;
},
}
},
_ = &mut self.stop_signal_rx => break,
}
}
}
}

93
library/src/connection/connection_channel/data_channel_receiver.rs
View File

@ -1,59 +1,66 @@
use uuid::Uuid;
use tokio::sync::oneshot;
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::packet::definition::PacketType;
use crate::connection::packet::base_packet::BasePacket;
use crate::connection::socket::socket_stream::ReadHalf;
use crate::connection::connection_channel::data_packet_channel::PacketSender;
use crate::connection::connection_channel::data_channel_receive_thread::ReceiveThread;
pub struct Receiver {
pub struct DataChannelReceiver {
node_id: Uuid,
socket: ReadHalf,
stop_signal: oneshot::Receiver<()>,
data_packet_channel: PacketSender,
stop_signal_tx: Option<oneshot::Sender<()>>,
pub alive_reply_packet: UnboundedReceiver<BasePacket>,
pub file_transfer_reply_packet: UnboundedReceiver<BasePacket>,
pub result_packet: UnboundedReceiver<BasePacket>,
pub still_process_reply_packet: UnboundedReceiver<BasePacket>,
pub task_info_reply_packet: UnboundedReceiver<BasePacket>,
}
impl Receiver {
pub fn new(node_id: Uuid, socket: ReadHalf, stop_signal: oneshot::Receiver<()>, data_packet_channel: PacketSender) -> Self {
impl DataChannelReceiver {
pub fn new(node_id: Uuid, socket_rx: ReadHalf) -> Self {
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let (alive_reply_packet_tx, alive_reply_packet_rx) = mpsc::unbounded_channel();
let (file_transfer_reply_packet_tx, file_transfer_reply_packet_rx) = mpsc::unbounded_channel();
let (result_packet_tx, result_packet_rx) = mpsc::unbounded_channel();
let (still_process_reply_packet_tx, still_process_reply_packet_rx) = mpsc::unbounded_channel();
let (task_info_reply_packet_tx, task_info_reply_packet_rx) = mpsc::unbounded_channel();
let receiver_tx = ReceiverTX {
alive_reply_packet: alive_reply_packet_tx,
file_transfer_reply_packet: file_transfer_reply_packet_tx,
result_packet: result_packet_tx,
still_process_reply_packet: still_process_reply_packet_tx,
task_info_reply_packet: task_info_reply_packet_tx,
};
let mut receive_thread = ReceiveThread::new(node_id, socket_rx, receiver_tx, stop_signal_rx);
tokio::spawn(async move {
receive_thread.run().await;
});
Self {
node_id,
socket,
stop_signal,
data_packet_channel,
stop_signal_tx: Some(stop_signal_tx),
alive_reply_packet: alive_reply_packet_rx,
file_transfer_reply_packet: file_transfer_reply_packet_rx,
result_packet: result_packet_rx,
still_process_reply_packet: still_process_reply_packet_rx,
task_info_reply_packet: task_info_reply_packet_rx,
}
}
pub async fn run(&mut self) {
loop {
tokio::select! {
packet = self.socket.receive_packet() => {
match packet {
Ok(packet) => {
let packet_type = PacketType::parse_packet_type(&packet.clone_id_byte());
let result = match packet_type {
PacketType::AliveReplyPacket => self.data_packet_channel.alive_reply_packet.send(packet),
PacketType::FileTransferReplyPacket => self.data_packet_channel.file_transfer_reply_packet.send(packet),
PacketType::ResultPacket => self.data_packet_channel.result_packet.send(packet),
PacketType::StillProcessReplyPacket => self.data_packet_channel.still_process_reply_packet.send(packet),
PacketType::TaskInfoReplyPacket => self.data_packet_channel.task_info_reply_packet.send(packet),
_ => {
Logger::append_node_log(self.node_id, LogLevel::WARNING, "Data Channel Receiver: Receive unknown packet.".to_string()).await;
Ok(())
},
};
if result.is_err() {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel Receiver: Client disconnect.".to_string()).await;
break;
}
},
Err(_) => {
Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel Receiver: Client disconnect.".to_string()).await;
break;
},
}
},
_ = &mut self.stop_signal => break,
}
pub async fn disconnect(&mut self) {
match self.stop_signal_tx.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel: Destroyed Receiver successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Data Channel: Failed to destroy Receiver.".to_string()).await,
}
}
}
pub struct ReceiverTX {
pub alive_reply_packet: UnboundedSender<BasePacket>,
pub file_transfer_reply_packet: UnboundedSender<BasePacket>,
pub result_packet: UnboundedSender<BasePacket>,
pub still_process_reply_packet: UnboundedSender<BasePacket>,
pub task_info_reply_packet: UnboundedSender<BasePacket>,
}

48
library/src/connection/connection_channel/data_channel_sender.rs Normal file
View File

@ -0,0 +1,48 @@
use uuid::Uuid;
use tokio::sync::{mpsc, oneshot};
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::socket::socket_stream::WriteHalf;
use crate::connection::connection_channel::send_thread::SendThread;
pub type SenderTX = mpsc::UnboundedSender<Box<dyn Packet+Send>>;
pub type SenderRX = mpsc::UnboundedReceiver<Box<dyn Packet+Send>>;
pub struct DataChannelSender {
node_id: Uuid,
sender_tx: SenderTX,
stop_signal_tx: Option<oneshot::Sender<()>>,
}
impl DataChannelSender {
pub fn new(node_id: Uuid, socket_tx: WriteHalf) -> Self {
let (sender_tx, sender_rx) = mpsc::unbounded_channel();
let (stop_signal_tx, stop_signal_rx) = oneshot::channel();
let mut send_thread = SendThread::new(node_id, socket_tx, sender_rx, stop_signal_rx);
tokio::spawn(async move {
send_thread.run().await;
});
Self {
node_id,
sender_tx,
stop_signal_tx: Some(stop_signal_tx),
}
}
pub async fn disconnect(&mut self) {
match self.stop_signal_tx.take() {
Some(stop_signal) => {
let _ = stop_signal.send(());
Logger::append_node_log(self.node_id, LogLevel::INFO, "Data Channel: Destroyed Sender successfully.".to_string()).await;
},
None => Logger::append_node_log(self.node_id, LogLevel::ERROR, "Data Channel: Failed to destroy Sender.".to_string()).await,
}
}
pub async fn send<T: Packet + Send + 'static>(&mut self, packet: T) {
let packet: Box<dyn Packet + Send + 'static> = Box::new(packet);
if let Err(err) = self.sender_tx.send(packet) {
Logger::append_node_log(self.node_id, LogLevel::ERROR, format!("Data Channel: Unable to submit packet to Send Thread.\nReason: {}.", err)).await;
}
}
}

46
library/src/connection/connection_channel/data_packet_channel.rs
View File

@ -1,46 +0,0 @@
use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
use crate::connection::packet::base_packet::BasePacket;
pub struct DataPacketChannel;
impl DataPacketChannel {
pub fn split() -> (PacketSender, PacketReceiver) {
let (alive_reply_packet_tx, alive_reply_packet_rx) = mpsc::unbounded_channel();
let (file_transfer_reply_packet_tx, file_transfer_reply_packet_rx) = mpsc::unbounded_channel();
let (result_packet_tx, result_packet_rx) = mpsc::unbounded_channel();
let (still_process_reply_packet_tx, still_process_reply_packet_rx) = mpsc::unbounded_channel();
let (task_info_reply_packet_tx, task_info_reply_packet_rx) = mpsc::unbounded_channel();
(
PacketSender {
alive_reply_packet: alive_reply_packet_tx,
file_transfer_reply_packet: file_transfer_reply_packet_tx,
result_packet: result_packet_tx,
still_process_reply_packet: still_process_reply_packet_tx,
task_info_reply_packet: task_info_reply_packet_tx,
},
PacketReceiver {
alive_reply_packet: alive_reply_packet_rx,
file_transfer_reply_packet: file_transfer_reply_packet_rx,
result_packet: result_packet_rx,
still_process_reply_packet: still_process_reply_packet_rx,
task_info_reply_packet: task_info_reply_packet_rx,
},
)
}
}
pub struct PacketSender {
pub alive_reply_packet: UnboundedSender<BasePacket>,
pub file_transfer_reply_packet: UnboundedSender<BasePacket>,
pub result_packet: UnboundedSender<BasePacket>,
pub still_process_reply_packet: UnboundedSender<BasePacket>,
pub task_info_reply_packet: UnboundedSender<BasePacket>,
}
pub struct PacketReceiver {
pub alive_reply_packet: UnboundedReceiver<BasePacket>,
pub file_transfer_reply_packet: UnboundedReceiver<BasePacket>,
pub result_packet: UnboundedReceiver<BasePacket>,
pub still_process_reply_packet: UnboundedReceiver<BasePacket>,
pub task_info_reply_packet: UnboundedReceiver<BasePacket>,
}

8
library/src/connection/connection_channel/mod.rs
View File

@ -1,7 +1,9 @@
pub mod control_channel;
pub mod control_channel_receive_thread;
pub mod control_channel_receiver;
pub mod control_packet_channel;
pub mod control_channel_sender;
pub mod data_channel;
pub mod data_channel_receive_thread;
pub mod data_channel_receiver;
pub mod data_packet_channel;
pub mod sender;
pub mod data_channel_sender;
pub mod send_thread;

46
library/src/connection/connection_channel/send_thread.rs Normal file
View File

@ -0,0 +1,46 @@
use uuid::Uuid;
use tokio::select;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::socket::socket_stream::WriteHalf;
type SenderRX = mpsc::UnboundedReceiver<Box<dyn Packet + Send>>;
pub struct SendThread {
node_id: Uuid,
socket_tx: WriteHalf,
sender_rx: SenderRX,
stop_signal_rx: oneshot::Receiver<()>,
}
impl SendThread {
pub fn new(node_id: Uuid, socket_tx: WriteHalf, sender_rx: SenderRX, stop_signal_rx: oneshot::Receiver<()>) -> Self {
Self {
node_id,
socket_tx,
sender_rx,
stop_signal_rx,
}
}
pub async fn run(&mut self) {
loop {
select! {
biased;
reply = self.sender_rx.recv() => {
match reply {
Some(packet) => {
if self.socket_tx.send_packet(packet).await.is_err() {
Logger::append_node_log(self.node_id, LogLevel::ERROR, "Send Thread: Failed to send packet.".to_string()).await;
}
},
None => break,
}
},
_ = &mut self.stop_signal_rx => break,
}
}
}
}

34
library/src/connection/connection_channel/sender.rs
View File

@ -1,34 +0,0 @@
use uuid::Uuid;
use tokio::sync::mpsc;
use crate::utils::logger::{Logger, LogLevel};
use crate::connection::packet::definition::Packet;
use crate::connection::socket::socket_stream::WriteHalf;
pub struct Sender {
node_id: Uuid,
socket: WriteHalf,
receiver: mpsc::UnboundedReceiver<Option<Box<dyn Packet + Send>>>
}
impl Sender {
pub fn new(node_id: Uuid, socket: WriteHalf, receiver: mpsc::UnboundedReceiver<Option<Box<dyn Packet + Send>>>) -> Self {
Self {
node_id,
socket,
receiver,
}
}
pub async fn run(&mut self) {
while let Some(packet) = self.receiver.recv().await {
match packet {
Some(packet) => {
if self.socket.send_packet(packet).await.is_err() {
Logger::append_node_log(self.node_id, LogLevel::ERROR, "Sender: Failed to send packet.".to_string()).await;
}
},
None => break,
}
}
}
}

17
library/src/manager/file_manager.rs
View File

@ -62,14 +62,11 @@ impl FileManager {
tokio::spawn(async {
Self::post_processing().await;
});
}
pub async fn terminate() {
Self::instance_mut().await.terminate = true;
Self::cleanup().await;
Logger::append_system_log(LogLevel::INFO, "File Manager: Online.".to_string()).await;
}
async fn initialize() {
Logger::append_system_log(LogLevel::INFO, "File Manager: Initializing.".to_string()).await;
let folders = ["SavedModel", "SavedFile", "PreProcessing", "PostProcessing", "Result"];
for &folder_name in &folders {
match fs::create_dir(folder_name).await {
@ -81,9 +78,18 @@ impl FileManager {
Ok(_) => Logger::append_system_log(LogLevel::INFO, "File Manager: GStreamer initialization successfully.".to_string()).await,
Err(err) => Logger::append_system_log(LogLevel::ERROR, format!("File Manager: GStreamer initialization failed.\nReason: {}.", err)).await,
}
Logger::append_system_log(LogLevel::INFO, "File Manager: Initialization completed.".to_string()).await;
}
pub async fn terminate() {
Logger::append_system_log(LogLevel::INFO, "File Manager: Terminating.".to_string()).await;
Self::instance_mut().await.terminate = true;
Self::cleanup().await;
Logger::append_system_log(LogLevel::INFO, "File Manager: Termination complete.".to_string()).await;
}
async fn cleanup() {
Logger::append_system_log(LogLevel::INFO, "File Manager: Cleaning up.".to_string()).await;
let folders = ["SavedModel", "SavedFile", "PreProcessing", "PostProcessing", "Result"];
for &folder_name in &folders {
match fs::remove_dir_all(folder_name).await {
@ -91,6 +97,7 @@ impl FileManager {
Err(err) => Logger::append_system_log(LogLevel::ERROR, format!("File Manager: Cannot delete {} folder.\nReason: {}.", folder_name, err)).await
}
};
Logger::append_system_log(LogLevel::INFO, "File Manager: Cleanup completed.".to_string()).await;
}
pub async fn add_pre_process_task(task: Task) {

154
library/src/manager/node.rs
View File

@ -23,55 +23,57 @@ use crate::connection::packet::alive_packet::AlivePacket;
use crate::connection::socket::socket_stream::SocketStream;
use crate::connection::packet::confirm_packet::ConfirmPacket;
use crate::manager::utils::node_information::NodeInformation;
use crate::connection::connection_channel::data_packet_channel;
use crate::connection::packet::task_info_packet::TaskInfoPacket;
use crate::connection::packet::file_body_packet::FileBodyPacket;
use crate::connection::connection_channel::control_packet_channel;
use crate::connection::packet::file_header_packet::FileHeaderPacket;
use crate::manager::utils::file_transfer_result::FileTransferResult;
use crate::connection::connection_channel::data_channel::DataChannel;
use crate::connection::connection_channel::data_channel_sender::DataChannelSender;
use crate::connection::packet::still_process_packet::StillProcessPacket;
use crate::connection::connection_channel::control_channel::ControlChannel;
use crate::connection::connection_channel::control_channel_sender::ControlChannelSender;
use crate::connection::packet::data_channel_port_packet::DataChannelPortPacket;
use crate::connection::connection_channel::control_channel::ControlChannel;
use crate::connection::connection_channel::control_channel_receiver::ControlChannelReceiver;
use crate::connection::connection_channel::data_channel::DataChannel;
use crate::connection::connection_channel::data_channel_receiver::DataChannelReceiver;
pub struct Node {
uuid: Uuid,
information: NodeInformation,
terminate: bool,
information: NodeInformation,
idle_unused: Performance,
realtime_usage: Performance,
image_task: VecDeque<ImageTask>,
previous_task: Option<ImageTask>,
control_channel: ControlChannel,
data_channel: Option<DataChannel>,
control_packet_channel: control_packet_channel::PacketReceiver,
data_packet_channel: Option<data_packet_channel::PacketReceiver>,
control_channel_sender: ControlChannelSender,
control_channel_receiver: ControlChannelReceiver,
data_channel_sender: Option<DataChannelSender>,
data_channel_receiver: Option<DataChannelReceiver>,
}
impl Node {
pub async fn new(uuid: Uuid, socket_stream: SocketStream) -> Option<Self> {
let config = Config::now().await;
let (mut control_channel, mut control_packet_channel) = ControlChannel::new(uuid, socket_stream);
let (mut control_channel_sender, mut control_channel_receiver) = ControlChannel::new(uuid, socket_stream);
select! {
biased;
reply = control_packet_channel.node_information_packet.recv() => {
reply = control_channel_receiver.node_information_packet.recv() => {
match &reply {
Some(packet) => {
match serde_json::from_slice::<NodeInformation>(&packet.as_data_byte()) {
Ok(information) => {
control_channel.send(ConfirmPacket::new()).await;
control_channel_sender.send(ConfirmPacket::new()).await;
let node = Self {
uuid,
information,
terminate: false,
information,
idle_unused: Performance::default(),
realtime_usage: Performance::default(),
image_task: VecDeque::new(),
previous_task: None,
control_channel,
data_channel: None,
control_packet_channel,
data_packet_channel: None,
control_channel_sender,
control_channel_receiver,
data_channel_sender: None,
data_channel_receiver: None,
};
Some(node)
},
@ -103,18 +105,22 @@ impl Node {
pub async fn terminate(node: Arc<RwLock<Node>>) {
let uuid = node.read().await.uuid;
Logger::append_node_log(uuid, LogLevel::INFO, "Node: Terminating node.".to_string()).await;
let image_task = {
let mut node = node.write().await;
node.terminate = true;
node.control_channel.disconnect().await;
if let Some(data_channel) = &mut node.data_channel {
data_channel.disconnect().await;
node.control_channel_sender.disconnect().await;
node.control_channel_receiver.disconnect().await;
if let Some(data_channel_sender) = &mut node.data_channel_sender {
data_channel_sender.disconnect().await;
}
if let Some(data_channel_receiver) = &mut node.data_channel_receiver {
data_channel_receiver.disconnect().await;
}
mem::take(&mut node.image_task)
};
TaskManager::redistribute_task(image_task).await;
NodeCluster::remove_node(uuid).await;
Logger::append_node_log(uuid, LogLevel::INFO, "Node: Terminating node.".to_string()).await;
}
async fn update_performance(node: Arc<RwLock<Node>>) {
@ -134,13 +140,13 @@ impl Node {
let mut node = node.write().await;
select! {
biased;
reply = node.control_packet_channel.performance_packet.recv() => {
reply = node.control_channel_receiver.performance_packet.recv() => {
match &reply {
Some(reply_packet) => {
match serde_json::from_slice::<Performance>(reply_packet.as_data_byte()) {
Ok(performance) => {
node.realtime_usage = performance;
node.control_channel.send(ConfirmPacket::new()).await;
node.control_channel_sender.send(ConfirmPacket::new()).await;
timer = Instant::now();
},
Err(_) => continue,
@ -183,8 +189,8 @@ impl Node {
return;
}
if polling_timer.elapsed() > polling_interval * polling_times {
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(StillProcessPacket::new()).await,
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(StillProcessPacket::new()).await,
None => {
data_channel_available = false;
break;
@ -192,17 +198,17 @@ impl Node {
}
polling_times += 1;
}
match &mut node.write().await.data_packet_channel {
Some(data_packet_channel) => {
match &mut node.write().await.data_channel_receiver {
Some(data_channel_receiver) => {
select! {
biased;
reply = data_packet_channel.still_process_reply_packet.recv() => {
reply = data_channel_receiver.still_process_reply_packet.recv() => {
match &reply {
Some(_) => timeout_timer = Instant::now(),
None => continue,
}
},
reply = data_packet_channel.result_packet.recv() => {
reply = data_channel_receiver.result_packet.recv() => {
match &reply {
Some(reply_packet) => {
if let Ok(task_result) = serde_json::from_slice::<TaskResult>(reply_packet.as_data_byte()) {
@ -218,7 +224,7 @@ impl Node {
},
_ = sleep(Duration::from_millis(config.internal_timestamp)) => continue,
}
},
}
None => {
data_channel_available = false;
break;
@ -256,8 +262,8 @@ impl Node {
break;
}
if timer.elapsed() > polling_interval * polling_times {
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(AlivePacket::new()).await,
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(AlivePacket::new()).await,
None => {
data_channel_available = false;
break;
@ -265,11 +271,11 @@ impl Node {
}
polling_times += 1;
}
match &mut node.write().await.data_packet_channel {
Some(data_packet_channel) => {
match &mut node.write().await.data_channel_receiver {
Some(data_channel_receiver) => {
select! {
biased;
reply = data_packet_channel.alive_reply_packet.recv() => {
reply = data_channel_receiver.alive_reply_packet.recv() => {
match &reply {
Some(_) => timeout_timer = Instant::now(),
None => continue,
@ -277,7 +283,7 @@ impl Node {
},
_ = sleep(Duration::from_millis(config.internal_timestamp)) => continue,
}
},
}
None => {
data_channel_available = false;
break;
@ -295,12 +301,12 @@ impl Node {
}
async fn create_data_channel(node: Arc<RwLock<Node>>) {
let uuid = node.read().await.uuid;
let config = Config::now().await;
let (listener, port) = loop {
let port = match PortPool::allocate_port().await {
Some(port) => port,
None => {
let uuid = node.read().await.uuid;
Logger::append_node_log(uuid, LogLevel::WARNING, "Node: No available port for Data Channel".to_string()).await;
sleep(Duration::from_secs(config.bind_retry_duration)).await;
continue;
@ -321,10 +327,11 @@ impl Node {
let polling_interval = Duration::from_millis(config.polling_interval);
let (stream, address) = loop {
if timer.elapsed() > timeout_duration {
PortPool::free_port(port).await;
return;
}
if timer.elapsed() > polling_times * polling_interval {
node.write().await.control_channel.send(DataChannelPortPacket::new(port)).await;
node.write().await.control_channel_sender.send(DataChannelPortPacket::new(port)).await;
polling_times += 1;
}
select! {
@ -333,7 +340,7 @@ impl Node {
match connection {
Ok(connection) => break connection,
Err(_) => {
node.write().await.control_channel.send(DataChannelPortPacket::new(port)).await;
node.write().await.control_channel_sender.send(DataChannelPortPacket::new(port)).await;
continue;
},
}
@ -342,15 +349,15 @@ impl Node {
}
};
let socket_stream = SocketStream::new(stream, address);
let (data_channel_sender, data_channel_receiver) = DataChannel::new(uuid, socket_stream);
let mut node = node.write().await;
let (data_channel, data_packet_channel) = DataChannel::new(node.uuid, socket_stream);
node.data_channel = Some(data_channel);
node.data_packet_channel = Some(data_packet_channel);
Logger::append_node_log(node.uuid, LogLevel::INFO, "Node: Create Data channel successfully.".to_string()).await;
node.data_channel_sender = Some(data_channel_sender);
node.data_channel_receiver = Some(data_channel_receiver);
Logger::append_node_log(uuid, LogLevel::INFO, "Node: Create Data channel successfully.".to_string()).await;
}
async fn transfer_task(node: Arc<RwLock<Node>>, image_task: &ImageTask) -> Result<(), String> {
if node.write().await.data_channel.is_some() {
if node.write().await.data_channel_sender.is_some() {
let should_transfer_model = if let Some(last_task) = &node.read().await.previous_task {
image_task.task_uuid != last_task.task_uuid
} else {
@ -363,7 +370,7 @@ impl Node {
Node::transfer_file(node.clone(), &image_task.image_filename, &image_task.image_filepath).await?;
} else {
Node::create_data_channel(node).await;
return Err("Node: Data Channel is not available.".to_string())
Err("Node: Data Channel is not available.".to_string())?
}
Ok(())
}
@ -376,39 +383,40 @@ impl Node {
let timeout_duration = Duration::from_secs(config.control_channel_timeout);
while time.elapsed() < timeout_duration {
if time.elapsed() > polling_interval * polling_times {
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(TaskInfoPacket::new(TaskInfo::new(&image_task))).await,
None => return Err("Node: Data Channel is not available.".to_string()),
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(TaskInfoPacket::new(TaskInfo::new(&image_task))).await,
None => Err("Node: Data Channel is not available.".to_string())?,
}
polling_times += 1;
}
match &mut node.write().await.data_packet_channel {
Some(data_packet_channel) => {
match &mut node.write().await.data_channel_receiver {
Some(data_channel_receiver) => {
select! {
reply = data_packet_channel.task_info_reply_packet.recv() => {
reply = data_channel_receiver.task_info_reply_packet.recv() => {
return match &reply {
Some(_) => Ok(()),
None => return Err("Node: An error occurred while receive packet.".to_string()),
None => Err("Node: An error occurred while receive packet.".to_string()),
}
}
_ = sleep(Duration::from_millis(config.internal_timestamp)) => continue,
}
},
None => return Err("Node: Data Channel is not available.".to_string()),
}
None => Err("Node: Data Channel is not available.".to_string())?,
}
}
Err("Node: Task Info retransmission limit reached.".to_string())
}
#[allow(unused_assignments)]
async fn transfer_file(node: Arc<RwLock<Node>>, filename: &String, filepath: &PathBuf) -> Result<(), String> {
let config = Config::now().await;
let filesize = match fs::metadata(&filepath).await {
Ok(metadata) => metadata.len(),
Err(err) => return Err(format!("Node: Cannot read file {}.\nReason: {}", filepath.display(), err)),
Err(err) => Err(format!("Node: Cannot read file {}.\nReason: {}", filepath.display(), err))?,
};
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(FileHeaderPacket::new(filename.clone(), filesize as usize)).await,
None => return Err("Node: Data channel is not available.".to_string()),
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(FileHeaderPacket::new(filename.clone(), filesize as usize)).await,
None => Err("Node: Data channel is not available.".to_string())?,
}
let file = File::open(filepath.clone()).await;
let mut sequence_number = 0_usize;
@ -425,28 +433,28 @@ impl Node {
}
let mut data = sequence_number.to_be_bytes().to_vec();
data.extend_from_slice(&buffer[..bytes_read]);
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(FileBodyPacket::new(data.clone())).await,
None => return Err("Node: Data channel is not available.".to_string()),
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(FileBodyPacket::new(data.clone())).await,
None => Err("Node: Data channel is not available.".to_string())?,
}
sent_packets.insert(sequence_number, data);
sequence_number += 1;
},
Err(_) => return Err(format!("Node: An error occurred while reading {} file", filepath.display())),
Err(_) => Err(format!("Node: An error occurred while reading {} file", filepath.display()))?,
}
}
},
Err(err) => return Err(format!("Node: Cannot read file {}.\nReason: {}", filepath.display(), err)),
Err(err) => Err(format!("Node: Cannot read file {}.\nReason: {}", filepath.display(), err))?,
}
let time = Instant::now();
let timeout_duration = Duration::from_secs(config.file_transfer_timeout);
let mut require_resend = Vec::new();
while time.elapsed() < timeout_duration {
match &mut node.write().await.data_packet_channel {
Some(data_packet_channel) => {
match &mut node.write().await.data_channel_receiver {
Some(data_channel_receiver) => {
select! {
biased;
reply = data_packet_channel.file_transfer_reply_packet.recv() => {
reply = data_channel_receiver.file_transfer_reply_packet.recv() => {
match &reply {
Some(reply_packet) => {
match FileTransferResult::parse_from_packet(reply_packet).into() {
@ -454,19 +462,19 @@ impl Node {
None => return Ok(()),
}
},
None => return Err("Node: An error occurred while receive packet.".to_string()),
None => Err("Node: An error occurred while receive packet.".to_string())?,
}
},
_ = sleep(Duration::from_millis(config.internal_timestamp)) => continue,
}
},
None => return Err("Node: Data channel is not available.".to_string()),
}
None => Err("Node: Data channel is not available.".to_string())?,
}
for missing_chunk in require_resend {
for missing_chunk in &require_resend {
if let Some(data) = sent_packets.get(&missing_chunk) {
match &mut node.write().await.data_channel {
Some(data_channel) => data_channel.send(FileBodyPacket::new(data.clone())).await,
None => return Err("Node: Data channel is not available.".to_string()),
match &mut node.write().await.data_channel_sender {
Some(data_channel_sender) => data_channel_sender.send(FileBodyPacket::new(data.clone())).await,
None => Err("Node: Data channel is not available.".to_string())?,
}
}
}

12
library/src/manager/node_cluster.rs
View File

@ -8,6 +8,7 @@ use futures::stream::{self, StreamExt};
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::manager::node::Node;
use crate::utils::config::Config;
use crate::utils::logger::{Logger, LogLevel};
lazy_static! {
static ref GLOBAL_CLUSTER: RwLock<NodeCluster> = RwLock::new(NodeCluster::new());
@ -38,10 +39,6 @@ impl NodeCluster {
GLOBAL_CLUSTER.write().await
}
pub async fn terminate() {
Self::instance_mut().await.terminate = true;
}
pub async fn run() {
tokio::spawn(async {
let config = Config::now().await;
@ -61,6 +58,13 @@ impl NodeCluster {
sleep(Duration::from_millis(config.internal_timestamp)).await;
}
});
Logger::append_system_log(LogLevel::INFO, "Node Cluster: Online.".to_string()).await;
}
pub async fn terminate() {
Logger::append_system_log(LogLevel::INFO, "Node Cluster: Terminating.".to_string()).await;
Self::instance_mut().await.terminate = true;
Logger::append_system_log(LogLevel::INFO, "Node Cluster: Termination complete.".to_string()).await;
}
pub async fn add_node(node: Node) {

36
library/src/web/server.rs
View File

@ -1,26 +1,43 @@
use uuid::Uuid;
use lazy_static::lazy_static;
use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
use crate::manager::node::Node;
use crate::utils::logger::{Logger, LogLevel};
use crate::manager::node_cluster::NodeCluster;
use crate::connection::socket::node_socket::NodeSocket;
lazy_static!{
static ref GLOBAL_SERVER: RwLock<Server> = RwLock::new(Server::new());
}
pub struct Server {
node_socket: NodeSocket,
terminate: bool,
}
impl Server {
async fn new() -> Self {
Logger::append_system_log(LogLevel::INFO, "Server online.".to_string()).await;
fn new() -> Self {
Self {
node_socket: NodeSocket::new().await,
terminate: false,
}
}
pub fn run(mut self) {
pub async fn instance() -> RwLockReadGuard<'static, Self> {
GLOBAL_SERVER.read().await
}
pub async fn instance_mut() -> RwLockWriteGuard<'static, Self> {
GLOBAL_SERVER.write().await
}
pub async fn run() {
tokio::spawn(async move {
loop {
if Self::instance().await.terminate {
return;
}
let node_id = Uuid::new_v4();
let (socket_stream, node_ip) = self.node_socket.get_connection().await;
let mut node_socket = NodeSocket::new().await;
let (socket_stream, node_ip) = node_socket.get_connection().await;
let node = Node::new(node_id, socket_stream).await;
if let Some(node) = node {
NodeCluster::add_node(node).await;
@ -28,5 +45,12 @@ impl Server {
}
}
});
Logger::append_system_log(LogLevel::INFO, "Server: Online.".to_string()).await;
}
pub async fn terminate() {
Logger::append_system_log(LogLevel::INFO, "Server: Terminating.".to_string()).await;
Self::instance_mut().await.terminate = true;
Logger::append_system_log(LogLevel::INFO, "Server: Termination complete.".to_string()).await;
}
}