Update dependencies and refactor

Cargo.toml

@@ -9,25 +9,24 @@ name = "deucalion"
 crate-type = ["dylib"]
 
 [dependencies]
-tokio = { version = "0.2", features = ["io-util", "sync", "rt-core", "rt-threaded", "blocking"] }
-tokio-util = { version = "0.2", features = ["codec"] }
-parity-tokio-ipc = "0.7"
+tokio = { version = "1.25.0", features = ["io-util", "sync", "rt", "rt-multi-thread", "macros"] }
+tokio-util = { version = "0.7.4", features = ["codec", "compat"] }
+parity-tokio-ipc = "0.9"
 futures = "0.3"
-failure = "0.1.6"
-failure_derive = "0.1.6"
-bytes = "0.5"
-stream-cancel = "0.5"
+failure = "0.1.8"
+failure_derive = "0.1.8"
+bytes = "1.4.0"
+stream-cancel = "0.8.1"
 log = "0.4"
-simplelog = "^0.7.4"
-crossbeam-channel = "0.4"
-detour = "0.8.0"
-parking_lot = "0.10.0"
-pelite = "0.9"
-twoway = "0.2.1"
-region = "2.2.0"
+simplelog = "0.12"
+detour = { git = "https://github.com/Hpmason/detour-rs", branch = "fix-nightly1.67.0-changes" }
+parking_lot = "0.12.1"
+pelite = "0.10"
+twoway = "0.2.2"
+region = "3.0"
+once_cell = "1.17"
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = [
-  "minwindef", "minwinbase", "processthreadsapi", "winerror",
-  "winsock2", "libloaderapi", "consoleapi", "wincon"
+  "minwindef", "minwinbase", "processthreadsapi", "libloaderapi", "consoleapi", "wincon"
 ] }

src/hook/mod.rs

@@ -1,5 +1,12 @@
-use crossbeam_channel as channel;
 use failure::Error;
+use tokio::sync::mpsc;
+
+use crate::procloader::{get_ffxiv_handle, sig_scan_helper};
+use pelite::pattern;
+use pelite::pe::PeView;
+
+use std::sync::Arc;
+use tokio::sync::Mutex;
 
 use crate::rpc;
 
@@ -9,23 +16,32 @@ mod waitgroup;
 pub struct State {
     rzp_hook: recvzonepacket::Hook,
     wg: waitgroup::WaitGroup,
-    pub broadcast_rx: channel::Receiver<rpc::Payload>,
+    pub broadcast_rx: Arc<Mutex<mpsc::UnboundedReceiver<rpc::Payload>>>,
 }
 
 impl State {
     pub fn new() -> Result<State, Error> {
-        let (broadcast_tx, broadcast_rx) = channel::unbounded::<rpc::Payload>();
+        let (broadcast_tx, broadcast_rx) = mpsc::unbounded_channel::<rpc::Payload>();
 
         let wg = waitgroup::WaitGroup::new();
         let hs = State {
             rzp_hook: recvzonepacket::Hook::new(broadcast_tx.clone(), wg.clone())?,
             wg,
-            broadcast_rx,
+            broadcast_rx: Arc::new(Mutex::new(broadcast_rx)),
         };
-        hs.rzp_hook.setup()?;
         Ok(hs)
     }
 
+    pub fn initialize_recv_hook(&self, sig_str: String) -> Result<(), Error> {
+        let pat = pattern::parse(&sig_str)?;
+        let sig: &[pattern::Atom] = &pat;
+        let handle_ffxiv = get_ffxiv_handle()?;
+        let pe_view = unsafe { PeView::module(handle_ffxiv) };
+        let recvzonepacket_rva = sig_scan_helper("RecvZonePacket", sig, pe_view, 1)?;
+
+        self.rzp_hook.setup(recvzonepacket_rva)
+    }
+
     pub fn shutdown(&self) {
         self.rzp_hook.shutdown();
         // Wait for any hooks to finish what they're doing

src/hook/recvzonepacket.rs

@@ -1,82 +1,77 @@
 use std::mem;
 use std::ptr;
 
-use parking_lot::RwLock;
-use std::sync::Arc;
+use once_cell::sync::OnceCell;
 
-use failure::Error;
+use failure::{format_err, Error};
 
-use crossbeam_channel as channel;
 use detour;
 use detour::static_detour;
+use tokio::sync::mpsc;
 
 use crate::hook::waitgroup;
 use crate::rpc;
 
-use pelite::pattern as pat;
-use pelite::pe64::PeView;
-
-use crate::procloader::{get_ffxiv_handle, sig_scan_helper};
+use crate::procloader::get_ffxiv_handle;
 
 use log::info;
 
 static_detour! {
     static RecvZonePacket: unsafe extern "system" fn(*const u8, *const usize) -> usize;
 }
 
-const RECVZONEPACKET_SIG: &[pat::Atom] = pat!("E8 $ { ' } 84 C0 0F 85 ? ? ? ? 44 0F B6 64 24 ?");
+// const RECVZONEPACKET_SIG: &[pat::Atom] = pat!("E8 $ { ' } 84 C0 0F 85 ? ? ? ? 44 0F B6 64 24 ?");
 
 #[derive(Clone)]
 pub struct Hook {
-    call_home_tx: channel::Sender<rpc::Payload>,
-
-    hook: Arc<
-        RwLock<
-            &'static detour::StaticDetour<
-                unsafe extern "system" fn(*const u8, *const usize) -> usize,
-            >,
-        >,
+    data_tx: mpsc::UnboundedSender<rpc::Payload>,
+
+    hook: OnceCell<
+        &'static detour::StaticDetour<unsafe extern "system" fn(*const u8, *const usize) -> usize>,
     >,
     wg: waitgroup::WaitGroup,
 }
 
 impl Hook {
     pub fn new(
-        call_home_tx: channel::Sender<rpc::Payload>,
+        data_tx: mpsc::UnboundedSender<rpc::Payload>,
         wg: waitgroup::WaitGroup,
     ) -> Result<Hook, Error> {
-        let handle_ffxiv = get_ffxiv_handle()?;
-        let pe_view = unsafe { PeView::module(handle_ffxiv) };
-        let rzp_offset = sig_scan_helper("RecvZonePacket", RECVZONEPACKET_SIG, pe_view, 1)?;
-        let ptr_rzp = handle_ffxiv.wrapping_offset(rzp_offset);
-
-        let hook = unsafe {
-            let rzp: unsafe extern "system" fn(*const u8, *const usize) -> usize =
-                mem::transmute(ptr_rzp as *const ());
-            RecvZonePacket.initialize(rzp, |_, _| 0)?
-        };
         Ok(Hook {
-            call_home_tx: call_home_tx,
+            data_tx: data_tx,
 
-            hook: Arc::new(RwLock::new(hook)),
+            hook: OnceCell::new(),
             wg: wg,
         })
     }
 
-    pub fn setup(&self) -> Result<(), Error> {
+    pub fn setup(&self, recvzonepacket_rva: isize) -> Result<(), Error> {
+        let ptr_rzp = get_ffxiv_handle()?.wrapping_offset(recvzonepacket_rva);
+
+        let self_clone = self.clone();
+
+        let hook = unsafe {
+            let rzp: unsafe extern "system" fn(*const u8, *const usize) -> usize =
+                mem::transmute(ptr_rzp as *const ());
+            RecvZonePacket.initialize(rzp, move |a, b| self_clone.recv_zone_packet(a, b))?
+        };
+        self.hook
+            .set(hook)
+            .map_err(|_| format_err!("Failed to set up the hook."))?;
         unsafe {
-            let self_clone = self.clone();
-            let hook = self.hook.write();
-            hook.set_detour(move |a, b| self_clone.recv_zone_packet(a, b));
-            hook.enable()?;
+            self.hook.get_unchecked().enable()?;
         }
         Ok(())
     }
 
     unsafe fn recv_zone_packet(&self, this: *const u8, a2: *const usize) -> usize {
         let _guard = self.wg.add();
 
-        let ret = self.hook.read().call(this, a2);
+        let ret = self
+            .hook
+            .get()
+            .expect("Hook function was called without a valid hook")
+            .call(this, a2);
 
         let ptr_received_packet: *const u8 = *(a2.add(2)) as *const u8;
 
@@ -105,7 +100,7 @@ impl Hook {
         dst[4..8].clone_from_slice(&target_actor_id.to_le_bytes());
         ptr::copy(ptr_data, dst[8..].as_mut_ptr(), data_len as usize);
 
-        let _ = self.call_home_tx.send(rpc::Payload {
+        let _ = self.data_tx.send(rpc::Payload {
             op: rpc::MessageOps::Recv,
             ctx: 0,
             data: dst,
@@ -115,7 +110,9 @@ impl Hook {
 
     pub fn shutdown(&self) {
         unsafe {
-            let _ = self.hook.write().disable();
+            if let Some(hook) = self.hook.get() {
+                let _ = hook.disable();
+            };
         }
     }
 }

src/lib.rs

@@ -1,9 +1,6 @@
 use std::io::{self, Read};
 use std::panic;
 
-use channel::select;
-use crossbeam_channel as channel;
-
 #[cfg(windows)]
 use winapi::shared::minwindef::*;
 use winapi::um::libloaderapi;
@@ -15,13 +12,12 @@ use winapi::um::consoleapi;
 #[cfg(debug_assertions)]
 use winapi::um::wincon;
 
-use failure::{Error, ResultExt};
+use failure::{format_err, Error, ResultExt};
 
 use std::sync::Arc;
-use std::thread;
 
-use tokio::runtime;
-use tokio::sync::{mpsc, Mutex};
+use tokio::select;
+use tokio::sync::{mpsc, oneshot, Mutex};
 
 mod hook;
 mod procloader;
@@ -31,64 +27,70 @@ mod server;
 use log::{debug, error, info};
 use simplelog::{self, LevelFilter, SimpleLogger};
 
-fn handle_payload(payload: rpc::Payload) {
+fn handle_payload(payload: rpc::Payload, hs: Arc<hook::State>) -> Result<(), Error> {
     debug!("Received payload: {:?}", payload);
     match payload.op {
         rpc::MessageOps::Debug => {
             debug!("{:?}", payload);
         }
+        rpc::MessageOps::Recv => {
+            if let Err(e) = parse_sig_and_initialize_hook(hs, payload.data) {
+                let err = format_err!("Error setting up hook: {:?}", e);
+                debug!("{:?}", err);
+                return Err(err);
+            }
+        }
         _ => {}
     };
+    Ok(())
 }
 
-fn main_with_result() -> Result<(), Error> {
-    let pid = unsafe { processthreadsapi::GetCurrentProcessId() };
-    let pipe_name = format!(r"\\.\pipe\deucalion-{}", pid as u32);
+fn parse_sig_and_initialize_hook(hs: Arc<hook::State>, data: Vec<u8>) -> Result<(), Error> {
+    let sig_str = String::from_utf8(data)?;
+    hs.initialize_recv_hook(sig_str)
+}
+
+#[tokio::main]
+async fn main_with_result() -> Result<(), Error> {
+    let hs = Arc::new(hook::State::new().context("Error setting up the hook")?);
+    let hs_clone = hs.clone();
 
     let (signal_tx, signal_rx) = mpsc::channel(1);
     let state = Arc::new(Mutex::new(server::Shared::new(signal_tx)));
-    let mut rt = runtime::Runtime::new()?;
-
-    let msg_loop_state = state.clone();
-
-    info!("Installing hook...");
-    let hs = hook::State::new().context("Error setting up the hook")?;
-    info!("Installed hook");
-
-    let (shutdown_tx, shutdown_rx) = channel::bounded::<()>(0);
-    let msg_thread_handle = thread::spawn(move || {
-        match runtime::Builder::new().basic_scheduler().build() {
-            Ok(mut msg_loop_rt) => msg_loop_rt.block_on(async {
-                loop {
-                    select! {
-                        recv(hs.broadcast_rx) -> res => {
-                            if let Ok(payload) = res {
-                                msg_loop_state.lock().await.broadcast(payload).await;
-                            }
-                        },
-                        recv(shutdown_rx) -> _ => {
-                            hs.shutdown();
-                            return ();
-                        }
+    let state_clone = state.clone();
+
+    let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>();
+    let msg_loop_future = tokio::spawn(async move {
+        loop {
+            let mut broadcast_rx = hs.broadcast_rx.lock().await;
+            select! {
+                res = broadcast_rx.recv() => {
+                    if let Some(payload) = res {
+                        state_clone.lock().await.broadcast(payload).await;
                     }
+                },
+                _ = &mut shutdown_rx => {
+                    hs.shutdown();
+                    return ();
                 }
-            }),
-            Err(e) => error!("Error spawning tokio runtime: {:?}", e),
-        };
-        ()
+            }
+        }
     });
 
-    rt.block_on(server::run(
-        pipe_name,
-        state,
-        signal_rx,
-        move |payload: rpc::Payload| {
-            handle_payload(payload);
-        },
-    ))?;
+    let pid = unsafe { processthreadsapi::GetCurrentProcessId() };
+    let pipe_name = format!(r"\\.\pipe\deucalion-{}", pid as u32);
+
+    if let Err(e) = server::run(pipe_name, state, signal_rx, move |payload: rpc::Payload| {
+        handle_payload(payload, hs_clone.clone())
+    })
+    .await
+    {
+        error!("Server encountered error running: {:?}", e)
+    }
 
+    // Signal the msg loop to exit and shut down the hook
     drop(shutdown_tx);
-    msg_thread_handle.join().unwrap();
+    msg_loop_future.await?;
 
     info!("Hook shutdown initiated...");
     info!("Shut down!");

src/procloader.rs

@@ -9,8 +9,8 @@ use failure::{format_err, Error};
 use failure_derive::Fail;
 
 use pelite::pattern as pat;
-use pelite::pe64::image::{Rva, Va};
-use pelite::pe64::Pe;
+use pelite::pe::image::{Rva, Va};
+use pelite::pe::Pe;
 use std::mem;
 
 use std::convert::TryInto;

src/rpc.rs

@@ -79,8 +79,7 @@ impl Decoder for PayloadCodec {
     }
 }
 
-impl Encoder for PayloadCodec {
-    type Item = Payload;
+impl Encoder<Payload> for PayloadCodec {
     type Error = io::Error;
 
     fn encode(&mut self, data: Payload, mut dst: &mut BytesMut) -> Result<(), io::Error> {