weyne/rustdesk
1
0
Fork 0
mirror of https://github.com/weyne85/rustdesk.git synced 2025-10-29 17:00:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into dev

Browse Source
This commit is contained in:
xxrl
2022-11-20 13:27:39 +08:00
parent 4ce4830274 8b20237096
commit edfb22d9c2
107 changed files with 5470 additions and 1631 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
src/server/connection.rs
View File

@@ -26,10 +26,9 @@ use hbb_common::{
use scrap::android::call_main_service_mouse_input;
use serde_json::{json, value::Value};
use sha2::{Digest, Sha256};
use std::sync::{
atomic::{AtomicI64, Ordering},
mpsc as std_mpsc,
};
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use std::sync::atomic::Ordering;
use std::sync::{atomic::AtomicI64, mpsc as std_mpsc};
#[cfg(not(any(target_os = "android", target_os = "ios")))]
use system_shutdown;
@@ -234,8 +233,12 @@ impl Connection {
#[cfg(not(any(target_os = "android", target_os = "ios")))]
std::thread::spawn(move || Self::handle_input(rx_input, tx_cloned));
let mut second_timer = time::interval(Duration::from_secs(1));
#[cfg(windows)]
let mut last_uac = false;
#[cfg(windows)]
let mut last_foreground_window_elevated = false;
#[cfg(windows)]
let is_installed = crate::platform::is_installed();
loop {
tokio::select! {
@@ -339,6 +342,12 @@ impl Connection {
};
conn.send(msg_out).await;
}
#[cfg(windows)]
ipc::Data::DataPortableService(ipc::DataPortableService::RequestStart) => {
if let Err(e) = crate::portable_service::client::start_portable_service() {
log::error!("Failed to start portable service from cm:{:?}", e);
}
}
_ => {}
}
},
@@ -415,23 +424,36 @@ impl Connection {
}
},
_ = second_timer.tick() => {
let uac = crate::video_service::IS_UAC_RUNNING.lock().unwrap().clone();
if last_uac != uac {
last_uac = uac;
let mut misc = Misc::new();
misc.set_uac(uac);
let mut msg = Message::new();
msg.set_misc(misc);
conn.inner.send(msg.into());
}
let foreground_window_elevated = crate::video_service::IS_FOREGROUND_WINDOW_ELEVATED.lock().unwrap().clone();
if last_foreground_window_elevated != foreground_window_elevated {
last_foreground_window_elevated = foreground_window_elevated;
let mut misc = Misc::new();
misc.set_foreground_window_elevated(foreground_window_elevated);
let mut msg = Message::new();
msg.set_misc(misc);
conn.inner.send(msg.into());
#[cfg(windows)]
{
if !is_installed {
let portable_service_running = crate::portable_service::client::PORTABLE_SERVICE_RUNNING.lock().unwrap().clone();
let uac = crate::video_service::IS_UAC_RUNNING.lock().unwrap().clone();
if last_uac != uac {
last_uac = uac;
if !uac || !portable_service_running{
let mut misc = Misc::new();
misc.set_uac(uac);
let mut msg = Message::new();
msg.set_misc(misc);
conn.inner.send(msg.into());
}
}
let foreground_window_elevated = crate::video_service::IS_FOREGROUND_WINDOW_ELEVATED.lock().unwrap().clone();
if last_foreground_window_elevated != foreground_window_elevated {
last_foreground_window_elevated = foreground_window_elevated;
if !foreground_window_elevated || !portable_service_running {
let mut misc = Misc::new();
misc.set_foreground_window_elevated(foreground_window_elevated);
let mut msg = Message::new();
msg.set_misc(misc);
conn.inner.send(msg.into());
}
}
let show_elevation = !portable_service_running;
conn.send_to_cm(ipc::Data::DataPortableService(ipc::DataPortableService::CmShowElevation(show_elevation)));
}
}
}
_ = test_delay_timer.tick() => {

8
src/server/dbus.rs
View File

@@ -6,7 +6,9 @@
use dbus::blocking::Connection;
use dbus_crossroads::{Crossroads, IfaceBuilder};
use hbb_common::{log};
use std::{error::Error, fmt, time::Duration, collections::HashMap};
use std::{error::Error, fmt, time::Duration};
#[cfg(feature = "flutter")]
use std::collections::HashMap;
const DBUS_NAME: &str = "org.rustdesk.rustdesk";
const DBUS_PREFIX: &str = "/dbus";
@@ -65,7 +67,7 @@ fn handle_client_message(builder: &mut IfaceBuilder<()>) {
DBUS_METHOD_NEW_CONNECTION,
(DBUS_METHOD_NEW_CONNECTION_ID,),
(DBUS_METHOD_RETURN,),
move |_, _, (peer_id,): (String,)| {
move |_, _, (_peer_id,): (String,)| {
#[cfg(feature = "flutter")]
{
use crate::flutter::{self, APP_TYPE_MAIN};
@@ -77,7 +79,7 @@ fn handle_client_message(builder: &mut IfaceBuilder<()>) {
{
let data = HashMap::from([
("name", "new_connection"),
("peer_id", peer_id.as_str())
("peer_id", _peer_id.as_str())
]);
if !stream.add(serde_json::ser::to_string(&data).unwrap_or("".to_string())) {
log::error!("failed to add dbus message to flutter global dbus stream.");

116
src/server/input_service.rs
View File

@@ -13,6 +13,8 @@ use std::{
time::Instant,
};
const INVALID_CURSOR_POS: i32 = i32::MIN;
#[derive(Default)]
struct StateCursor {
hcursor: u64,
@@ -28,14 +30,33 @@ impl super::service::Reset for StateCursor {
}
}
#[derive(Default)]
struct StatePos {
cursor_pos: (i32, i32),
}
impl Default for StatePos {
fn default() -> Self {
Self {
cursor_pos: (INVALID_CURSOR_POS, INVALID_CURSOR_POS),
}
}
}
impl super::service::Reset for StatePos {
fn reset(&mut self) {
self.cursor_pos = (0, 0);
self.cursor_pos = (INVALID_CURSOR_POS, INVALID_CURSOR_POS);
}
}
impl StatePos {
#[inline]
fn is_valid(&self) -> bool {
self.cursor_pos.0 != INVALID_CURSOR_POS
}
#[inline]
fn is_moved(&self, x: i32, y: i32) -> bool {
self.is_valid() && (self.cursor_pos.0 != x || self.cursor_pos.1 != y)
}
}
@@ -105,7 +126,7 @@ pub fn new_pos() -> GenericService {
}
fn update_last_cursor_pos(x: i32, y: i32) {
let mut lock = LATEST_CURSOR_POS.lock().unwrap();
let mut lock = LATEST_SYS_CURSOR_POS.lock().unwrap();
if lock.1 .0 != x || lock.1 .1 != y {
(lock.0, lock.1) = (Instant::now(), (x, y))
}
@@ -114,8 +135,7 @@ fn update_last_cursor_pos(x: i32, y: i32) {
fn run_pos(sp: GenericService, state: &mut StatePos) -> ResultType<()> {
if let Some((x, y)) = crate::get_cursor_pos() {
update_last_cursor_pos(x, y);
if state.cursor_pos.0 != x || state.cursor_pos.1 != y {
state.cursor_pos = (x, y);
if state.is_moved(x, y) {
let mut msg_out = Message::new();
msg_out.set_cursor_position(CursorPosition {
x,
@@ -124,7 +144,7 @@ fn run_pos(sp: GenericService, state: &mut StatePos) -> ResultType<()> {
});
let exclude = {
let now = get_time();
let lock = LATEST_INPUT_CURSOR.lock().unwrap();
let lock = LATEST_PEER_INPUT_CURSOR.lock().unwrap();
if now - lock.time < 300 {
lock.conn
} else {
@@ -133,6 +153,7 @@ fn run_pos(sp: GenericService, state: &mut StatePos) -> ResultType<()> {
};
sp.send_without(msg_out, exclude);
}
state.cursor_pos = (x, y);
}
sp.snapshot(|sps| {
@@ -182,12 +203,13 @@ lazy_static::lazy_static! {
Arc::new(Mutex::new(Enigo::new()))
};
static ref KEYS_DOWN: Arc<Mutex<HashMap<u64, Instant>>> = Default::default();
static ref LATEST_INPUT_CURSOR: Arc<Mutex<Input>> = Default::default();
static ref LATEST_CURSOR_POS: Arc<Mutex<(Instant, (i32, i32))>> = Arc::new(Mutex::new((Instant::now().sub(MOUSE_MOVE_PROTECTION_TIMEOUT), (0, 0))));
static ref LATEST_PEER_INPUT_CURSOR: Arc<Mutex<Input>> = Default::default();
static ref LATEST_SYS_CURSOR_POS: Arc<Mutex<(Instant, (i32, i32))>> = Arc::new(Mutex::new((Instant::now().sub(MOUSE_MOVE_PROTECTION_TIMEOUT), (0, 0))));
}
static EXITING: AtomicBool = AtomicBool::new(false);
const MOUSE_MOVE_PROTECTION_TIMEOUT: Duration = Duration::from_millis(1_000);
// Actual diff of (x,y) is (1,1) here. But 5 may be tolerant.
const MOUSE_ACTIVE_DISTANCE: i32 = 5;
// mac key input must be run in main thread, otherwise crash on >= osx 10.15
@@ -258,14 +280,30 @@ fn get_modifier_state(key: Key, en: &mut Enigo) -> bool {
}
pub fn handle_mouse(evt: &MouseEvent, conn: i32) {
if !active_mouse_(conn) {
return;
}
let evt_type = evt.mask & 0x7;
if evt_type == 0 {
let time = get_time();
*LATEST_PEER_INPUT_CURSOR.lock().unwrap() = Input {
time,
conn,
x: evt.x,
y: evt.y,
};
}
#[cfg(target_os = "macos")]
if !*IS_SERVER {
// having GUI, run main GUI thread, otherwise crash
let evt = evt.clone();
QUEUE.exec_async(move || handle_mouse_(&evt, conn));
QUEUE.exec_async(move || handle_mouse_(&evt));
return;
}
handle_mouse_(evt, conn);
#[cfg(windows)]
crate::portable_service::client::handle_mouse(evt);
#[cfg(not(windows))]
handle_mouse_(evt);
}
pub fn fix_key_down_timeout_loop() {
@@ -373,55 +411,58 @@ fn fix_modifiers(modifiers: &[EnumOrUnknown<ControlKey>], en: &mut Enigo, ck: i3
}
}
fn is_mouse_active_by_conn(conn: i32) -> bool {
fn active_mouse_(conn: i32) -> bool {
// out of time protection
if LATEST_CURSOR_POS.lock().unwrap().0.elapsed() > MOUSE_MOVE_PROTECTION_TIMEOUT {
if LATEST_SYS_CURSOR_POS.lock().unwrap().0.elapsed() > MOUSE_MOVE_PROTECTION_TIMEOUT {
return true;
}
let mut last_input = LATEST_INPUT_CURSOR.lock().unwrap();
// last conn input may be protected
if last_input.conn != conn {
if LATEST_PEER_INPUT_CURSOR.lock().unwrap().conn != conn {
return false;
}
// check if input is in valid range
let in_actived_dist = |a: i32, b: i32| -> bool { (a - b).abs() < MOUSE_ACTIVE_DISTANCE };
// Check if input is in valid range
match crate::get_cursor_pos() {
Some((x, y)) => {
let is_same_input = (last_input.x - x).abs() < MOUSE_ACTIVE_DISTANCE
&& (last_input.y - y).abs() < MOUSE_ACTIVE_DISTANCE;
if !is_same_input {
last_input.x = -MOUSE_ACTIVE_DISTANCE * 2;
last_input.y = -MOUSE_ACTIVE_DISTANCE * 2;
let (last_in_x, last_in_y) = {
let lock = LATEST_PEER_INPUT_CURSOR.lock().unwrap();
(lock.x, lock.y)
};
let mut can_active = in_actived_dist(last_in_x, x) && in_actived_dist(last_in_y, y);
// The cursor may not have been moved to last input position if system is busy now.
// While this is not a common case, we check it again after some time later.
if !can_active {
// 10 micros may be enough for system to move cursor.
// We do not care about the situation which system is too slow(more than 10 micros is required).
std::thread::sleep(std::time::Duration::from_micros(10));
// Sleep here can also somehow suppress delay accumulation.
if let Some((x2, y2)) = crate::get_cursor_pos() {
can_active = in_actived_dist(last_in_x, x2) && in_actived_dist(last_in_y, y2);
}
}
is_same_input
if !can_active {
let mut lock = LATEST_PEER_INPUT_CURSOR.lock().unwrap();
lock.x = INVALID_CURSOR_POS / 2;
lock.y = INVALID_CURSOR_POS / 2;
}
can_active
}
None => true,
}
}
fn handle_mouse_(evt: &MouseEvent, conn: i32) {
pub fn handle_mouse_(evt: &MouseEvent) {
if EXITING.load(Ordering::SeqCst) {
return;
}
if !is_mouse_active_by_conn(conn) {
return;
}
#[cfg(windows)]
crate::platform::windows::try_change_desktop();
let buttons = evt.mask >> 3;
let evt_type = evt.mask & 0x7;
if evt_type == 0 {
let time = get_time();
*LATEST_INPUT_CURSOR.lock().unwrap() = Input {
time,
conn,
x: evt.x,
y: evt.y,
};
}
let mut en = ENIGO.lock().unwrap();
#[cfg(not(target_os = "macos"))]
let mut to_release = Vec::new();
@@ -677,6 +718,9 @@ pub fn handle_key(evt: &KeyEvent) {
QUEUE.exec_async(move || handle_key_(&evt));
return;
}
#[cfg(windows)]
crate::portable_service::client::handle_key(evt);
#[cfg(not(windows))]
handle_key_(evt);
}
@@ -928,7 +972,7 @@ fn legacy_keyboard_mode(evt: &KeyEvent) {
}
}
fn handle_key_(evt: &KeyEvent) {
pub fn handle_key_(evt: &KeyEvent) {
if EXITING.load(Ordering::SeqCst) {
return;
}

800
src/server/portable_service.rs Normal file
View File

@@ -0,0 +1,800 @@
use core::slice;
use hbb_common::{
allow_err,
anyhow::anyhow,
bail,
config::Config,
log,
message_proto::{KeyEvent, MouseEvent},
protobuf::Message,
tokio::{self, sync::mpsc},
ResultType,
};
use scrap::{Capturer, Frame, TraitCapturer};
use shared_memory::*;
use std::{
mem::size_of,
ops::{Deref, DerefMut},
sync::{Arc, Mutex},
time::Duration,
};
use winapi::{
shared::minwindef::{BOOL, FALSE, TRUE},
um::winuser::{self, CURSORINFO, PCURSORINFO},
};
use crate::{
ipc::{self, new_listener, Connection, Data, DataPortableService},
video_service::get_current_display,
};
use super::video_qos;
const SIZE_COUNTER: usize = size_of::<i32>() * 2;
const FRAME_ALIGN: usize = 64;
const ADDR_CURSOR_PARA: usize = 0;
const ADDR_CURSOR_COUNTER: usize = ADDR_CURSOR_PARA + size_of::<CURSORINFO>();
const ADDR_CAPTURER_PARA: usize = ADDR_CURSOR_COUNTER + SIZE_COUNTER;
const ADDR_CAPTURE_FRAME_SIZE: usize = ADDR_CAPTURER_PARA + size_of::<CapturerPara>();
const ADDR_CAPTURE_WOULDBLOCK: usize = ADDR_CAPTURE_FRAME_SIZE + size_of::<i32>();
const ADDR_CAPTURE_FRAME_COUNTER: usize = ADDR_CAPTURE_WOULDBLOCK + size_of::<i32>();
const ADDR_CAPTURE_FRAME: usize =
(ADDR_CAPTURE_FRAME_COUNTER + SIZE_COUNTER + FRAME_ALIGN - 1) / FRAME_ALIGN * FRAME_ALIGN;
const IPC_PROFIX: &str = "_portable_service";
pub const SHMEM_NAME: &str = "_portable_service";
const MAX_NACK: usize = 3;
const MAX_DXGI_FAIL_TIME: usize = 5;
pub struct SharedMemory {
inner: Shmem,
}
unsafe impl Send for SharedMemory {}
unsafe impl Sync for SharedMemory {}
impl Deref for SharedMemory {
type Target = Shmem;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl DerefMut for SharedMemory {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.inner
}
}
impl SharedMemory {
pub fn create(name: &str, size: usize) -> ResultType<Self> {
let flink = Self::flink(name.to_string());
let shmem = match ShmemConf::new()
.size(size)
.flink(&flink)
.force_create_flink()
.create()
{
Ok(m) => m,
Err(ShmemError::LinkExists) => {
bail!(
"Unable to force create shmem flink {}, which should not happen.",
flink
)
}
Err(e) => {
bail!("Unable to create shmem flink {} : {}", flink, e);
}
};
log::info!("Create shared memory, size:{}, flink:{}", size, flink);
Self::set_all_perm(&flink);
Ok(SharedMemory { inner: shmem })
}
pub fn open_existing(name: &str) -> ResultType<Self> {
let flink = Self::flink(name.to_string());
let shmem = match ShmemConf::new().flink(&flink).allow_raw(true).open() {
Ok(m) => m,
Err(e) => {
bail!("Unable to open existing shmem flink {} : {}", flink, e);
}
};
log::info!("open existing shared memory, flink:{:?}", flink);
Ok(SharedMemory { inner: shmem })
}
pub fn write(&self, addr: usize, data: &[u8]) {
unsafe {
assert!(addr + data.len() <= self.inner.len());
let ptr = self.inner.as_ptr().add(addr);
let shared_mem_slice = slice::from_raw_parts_mut(ptr, data.len());
shared_mem_slice.copy_from_slice(data);
}
}
fn flink(name: String) -> String {
let mut shmem_flink = format!("shared_memory{}", name);
if cfg!(windows) {
let df = "C:\\ProgramData";
let df = if std::path::Path::new(df).exists() {
df.to_owned()
} else {
std::env::var("TEMP").unwrap_or("C:\\Windows\\TEMP".to_owned())
};
let df = format!("{}\\{}", df, *hbb_common::config::APP_NAME.read().unwrap());
std::fs::create_dir(&df).ok();
shmem_flink = format!("{}\\{}", df, shmem_flink);
} else {
shmem_flink = Config::ipc_path("").replace("ipc", "") + &shmem_flink;
}
return shmem_flink;
}
fn set_all_perm(_p: &str) {
#[cfg(not(windows))]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(_p, std::fs::Permissions::from_mode(0o0777)).ok();
}
}
}
mod utils {
use core::slice;
use std::mem::size_of;
pub fn i32_to_vec(i: i32) -> Vec<u8> {
i.to_ne_bytes().to_vec()
}
pub fn ptr_to_i32(ptr: *const u8) -> i32 {
unsafe {
let v = slice::from_raw_parts(ptr, size_of::<i32>());
i32::from_ne_bytes([v[0], v[1], v[2], v[3]])
}
}
pub fn counter_ready(counter: *const u8) -> bool {
unsafe {
let wptr = counter;
let rptr = counter.add(size_of::<i32>());
let iw = ptr_to_i32(wptr);
let ir = ptr_to_i32(rptr);
if ir != iw {
std::ptr::copy_nonoverlapping(wptr, rptr as *mut _, size_of::<i32>());
true
} else {
false
}
}
}
pub fn counter_equal(counter: *const u8) -> bool {
unsafe {
let wptr = counter;
let rptr = counter.add(size_of::<i32>());
let iw = ptr_to_i32(wptr);
let ir = ptr_to_i32(rptr);
iw == ir
}
}
pub fn increase_counter(counter: *mut u8) {
unsafe {
let wptr = counter;
let rptr = counter.add(size_of::<i32>());
let iw = ptr_to_i32(counter);
let ir = ptr_to_i32(counter);
let v = i32_to_vec(iw + 1);
std::ptr::copy_nonoverlapping(v.as_ptr(), wptr, size_of::<i32>());
if ir == iw + 1 {
let v = i32_to_vec(iw);
std::ptr::copy_nonoverlapping(v.as_ptr(), rptr, size_of::<i32>());
}
}
}
pub fn align(v: usize, align: usize) -> usize {
(v + align - 1) / align * align
}
}
// functions called in seperate SYSTEM user process.
pub mod server {
use super::*;
lazy_static::lazy_static! {
static ref EXIT: Arc<Mutex<bool>> = Default::default();
}
pub fn run_portable_service() {
let shmem = Arc::new(SharedMemory::open_existing(SHMEM_NAME).unwrap());
let shmem1 = shmem.clone();
let shmem2 = shmem.clone();
let mut threads = vec![];
threads.push(std::thread::spawn(|| {
run_get_cursor_info(shmem1);
}));
threads.push(std::thread::spawn(|| {
run_capture(shmem2);
}));
threads.push(std::thread::spawn(|| {
run_ipc_client();
}));
threads.push(std::thread::spawn(|| {
run_exit_check();
}));
for th in threads.drain(..) {
th.join().unwrap();
log::info!("thread joined");
}
}
fn run_exit_check() {
loop {
if EXIT.lock().unwrap().clone() {
std::thread::sleep(Duration::from_millis(50));
std::process::exit(0);
}
std::thread::sleep(Duration::from_millis(50));
}
}
fn run_get_cursor_info(shmem: Arc<SharedMemory>) {
loop {
if EXIT.lock().unwrap().clone() {
break;
}
unsafe {
let para = shmem.as_ptr().add(ADDR_CURSOR_PARA) as *mut CURSORINFO;
(*para).cbSize = size_of::<CURSORINFO>() as _;
let result = winuser::GetCursorInfo(para);
if result == TRUE {
utils::increase_counter(shmem.as_ptr().add(ADDR_CURSOR_COUNTER));
}
}
// more frequent in case of `Error of mouse_cursor service`
std::thread::sleep(Duration::from_millis(15));
}
}
fn run_capture(shmem: Arc<SharedMemory>) {
let mut c = None;
let mut last_current_display = usize::MAX;
let mut last_use_yuv = false;
let mut last_timeout_ms: i32 = 33;
let mut spf = Duration::from_millis(last_timeout_ms as _);
let mut first_frame_captured = false;
let mut dxgi_failed_times = 0;
loop {
if EXIT.lock().unwrap().clone() {
break;
}
unsafe {
let para_ptr = shmem.as_ptr().add(ADDR_CAPTURER_PARA);
let para = para_ptr as *const CapturerPara;
let current_display = (*para).current_display;
let use_yuv = (*para).use_yuv;
let use_yuv_set = (*para).use_yuv_set;
let timeout_ms = (*para).timeout_ms;
if !use_yuv_set {
c = None;
std::thread::sleep(spf);
continue;
}
if c.is_none() {
*crate::video_service::CURRENT_DISPLAY.lock().unwrap() = current_display;
let (_, _current, display) = get_current_display().unwrap();
match Capturer::new(display, use_yuv) {
Ok(mut v) => {
c = {
last_current_display = current_display;
last_use_yuv = use_yuv;
first_frame_captured = false;
if dxgi_failed_times > MAX_DXGI_FAIL_TIME {
dxgi_failed_times = 0;
v.set_gdi();
}
Some(v)
}
}
Err(e) => {
log::error!("Failed to create gdi capturer:{:?}", e);
std::thread::sleep(std::time::Duration::from_secs(1));
continue;
}
}
} else {
if current_display != last_current_display || use_yuv != last_use_yuv {
log::info!(
"display:{}->{}, use_yuv:{}->{}",
last_current_display,
current_display,
last_use_yuv,
use_yuv
);
c = None;
continue;
}
if timeout_ms != last_timeout_ms
&& timeout_ms >= 1000 / video_qos::MAX_FPS as i32
&& timeout_ms <= 1000 / video_qos::MIN_FPS as i32
{
last_timeout_ms = timeout_ms;
spf = Duration::from_millis(timeout_ms as _);
}
}
if first_frame_captured {
if !utils::counter_equal(shmem.as_ptr().add(ADDR_CAPTURE_FRAME_COUNTER)) {
std::thread::sleep(spf);
continue;
}
}
match c.as_mut().unwrap().frame(spf) {
Ok(f) => {
let len = f.0.len();
let len_slice = utils::i32_to_vec(len as _);
shmem.write(ADDR_CAPTURE_FRAME_SIZE, &len_slice);
shmem.write(ADDR_CAPTURE_FRAME, f.0);
shmem.write(ADDR_CAPTURE_WOULDBLOCK, &utils::i32_to_vec(TRUE));
utils::increase_counter(shmem.as_ptr().add(ADDR_CAPTURE_FRAME_COUNTER));
first_frame_captured = true;
dxgi_failed_times = 0;
}
Err(e) => {
if e.kind() != std::io::ErrorKind::WouldBlock {
// DXGI_ERROR_INVALID_CALL after each success on Microsoft GPU driver
// log::error!("capture frame failed:{:?}", e);
if crate::platform::windows::desktop_changed() {
crate::platform::try_change_desktop();
c = None;
std::thread::sleep(spf);
continue;
}
if !c.as_ref().unwrap().is_gdi() {
dxgi_failed_times += 1;
}
if dxgi_failed_times > MAX_DXGI_FAIL_TIME {
c = None;
shmem.write(ADDR_CAPTURE_WOULDBLOCK, &utils::i32_to_vec(FALSE));
std::thread::sleep(spf);
}
} else {
shmem.write(ADDR_CAPTURE_WOULDBLOCK, &utils::i32_to_vec(TRUE));
}
}
}
}
}
}
#[tokio::main(flavor = "current_thread")]
async fn run_ipc_client() {
use DataPortableService::*;
let postfix = IPC_PROFIX;
match ipc::connect(1000, postfix).await {
Ok(mut stream) => {
let mut timer = tokio::time::interval(Duration::from_secs(1));
let mut nack = 0;
loop {
tokio::select! {
res = stream.next() => {
match res {
Err(err) => {
log::error!(
"ipc{} connection closed: {}",
postfix,
err
);
break;
}
Ok(Some(Data::DataPortableService(data))) => match data {
Ping => {
allow_err!(
stream
.send(&Data::DataPortableService(Pong))
.await
);
}
Pong => {
nack = 0;
}
ConnCount(Some(n)) => {
if n == 0 {
log::info!("Connnection count equals 0, exit");
stream.send(&Data::DataPortableService(WillClose)).await.ok();
break;
}
}
Mouse(v) => {
if let Ok(evt) = MouseEvent::parse_from_bytes(&v) {
crate::input_service::handle_mouse_(&evt);
}
}
Key(v) => {
if let Ok(evt) = KeyEvent::parse_from_bytes(&v) {
crate::input_service::handle_key_(&evt);
}
}
_ => {}
},
_ => {}
}
}
_ = timer.tick() => {
nack+=1;
if nack > MAX_NACK {
log::info!("max ping nack, exit");
break;
}
stream.send(&Data::DataPortableService(Ping)).await.ok();
stream.send(&Data::DataPortableService(ConnCount(None))).await.ok();
}
}
}
}
Err(e) => {
log::error!("Failed to connect portable service ipc:{:?}", e);
}
}
*EXIT.lock().unwrap() = true;
}
}
// functions called in main process.
pub mod client {
use hbb_common::anyhow::Context;
use super::*;
lazy_static::lazy_static! {
pub static ref PORTABLE_SERVICE_RUNNING: Arc<Mutex<bool>> = Default::default();
static ref SHMEM: Arc<Mutex<Option<SharedMemory>>> = Default::default();
static ref SENDER : Mutex<mpsc::UnboundedSender<ipc::Data>> = Mutex::new(client::start_ipc_server());
}
pub(crate) fn start_portable_service() -> ResultType<()> {
log::info!("start portable service");
if PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
bail!("already running");
}
if SHMEM.lock().unwrap().is_none() {
let displays = scrap::Display::all()?;
if displays.is_empty() {
bail!("no display available!");
}
let mut max_pixel = 0;
let align = 64;
for d in displays {
let pixel = utils::align(d.width(), align) * utils::align(d.height(), align);
if max_pixel < pixel {
max_pixel = pixel;
}
}
let shmem_size = utils::align(ADDR_CAPTURE_FRAME + max_pixel * 4, align);
// os error 112, no enough space
*SHMEM.lock().unwrap() = Some(crate::portable_service::SharedMemory::create(
crate::portable_service::SHMEM_NAME,
shmem_size,
)?);
shutdown_hooks::add_shutdown_hook(drop_portable_service_shared_memory);
}
let mut option = SHMEM.lock().unwrap();
let shmem = option.as_mut().unwrap();
unsafe {
libc::memset(shmem.as_ptr() as _, 0, shmem.len() as _);
}
if crate::platform::run_background(
&std::env::current_exe()?.to_string_lossy().to_string(),
"--portable-service",
)
.is_err()
{
*SHMEM.lock().unwrap() = None;
bail!("Failed to run portable service process");
}
let _sender = SENDER.lock().unwrap();
Ok(())
}
pub extern "C" fn drop_portable_service_shared_memory() {
log::info!("drop shared memory");
*SHMEM.lock().unwrap() = None;
}
pub struct CapturerPortable;
impl CapturerPortable {
pub fn new(current_display: usize, use_yuv: bool) -> Self
where
Self: Sized,
{
let mut option = SHMEM.lock().unwrap();
let shmem = option.as_mut().unwrap();
Self::set_para(
shmem,
CapturerPara {
current_display,
use_yuv,
use_yuv_set: false,
timeout_ms: 33,
},
);
shmem.write(ADDR_CAPTURE_WOULDBLOCK, &utils::i32_to_vec(TRUE));
CapturerPortable {}
}
fn set_para(shmem: &mut SharedMemory, para: CapturerPara) {
let para_ptr = &para as *const CapturerPara as *const u8;
let para_data;
unsafe {
para_data = slice::from_raw_parts(para_ptr, size_of::<CapturerPara>());
}
shmem.write(ADDR_CAPTURER_PARA, para_data);
}
}
impl TraitCapturer for CapturerPortable {
fn set_use_yuv(&mut self, use_yuv: bool) {
let mut option = SHMEM.lock().unwrap();
let shmem = option.as_mut().unwrap();
unsafe {
let para_ptr = shmem.as_ptr().add(ADDR_CAPTURER_PARA);
let para = para_ptr as *const CapturerPara;
Self::set_para(
shmem,
CapturerPara {
current_display: (*para).current_display,
use_yuv,
use_yuv_set: true,
timeout_ms: (*para).timeout_ms,
},
);
}
}
fn frame<'a>(&'a mut self, timeout: Duration) -> std::io::Result<Frame<'a>> {
let mut option = SHMEM.lock().unwrap();
let shmem = option.as_mut().unwrap();
unsafe {
let base = shmem.as_ptr();
let para_ptr = base.add(ADDR_CAPTURER_PARA);
let para = para_ptr as *const CapturerPara;
if timeout.as_millis() != (*para).timeout_ms as _ {
Self::set_para(
shmem,
CapturerPara {
current_display: (*para).current_display,
use_yuv: (*para).use_yuv,
use_yuv_set: (*para).use_yuv_set,
timeout_ms: timeout.as_millis() as _,
},
);
}
if utils::counter_ready(base.add(ADDR_CAPTURE_FRAME_COUNTER)) {
let frame_len_ptr = base.add(ADDR_CAPTURE_FRAME_SIZE);
let frame_len = utils::ptr_to_i32(frame_len_ptr);
let frame_ptr = base.add(ADDR_CAPTURE_FRAME);
let data = slice::from_raw_parts(frame_ptr, frame_len as usize);
Ok(Frame(data))
} else {
let ptr = base.add(ADDR_CAPTURE_WOULDBLOCK);
let wouldblock = utils::ptr_to_i32(ptr);
if wouldblock == TRUE {
Err(std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"wouldblock error".to_string(),
))
} else {
Err(std::io::Error::new(
std::io::ErrorKind::Other,
"other error".to_string(),
))
}
}
}
}
// control by itself
fn is_gdi(&self) -> bool {
true
}
fn set_gdi(&mut self) -> bool {
true
}
}
pub(super) fn start_ipc_server() -> mpsc::UnboundedSender<Data> {
let (tx, rx) = mpsc::unbounded_channel::<Data>();
std::thread::spawn(move || start_ipc_server_async(rx));
tx
}
#[tokio::main(flavor = "current_thread")]
async fn start_ipc_server_async(rx: mpsc::UnboundedReceiver<Data>) {
use DataPortableService::*;
let rx = Arc::new(tokio::sync::Mutex::new(rx));
let postfix = IPC_PROFIX;
#[cfg(feature = "flutter")]
let quick_support = {
let args: Vec<_> = std::env::args().collect();
args.contains(&"--quick_support".to_string())
};
#[cfg(not(feature = "flutter"))]
let quick_support = std::env::current_exe()
.unwrap_or("".into())
.to_string_lossy()
.to_lowercase()
.ends_with("qs.exe");
match new_listener(postfix).await {
Ok(mut incoming) => loop {
{
tokio::select! {
Some(result) = incoming.next() => {
match result {
Ok(stream) => {
log::info!("Got portable service ipc connection");
let rx_clone = rx.clone();
tokio::spawn(async move {
let mut stream = Connection::new(stream);
let postfix = postfix.to_owned();
let mut timer = tokio::time::interval(Duration::from_secs(1));
let mut nack = 0;
let mut rx = rx_clone.lock().await;
loop {
tokio::select! {
res = stream.next() => {
match res {
Err(err) => {
log::info!(
"ipc{} connection closed: {}",
postfix,
err
);
break;
}
Ok(Some(Data::DataPortableService(data))) => match data {
Ping => {
stream.send(&Data::DataPortableService(Pong)).await.ok();
}
Pong => {
nack = 0;
*PORTABLE_SERVICE_RUNNING.lock().unwrap() = true;
},
ConnCount(None) => {
if !quick_support {
let cnt = crate::server::CONN_COUNT.lock().unwrap().clone();
stream.send(&Data::DataPortableService(ConnCount(Some(cnt)))).await.ok();
}
},
WillClose => {
log::info!("portable service will close");
break;
}
_=>{}
}
_=>{}
}
}
_ = timer.tick() => {
nack+=1;
if nack > MAX_NACK {
// In fact, this will not happen, ipc will be closed before max nack.
log::error!("max ipc nack");
break;
}
stream.send(&Data::DataPortableService(Ping)).await.ok();
}
Some(data) = rx.recv() => {
allow_err!(stream.send(&data).await);
}
}
}
*PORTABLE_SERVICE_RUNNING.lock().unwrap() = false;
});
}
Err(err) => {
log::error!("Couldn't get portable client: {:?}", err);
}
}
}
}
}
},
Err(err) => {
log::error!("Failed to start portable service ipc server: {}", err);
}
}
}
fn ipc_send(data: Data) -> ResultType<()> {
let sender = SENDER.lock().unwrap();
sender
.send(data)
.map_err(|e| anyhow!("ipc send error:{:?}", e))
}
fn get_cursor_info_(shmem: &mut SharedMemory, pci: PCURSORINFO) -> BOOL {
unsafe {
let shmem_addr_para = shmem.as_ptr().add(ADDR_CURSOR_PARA);
if utils::counter_ready(shmem.as_ptr().add(ADDR_CURSOR_COUNTER)) {
std::ptr::copy_nonoverlapping(shmem_addr_para, pci as _, size_of::<CURSORINFO>());
return TRUE;
}
FALSE
}
}
fn handle_mouse_(evt: &MouseEvent) -> ResultType<()> {
let mut v = vec![];
evt.write_to_vec(&mut v)?;
ipc_send(Data::DataPortableService(DataPortableService::Mouse(v)))
}
fn handle_key_(evt: &KeyEvent) -> ResultType<()> {
let mut v = vec![];
evt.write_to_vec(&mut v)?;
ipc_send(Data::DataPortableService(DataPortableService::Key(v)))
}
pub fn create_capturer(
current_display: usize,
display: scrap::Display,
use_yuv: bool,
portable_service_running: bool,
) -> ResultType<Box<dyn TraitCapturer>> {
if portable_service_running != PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
log::info!("portable service status mismatch");
}
if portable_service_running {
log::info!("Create shared memeory capturer");
return Ok(Box::new(CapturerPortable::new(current_display, use_yuv)));
} else {
log::debug!("Create capturer dxgi|gdi");
return Ok(Box::new(
Capturer::new(display, use_yuv).with_context(|| "Failed to create capturer")?,
));
}
}
pub fn get_cursor_info(pci: PCURSORINFO) -> BOOL {
if PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
get_cursor_info_(&mut SHMEM.lock().unwrap().as_mut().unwrap(), pci)
} else {
unsafe { winuser::GetCursorInfo(pci) }
}
}
pub fn handle_mouse(evt: &MouseEvent) {
if PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
handle_mouse_(evt).ok();
} else {
crate::input_service::handle_mouse_(evt);
}
}
pub fn handle_key(evt: &KeyEvent) {
if PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
handle_key_(evt).ok();
} else {
crate::input_service::handle_key_(evt);
}
}
}
#[repr(C)]
struct CapturerPara {
current_display: usize,
use_yuv: bool,
use_yuv_set: bool,
timeout_ms: i32,
}

12
src/server/video_qos.rs
View File

@@ -1,8 +1,8 @@
use super::*;
use std::time::Duration;
const FPS: u8 = 30;
const MIN_FPS: u8 = 10;
const MAX_FPS: u8 = 120;
pub const FPS: u8 = 30;
pub const MIN_FPS: u8 = 10;
pub const MAX_FPS: u8 = 120;
trait Percent {
fn as_percent(&self) -> u32;
}
@@ -225,11 +225,7 @@ impl VideoQoS {
}
pub fn check_abr_config(&mut self) -> bool {
self.enable_abr = if let Some(v) = Config2::get().options.get("enable-abr") {
v != "N"
} else {
true // default is true
};
self.enable_abr = "N" != Config::get_option("enable-abr");
self.enable_abr
}

65
src/server/video_service.rs
View File

@@ -19,21 +19,26 @@
// https://slhck.info/video/2017/03/01/rate-control.html
use super::{video_qos::VideoQoS, *};
#[cfg(windows)]
use crate::portable_service::client::PORTABLE_SERVICE_RUNNING;
use hbb_common::tokio::sync::{
mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
Mutex as TokioMutex,
};
#[cfg(not(windows))]
use scrap::Capturer;
use scrap::{
codec::{Encoder, EncoderCfg, HwEncoderConfig},
record::{Recorder, RecorderContext},
vpxcodec::{VpxEncoderConfig, VpxVideoCodecId},
Capturer, Display, TraitCapturer,
Display, TraitCapturer,
};
#[cfg(windows)]
use std::sync::Once;
use std::{
collections::HashSet,
io::ErrorKind::WouldBlock,
ops::{Deref, DerefMut},
sync::Once,
time::{self, Duration, Instant},
};
#[cfg(windows)]
@@ -48,7 +53,7 @@ pub const SCRAP_X11_REF_URL: &str = "https://rustdesk.com/docs/en/manual/linux/#
pub const NAME: &'static str = "video";
lazy_static::lazy_static! {
static ref CURRENT_DISPLAY: Arc<Mutex<usize>> = Arc::new(Mutex::new(usize::MAX));
pub static ref CURRENT_DISPLAY: Arc<Mutex<usize>> = Arc::new(Mutex::new(usize::MAX));
static ref LAST_ACTIVE: Arc<Mutex<Instant>> = Arc::new(Mutex::new(Instant::now()));
static ref SWITCH: Arc<Mutex<bool>> = Default::default();
static ref FRAME_FETCHED_NOTIFIER: (UnboundedSender<(i32, Option<Instant>)>, Arc<TokioMutex<UnboundedReceiver<(i32, Option<Instant>)>>>) = {
@@ -187,6 +192,8 @@ fn create_capturer(
privacy_mode_id: i32,
display: Display,
use_yuv: bool,
current: usize,
_portable_service_running: bool,
) -> ResultType<Box<dyn TraitCapturer>> {
#[cfg(not(windows))]
let c: Option<Box<dyn TraitCapturer>> = None;
@@ -243,17 +250,23 @@ fn create_capturer(
}
}
let c = match c {
Some(c1) => c1,
match c {
Some(c1) => return Ok(c1),
None => {
let c1 =
Capturer::new(display, use_yuv).with_context(|| "Failed to create capturer")?;
log::debug!("Create capturer dxgi|gdi");
Box::new(c1)
#[cfg(windows)]
return crate::portable_service::client::create_capturer(
current,
display,
use_yuv,
_portable_service_running,
);
#[cfg(not(windows))]
return Ok(Box::new(
Capturer::new(display, use_yuv).with_context(|| "Failed to create capturer")?,
));
}
};
Ok(c)
}
#[cfg(windows)]
@@ -276,8 +289,8 @@ fn ensure_close_virtual_device() -> ResultType<()> {
pub fn test_create_capturer(privacy_mode_id: i32, timeout_millis: u64) -> bool {
let test_begin = Instant::now();
while test_begin.elapsed().as_millis() < timeout_millis as _ {
if let Ok((_, _, display)) = get_current_display() {
if let Ok(_) = create_capturer(privacy_mode_id, display, true) {
if let Ok((_, current, display)) = get_current_display() {
if let Ok(_) = create_capturer(privacy_mode_id, display, true, current, false) {
return true;
}
}
@@ -326,7 +339,7 @@ impl DerefMut for CapturerInfo {
}
}
fn get_capturer(use_yuv: bool) -> ResultType<CapturerInfo> {
fn get_capturer(use_yuv: bool, portable_service_running: bool) -> ResultType<CapturerInfo> {
#[cfg(target_os = "linux")]
{
if !scrap::is_x11() {
@@ -368,7 +381,13 @@ fn get_capturer(use_yuv: bool) -> ResultType<CapturerInfo> {
} else {
log::info!("In privacy mode, the peer side cannot watch the screen");
}
let capturer = create_capturer(captuerer_privacy_mode_id, display, use_yuv)?;
let capturer = create_capturer(
captuerer_privacy_mode_id,
display,
use_yuv,
current,
portable_service_running,
)?;
Ok(CapturerInfo {
origin,
width,
@@ -388,8 +407,12 @@ fn run(sp: GenericService) -> ResultType<()> {
// ensure_inited() is needed because release_resouce() may be called.
#[cfg(target_os = "linux")]
super::wayland::ensure_inited()?;
#[cfg(windows)]
let last_portable_service_running = PORTABLE_SERVICE_RUNNING.lock().unwrap().clone();
#[cfg(not(windows))]
let last_portable_service_running = false;
let mut c = get_capturer(true)?;
let mut c = get_capturer(true, last_portable_service_running)?;
let mut video_qos = VIDEO_QOS.lock().unwrap();
video_qos.set_size(c.width as _, c.height as _);
@@ -497,10 +520,16 @@ fn run(sp: GenericService) -> ResultType<()> {
if codec_name != Encoder::current_hw_encoder_name() {
bail!("SWITCH");
}
#[cfg(windows)]
if last_portable_service_running != PORTABLE_SERVICE_RUNNING.lock().unwrap().clone() {
bail!("SWITCH");
}
check_privacy_mode_changed(&sp, c.privacy_mode_id)?;
#[cfg(windows)]
{
if crate::platform::windows::desktop_changed() {
if crate::platform::windows::desktop_changed()
&& !PORTABLE_SERVICE_RUNNING.lock().unwrap().clone()
{
bail!("Desktop changed");
}
}
@@ -529,7 +558,7 @@ fn run(sp: GenericService) -> ResultType<()> {
frame_controller.set_send(now, send_conn_ids);
}
scrap::Frame::RAW(data) => {
if (data.len() != 0) {
if data.len() != 0 {
let send_conn_ids =
handle_one_frame(&sp, data, ms, &mut encoder, recorder.clone())?;
frame_controller.set_send(now, send_conn_ids);
@@ -873,7 +902,7 @@ pub(super) fn get_current_display_2(mut all: Vec<Display>) -> ResultType<(usize,
return Ok((n, current, all.remove(current)));
}
fn get_current_display() -> ResultType<(usize, usize, Display)> {
pub fn get_current_display() -> ResultType<(usize, usize, Display)> {
get_current_display_2(try_get_displays()?)
}