EvanZhu0721 / 公开仓库

SBMS

镜像已同步

SBMS complete the calculation rules for the default logical desktop topology relationship and the physical dimensions of the displays in the Windows multi-screen collaboration system.

RustGPL-3.0main来源 commit a37d6c4b2026/10/15 07:10:15 同步

上游 GitHub

src/renderer.rs20.6 KB

1use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle};

2use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

3use std::sync::{Arc, mpsc};

4use std::thread::{self, JoinHandle};

5use std::time::{Duration, Instant};

7use windows::Win32::Foundation::{

8 HANDLE, HINSTANCE, HWND, LPARAM, LRESULT, RECT, WAIT_FAILED, WPARAM,

9};

10use windows::Win32::System::LibraryLoader::GetModuleHandleW;

11use windows::Win32::System::Threading::{CreateEventW, INFINITE, SetEvent};

12use windows::Win32::UI::WindowsAndMessaging::{

13 CS_HREDRAW, CS_VREDRAW, CreateWindowExW, DefWindowProcW, DestroyWindow, DispatchMessageW, MSG,

14 MWMO_INPUTAVAILABLE, MsgWaitForMultipleObjectsEx, PM_REMOVE, PeekMessageW, PostQuitMessage,

15 QS_ALLINPUT, RegisterClassW, SW_SHOW, ShowWindow, TranslateMessage, UnregisterClassW,

16 WINDOW_EX_STYLE, WM_CLOSE, WM_DESTROY, WM_QUIT, WNDCLASSW, WS_EX_NOACTIVATE, WS_EX_TOOLWINDOW,

17 WS_EX_TOPMOST, WS_POPUP,

18};

19use windows::core::{PCWSTR, w};

21use crate::display::Display;

22use crate::gpu_renderer::{GpuRendererConfig, GpuRendererError, run_gpu_renderer};

23use crate::input::{InputGuard, flush_movement, handle_message};

25const START_TIMEOUT: Duration = Duration::from_secs(10);

26const START_CANCEL_POLL_INTERVAL: Duration = Duration::from_millis(50);

27#[cfg(not(test))]

28const STOP_TIMEOUT: Duration = Duration::from_secs(2);

29#[cfg(test)]

30const STOP_TIMEOUT: Duration = Duration::from_millis(20);

31static WINDOW_CLASS_SEQUENCE: AtomicU64 = AtomicU64::new(1);

33#[derive(Clone, Debug)]

34pub enum RendererEvent {

35 Fps(u32),

36 TopologyLost,

37 Failed(String),

38}

40pub type RendererReporter = Arc<dyn Fn(RendererEvent) + Send + Sync + 'static>;

42pub struct Renderer {

43 stop: Arc<AtomicBool>,

44 wake_event: Arc<OwnedHandle>,

45 done: mpsc::Receiver<Result<(), String>>,

46 thread: Option<JoinHandle<()>>,

47}

49#[derive(Debug, Eq, PartialEq)]

50enum StartWait {

51 Ready,

52 Failed(String),

53 Cancelled,

54 TimedOut,

55 Disconnected,

56}

58#[derive(Debug, Eq, PartialEq)]

59pub(crate) enum RendererStartError {

60 Cancelled { cleanup_error: Option<String> },

61 Failed(String),

62}

64impl RendererStartError {

65 fn into_message(self) -> String {

66 match self {

67 Self::Cancelled {

68 cleanup_error: None,

69 } => "renderer startup was cancelled".into(),

70 Self::Cancelled {

71 cleanup_error: Some(error),

72 } => format!("renderer startup was cancelled; cleanup was incomplete: {error}"),

73 Self::Failed(error) => error,

74 }

75 }

76}

78impl Renderer {

79 pub fn start_with_reporter(

80 target: Display,

81 source: Display,

82 reporter: RendererReporter,

83 ) -> Result<Self, String> {

84 let cancel = AtomicBool::new(false);

85 Self::start_with_reporter_cancellable(target, source, reporter, &cancel)

86 .map_err(RendererStartError::into_message)

87 }

89 pub(crate) fn start_with_reporter_cancellable(

90 target: Display,

91 source: Display,

92 reporter: RendererReporter,

93 cancel: &AtomicBool,

94 ) -> Result<Self, RendererStartError> {

95 let stop = Arc::new(AtomicBool::new(false));

96 let worker_stop = Arc::clone(&stop);

97 let wake_event = Arc::new(create_wake_event().map_err(RendererStartError::Failed)?);

98 let worker_wake_event = Arc::clone(&wake_event);

99 let (ready_tx, ready_rx) = mpsc::channel();

100 let (done_tx, done_rx) = mpsc::sync_channel(1);

101 let thread = thread::spawn(move || {

102 let result = run(

103 target.rect,

104 source.rect,

105 &worker_stop,

106 &worker_wake_event,

107 &ready_tx,

108 &reporter,

109 );

110 if let Err(error) = &result {

111 let _ = ready_tx.send(Err(error.clone()));

112 }

113 let failure = result.as_ref().err().cloned();

114 if let Some(error) = failure {

115 reporter(RendererEvent::Failed(error));

116 }

117 let _ = done_tx.send(result);

118 });

119

120 match wait_for_start(&ready_rx, cancel, START_TIMEOUT) {

121 StartWait::Ready => Ok(Self {

122 stop,

123 wake_event,

124 done: done_rx,

125 thread: Some(thread),

126 }),

127 StartWait::Failed(error) => {

128 stop.store(true, Ordering::Release);

129 signal_event(&wake_event);

130 let _ = thread.join();

131 Err(RendererStartError::Failed(error))

132 }

133 StartWait::Cancelled => Err(RendererStartError::Cancelled {

134 cleanup_error: abort_start(&stop, &wake_event, &done_rx, thread).err(),

135 }),

136 StartWait::TimedOut => {

137 let cleanup = abort_start(&stop, &wake_event, &done_rx, thread)

138 .err()

139 .map(|error| format!("; cleanup was incomplete: {error}"))

140 .unwrap_or_default();

141 Err(RendererStartError::Failed(format!(

142 "renderer did not receive and draw a first frame within 10 seconds{cleanup}"

143 )))

144 }

145 StartWait::Disconnected => {

146 let cleanup = abort_start(&stop, &wake_event, &done_rx, thread)

147 .err()

148 .map(|error| format!("; cleanup was incomplete: {error}"))

149 .unwrap_or_default();

150 Err(RendererStartError::Failed(format!(

151 "renderer exited before drawing its first frame{cleanup}"

152 )))

153 }

154 }

155 }

156

157 pub fn request_stop(&self) {

158 self.stop.store(true, Ordering::Release);

159 signal_event(&self.wake_event);

160 }

161

162 pub fn stop(&mut self) -> Result<(), String> {

163 if self.thread.is_none() {

164 return Ok(());

165 }

166 self.request_stop();

167 let (result, exceeded_timeout) = match self.done.recv_timeout(STOP_TIMEOUT) {

168 Ok(result) => (result, false),

169 Err(mpsc::RecvTimeoutError::Timeout) => {

170 // A live renderer owns HWND, input and D3D resources. Keep waiting

171 // rather than detaching it and allowing the display topology to be

172 // torn down underneath the worker.

173 let result = self

174 .done

175 .recv()

176 .map_err(|_| "renderer stopped without reporting a result".to_string())?;

177 (result, true)

178 }

179 Err(mpsc::RecvTimeoutError::Disconnected) => {

180 let thread = self.thread.take().expect("renderer thread disappeared");

181 thread

182 .join()

183 .map_err(|_| "renderer thread panicked".to_string())?;

184 return Err("renderer stopped without reporting a result".into());

185 }

186 };

187 let thread = self.thread.take().expect("renderer thread disappeared");

188 thread

189 .join()

190 .map_err(|_| "renderer thread panicked".to_string())?;

191 match (result, exceeded_timeout) {

192 (Ok(()), false) => Ok(()),

193 (Ok(()), true) => Err("renderer needed more than 2 seconds to stop".into()),

194 (Err(error), false) => Err(error),

195 (Err(error), true) => Err(format!(

196 "renderer needed more than 2 seconds to stop; renderer failed: {error}"

197 )),

198 }

199 }

200}

201

202fn wait_for_start(

203 ready: &mpsc::Receiver<Result<(), String>>,

204 cancel: &AtomicBool,

205 timeout: Duration,

206) -> StartWait {

207 let deadline = Instant::now() + timeout;

208 loop {

209 match ready.try_recv() {

210 Ok(Ok(())) => return StartWait::Ready,

211 Ok(Err(error)) => return StartWait::Failed(error),

212 Err(mpsc::TryRecvError::Disconnected) => return StartWait::Disconnected,

213 Err(mpsc::TryRecvError::Empty) => {}

214 }

215 if cancel.load(Ordering::Acquire) {

216 return StartWait::Cancelled;

217 }

218 let now = Instant::now();

219 if now >= deadline {

220 return StartWait::TimedOut;

221 }

222 let wait = (deadline - now).min(START_CANCEL_POLL_INTERVAL);

223 match ready.recv_timeout(wait) {

224 Ok(Ok(())) => return StartWait::Ready,

225 Ok(Err(error)) => return StartWait::Failed(error),

226 Err(mpsc::RecvTimeoutError::Disconnected) => return StartWait::Disconnected,

227 Err(mpsc::RecvTimeoutError::Timeout) => {}

228 }

229 }

230}

231

232fn abort_start(

233 stop: &AtomicBool,

234 wake_event: &OwnedHandle,

235 done: &mpsc::Receiver<Result<(), String>>,

236 thread: JoinHandle<()>,

237) -> Result<(), String> {

238 stop.store(true, Ordering::Release);

239 signal_event(wake_event);

240 let exceeded_timeout = match done.recv_timeout(STOP_TIMEOUT) {

241 Ok(_) | Err(mpsc::RecvTimeoutError::Disconnected) => false,

242 Err(mpsc::RecvTimeoutError::Timeout) => {

243 // Startup failure cannot hand an unowned live renderer to the caller.

244 // Wait for definitive termination before mapping rollback changes the

245 // virtual display or physical topology.

246 let _ = done.recv();

247 true

248 }

249 };

250 thread

251 .join()

252 .map_err(|_| "renderer thread panicked while stopping".to_string())?;

253 if exceeded_timeout {

254 Err("renderer needed more than 2 seconds to stop".into())

255 } else {

256 Ok(())

257 }

258}

259

260impl Drop for Renderer {

261 fn drop(&mut self) {

262 let _ = self.stop();

263 }

264}

265

266fn run(

267 target: RECT,

268 source: RECT,

269 stop: &AtomicBool,

270 wake_event: &OwnedHandle,

271 ready: &mpsc::Sender<Result<(), String>>,

272 reporter: &RendererReporter,

273) -> Result<(), String> {

274 let instance = unsafe { GetModuleHandleW(None) }

275 .map_err(|error| format!("GetModuleHandleW failed: {error}"))?;

276 let instance = HINSTANCE(instance.0);

277 let class_name_storage =

278 window_class_name(WINDOW_CLASS_SEQUENCE.fetch_add(1, Ordering::Relaxed));

279 let class_name = PCWSTR(class_name_storage.as_ptr());

280 let class = WNDCLASSW {

281 style: CS_HREDRAW | CS_VREDRAW,

282 lpfnWndProc: Some(window_proc),

283 hInstance: instance,

284 lpszClassName: class_name,

285 ..Default::default()

286 };

287 if unsafe { RegisterClassW(&class) } == 0 {

288 return Err(format!(

289 "RegisterClassW failed: {}",

290 windows::core::Error::from_thread()

291 ));

292 }

293

294 let width = target.right - target.left;

295 let height = target.bottom - target.top;

296 let window = unsafe {

297 CreateWindowExW(

298 WINDOW_EX_STYLE(WS_EX_TOPMOST.0 | WS_EX_TOOLWINDOW.0 | WS_EX_NOACTIVATE.0),

299 class_name,

300 w!("SBMS"),

301 WS_POPUP,

302 target.left,

303 target.top,

304 width,

305 height,

306 None,

307 None,

308 Some(instance),

309 None,

310 )

311 }

312 .map_err(|error| {

313 unsafe {

314 let _ = UnregisterClassW(class_name, Some(instance));

315 }

316 format!("CreateWindowExW failed: {error}")

317 })?;

318 unsafe {

319 let _ = ShowWindow(window, SW_SHOW);

320 }

321

322 let input = match InputGuard::start(window, target, source, instance) {

323 Ok(input) => input,

324 Err(error) => {

325 unsafe {

326 let _ = DestroyWindow(window);

327 let _ = UnregisterClassW(class_name, Some(instance));

328 }

329 return Err(error);

330 }

331 };

332 let result = thread::scope(|scope| {

333 let (draw_done_tx, draw_done_rx) = mpsc::sync_channel(1);

334 let window_value = window.0 as usize;

335 let draw_thread = scope.spawn(move || {

336 let window = HWND(window_value as *mut _);

337 let result = draw_frames(window, width, height, source, stop, ready, reporter);

338 let _ = draw_done_tx.send(result);

339 signal_event(wake_event);

340 });

341 let draw_result = loop {

342 if let Ok(result) = draw_done_rx.try_recv() {

343 break result;

344 }

345 if stop.load(Ordering::Acquire) {

346 break draw_done_rx

347 .recv()

348 .map_err(|_| "renderer draw worker exited without a result".to_string())?;

349 }

350 if !pump_messages() {

351 stop.store(true, Ordering::Release);

352 continue;

353 }

354 flush_movement();

355 if let Err(error) = wait_for_message(wake_event) {

356 stop.store(true, Ordering::Release);

357 let _ = draw_done_rx.recv();

358 break Err(error);

359 }

360 };

361 draw_thread

362 .join()

363 .map_err(|_| "renderer draw worker panicked".to_string())?;

364 draw_result

365 });

366 drop(input);

367 unsafe {

368 let _ = DestroyWindow(window);

369 let _ = UnregisterClassW(class_name, Some(instance));

370 }

371 result

372}

373

374fn create_wake_event() -> Result<OwnedHandle, String> {

375 let handle = unsafe { CreateEventW(None, false, false, PCWSTR::null()) }

376 .map_err(|error| format!("CreateEventW(renderer wake) failed: {error}"))?;

377 Ok(unsafe { OwnedHandle::from_raw_handle(handle.0) })

378}

379

380fn wait_for_message(wake_event: &OwnedHandle) -> Result<(), String> {

381 let handle = HANDLE(wake_event.as_raw_handle());

382 let result = unsafe {

383 MsgWaitForMultipleObjectsEx(

384 Some(std::slice::from_ref(&handle)),

385 INFINITE,

386 QS_ALLINPUT,

387 MWMO_INPUTAVAILABLE,

388 )

389 };

390 if result == WAIT_FAILED {

391 return Err(format!(

392 "MsgWaitForMultipleObjectsEx failed: {}",

393 windows::core::Error::from_thread()

394 ));

395 }

396 Ok(())

397}

398

399fn signal_event(wake_event: &OwnedHandle) {

400 let _ = unsafe { SetEvent(HANDLE(wake_event.as_raw_handle())) };

401}

402

403fn draw_frames(

404 window: HWND,

405 target_width: i32,

406 target_height: i32,

407 source: RECT,

408 stop: &AtomicBool,

409 ready: &mpsc::Sender<Result<(), String>>,

410 reporter: &RendererReporter,

411) -> Result<(), String> {

412 let mut first_frame = true;

413 let mut frames = 0_u32;

414 let mut sample_started = Instant::now();

415 let result = run_gpu_renderer(

416 GpuRendererConfig {

417 target_window: window,

418 target_width: target_width as u32,

419 target_height: target_height as u32,

420 source_rect: source,

421 },

422 stop,

423 || {

424 frames = frames.saturating_add(1);

425 if first_frame {

426 ready

427 .send(Ok(()))

428 .map_err(|_| "mapping start was cancelled".to_string())?;

429 first_frame = false;

430 }

431 let elapsed = sample_started.elapsed();

432 if elapsed >= Duration::from_secs(1) {

433 reporter(RendererEvent::Fps(measured_fps(frames, elapsed)));

434 frames = 0;

435 sample_started = Instant::now();

436 }

437 Ok(())

438 },

439 );

440 finish_gpu_renderer(result, reporter)

441}

442

443fn finish_gpu_renderer(

444 result: Result<(), GpuRendererError>,

445 reporter: &RendererReporter,

446) -> Result<(), String> {

447 match result {

448 Ok(()) => Ok(()),

449 Err(GpuRendererError::AccessLost) => {

450 reporter(RendererEvent::TopologyLost);

451 Ok(())

452 }

453 Err(error) => Err(error.to_string()),

454 }

455}

456

457fn measured_fps(frames: u32, elapsed: Duration) -> u32 {

458 if elapsed.is_zero() {

459 return 0;

460 }

461 ((frames as f64 / elapsed.as_secs_f64()).round()).clamp(0.0, u32::MAX as f64) as u32

462}

463

464fn window_class_name(sequence: u64) -> Vec<u16> {

465 format!("SBMSMirrorWindow-{}-{sequence}", std::process::id())

466 .encode_utf16()

467 .chain(Some(0))

468 .collect()

469}

470

471fn pump_messages() -> bool {

472 let mut message = MSG::default();

473 while unsafe { PeekMessageW(&mut message, None, 0, 0, PM_REMOVE) }.as_bool() {

474 if message.message == WM_QUIT {

475 return false;

476 }

477 unsafe {

478 let _ = TranslateMessage(&message);

479 DispatchMessageW(&message);

480 }

481 }

482 true

483}

484

485unsafe extern "system" fn window_proc(

486 window: HWND,

487 message: u32,

488 wparam: WPARAM,

489 lparam: LPARAM,

490) -> LRESULT {

491 if let Some(result) = handle_message(message, wparam, lparam) {

492 return result;

493 }

494 match message {

495 WM_CLOSE => {

496 unsafe {

497 let _ = DestroyWindow(window);

498 }

499 LRESULT(0)

500 }

501 WM_DESTROY => {

502 unsafe { PostQuitMessage(0) };

503 LRESULT(0)

504 }

505 _ => unsafe { DefWindowProcW(window, message, wparam, lparam) },

506 }

507}

508

509#[cfg(test)]

510mod tests {

511 use super::{

512 Renderer, RendererEvent, RendererReporter, StartWait, create_wake_event,

513 finish_gpu_renderer, measured_fps, signal_event, wait_for_message, wait_for_start,

514 window_class_name,

515 };

516 use crate::gpu_renderer::GpuRendererError;

517 use std::sync::Arc;

518 use std::sync::atomic::{AtomicBool, Ordering};

519 use std::sync::mpsc;

520 use std::thread;

521 use std::time::Duration;

522

523 #[test]

524 fn fps_uses_actual_sample_duration() {

525 assert_eq!(measured_fps(60, Duration::from_secs(1)), 60);

526 assert_eq!(measured_fps(90, Duration::from_millis(1500)), 60);

527 }

528

529 #[test]

530 fn fps_reports_idle_and_handles_zero_duration() {

531 assert_eq!(measured_fps(0, Duration::from_secs(1)), 0);

532 assert_eq!(measured_fps(1, Duration::ZERO), 0);

533 }

534

535 #[test]

536 fn access_lost_reports_recoverable_topology_loss() {

537 let (event_tx, event_rx) = mpsc::channel();

538 let reporter: RendererReporter = Arc::new(move |event| {

539 event_tx.send(event).unwrap();

540 });

541

542 assert!(finish_gpu_renderer(Err(GpuRendererError::AccessLost), &reporter).is_ok());

543 assert!(matches!(

544 event_rx.recv().unwrap(),

545 RendererEvent::TopologyLost

546 ));

547 }

548

549 #[test]

550 fn renderer_window_classes_are_unique_and_null_terminated() {

551 let first = window_class_name(7);

552 let second = window_class_name(8);

553 assert_ne!(first, second);

554 assert_eq!(first.last(), Some(&0));

555 assert_eq!(second.last(), Some(&0));

556 }

557

558 #[test]

559 fn ready_frame_wins_a_simultaneous_start_cancellation() {

560 let (ready_tx, ready_rx) = mpsc::channel();

561 ready_tx.send(Ok(())).unwrap();

562 let cancel = AtomicBool::new(true);

563 assert_eq!(

564 wait_for_start(&ready_rx, &cancel, Duration::from_secs(1)),

565 StartWait::Ready

566 );

567 }

568

569 #[test]

570 fn renderer_start_wait_distinguishes_cancel_timeout_and_disconnect() {

571 let (_ready_tx, ready_rx) = mpsc::channel();

572 let cancel = AtomicBool::new(true);

573 assert_eq!(

574 wait_for_start(&ready_rx, &cancel, Duration::from_secs(1)),

575 StartWait::Cancelled

576 );

577

578 let (_ready_tx, ready_rx) = mpsc::channel();

579 let cancel = AtomicBool::new(false);

580 assert_eq!(

581 wait_for_start(&ready_rx, &cancel, Duration::from_millis(1)),

582 StartWait::TimedOut

583 );

584

585 let (ready_tx, ready_rx) = mpsc::channel::<Result<(), String>>();

586 drop(ready_tx);

587 assert_eq!(

588 wait_for_start(&ready_rx, &cancel, Duration::from_secs(1)),

589 StartWait::Disconnected

590 );

591 }

592

593 #[test]

594 fn event_releases_message_wait() {

595 let wake_event = Arc::new(create_wake_event().unwrap());

596 let waiting_event = Arc::clone(&wake_event);

597 let (ready_tx, ready_rx) = mpsc::sync_channel(1);

598 let (done_tx, done_rx) = mpsc::sync_channel(1);

599 let thread = thread::spawn(move || {

600 ready_tx.send(()).unwrap();

601 done_tx.send(wait_for_message(&waiting_event)).unwrap();

602 });

603

604 ready_rx.recv().unwrap();

605 signal_event(&wake_event);

606 assert_eq!(

607 done_rx.recv_timeout(Duration::from_millis(200)).unwrap(),

608 Ok(())

609 );

610 thread.join().unwrap();

611 }

612

613 #[test]

614 fn stop_waits_for_actual_thread_exit_after_reporting_timeout() {

615 let stop = Arc::new(AtomicBool::new(false));

616 let worker_stop = Arc::clone(&stop);

617 let wake_event = Arc::new(create_wake_event().unwrap());

618 let (done_tx, done_rx) = mpsc::sync_channel(1);

619 let thread = thread::spawn(move || {

620 while !worker_stop.load(Ordering::Acquire) {

621 thread::yield_now();

622 }

623 thread::sleep(Duration::from_millis(40));

624 done_tx.send(Ok(())).unwrap();

625 });

626 let mut renderer = Renderer {

627 stop,

628 wake_event,

629 done: done_rx,

630 thread: Some(thread),

631 };

632

633 let error = renderer.stop().unwrap_err();

634 assert!(error.contains("needed more than"));

635 assert!(renderer.thread.is_none());

636 }

637}

638