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/controller.rs43.6 KB

1use std::collections::HashMap;

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

3use std::sync::mpsc::{self, Sender};

4use std::sync::{Arc, Mutex, Weak};

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

7use crate::diagnostics::{self, Level, MappingSessionId};

8use crate::display::active_displays;

9use crate::geometry::{AspectRatio, Rotation};

10use crate::mapping::{

11 MappingError, MappingEvent, MappingGroupInfo, MappingGroupRequest, MappingPlan,

12 MappingReporter, MappingRoute, MappingSession, native_topology_rotation,

13};

14use crate::renderer::RendererEvent;

15use crate::sunshine::{self, CaptureBackend, DeploymentEvent};

17const DESKTOP_DUPLICATION_PROCESS_LIMIT: usize = 4;

18const MAX_RECONNECT_ATTEMPTS: u8 = 2;

20#[derive(Clone, Debug)]

21pub struct DisplayOption {

22 pub id: String,

23 pub name: String,

24 pub label: String,

25 pub native_width: u32,

26 pub native_height: u32,

27 pub detected_physical_width_mm: Option<f64>,

28 pub detected_physical_height_mm: Option<f64>,

29 pub physical_width_mm: Option<f64>,

30 pub physical_height_mm: Option<f64>,

31 pub physical_override_inches: Option<f64>,

32 pub physical_override_aspect_ratio: Option<AspectRatio>,

33 pub rotation: Rotation,

34}

36#[derive(Clone, Debug)]

37pub enum ControllerEvent {

38 Displays(Vec<DisplayOption>),

39 GroupReady(MappingGroupInfo),

40 Fps {

41 id: u32,

42 fps: u32,

43 },

44 Sunshine {

45 id: u32,

46 display_id: String,

47 requested_port: u16,

48 port: Option<u16>,

49 error: Option<String>,

50 },

51 State {

52 state: &'static str,

53 detail: String,

54 running: bool,

55 busy: bool,

56 error: String,

57 restart_available: bool,

58 },

59}

61#[derive(Clone, Copy, Debug, Eq, PartialEq)]

62enum StartReason {

63 User,

64 ManualRestart,

65 Reconnect { attempt: u8 },

66}

68impl StartReason {

69 fn reconnect_attempt(self) -> Option<u8> {

70 match self {

71 Self::Reconnect { attempt } => Some(attempt),

72 Self::User | Self::ManualRestart => None,

73 }

74 }

75}

77enum Command {

78 Refresh,

79 Start {

80 plan: MappingPlan,

81 cancel: Arc<AtomicBool>,

82 reason: StartReason,

83 },

84 Stop,

85 RestartMapping {

86 cancel: Arc<AtomicBool>,

87 },

88 Shutdown,

89}

91enum Message {

92 Command(Command),

93 Mapping {

94 generation: u64,

95 event: MappingEvent,

96 },

97 Sunshine(DeploymentEvent),

98}

100#[derive(Clone)]

101pub struct ControllerSender {

102 tx: Sender<Message>,

103 startup_cancellations: Arc<Mutex<Vec<Weak<AtomicBool>>>>,

104 shutdown_requested: Arc<AtomicBool>,

105}

106

107pub struct Controller {

108 sender: ControllerSender,

109 shutdown_requested: Arc<AtomicBool>,

110 worker: Option<JoinHandle<()>>,

111}

112

113struct ActiveRun {

114 session: MappingSession,

115 reporter: MappingReporter,

116 generation: u64,

117 session_id: MappingSessionId,

118 sunshine_ports: HashMap<u32, u16>,

119 sunshine_capture: CaptureBackend,

120 cancel: Arc<AtomicBool>,

121}

122

123impl ControllerSender {

124 fn new(tx: Sender<Message>, shutdown_requested: Arc<AtomicBool>) -> Self {

125 Self {

126 tx,

127 startup_cancellations: Arc::new(Mutex::new(Vec::new())),

128 shutdown_requested,

129 }

130 }

131

132 pub fn refresh(&self) {

133 let _ = self.tx.send(Message::Command(Command::Refresh));

134 }

135

136 pub fn start(&self, plan: MappingPlan) {

137 self.enqueue_start(plan, StartReason::User);

138 }

139

140 pub fn restart_mapping(&self) {

141 let cancel = self.startup_cancellation();

142 let _ = self

143 .tx

144 .send(Message::Command(Command::RestartMapping { cancel }));

145 }

146

147 fn enqueue_start(&self, plan: MappingPlan, reason: StartReason) {

148 let cancel = self.startup_cancellation();

149 let _ = self.tx.send(Message::Command(Command::Start {

150 plan,

151 cancel,

152 reason,

153 }));

154 }

155

156 fn startup_cancellation(&self) -> Arc<AtomicBool> {

157 let cancel = Arc::new(AtomicBool::new(false));

158 let mut cancellations = self

159 .startup_cancellations

160 .lock()

161 .expect("startup cancellation registry poisoned");

162 cancellations.retain(|cancel| cancel.strong_count() > 0);

163 cancellations.push(Arc::downgrade(&cancel));

164 if self.shutdown_requested.load(Ordering::Acquire) {

165 cancel.store(true, Ordering::Release);

166 }

167 cancel

168 }

169

170 pub fn stop(&self) {

171 self.cancel_startups();

172 let _ = self.tx.send(Message::Command(Command::Stop));

173 }

174

175 fn cancel_startups(&self) {

176 let mut cancellations = self

177 .startup_cancellations

178 .lock()

179 .expect("startup cancellation registry poisoned");

180 cancellations.retain(|cancel| {

181 if let Some(cancel) = cancel.upgrade() {

182 cancel.store(true, Ordering::Release);

183 true

184 } else {

185 false

186 }

187 });

188 }

189}

190

191impl Controller {

192 pub fn spawn(emit: impl Fn(ControllerEvent) + Send + 'static) -> Self {

193 let (tx, rx) = mpsc::channel();

194 let shutdown_requested = Arc::new(AtomicBool::new(false));

195 let sender = ControllerSender::new(tx, Arc::clone(&shutdown_requested));

196 let worker_tx = sender.tx.clone();

197 let worker_shutdown = Arc::clone(&shutdown_requested);

198 let worker = thread::spawn(move || {

199 let sunshine_tx = worker_tx.clone();

200 let sunshine = sunshine::Supervisor::spawn(move |event| {

201 let _ = sunshine_tx.send(Message::Sunshine(event));

202 });

203 let sunshine_sender = sunshine.sender();

204 sunshine_sender.stop_all();

205 let mut active = None;

206 let mut restart_plan = None::<MappingPlan>;

207 let mut next_generation = 0_u64;

208 while let Ok(message) = rx.recv() {

209 match message {

210 Message::Command(Command::Refresh) => refresh(&emit),

211 Message::Command(Command::Start {

212 plan,

213 cancel,

214 reason,

215 }) => {

216 if active.is_some() {

217 continue;

218 }

219 if reason == StartReason::User {

220 restart_plan = None;

221 }

222 let sunshine_ports = match plan

223 .groups

224 .iter()

225 .enumerate()

226 .map(|(slot, group)| {

227 sunshine::default_port_for_slot(slot).map(|port| (group.id, port))

228 })

229 .collect::<Result<HashMap<_, _>, _>>()

230 {

231 Ok(ports) => ports,

232 Err(error) => {

233 diagnostics::log(

234 Level::Error,

235 "controller",

236 "sunshine-ports",

237 None,

238 error.as_str(),

239 );

240 if let Some(attempt) = reason.reconnect_attempt() {

241 if !queue_reconnect_after_failure(

242 attempt,

243 &plan,

244 error.as_str(),

245 &worker_tx,

246 &emit,

247 None,

248 &cancel,

249 ) {

250 restart_plan = Some(plan);

251 }

252 continue;

253 }

254 if reason == StartReason::ManualRestart {

255 restart_plan = Some(plan);

256 emit(restartable_state("Manual restart failed", error));

257 continue;

258 }

259 emit(state(

260 "Stopped",

261 "Couldn’t reserve streaming ports".into(),

262 false,

263 false,

264 error,

265 ));

266 continue;

267 }

268 };

269 let sunshine_capture = capture_backend_for_plan(&plan);

270 next_generation = next_generation.wrapping_add(1);

271 let generation = next_generation;

272 let session_id = diagnostics::new_mapping_session_id();

273 diagnostics::log(

274 Level::Info,

275 "controller",

276 if reason.reconnect_attempt().is_some() {

277 "reconnect-start"

278 } else {

279 "start"

280 },

281 Some(session_id),

282 format!("mapping plan requested with {} group(s)", plan.groups.len()),

283 );

284 diagnostics::log(

285 Level::Info,

286 "controller",

287 "sunshine-capture",

288 Some(session_id),

289 format!(

290

                                "selected capture={} for {} managed Sunshine instance(s); projected DDA processes={} (limit={})",

291 sunshine_capture.as_str(),

292 plan.groups.len(),

293 projected_desktop_duplication_processes(&plan),

294 DESKTOP_DUPLICATION_PROCESS_LIMIT,

295 ),

296 );

297 for group in &plan.groups {

298 diagnostics::log(

299 Level::Info,

300 "controller",

301 "plan",

302 Some(session_id),

303 format_mapping_group_plan(group),

304 );

305 }

306 if let Some(attempt) = reason.reconnect_attempt() {

307 emit(state(

308 "Reconnecting",

309 format!("Reconnect attempt {attempt} of {MAX_RECONNECT_ATTEMPTS}…"),

310 false,

311 true,

312 "",

313 ));

314 } else {

315 emit(state(

316 "Starting",

317 if reason == StartReason::ManualRestart {

318 "Restarting mapping…".into()

319 } else {

320 "Creating virtual display…".into()

321 },

322 false,

323 true,

324 "",

325 ));

326 }

327 let report_tx = worker_tx.clone();

328 let reporter: MappingReporter = Arc::new(move |event| {

329 if let MappingEvent::Topology { message } = &event {

330 diagnostics::log(

331 Level::Debug,

332 "mapping",

333 "topology",

334 Some(session_id),

335 message,

336 );

337 return;

338 }

339 if let MappingEvent::Renderer { id, event } = &event {

340 match event {

341 RendererEvent::Failed(error) => diagnostics::log(

342 Level::Error,

343 "renderer",

344 "renderer",

345 Some(session_id),

346 format!("group {id}: {error}"),

347 ),

348 RendererEvent::TopologyLost => diagnostics::log(

349 Level::Warn,

350 "renderer",

351 "topology-lost",

352 Some(session_id),

353 format!(

354 "group {id}: desktop duplication topology was lost"

355 ),

356 ),

357 RendererEvent::Fps(_) => {}

358 }

359 }

360 let _ = report_tx.send(Message::Mapping { generation, event });

361 });

362 match MappingSession::start_plan_with_reporter_cancellable(

363 plan.clone(),

364 Arc::clone(&reporter),

365 &cancel,

366 ) {

367 Ok(mut started) => {

368 if worker_shutdown.load(Ordering::Acquire)

369 || cancel.load(Ordering::Acquire)

370 {

371 match started.stop() {

372 Ok(()) => {

373 diagnostics::log(

374 Level::Info,

375 "controller",

376 "cancelled",

377 Some(session_id),

378 "mapping completed while cancellation was pending and was stopped",

379 );

380 if !worker_shutdown.load(Ordering::Acquire) {

381 emit(state(

382 "Stopped",

383 "Choose a display to start".into(),

384 false,

385 false,

386 "",

387 ));

388 }

389 }

390 Err(error) => {

391 diagnostics::log(

392 Level::Error,

393 "controller",

394 error.stage(),

395 Some(session_id),

396 error.to_string(),

397 );

398 if !worker_shutdown.load(Ordering::Acquire) {

399 emit(state(

400 "Stopped",

401 "Cleanup completed with errors".into(),

402 false,

403 false,

404 error.to_string(),

405 ));

406 }

407 }

408 }

409 continue;

410 }

411 active = Some(ActiveRun {

412 session: started,

413 reporter,

414 generation,

415 session_id,

416 sunshine_ports,

417 sunshine_capture,

418 cancel,

419 });

420 restart_plan = None;

421 diagnostics::log(

422 Level::Info,

423 "controller",

424 "running",

425 Some(session_id),

426 "mapping is active",

427 );

428 emit(state(

429 "Running",

430 "Mapping is active".into(),

431 true,

432 false,

433 "",

434 ));

435 }

436 Err(error) => {

437 if worker_shutdown.load(Ordering::Acquire) {

438 let clean_cancel = error.is_clean_cancellation();

439 diagnostics::log(

440 if clean_cancel {

441 Level::Info

442 } else {

443 Level::Error

444 },

445 "controller",

446 if clean_cancel {

447 "shutdown"

448 } else {

449 error.stage()

450 },

451 Some(session_id),

452 if clean_cancel {

453 "mapping startup cancelled during shutdown".into()

454 } else {

455 error.to_string()

456 },

457 );

458 continue;

459 }

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

461 if error.is_clean_cancellation() {

462 diagnostics::log(

463 Level::Info,

464 "controller",

465 "cancelled",

466 Some(session_id),

467 "mapping startup cancelled",

468 );

469 emit(state(

470 "Stopped",

471 "Choose a display to start".into(),

472 false,

473 false,

474 "",

475 ));

476 } else {

477 diagnostics::log(

478 Level::Error,

479 "controller",

480 error.stage(),

481 Some(session_id),

482 error.to_string(),

483 );

484 emit(state(

485 "Stopped",

486 "Cleanup completed with errors".into(),

487 false,

488 false,

489 error.to_string(),

490 ));

491 }

492 continue;

493 }

494 diagnostics::log(

495 Level::Error,

496 "controller",

497 error.stage(),

498 Some(session_id),

499 error.to_string(),

500 );

501 if let Some(attempt) = reason.reconnect_attempt() {

502 if !queue_reconnect_after_failure(

503 attempt,

504 &plan,

505 error.to_string(),

506 &worker_tx,

507 &emit,

508 Some(session_id),

509 &cancel,

510 ) {

511 restart_plan = Some(plan);

512 }

513 continue;

514 }

515 if reason == StartReason::ManualRestart {

516 restart_plan = Some(plan);

517 emit(restartable_state(

518 "Manual restart failed",

519 error.to_string(),

520 ));

521 continue;

522 }

523 emit(state(

524 "Stopped",

525 "Couldn’t start mapping".into(),

526 false,

527 false,

528 error.to_string(),

529 ));

530 }

531 }

532 }

533 Message::Command(Command::Stop) => {

534 restart_plan = None;

535 stop_with_state(&mut active, &sunshine_sender, &emit);

536 }

537 Message::Command(Command::RestartMapping { cancel }) => {

538 if active.is_none()

539 && !cancel.load(Ordering::Acquire)

540 && let Some(plan) = restart_plan.take()

541 {

542 let _ = worker_tx.send(Message::Command(Command::Start {

543 plan,

544 cancel,

545 reason: StartReason::ManualRestart,

546 }));

547 }

548 }

549 Message::Command(Command::Shutdown) => {

550 stop_with_state(&mut active, &sunshine_sender, &emit);

551 break;

552 }

553 Message::Mapping { generation, event }

554 if active

555 .as_ref()

556 .is_some_and(|active| active.generation == generation) =>

557 {

558 match event {

559 MappingEvent::GroupReady(group) => {

560 let active = active.as_ref().expect("guard checked active run");

561 if let (Some(display_id), Some(port)) = (

562 group.sunshine_id.clone(),

563 active.sunshine_ports.get(&group.id).copied(),

564 ) {

565 sunshine_sender.start(

566 generation,

567 group.id,

568 display_id,

569 port,

570 active.sunshine_capture,

571 );

572 }

573 emit(ControllerEvent::GroupReady(group));

574 }

575 MappingEvent::Renderer {

576 id,

577 event: RendererEvent::Fps(fps),

578 } => {

579 emit(ControllerEvent::Fps {

580 id,

581 fps: fps.min(999),

582 });

583 }

584 MappingEvent::Renderer {

585 id,

586 event: RendererEvent::TopologyLost,

587 } => {

588 let active_run = active.as_mut().expect("guard checked active run");

589 if active_run.cancel.load(Ordering::Acquire) {

590 continue;

591 }

592 emit(state(

593 "Reconnecting",

594 format!("Output {} changed · rebinding renderer…", id + 1),

595 false,

596 true,

597 "",

598 ));

599 let mut recovery = active_run.session.recover_group_topology(

600 id,

601 Arc::clone(&active_run.reporter),

602 &active_run.cancel,

603 );

604 if recovery.is_err() && !active_run.cancel.load(Ordering::Acquire) {

605 diagnostics::log(

606 Level::Warn,

607 "controller",

608 "renderer-retry",

609 Some(active_run.session_id),

610 format!(

611 "group {id}: first in-place renderer recovery failed; retrying once"

612 ),

613 );

614 recovery = active_run.session.recover_group_topology(

615 id,

616 Arc::clone(&active_run.reporter),

617 &active_run.cancel,

618 );

619 }

620 match recovery {

621 Ok(_) => {

622 diagnostics::log(

623 Level::Info,

624 "controller",

625 "renderer-recovered",

626 Some(active_run.session_id),

627 format!(

628 "group {id}: renderer rebound without recreating the virtual display"

629 ),

630 );

631 emit(state(

632 "Running",

633 format!("Output {} recovered", id + 1),

634 true,

635 false,

636 "",

637 ));

638 }

639 Err(error) => {

640 diagnostics::log(

641 Level::Error,

642 "controller",

643 error.stage(),

644 Some(active_run.session_id),

645 error.to_string(),

646 );

647 emit(state(

648 "Degraded",

649 format!("Output {} needs attention", id + 1),

650 true,

651 false,

652 error.to_string(),

653 ));

654 }

655 }

656 }

657 MappingEvent::Renderer {

658 id,

659 event: RendererEvent::Failed(error),

660 } => {

661 let (stopped_session_id, cleanup) =

662 stop_active(&mut active, &sunshine_sender)

663 .expect("guard checked active run");

664 let session_id = Some(stopped_session_id);

665 let cleanup_error = cleanup.err().map(|cleanup| {

666 diagnostics::log(

667 Level::Error,

668 "controller",

669 cleanup.stage(),

670 session_id,

671 cleanup.to_string(),

672 );

673 cleanup.to_string()

674 });

675 let error = match cleanup_error {

676 Some(cleanup) => {

677 format!("{error}; cleanup also failed: {cleanup}")

678 }

679 None => error,

680 };

681 emit(state(

682 "Stopped",

683 format!("Output {} stopped unexpectedly", id + 1),

684 false,

685 false,

686 error,

687 ));

688 }

689 MappingEvent::Topology { .. } => {}

690 }

691 }

692 Message::Sunshine(event)

693 if active

694 .as_ref()

695 .is_some_and(|active| active.generation == event.generation) =>

696 {

697 emit(ControllerEvent::Sunshine {

698 id: event.group_id,

699 display_id: event.display_id,

700 requested_port: event.requested_port,

701 port: event.port,

702 error: event.error,

703 });

704 }

705 Message::Sunshine(_) => {}

706 Message::Mapping { .. } => {}

707 }

708 }

709 sunshine.shutdown();

710 });

711 Self {

712 sender,

713 shutdown_requested,

714 worker: Some(worker),

715 }

716 }

717

718 pub fn sender(&self) -> ControllerSender {

719 self.sender.clone()

720 }

721

722 pub fn shutdown(mut self) {

723 self.shutdown_requested.store(true, Ordering::Release);

724 self.sender.cancel_startups();

725 let _ = self.sender.tx.send(Message::Command(Command::Shutdown));

726 if let Some(worker) = self.worker.take() {

727 let _ = worker.join();

728 }

729 }

730}

731

732fn format_mapping_group_plan(group: &MappingGroupRequest) -> String {

733 let route = match &group.route {

734 MappingRoute::Mirror { target } => format!("mirror target={target}"),

735 MappingRoute::StreamOnly => "stream_only".into(),

736 };

737 format!(

738 "group {} route={} native={}x{}@{}/{} orientation={:?} topology_rotation={:?}",

739 group.id,

740 route,

741 group.mode.width,

742 group.mode.height,

743 group.mode.refresh_numerator,

744 group.mode.refresh_denominator,

745 group.rotation,

746 native_topology_rotation(group.rotation)

747 )

748}

749

750fn capture_backend_for_plan(plan: &MappingPlan) -> CaptureBackend {

751 if projected_desktop_duplication_processes(plan) > DESKTOP_DUPLICATION_PROCESS_LIMIT {

752 CaptureBackend::Wgc

753 } else {

754 CaptureBackend::Auto

755 }

756}

757

758fn projected_desktop_duplication_processes(plan: &MappingPlan) -> usize {

759 let renderer_processes = usize::from(

760 plan.groups

761 .iter()

762 .any(|group| matches!(group.route, MappingRoute::Mirror { .. })),

763 );

764 renderer_processes + plan.groups.len()

765}

766

767fn refresh(emit: &impl Fn(ControllerEvent)) {

768 match active_displays() {

769 Ok(displays) => {

770 let displays = displays

771 .into_iter()

772 .filter(|display| !display.virtual_display)

773 .map(|display| DisplayOption {

774 id: display.id,

775 name: display.name.clone(),

776 native_width: display.native_width,

777 native_height: display.native_height,

778 detected_physical_width_mm: display.physical_width_mm,

779 detected_physical_height_mm: display.physical_height_mm,

780 physical_width_mm: display.physical_width_mm,

781 physical_height_mm: display.physical_height_mm,

782 physical_override_inches: None,

783 physical_override_aspect_ratio: None,

784 rotation: display.rotation,

785 label: format!(

786 "{} · {}×{}{}",

787 display.name,

788 display.rect.right - display.rect.left,

789 display.rect.bottom - display.rect.top,

790 if display.primary { " · Primary" } else { "" }

791 ),

792 })

793 .collect();

794 emit(ControllerEvent::Displays(displays));

795 }

796 Err(error) => {

797 diagnostics::log(

798 Level::Error,

799 "controller",

800 "display-discovery",

801 None,

802 error.to_string(),

803 );

804 emit(state(

805 "Stopped",

806 "Display discovery failed".into(),

807 false,

808 false,

809 error.to_string(),

810 ));

811 }

812 }

813}

814

815fn stop_active(

816 active: &mut Option<ActiveRun>,

817 sunshine_sender: &sunshine::SupervisorSender,

818) -> Option<(MappingSessionId, Result<(), MappingError>)> {

819 sunshine_sender.stop_all();

820 let mut active = active.take()?;

821 let session_id = active.session_id;

822 Some((session_id, active.session.stop()))

823}

824

825fn stop_with_state(

826 active: &mut Option<ActiveRun>,

827 sunshine_sender: &sunshine::SupervisorSender,

828 emit: &impl Fn(ControllerEvent),

829) {

830 if active.is_none() {

831 let _ = stop_active(active, sunshine_sender);

832 return;

833 }

834 emit(state(

835 "Stopping",

836 "Restoring windows and display topology…".into(),

837 true,

838 true,

839 "",

840 ));

841 let (session_id, result) =

842 stop_active(active, sunshine_sender).expect("active run was checked before cleanup");

843 let stopped_session_id = Some(session_id);

844 match result {

845 Ok(()) => {

846 diagnostics::log(

847 Level::Info,

848 "controller",

849 "stop",

850 stopped_session_id,

851 "mapping stopped",

852 );

853 emit(state(

854 "Stopped",

855 "Choose a display to start".into(),

856 false,

857 false,

858 "",

859 ));

860 }

861 Err(error) => {

862 diagnostics::log(

863 Level::Error,

864 "controller",

865 error.stage(),

866 stopped_session_id,

867 error.to_string(),

868 );

869 emit(state(

870 "Stopped",

871 "Cleanup completed with errors".into(),

872 false,

873 false,

874 error.to_string(),

875 ));

876 }

877 }

878}

879

880fn queue_reconnect_after_failure(

881 attempt: u8,

882 plan: &MappingPlan,

883 error: impl Into<String>,

884 sender: &Sender<Message>,

885 emit: &impl Fn(ControllerEvent),

886 session_id: Option<MappingSessionId>,

887 cancel: &Arc<AtomicBool>,

888) -> bool {

889 let error = error.into();

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

891 return false;

892 }

893 if let Some(next_attempt) = next_reconnect_attempt(attempt) {

894 diagnostics::log(

895 Level::Warn,

896 "controller",

897 "reconnect-retry",

898 session_id,

899 format!(

900 "reconnect attempt {attempt} failed; retrying with attempt {next_attempt}: {error}"

901 ),

902 );

903 let _ = sender.send(Message::Command(Command::Start {

904 plan: plan.clone(),

905 cancel: Arc::clone(cancel),

906 reason: StartReason::Reconnect {

907 attempt: next_attempt,

908 },

909 }));

910 true

911 } else {

912 diagnostics::log(

913 Level::Error,

914 "controller",

915 "reconnect-exhausted",

916 session_id,

917 format!("automatic reconnect failed after {MAX_RECONNECT_ATTEMPTS} attempts: {error}"),

918 );

919 emit(restartable_state(

920 "Automatic reconnect timed out",

921 format!("Automatic reconnect failed after {MAX_RECONNECT_ATTEMPTS} attempts: {error}"),

922 ));

923 false

924 }

925}

926

927fn next_reconnect_attempt(attempt: u8) -> Option<u8> {

928 (attempt < MAX_RECONNECT_ATTEMPTS).then_some(attempt + 1)

929}

930

931fn state(

932 state: &'static str,

933 detail: String,

934 running: bool,

935 busy: bool,

936 error: impl Into<String>,

937) -> ControllerEvent {

938 ControllerEvent::State {

939 state,

940 detail,

941 running,

942 busy,

943 error: error.into(),

944 restart_available: false,

945 }

946}

947

948fn restartable_state(detail: impl Into<String>, error: impl Into<String>) -> ControllerEvent {

949 ControllerEvent::State {

950 state: "Stopped",

951 detail: detail.into(),

952 running: false,

953 busy: false,

954 error: error.into(),

955 restart_available: true,

956 }

957}

958

959#[cfg(test)]

960mod tests {

961 use super::*;

962 use crate::session_gate::VirtualMode;

963

964 fn plan(routes: Vec<MappingRoute>) -> MappingPlan {

965 MappingPlan {

966 groups: routes

967 .into_iter()

968 .enumerate()

969 .map(|(id, route)| MappingGroupRequest {

970 id: id as u32,

971 mode: VirtualMode::default(),

972 rotation: crate::geometry::Rotation::Deg0,

973 route,

974 })

975 .collect(),

976 }

977 }

978

979 #[test]

980 fn stays_on_auto_at_the_desktop_duplication_limit() {

981 let plan = plan(vec![

982 MappingRoute::Mirror {

983 target: "display-a".into(),

984 },

985 MappingRoute::StreamOnly,

986 MappingRoute::StreamOnly,

987 ]);

988

989 assert_eq!(capture_backend_for_plan(&plan), CaptureBackend::Auto);

990 }

991

992 #[test]

993 fn uses_wgc_when_a_mirror_renderer_pushes_four_sunshine_instances_over_the_limit() {

994 let plan = plan(vec![

995 MappingRoute::Mirror {

996 target: "display-a".into(),

997 },

998 MappingRoute::StreamOnly,

999 MappingRoute::StreamOnly,

1000 MappingRoute::StreamOnly,

1001 ]);

1002

1003 assert_eq!(capture_backend_for_plan(&plan), CaptureBackend::Wgc);

1004 }

1005

1006 #[test]

1007 fn uses_wgc_for_five_stream_only_instances() {

1008 let plan = plan(vec![

1009 MappingRoute::StreamOnly,

1010 MappingRoute::StreamOnly,

1011 MappingRoute::StreamOnly,

1012 MappingRoute::StreamOnly,

1013 MappingRoute::StreamOnly,

1014 ]);

1015

1016 assert_eq!(capture_backend_for_plan(&plan), CaptureBackend::Wgc);

1017 }

1018

1019 #[test]

1020 fn keeps_four_stream_only_instances_on_auto() {

1021 let plan = plan(vec![

1022 MappingRoute::StreamOnly,

1023 MappingRoute::StreamOnly,

1024 MappingRoute::StreamOnly,

1025 MappingRoute::StreamOnly,

1026 ]);

1027

1028 assert_eq!(capture_backend_for_plan(&plan), CaptureBackend::Auto);

1029 }

1030

1031 #[test]

1032 fn stop_cancels_a_queued_start_before_the_worker_receives_stop() {

1033 let (tx, rx) = mpsc::channel();

1034 let sender = ControllerSender::new(tx, Arc::new(AtomicBool::new(false)));

1035 sender.start(plan(vec![MappingRoute::StreamOnly]));

1036 let Message::Command(Command::Start { cancel, .. }) = rx.recv().unwrap() else {

1037 panic!("expected start command");

1038 };

1039

1040 assert!(!cancel.load(Ordering::Acquire));

1041 sender.stop();

1042 assert!(cancel.load(Ordering::Acquire));

1043 assert!(matches!(

1044 rx.recv().unwrap(),

1045 Message::Command(Command::Stop)

1046 ));

1047 }

1048

1049 #[test]

1050 fn a_start_sent_after_stop_gets_a_fresh_cancellation_token() {

1051 let (tx, rx) = mpsc::channel();

1052 let sender = ControllerSender::new(tx, Arc::new(AtomicBool::new(false)));

1053 sender.stop();

1054 sender.start(plan(vec![MappingRoute::StreamOnly]));

1055 assert!(matches!(

1056 rx.recv().unwrap(),

1057 Message::Command(Command::Stop)

1058 ));

1059 let Message::Command(Command::Start { cancel, .. }) = rx.recv().unwrap() else {

1060 panic!("expected start command");

1061 };

1062

1063 assert!(!cancel.load(Ordering::Acquire));

1064 }

1065

1066 #[test]

1067 fn a_start_created_after_shutdown_is_already_cancelled() {

1068 let (tx, rx) = mpsc::channel();

1069 let shutdown = Arc::new(AtomicBool::new(false));

1070 let sender = ControllerSender::new(tx, Arc::clone(&shutdown));

1071 shutdown.store(true, Ordering::Release);

1072 sender.start(plan(vec![MappingRoute::StreamOnly]));

1073 let Message::Command(Command::Start { cancel, .. }) = rx.recv().unwrap() else {

1074 panic!("expected start command");

1075 };

1076

1077 assert!(cancel.load(Ordering::Acquire));

1078 }

1079

1080 #[test]

1081 fn reconnect_retries_at_most_twice() {

1082 assert_eq!(next_reconnect_attempt(1), Some(2));

1083 assert_eq!(next_reconnect_attempt(2), None);

1084 }

1085

1086 #[test]

1087 fn stop_cancels_a_queued_manual_restart() {

1088 let (tx, rx) = mpsc::channel();

1089 let sender = ControllerSender::new(tx, Arc::new(AtomicBool::new(false)));

1090 sender.restart_mapping();

1091 let Message::Command(Command::RestartMapping { cancel }) = rx.recv().unwrap() else {

1092 panic!("expected restart command");

1093 };

1094

1095 assert!(!cancel.load(Ordering::Acquire));

1096 sender.stop();

1097 assert!(cancel.load(Ordering::Acquire));

1098 }

1099

1100 #[test]

1101 fn cancelled_reconnect_is_not_queued() {

1102 let (tx, rx) = mpsc::channel();

1103 let cancel = Arc::new(AtomicBool::new(true));

1104 let retry_plan = plan(vec![MappingRoute::StreamOnly]);

1105

1106 assert!(!queue_reconnect_after_failure(

1107 1,

1108 &retry_plan,

1109 "topology still unavailable",

1110 &tx,

1111 &|_| {},

1112 None,

1113 &cancel,

1114 ));

1115 assert!(rx.try_recv().is_err());

1116 }

1117}

1118