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/mapping.rs46.5 KB

1use std::collections::HashSet;

2use std::error::Error;

3use std::fmt::{Display as FmtDisplay, Formatter};

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

5use std::sync::{Arc, Mutex};

6use std::thread;

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

9use serde::{Deserialize, Serialize};

10use windows::core::HRESULT;

12use crate::display::{

13 Display, active_display_topology, active_displays, apply_display_mode,

14 restore_display_topology, unique_physical_display,

15};

16use crate::geometry::Rotation;

17use crate::limits::MAX_OUTPUTS;

18use crate::renderer::{Renderer, RendererEvent, RendererReporter, RendererStartError};

19use crate::session_gate::{SessionGate, VirtualDisplayConfig, VirtualMode};

20use crate::virtual_display::VirtualDisplay;

21use crate::window_migration::WindowMigration;

23const BASE_TOPOLOGY_TIMEOUT_SECS: u64 = 15;

24const TOPOLOGY_TIMEOUT_SECS_PER_GROUP_OVER_EIGHT: u64 = 2;

25const TOPOLOGY_SETTLE: Duration = Duration::from_millis(750);

26const POLL_INTERVAL: Duration = Duration::from_millis(100);

27const RECOVERY_STABLE_INTERVAL: Duration = Duration::from_millis(500);

28const RECOVERY_TIMEOUT: Duration = Duration::from_secs(7);

29const RECOVERY_STABLE_SAMPLES: u8 = 3;

30const ERROR_CANCELLED_HRESULT: HRESULT = HRESULT(0x800704C7_u32 as i32);

32#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]

33pub struct MappingPlan {

34 pub groups: Vec<MappingGroupRequest>,

35}

37#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]

38pub struct MappingGroupRequest {

39 /// Stable zero-based output slot. The value is also the IDD connector index.

40 pub id: u32,

41 pub mode: VirtualMode,

42 #[serde(default)]

43 pub rotation: Rotation,

44 pub route: MappingRoute,

45}

47#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]

48#[serde(tag = "kind", rename_all = "snake_case")]

49pub enum MappingRoute {

50 Mirror { target: String },

51 StreamOnly,

52}

54#[derive(Clone, Debug, Eq, PartialEq)]

55pub struct MappingGroupInfo {

56 pub id: u32,

57 pub mode: VirtualMode,

58 pub rotation: Rotation,

59 pub route: MappingRoute,

60 pub source_id: String,

61 pub source_device_name: String,

62 pub sunshine_id: Option<String>,

63}

65#[derive(Clone, Debug)]

66pub enum MappingEvent {

67 GroupReady(MappingGroupInfo),

68 Renderer { id: u32, event: RendererEvent },

69 Topology { message: String },

70}

72pub type MappingReporter = Arc<dyn Fn(MappingEvent) + Send + Sync + 'static>;

74struct ActiveGroup {

75 info: MappingGroupInfo,

76 target_id: Option<String>,

77 renderer: Option<Renderer>,

78 renderer_event_gate: Arc<RendererEventGate>,

79 migration: Option<WindowMigration>,

80}

82enum RendererEventGateState {

83 Starting(Vec<RendererEvent>),

84 Active,

85 Discarded,

86}

88struct RendererEventGate {

89 id: u32,

90 reporter: MappingReporter,

91 state: Mutex<RendererEventGateState>,

92}

94impl RendererEventGate {

95 fn new(id: u32, reporter: MappingReporter) -> Arc<Self> {

96 Arc::new(Self {

97 id,

98 reporter,

99 state: Mutex::new(RendererEventGateState::Starting(Vec::new())),

100 })

101 }

102

103 fn report(&self, event: RendererEvent) {

104 let mut state = self

105 .state

106 .lock()

107 .unwrap_or_else(|poisoned| poisoned.into_inner());

108 match &mut *state {

109 RendererEventGateState::Starting(events) => events.push(event),

110 RendererEventGateState::Active => {

111 (self.reporter)(MappingEvent::Renderer { id: self.id, event })

112 }

113 RendererEventGateState::Discarded => {}

114 }

115 }

116

117 fn activate(&self) {

118 let mut state = self

119 .state

120 .lock()

121 .unwrap_or_else(|poisoned| poisoned.into_inner());

122 let buffered = match std::mem::replace(&mut *state, RendererEventGateState::Active) {

123 RendererEventGateState::Starting(events) => events,

124 RendererEventGateState::Active | RendererEventGateState::Discarded => Vec::new(),

125 };

126 for event in buffered {

127 (self.reporter)(MappingEvent::Renderer { id: self.id, event });

128 }

129 }

130

131 fn discard(&self) {

132 let mut state = self

133 .state

134 .lock()

135 .unwrap_or_else(|poisoned| poisoned.into_inner());

136 *state = RendererEventGateState::Discarded;

137 }

138}

139

140pub struct MappingSession {

141 groups: Vec<ActiveGroup>,

142 display: Option<VirtualDisplay>,

143 session_gate: Option<SessionGate>,

144 connector_indices: Vec<u32>,

145 physical_topology: Vec<Display>,

146}

147

148#[derive(Debug)]

149pub struct MappingError {

150 stage: &'static str,

151 group_id: Option<u32>,

152 message: String,

153 cleanup_incomplete: bool,

154}

155

156impl FmtDisplay for MappingError {

157 fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {

158 match self.group_id {

159 Some(id) => write!(formatter, "group {id} {}: {}", self.stage, self.message),

160 None => write!(formatter, "{}: {}", self.stage, self.message),

161 }

162 }

163}

164

165impl Error for MappingError {}

166

167impl MappingError {

168 pub fn stage(&self) -> &'static str {

169 self.stage

170 }

171

172 pub fn group_id(&self) -> Option<u32> {

173 self.group_id

174 }

175

176 pub fn message(&self) -> &str {

177 &self.message

178 }

179

180 pub(crate) fn is_clean_cancellation(&self) -> bool {

181 self.stage == "cancelled" && !self.cleanup_incomplete

182 }

183

184 fn record_cleanup_failure(&mut self, error: impl FmtDisplay) {

185 self.cleanup_incomplete = true;

186 self.message

187 .push_str(&format!("; rollback was incomplete: {error}"));

188 }

189}

190

191impl MappingPlan {

192 pub fn new(groups: Vec<MappingGroupRequest>) -> Result<Self, MappingError> {

193 let plan = Self { groups };

194 plan.validate()?;

195 Ok(plan)

196 }

197

198 pub fn validate(&self) -> Result<(), MappingError> {

199 if self.groups.is_empty() {

200 return Err(MappingError {

201 stage: "plan",

202 group_id: None,

203 message: "at least one mapping group is required".into(),

204 cleanup_incomplete: false,

205 });

206 }

207 if self.groups.len() > MAX_OUTPUTS {

208 return Err(MappingError {

209 stage: "plan",

210 group_id: None,

211 message: format!("at most {MAX_OUTPUTS} mapping groups are supported"),

212 cleanup_incomplete: false,

213 });

214 }

215

216 let mut ids = HashSet::with_capacity(self.groups.len());

217 let mut targets = HashSet::with_capacity(self.groups.len());

218 for group in &self.groups {

219 if group.id >= MAX_OUTPUTS as u32 {

220 return Err(group_stage(

221 group.id,

222 "plan",

223 format!(

224 "group id must be a connector index between 0 and {}",

225 MAX_OUTPUTS - 1

226 ),

227 ));

228 }

229 if !ids.insert(group.id) {

230 return Err(group_stage(

231 group.id,

232 "plan",

233 "mapping group id is duplicated",

234 ));

235 }

236 group

237 .mode

238 .validate()

239 .map_err(|error| group_stage(group.id, "mode", error))?;

240 if let MappingRoute::Mirror { target } = &group.route {

241 if target.trim().is_empty() {

242 return Err(group_stage(

243 group.id,

244 "plan",

245 "mirror target id must not be empty",

246 ));

247 }

248 if !targets.insert(target.to_ascii_lowercase()) {

249 return Err(group_stage(

250 group.id,

251 "plan",

252 "a physical target may only belong to one mapping group",

253 ));

254 }

255 }

256 }

257 Ok(())

258 }

259}

260

261impl MappingSession {

262 pub fn start_plan(plan: MappingPlan) -> Result<Self, MappingError> {

263 Self::start_plan_with_reporter(plan, Arc::new(|_| {}))

264 }

265

266 pub fn start_plan_with_reporter(

267 plan: MappingPlan,

268 reporter: MappingReporter,

269 ) -> Result<Self, MappingError> {

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

271 Self::start_plan_with_reporter_cancellable(plan, reporter, &cancel)

272 }

273

274 pub(crate) fn start_plan_with_reporter_cancellable(

275 plan: MappingPlan,

276 reporter: MappingReporter,

277 cancel: &AtomicBool,

278 ) -> Result<Self, MappingError> {

279 plan.validate()?;

280 check_start_cancelled(cancel)?;

281 let displays = active_displays().map_err(|error| stage("target", error))?;

282 check_start_cancelled(cancel)?;

283 for group in &plan.groups {

284 if let MappingRoute::Mirror { target } = &group.route {

285 mapping_target(&displays, target).map_err(|error| error.with_group(group.id))?;

286 }

287 }

288

289 let physical_topology = displays

290 .iter()

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

292 .cloned()

293 .collect();

294 let connector_indices = plan.groups.iter().map(|group| group.id).collect();

295 let gate_configs: Vec<_> = plan

296 .groups

297 .iter()

298 .map(|group| VirtualDisplayConfig {

299 connector_index: group.id,

300 mode: group.mode,

301 })

302 .collect();

303 check_start_cancelled(cancel)?;

304 let session_gate = SessionGate::create_many(&gate_configs)

305 .map_err(|error| stage("session-gate", error))?;

306 check_start_cancelled(cancel)?;

307 let display = match VirtualDisplay::create_cancellable(cancel) {

308 Ok(display) => display,

309 Err(error)

310 if cancel.load(Ordering::Acquire) && error.code() == ERROR_CANCELLED_HRESULT =>

311 {

312 return Err(start_cancelled());

313 }

314 Err(error) => return Err(stage("device", error)),

315 };

316

317 let mut session = Self {

318 groups: Vec::with_capacity(plan.groups.len()),

319 display: Some(display),

320 session_gate: Some(session_gate),

321 connector_indices,

322 physical_topology,

323 };

324

325 if let Err(error) = check_start_cancelled(cancel) {

326 return Err(rollback_start(session, error));

327 }

328 let sources = match wait_for_sources(&plan, &reporter, cancel) {

329 Ok(sources) => sources,

330 Err(error) => return Err(rollback_start(session, error)),

331 };

332 for (request, source) in plan.groups.iter().zip(&sources) {

333 let target_id = match &request.route {

334 MappingRoute::Mirror { target } => Some(target.clone()),

335 MappingRoute::StreamOnly => None,

336 };

337 session.groups.push(ActiveGroup {

338 info: MappingGroupInfo {

339 id: request.id,

340 mode: request.mode,

341 rotation: request.rotation,

342 route: request.route.clone(),

343 source_id: source.id.clone(),

344 source_device_name: source.device_name.clone(),

345 sunshine_id: source.sunshine_id.clone(),

346 },

347 target_id,

348 renderer: None,

349 renderer_event_gate: RendererEventGate::new(request.id, Arc::clone(&reporter)),

350 migration: None,

351 });

352 }

353

354 let stable_displays = match active_displays() {

355 Ok(displays) => displays,

356 Err(error) => return Err(rollback_start(session, stage("topology", error))),

357 };

358 for (index, request) in plan.groups.iter().enumerate() {

359 if let Err(error) = check_start_cancelled(cancel) {

360 return Err(rollback_start(session, error));

361 }

362 let MappingRoute::Mirror { target } = &request.route else {

363 continue;

364 };

365 let physical = match mapping_target(&stable_displays, target) {

366 Ok(target) => target,

367 Err(error) => {

368 return Err(rollback_start(session, error.with_group(request.id)));

369 }

370 };

371 let migration = match WindowMigration::start(&physical, &sources[index]) {

372 Ok(migration) => migration,

373 Err(_) if cancel.load(Ordering::Acquire) => {

374 return Err(rollback_start(session, start_cancelled()));

375 }

376 Err(error) => {

377 return Err(rollback_start(

378 session,

379 group_stage(request.id, "windows", error),

380 ));

381 }

382 };

383 session.groups[index].migration = Some(migration);

384 if let Err(error) = check_start_cancelled(cancel) {

385 return Err(rollback_start(session, error));

386 }

387 }

388

389 for (index, request) in plan.groups.iter().enumerate() {

390 if let Err(error) = check_start_cancelled(cancel) {

391 return Err(rollback_start(session, error));

392 }

393 let MappingRoute::Mirror { target } = &request.route else {

394 continue;

395 };

396 let physical = match mapping_target(&stable_displays, target) {

397 Ok(target) => target,

398 Err(error) => {

399 return Err(rollback_start(session, error.with_group(request.id)));

400 }

401 };

402 let event_gate = Arc::clone(&session.groups[index].renderer_event_gate);

403 let renderer_reporter: RendererReporter = Arc::new(move |event| {

404 event_gate.report(event);

405 });

406 let renderer = match Renderer::start_with_reporter_cancellable(

407 physical,

408 sources[index].clone(),

409 renderer_reporter,

410 cancel,

411 ) {

412 Ok(renderer) => renderer,

413 Err(RendererStartError::Cancelled { cleanup_error }) => {

414 let mut cause = start_cancelled();

415 if let Some(error) = cleanup_error {

416 cause.record_cleanup_failure(format!("renderer startup cleanup: {error}"));

417 }

418 return Err(rollback_start(session, cause));

419 }

420 Err(RendererStartError::Failed(error)) => {

421 return Err(rollback_start(

422 session,

423 group_stage(request.id, "first-frame", error),

424 ));

425 }

426 };

427 session.groups[index].renderer = Some(renderer);

428 session.groups[index].renderer_event_gate.activate();

429 if let Err(error) = check_start_cancelled(cancel) {

430 return Err(rollback_start(session, error));

431 }

432 }

433

434 for group in &session.groups {

435 reporter(MappingEvent::GroupReady(group.info.clone()));

436 }

437 Ok(session)

438 }

439

440 pub fn stop(&mut self) -> Result<(), MappingError> {

441 if self

442 .groups

443 .iter()

444 .all(|group| group.renderer.is_none() && group.migration.is_none())

445 && self.display.is_none()

446 && self.session_gate.is_none()

447 {

448 return Ok(());

449 }

450

451 let mut errors = Vec::new();

452 for group in self.groups.iter().rev() {

453 if let Some(renderer) = group.renderer.as_ref() {

454 renderer.request_stop();

455 }

456 }

457 for group in self.groups.iter_mut().rev() {

458 if let Some(mut renderer) = group.renderer.take()

459 && let Err(error) = renderer.stop()

460 {

461 errors.push(format!("group {} renderer-stop: {error}", group.info.id));

462 }

463 }

464 for group in self.groups.iter_mut().rev() {

465 if let Some(migration) = group.migration.as_mut() {

466 migration.prepare_restore();

467 }

468 }

469 for group in self.groups.iter_mut().rev() {

470 if let Some(migration) = group.migration.as_mut()

471 && let Err(error) = migration.restore()

472 {

473 errors.push(format!("group {} windows-restore: {error}", group.info.id));

474 }

475 }

476

477 self.display.take();

478 if let Err(error) = wait_for_sources_removed(&self.connector_indices) {

479 errors.push(format!("remove: {error}"));

480 }

481

482 thread::sleep(TOPOLOGY_SETTLE);

483 let topology_restored = match restore_display_topology(&self.physical_topology) {

484 Ok(()) => true,

485 Err(error) => {

486 errors.push(format!("topology-restore: {error}"));

487 false

488 }

489 };

490 if topology_restored {

491 thread::sleep(TOPOLOGY_SETTLE);

492 }

493

494 let final_displays = active_displays();

495 if let Err(error) = &final_displays {

496 errors.push(format!("windows-reconcile topology read: {error}"));

497 }

498 for group in self.groups.iter_mut().rev() {

499 let Some(migration) = group.migration.as_mut() else {

500 continue;

501 };

502 let target = group.target_id.as_ref().and_then(|target_id| {

503 final_displays

504 .as_ref()

505 .ok()

506 .and_then(|displays| unique_physical_display(displays, target_id).ok())

507 });

508 if let Err(error) = migration.reconcile_after_topology_change(target.as_ref()) {

509 errors.push(format!(

510 "group {} windows-reconcile: {error}",

511 group.info.id

512 ));

513 }

514 }

515 for group in &mut self.groups {

516 group.migration.take();

517 }

518 self.session_gate.take();

519

520 if errors.is_empty() {

521 Ok(())

522 } else {

523 Err(MappingError {

524 stage: "stop",

525 group_id: None,

526 message: errors.join("; "),

527 cleanup_incomplete: false,

528 })

529 }

530 }

531

532 pub fn groups(&self) -> impl ExactSizeIterator<Item = &MappingGroupInfo> {

533 self.groups.iter().map(|group| &group.info)

534 }

535

536 pub(crate) fn recover_group_topology(

537 &mut self,

538 group_id: u32,

539 reporter: MappingReporter,

540 cancel: &AtomicBool,

541 ) -> Result<MappingGroupInfo, MappingError> {

542 let group_index = self

543 .groups

544 .iter()

545 .position(|group| group.info.id == group_id)

546 .ok_or_else(|| {

547 group_stage(group_id, "recovery", "mapping group is no longer active")

548 })?;

549 let target_id = self.groups[group_index].target_id.clone().ok_or_else(|| {

550 group_stage(group_id, "recovery", "stream-only group has no renderer")

551 })?;

552

553 self.groups[group_index].renderer_event_gate.discard();

554

555 if let Some(mut renderer) = self.groups[group_index].renderer.take()

556 && let Err(error) = renderer.stop()

557 {

558 return Err(group_stage(group_id, "renderer-stop", error));

559 }

560

561 let group_info = self.groups[group_index].info.clone();

562 let displays = wait_for_stable_recovery_topology(&group_info, &target_id, cancel)?;

563 let target =

564 mapping_target(&displays, &target_id).map_err(|error| error.with_group(group_id))?;

565 let source = displays

566 .iter()

567 .find(|display| display.virtual_display && display.connector_index == group_id)

568 .cloned()

569 .ok_or_else(|| {

570 group_stage(

571 group_id,

572 "recovery",

573 "virtual source disappeared while recovering the renderer",

574 )

575 })?;

576

577 // Windows owns the runtime desktop layout. Commit the stable physical

578 // snapshot before rebuilding short-lived consumers so a renderer start

579 // failure cannot make a later Stop roll the user's position change back.

580 self.physical_topology = displays

581 .iter()

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

583 .cloned()

584 .collect();

585

586 if let Some(migration) = self.groups[group_index].migration.as_mut() {

587 migration

588 .refresh_topology(&target, &source)

589 .map_err(|error| group_stage(group_id, "window-scanner", error))?;

590 }

591

592 let event_gate = RendererEventGate::new(group_id, reporter);

593 let reporter_gate = Arc::clone(&event_gate);

594 let renderer_reporter: RendererReporter = Arc::new(move |event| {

595 reporter_gate.report(event);

596 });

597 let renderer = Renderer::start_with_reporter_cancellable(

598 target,

599 source.clone(),

600 renderer_reporter,

601 cancel,

602 )

603 .map_err(|error| match error {

604 RendererStartError::Cancelled { cleanup_error } => {

605 let mut error =

606 group_stage(group_id, "cancelled", "renderer recovery was cancelled");

607 if let Some(cleanup_error) = cleanup_error {

608 error.record_cleanup_failure(cleanup_error);

609 }

610 error

611 }

612 RendererStartError::Failed(error) => group_stage(group_id, "first-frame", error),

613 })?;

614

615 self.groups[group_index].info.source_id = source.id;

616 self.groups[group_index].info.source_device_name = source.device_name;

617 self.groups[group_index].info.sunshine_id = source.sunshine_id;

618 self.groups[group_index].renderer = Some(renderer);

619 self.groups[group_index].renderer_event_gate = Arc::clone(&event_gate);

620 event_gate.activate();

621 Ok(self.groups[group_index].info.clone())

622 }

623}

624

625impl MappingError {

626 fn with_group(mut self, group_id: u32) -> Self {

627 self.group_id = Some(group_id);

628 self

629 }

630}

631

632impl Drop for MappingSession {

633 fn drop(&mut self) {

634 let _ = self.stop();

635 }

636}

637

638fn rollback_start(mut session: MappingSession, mut cause: MappingError) -> MappingError {

639 if let Err(error) = session.stop() {

640 cause.record_cleanup_failure(error);

641 }

642 cause

643}

644

645fn mapping_target(displays: &[Display], target_id: &str) -> Result<Display, MappingError> {

646 unique_physical_display(displays, target_id).map_err(|error| stage("target", error))

647}

648

649fn wait_for_stable_recovery_topology(

650 group: &MappingGroupInfo,

651 target_id: &str,

652 cancel: &AtomicBool,

653) -> Result<Vec<Display>, MappingError> {

654 let group_id = group.id;

655 let deadline = Instant::now() + RECOVERY_TIMEOUT;

656 let mut stability = RecoveryStability::default();

657 loop {

658 check_start_cancelled(cancel)?;

659 let displays =

660 active_display_topology().map_err(|error| group_stage(group_id, "recovery", error))?;

661 let ready = unique_physical_display(&displays, target_id).is_ok()

662 && displays.iter().any(|display| {

663 display.virtual_display

664 && display.connector_index == group_id

665 && source_matches_mode(display, group.mode, group.rotation)

666 });

667 let signature = ready.then(|| recovery_topology_signature(&displays));

668 if stability.observe(signature) {

669 let hydrated =

670 active_displays().map_err(|error| group_stage(group_id, "recovery", error))?;

671 let hydrated_ready = unique_physical_display(&hydrated, target_id).is_ok()

672 && hydrated.iter().any(|display| {

673 display.virtual_display

674 && display.connector_index == group_id

675 && source_matches_mode(display, group.mode, group.rotation)

676 });

677 if hydrated_ready && stability.matches(&recovery_topology_signature(&hydrated)) {

678 return Ok(hydrated);

679 }

680 stability.reset();

681 }

682 if Instant::now() >= deadline {

683 return Err(group_stage(

684 group_id,

685 "recovery",

686 "display topology did not become stable within 7 seconds",

687 ));

688 }

689 thread::sleep(RECOVERY_STABLE_INTERVAL);

690 }

691}

692

693type RecoveryTopologySignature = Vec<(String, String, u32, i32, i32, i32, i32, u16, u32, u32)>;

694

695#[derive(Default)]

696struct RecoveryStability {

697 signature: Option<RecoveryTopologySignature>,

698 samples: u8,

699}

700

701impl RecoveryStability {

702 fn observe(&mut self, signature: Option<RecoveryTopologySignature>) -> bool {

703 if signature.is_some() && signature == self.signature {

704 self.samples = self.samples.saturating_add(1);

705 } else {

706 self.samples = u8::from(signature.is_some());

707 self.signature = signature;

708 }

709 self.samples >= RECOVERY_STABLE_SAMPLES

710 }

711

712 fn matches(&self, signature: &RecoveryTopologySignature) -> bool {

713 self.signature.as_ref() == Some(signature)

714 }

715

716 fn reset(&mut self) {

717 self.signature = None;

718 self.samples = 0;

719 }

720}

721

722fn recovery_topology_signature(displays: &[Display]) -> RecoveryTopologySignature {

723 let mut signature: Vec<_> = displays

724 .iter()

725 .map(|display| {

726 (

727 display.id.to_ascii_lowercase(),

728 display.device_name.to_ascii_lowercase(),

729 display.connector_index,

730 display.rect.left,

731 display.rect.top,

732 display.rect.right,

733 display.rect.bottom,

734 rotation_degrees(display.rotation),

735 display.refresh_numerator,

736 display.refresh_denominator,

737 )

738 })

739 .collect();

740 signature.sort_unstable();

741 signature

742}

743

744fn wait_for_sources(

745 plan: &MappingPlan,

746 reporter: &MappingReporter,

747 cancel: &AtomicBool,

748) -> Result<Vec<Display>, MappingError> {

749 let requested_ids: HashSet<_> = plan.groups.iter().map(|group| group.id).collect();

750 let topology_timeout = topology_timeout(plan.groups.len());

751 let deadline = Instant::now() + topology_timeout;

752 let mut mode_applied = HashSet::new();

753 let mut last_snapshot = None;

754 loop {

755 check_start_cancelled(cancel)?;

756 let displays = active_display_topology().map_err(|error| stage("topology", error))?;

757 let sources: Vec<_> = displays

758 .iter()

759 .filter(|display| display.virtual_display)

760 .cloned()

761 .collect();

762 let snapshot = format_topology_snapshot(plan, &sources);

763 if last_snapshot.as_deref() != Some(snapshot.as_str()) {

764 reporter(MappingEvent::Topology {

765 message: snapshot.clone(),

766 });

767 last_snapshot = Some(snapshot.clone());

768 }

769 if let Some(source) = sources

770 .iter()

771 .find(|source| !requested_ids.contains(&source.connector_index))

772 {

773 return Err(MappingError {

774 stage: "topology",

775 group_id: None,

776 message: format!(

777 "unexpected SBMS virtual connector {} is active",

778 source.connector_index

779 ),

780 cleanup_incomplete: false,

781 });

782 }

783 let mut seen = HashSet::new();

784 for source in &sources {

785 if !seen.insert(source.connector_index) {

786 return Err(group_stage(

787 source.connector_index,

788 "topology",

789 "more than one virtual source reported the same connector",

790 ));

791 }

792 }

793 if sources.len() == plan.groups.len() {

794 let mut device_names = HashSet::with_capacity(sources.len());

795 for source in &sources {

796 if !device_names.insert(source.device_name.to_ascii_lowercase()) {

797 return Err(group_stage(

798 source.connector_index,

799 "topology",

800 "virtual connectors were cloned onto one GDI source",

801 ));

802 }

803 }

804 }

805

806 let mut ordered = Vec::with_capacity(plan.groups.len());

807 let mut all_ready = true;

808 for group in &plan.groups {

809 let Some(source) = sources

810 .iter()

811 .find(|source| source.connector_index == group.id)

812 else {

813 all_ready = false;

814 continue;

815 };

816 if source_matches_mode(source, group.mode, group.rotation) {

817 ordered.push(source.clone());

818 continue;

819 }

820 all_ready = false;

821 if mode_applied.insert(group.id) {

822 reporter(MappingEvent::Topology {

823 message: format!(

824 "group {} applying mode on device={} requested={}",

825 group.id,

826 source.device_name,

827 format_requested_mode(group.mode, group.rotation)

828 ),

829 });

830 match apply_display_mode(source, group.mode) {

831 Ok(()) => reporter(MappingEvent::Topology {

832 message: format!(

833 "group {} mode apply accepted on device={}; waiting for topology",

834 group.id, source.device_name

835 ),

836 }),

837 Err(error) => {

838 reporter(MappingEvent::Topology {

839 message: format!(

840 "group {} mode apply rejected on device={}: {}",

841 group.id, source.device_name, error

842 ),

843 });

844 return Err(group_stage(group.id, "mode", error));

845 }

846 }

847 }

848 }

849 if all_ready && ordered.len() == plan.groups.len() {

850 let hydrated = active_displays().map_err(|error| stage("topology", error))?;

851 if let Some(ordered) = ordered_ready_sources(plan, &hydrated) {

852 return Ok(ordered);

853 }

854 }

855 if Instant::now() >= deadline {

856 let pending = format_pending_topology(plan, &sources);

857 reporter(MappingEvent::Topology {

858 message: format!(

859 "timeout after {} seconds; {snapshot}",

860 topology_timeout.as_secs()

861 ),

862 });

863 return Err(MappingError {

864 stage: "topology",

865 group_id: None,

866 message: format!(

867 "virtual displays were not ready after {} seconds; {pending}",

868 topology_timeout.as_secs()

869 ),

870 cleanup_incomplete: false,

871 });

872 }

873 thread::sleep(POLL_INTERVAL);

874 }

875}

876

877fn topology_timeout(group_count: usize) -> Duration {

878 let extra_groups = group_count.saturating_sub(8) as u64;

879 Duration::from_secs(

880 BASE_TOPOLOGY_TIMEOUT_SECS + extra_groups * TOPOLOGY_TIMEOUT_SECS_PER_GROUP_OVER_EIGHT,

881 )

882}

883

884fn check_start_cancelled(cancel: &AtomicBool) -> Result<(), MappingError> {

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

886 Err(start_cancelled())

887 } else {

888 Ok(())

889 }

890}

891

892fn start_cancelled() -> MappingError {

893 stage("cancelled", "mapping startup was cancelled")

894}

895

896fn ordered_ready_sources(plan: &MappingPlan, displays: &[Display]) -> Option<Vec<Display>> {

897 let sources: Vec<_> = displays

898 .iter()

899 .filter(|display| display.virtual_display)

900 .collect();

901 if sources.len() != plan.groups.len() {

902 return None;

903 }

904 let mut connector_indices = HashSet::with_capacity(sources.len());

905 let mut device_names = HashSet::with_capacity(sources.len());

906 if sources.iter().any(|source| {

907 !connector_indices.insert(source.connector_index)

908 || !device_names.insert(source.device_name.to_ascii_lowercase())

909 }) {

910 return None;

911 }

912 plan.groups

913 .iter()

914 .map(|group| {

915 sources

916 .iter()

917 .find(|source| {

918 source.connector_index == group.id

919 && source_matches_mode(source, group.mode, group.rotation)

920 })

921 .map(|source| (*source).clone())

922 })

923 .collect()

924}

925

926fn format_topology_snapshot(plan: &MappingPlan, sources: &[Display]) -> String {

927 plan.groups

928 .iter()

929 .map(|group| format_group_topology(group, sources))

930 .collect::<Vec<_>>()

931 .join("; ")

932}

933

934fn format_pending_topology(plan: &MappingPlan, sources: &[Display]) -> String {

935 plan.groups

936 .iter()

937 .filter(|group| {

938 !sources.iter().any(|source| {

939 source.connector_index == group.id

940 && source_matches_mode(source, group.mode, group.rotation)

941 })

942 })

943 .map(|group| format_group_topology(group, sources))

944 .collect::<Vec<_>>()

945 .join("; ")

946}

947

948fn format_group_topology(group: &MappingGroupRequest, sources: &[Display]) -> String {

949 let requested = format_requested_mode(group.mode, group.rotation);

950 let Some(source) = sources

951 .iter()

952 .find(|source| source.connector_index == group.id)

953 else {

954 return format!("group {} absent requested={requested}", group.id);

955 };

956 let state = if source_matches_mode(source, group.mode, group.rotation) {

957 "ready"

958 } else {

959 "observed"

960 };

961 format!(

962 "group {} {} device={} actual={} requested={}",

963 group.id,

964 state,

965 source.device_name,

966 format_observed_mode(source),

967 requested

968 )

969}

970

971fn format_requested_mode(mode: VirtualMode, rotation: Rotation) -> String {

972 format!(

973 "{}x{}@{}/{} orientation={} topology_rotation={}",

974 mode.width,

975 mode.height,

976 mode.refresh_numerator,

977 mode.refresh_denominator,

978 rotation_degrees(rotation),

979 rotation_degrees(native_topology_rotation(rotation))

980 )

981}

982

983fn format_observed_mode(source: &Display) -> String {

984 format!(

985 "{}x{}@{}/{} rotation={}",

986 source.rect.right - source.rect.left,

987 source.rect.bottom - source.rect.top,

988 source.refresh_numerator,

989 source.refresh_denominator,

990 rotation_degrees(source.rotation)

991 )

992}

993

994fn rotation_degrees(rotation: Rotation) -> u16 {

995 match rotation {

996 Rotation::Deg0 => 0,

997 Rotation::Deg90 => 90,

998 Rotation::Deg180 => 180,

999 Rotation::Deg270 => 270,

1000 }

1001}

1002

1003fn wait_for_sources_removed(connector_indices: &[u32]) -> Result<(), MappingError> {

1004 let deadline = Instant::now() + topology_timeout(connector_indices.len());

1005 loop {

1006 let displays = active_display_topology().map_err(|error| stage("remove", error))?;

1007 if !displays.iter().any(|display| {

1008 display.virtual_display && connector_indices.contains(&display.connector_index)

1009 }) {

1010 return Ok(());

1011 }

1012 if Instant::now() >= deadline {

1013 return Err(MappingError {

1014 stage: "remove",

1015 group_id: None,

1016 message: "virtual sources stayed active after the device handle closed".into(),

1017 cleanup_incomplete: false,

1018 });

1019 }

1020 thread::sleep(POLL_INTERVAL);

1021 }

1022}

1023

1024pub(crate) fn native_topology_rotation(_rotation: Rotation) -> Rotation {

1025 Rotation::Deg0

1026}

1027

1028fn source_matches_mode(

1029 source: &Display,

1030 requested: VirtualMode,

1031 requested_rotation: Rotation,

1032) -> bool {

1033 let width = source.rect.right - source.rect.left;

1034 let height = source.rect.bottom - source.rect.top;

1035 width == requested.width as i32

1036 && height == requested.height as i32

1037 && source.rotation == native_topology_rotation(requested_rotation)

1038 && u64::from(source.refresh_numerator) * u64::from(requested.refresh_denominator)

1039 == u64::from(requested.refresh_numerator) * u64::from(source.refresh_denominator)

1040}

1041

1042fn stage(stage: &'static str, error: impl FmtDisplay) -> MappingError {

1043 MappingError {

1044 stage,

1045 group_id: None,

1046 message: error.to_string(),

1047 cleanup_incomplete: false,

1048 }

1049}

1050

1051fn group_stage(group_id: u32, stage: &'static str, error: impl FmtDisplay) -> MappingError {

1052 MappingError {

1053 stage,

1054 group_id: Some(group_id),

1055 message: error.to_string(),

1056 cleanup_incomplete: false,

1057 }

1058}

1059

1060#[cfg(test)]

1061mod tests {

1062 use super::*;

1063 use std::sync::atomic::AtomicBool;

1064

1065 fn group(id: u32, route: MappingRoute) -> MappingGroupRequest {

1066 MappingGroupRequest {

1067 id,

1068 mode: VirtualMode::default(),

1069 rotation: Rotation::Deg0,

1070 route,

1071 }

1072 }

1073

1074 fn virtual_source(connector_index: u32) -> Display {

1075 Display {

1076 id: format!("virtual-{connector_index}"),

1077 connector_index,

1078 sunshine_id: Some(format!("sunshine-{connector_index}")),

1079 name: "SBMS".into(),

1080 device_name: format!(r"\\.\DISPLAY{}", connector_index + 20),

1081 rect: windows::Win32::Foundation::RECT {

1082 left: 0,

1083 top: 0,

1084 right: 3840,

1085 bottom: 2160,

1086 },

1087 native_width: 3840,

1088 native_height: 2160,

1089 physical_width_mm: None,

1090 physical_height_mm: None,

1091 rotation: Rotation::Deg0,

1092 refresh_numerator: 240,

1093 refresh_denominator: 1,

1094 primary: false,

1095 virtual_display: true,

1096 }

1097 }

1098

1099 #[test]

1100 fn mixed_plan_preserves_stable_connector_ids() {

1101 let plan = MappingPlan::new(vec![

1102 group(

1103 3,

1104 MappingRoute::Mirror {

1105 target: "physical-a".into(),

1106 },

1107 ),

1108 group(1, MappingRoute::StreamOnly),

1109 ])

1110 .unwrap();

1111

1112 assert_eq!(plan.groups[0].id, 3);

1113 assert_eq!(plan.groups[1].id, 1);

1114 }

1115

1116 #[test]

1117 fn invalid_plans_fail_before_starting_hardware() {

1118 assert!(MappingPlan::new(Vec::new()).is_err());

1119 assert!(

1120 MappingPlan::new(vec![

1121 group(0, MappingRoute::StreamOnly),

1122 group(0, MappingRoute::StreamOnly),

1123 ])

1124 .is_err()

1125 );

1126 assert!(

1127 MappingPlan::new(vec![

1128 group(

1129 0,

1130 MappingRoute::Mirror {

1131 target: "same".into(),

1132 },

1133 ),

1134 group(

1135 1,

1136 MappingRoute::Mirror {

1137 target: "SAME".into(),

1138 },

1139 ),

1140 ])

1141 .is_err()

1142 );

1143 assert!(

1144 MappingPlan::new(vec![group(MAX_OUTPUTS as u32, MappingRoute::StreamOnly,)]).is_err()

1145 );

1146 }

1147

1148 #[test]

1149 fn cancelled_start_stops_before_hardware_and_invalid_plan_still_wins() {

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

1151 let valid = MappingPlan {

1152 groups: vec![group(0, MappingRoute::StreamOnly)],

1153 };

1154 let Err(cancelled) =

1155 MappingSession::start_plan_with_reporter_cancellable(valid, Arc::new(|_| {}), &cancel)

1156 else {

1157 panic!("pre-cancelled mapping unexpectedly started");

1158 };

1159 assert!(cancelled.is_clean_cancellation());

1160 let mut message_only = start_cancelled();

1161 message_only

1162 .message

1163 .push_str("; rollback was incomplete: cleanup failed");

1164 assert!(message_only.is_clean_cancellation());

1165

1166 let mut incomplete = start_cancelled();

1167 incomplete.record_cleanup_failure("cleanup failed");

1168 assert!(!incomplete.is_clean_cancellation());

1169 assert!(incomplete.message().contains("rollback was incomplete"));

1170

1171 let invalid = MappingPlan { groups: Vec::new() };

1172 let Err(plan_error) = MappingSession::start_plan_with_reporter_cancellable(

1173 invalid,

1174 Arc::new(|_| {}),

1175 &cancel,

1176 ) else {

1177 panic!("invalid mapping unexpectedly started");

1178 };

1179 assert_eq!(plan_error.stage(), "plan");

1180 assert!(!plan_error.is_clean_cancellation());

1181 }

1182

1183 #[test]

1184 fn plan_json_is_an_explicit_core_interface() {

1185 let json = r#"{

1186 "groups": [

1187 {

1188 "id": 0,

1189 "mode": {

1190 "width": 3840,

1191 "height": 2160,

1192 "refresh_numerator": 240,

1193 "refresh_denominator": 1

1194 },

1195 "route": { "kind": "stream_only" }

1196 }

1197 ]

1198 }"#;

1199

1200 let plan: MappingPlan = serde_json::from_str(json).unwrap();

1201 plan.validate().unwrap();

1202 assert_eq!(plan.groups[0].route, MappingRoute::StreamOnly);

1203 assert_eq!(plan.groups[0].rotation, Rotation::Deg0);

1204 }

1205

1206 #[test]

1207 fn native_modes_keep_final_dimensions_and_rebase_orientation() {

1208 let native = VirtualMode {

1209 width: 1800,

1210 height: 2880,

1211 refresh_numerator: 120,

1212 refresh_denominator: 1,

1213 };

1214 assert_eq!(native.width, 1800);

1215 assert_eq!(native.height, 2880);

1216 assert_eq!(native_topology_rotation(Rotation::Deg0), Rotation::Deg0);

1217 assert_eq!(native_topology_rotation(Rotation::Deg90), Rotation::Deg0);

1218 assert_eq!(native_topology_rotation(Rotation::Deg180), Rotation::Deg0);

1219 assert_eq!(native_topology_rotation(Rotation::Deg270), Rotation::Deg0);

1220 }

1221

1222 #[test]

1223 fn native_portrait_source_satisfies_a_270_degree_request() {

1224 let requested = VirtualMode {

1225 width: 1800,

1226 height: 2880,

1227 refresh_numerator: 120,

1228 refresh_denominator: 1,

1229 };

1230 let source = Display {

1231 id: "virtual".into(),

1232 connector_index: 1,

1233 sunshine_id: None,

1234 name: "SBMS".into(),

1235 device_name: r"\\.\DISPLAY73".into(),

1236 rect: windows::Win32::Foundation::RECT {

1237 left: 0,

1238 top: 0,

1239 right: 1800,

1240 bottom: 2880,

1241 },

1242 native_width: 1800,

1243 native_height: 2880,

1244 physical_width_mm: None,

1245 physical_height_mm: None,

1246 rotation: Rotation::Deg0,

1247 refresh_numerator: 120,

1248 refresh_denominator: 1,

1249 primary: false,

1250 virtual_display: true,

1251 };

1252

1253 assert!(source_matches_mode(&source, requested, Rotation::Deg270));

1254 }

1255

1256 #[test]

1257 fn topology_timeout_scales_only_above_the_previous_eight_group_limit() {

1258 assert_eq!(topology_timeout(1), Duration::from_secs(15));

1259 assert_eq!(topology_timeout(8), Duration::from_secs(15));

1260 assert_eq!(topology_timeout(9), Duration::from_secs(17));

1261 assert_eq!(topology_timeout(16), Duration::from_secs(31));

1262 }

1263

1264 #[test]

1265 fn topology_snapshot_explains_an_absent_connector() {

1266 let plan = MappingPlan::new(vec![group(1, MappingRoute::StreamOnly)]).unwrap();

1267

1268 assert_eq!(

1269 format_topology_snapshot(&plan, &[]),

1270 "group 1 absent requested=3840x2160@240/1 orientation=0 topology_rotation=0"

1271 );

1272 }

1273

1274 #[test]

1275 fn hydrated_sources_follow_plan_order() {

1276 let plan = MappingPlan::new(vec![

1277 group(3, MappingRoute::StreamOnly),

1278 group(1, MappingRoute::StreamOnly),

1279 ])

1280 .unwrap();

1281 let displays = vec![virtual_source(1), virtual_source(3)];

1282

1283 let ordered = ordered_ready_sources(&plan, &displays).unwrap();

1284

1285 assert_eq!(

1286 ordered

1287 .iter()

1288 .map(|display| display.connector_index)

1289 .collect::<Vec<_>>(),

1290 vec![3, 1]

1291 );

1292 assert_eq!(ordered[0].sunshine_id.as_deref(), Some("sunshine-3"));

1293 }

1294

1295 #[test]

1296 fn hydrated_sources_reject_changed_or_cloned_topology() {

1297 let plan = MappingPlan::new(vec![

1298 group(0, MappingRoute::StreamOnly),

1299 group(1, MappingRoute::StreamOnly),

1300 ])

1301 .unwrap();

1302 let first = virtual_source(0);

1303 let mut second = virtual_source(1);

1304 second.device_name = first.device_name.clone();

1305 assert!(ordered_ready_sources(&plan, &[first.clone(), second]).is_none());

1306

1307 let mut wrong_mode = virtual_source(1);

1308 wrong_mode.rect.right = 1920;

1309 assert!(ordered_ready_sources(&plan, &[first, wrong_mode]).is_none());

1310 }

1311

1312 #[test]

1313 fn recovery_requires_three_identical_topology_samples() {

1314 let signature = recovery_topology_signature(&[virtual_source(0)]);

1315 let mut stability = RecoveryStability::default();

1316

1317 assert!(!stability.observe(Some(signature.clone())));

1318 assert!(!stability.observe(Some(signature.clone())));

1319 assert!(stability.observe(Some(signature)));

1320 }

1321

1322 #[test]

1323 fn recovery_stability_resets_when_position_changes_or_source_disappears() {

1324 let first = virtual_source(0);

1325 let first_signature = recovery_topology_signature(std::slice::from_ref(&first));

1326 let mut moved = first;

1327 moved.rect.left = -3840;

1328 moved.rect.right = 0;

1329 let moved_signature = recovery_topology_signature(&[moved]);

1330 let mut stability = RecoveryStability::default();

1331

1332 assert!(!stability.observe(Some(first_signature.clone())));

1333 assert!(!stability.observe(Some(first_signature)));

1334 assert!(!stability.observe(Some(moved_signature.clone())));

1335 assert!(!stability.observe(None));

1336 assert!(!stability.observe(Some(moved_signature.clone())));

1337 assert!(!stability.observe(Some(moved_signature.clone())));

1338 assert!(stability.observe(Some(moved_signature)));

1339 }

1340

1341 #[test]

1342 fn renderer_events_are_buffered_until_the_candidate_is_installed() {

1343 let (tx, rx) = std::sync::mpsc::channel();

1344 let reporter: MappingReporter = Arc::new(move |event| tx.send(event).unwrap());

1345 let gate = RendererEventGate::new(3, reporter);

1346

1347 gate.report(RendererEvent::TopologyLost);

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

1349 gate.activate();

1350 assert!(matches!(

1351 rx.recv().unwrap(),

1352 MappingEvent::Renderer {

1353 id: 3,

1354 event: RendererEvent::TopologyLost

1355 }

1356 ));

1357 }

1358

1359 #[test]

1360 fn renderer_events_from_discarded_candidates_stay_suppressed() {

1361 let (tx, rx) = std::sync::mpsc::channel();

1362 let reporter: MappingReporter = Arc::new(move |event| tx.send(event).unwrap());

1363 let gate = RendererEventGate::new(4, reporter);

1364

1365 gate.report(RendererEvent::TopologyLost);

1366 gate.discard();

1367 gate.report(RendererEvent::Failed("stale".into()));

1368

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

1370 }

1371}

1372