diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index 272eee671..0028896aa 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -259,7 +259,7 @@ async fn lineup_status( ) -> impl IntoResponse { let current_state = app_state .hd_scan_state - .load(std::sync::atomic::Ordering::SeqCst); + .load(std::sync::atomic::Ordering::Acquire); if current_state < 0 { axum::Json(json!({ "ScanInProgress": 0, @@ -301,11 +301,11 @@ async fn lineup_status( if final_state >= 100 { app_state .hd_scan_state - .store(-1, std::sync::atomic::Ordering::SeqCst); + .store(-1, std::sync::atomic::Ordering::Release); } else { app_state .hd_scan_state - .store(final_state, std::sync::atomic::Ordering::SeqCst); + .store(final_state, std::sync::atomic::Ordering::Release); } let found = (num_of_channels * usize::try_from(final_state).unwrap_or(1)) / 100; axum::Json(json!({ @@ -330,13 +330,13 @@ async fn lineup_post( "start" => { app_state .hd_scan_state - .store(0, std::sync::atomic::Ordering::SeqCst); + .store(0, std::sync::atomic::Ordering::Release); axum::http::StatusCode::OK.into_response() } "abort" => { app_state .hd_scan_state - .store(-1, std::sync::atomic::Ordering::SeqCst); + .store(-1, std::sync::atomic::Ordering::Release); axum::http::StatusCode::OK.into_response() } _ => axum::http::StatusCode::BAD_REQUEST.into_response(), diff --git a/backend/src/api/model/provider_lineup_manager.rs b/backend/src/api/model/provider_lineup_manager.rs index c9934736d..c7aa77567 100644 --- a/backend/src/api/model/provider_lineup_manager.rs +++ b/backend/src/api/model/provider_lineup_manager.rs @@ -319,7 +319,7 @@ impl MultiProviderLineup { let p = &pg[idx]; let result = p.try_allocate(grace, grace_period_timeout_secs).await; if !matches!(result, ProviderAllocation::Exhausted) { - index.store((idx + 1) % provider_count, Ordering::SeqCst); + index.store((idx + 1) % provider_count, Ordering::Release); return result; } @@ -331,7 +331,7 @@ impl MultiProviderLineup { } } - index.store(idx, Ordering::SeqCst); + index.store(idx, Ordering::Release); } } ProviderAllocation::Exhausted @@ -352,7 +352,7 @@ impl MultiProviderLineup { let p = &pg[idx]; let result = p.get_next(grace, grace_period_timeout_secs).await; if result.is_some() { - index.store((idx + 1) % provider_count, Ordering::SeqCst); + index.store((idx + 1) % provider_count, Ordering::Release); return result; } @@ -364,7 +364,7 @@ impl MultiProviderLineup { } } - index.store(idx, Ordering::SeqCst); + index.store(idx, Ordering::Release); } } None @@ -397,7 +397,7 @@ impl MultiProviderLineup { /// ``` async fn acquire(&self, with_grace: bool, grace_period_timeout_secs: u64) -> ProviderAllocation { let provider_count = self.providers.len(); - let start = self.index.fetch_add(1, Ordering::SeqCst) % provider_count; + let start = self.index.fetch_add(1, Ordering::Acquire) % provider_count; let mut idx = start; loop { @@ -414,7 +414,7 @@ impl MultiProviderLineup { if !matches!(allocation, ProviderAllocation::Exhausted) { if priority_group.is_exhausted().await { - self.index.store((idx + 1) % provider_count, Ordering::SeqCst); + self.index.store((idx + 1) % provider_count, Ordering::Release); } return allocation; } @@ -435,7 +435,7 @@ impl MultiProviderLineup { async fn get_next(&self, grace_period_timeout_secs: u64) -> Option> { let provider_count = self.providers.len(); - let start = self.index.fetch_add(1, Ordering::SeqCst) % provider_count; + let start = self.index.fetch_add(1, Ordering::Acquire) % provider_count; let mut idx = start; loop { @@ -452,7 +452,7 @@ impl MultiProviderLineup { if let Some(config) = allocation { if priority_group.is_exhausted().await { - self.index.store((idx + 1) % provider_count, Ordering::SeqCst); + self.index.store((idx + 1) % provider_count, Ordering::Release); } return Some(config); } @@ -682,8 +682,8 @@ impl ProviderLineupManager { let providers = self.providers.load(); let allocation = match Self::get_provider_config_by_name(input_name, &providers) { None => ProviderAllocation::Exhausted, // No Name matched, we don't have this provider - Some((lineup, _config)) => lineup.acquire(self.grace_period_millis.load(Ordering::SeqCst) > 0, - self.grace_period_timeout_secs.load(Ordering::SeqCst)).await + Some((lineup, _config)) => lineup.acquire(self.grace_period_millis.load(Ordering::Acquire) > 0, + self.grace_period_timeout_secs.load(Ordering::Acquire)).await }; Self::log_allocation(&allocation); allocation @@ -1059,14 +1059,14 @@ mod tests { let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { match lineup_clone.acquire(true, 5).await { - ProviderAllocation::Exhausted => exhausted.fetch_sub(1, Ordering::SeqCst), - ProviderAllocation::Available(_) => available.fetch_sub(1, Ordering::SeqCst), - ProviderAllocation::GracePeriod(_) => grace_period.fetch_sub(1, Ordering::SeqCst), + ProviderAllocation::Exhausted => exhausted.fetch_sub(1, Ordering::Acquire), + ProviderAllocation::Available(_) => available.fetch_sub(1, Ordering::Acquire), + ProviderAllocation::GracePeriod(_) => grace_period.fetch_sub(1, Ordering::Acquire), } }); } - assert_eq!(exhausted_count.load(Ordering::SeqCst), 0); - assert_eq!(available_count.load(Ordering::SeqCst), 0); - assert_eq!(grace_period_count.load(Ordering::SeqCst), 0); + assert_eq!(exhausted_count.load(Ordering::Acquire), 0); + assert_eq!(available_count.load(Ordering::Acquire), 0); + assert_eq!(grace_period_count.load(Ordering::Acquire), 0); } } diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 106d30740..91b4bac9f 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -50,7 +50,7 @@ where let chunk_len = chunk.len(); if self.current_offset >= chunk_len { if let Some(finished) = self.pending_writes.pop_front() { - self.size.fetch_add(finished.len(), Ordering::SeqCst); + self.size.fetch_add(finished.len(), Ordering::Acquire); } self.current_offset = 0; continue; @@ -96,7 +96,7 @@ where fn finalize(&mut self) { if !self.completed { self.completed = true; - let size = self.size.load(Ordering::SeqCst); + let size = self.size.load(Ordering::Acquire); debug!("Persisted {size} bytes to cache resource"); (self.callback)(size); } diff --git a/backend/src/api/model/streams/provider_stream_factory.rs b/backend/src/api/model/streams/provider_stream_factory.rs index 841c85727..cf30108c5 100644 --- a/backend/src/api/model/streams/provider_stream_factory.rs +++ b/backend/src/api/model/streams/provider_stream_factory.rs @@ -144,7 +144,7 @@ impl ProviderStreamFactoryOptions { self.range_bytes .as_ref() .as_ref() - .map(|atomic| atomic.load(Ordering::SeqCst)) + .map(|atomic| atomic.load(Ordering::Acquire)) } // pub fn get_range_bytes(&self) -> &Arc> { diff --git a/backend/src/api/scheduler.rs b/backend/src/api/scheduler.rs index d258696ca..3cfbfddab 100644 --- a/backend/src/api/scheduler.rs +++ b/backend/src/api/scheduler.rs @@ -119,7 +119,7 @@ mod tests { let expression = "0/1 * * * * * *"; // every second let runs = AtomicU8::new(0); - let run_me = || runs.fetch_add(1, Ordering::SeqCst); + let run_me = || runs.fetch_add(1, Ordering::Acquire); let start = std::time::Instant::now(); if let Ok(schedule) = Schedule::from_str(expression) { @@ -130,14 +130,14 @@ mod tests { tokio::time::sleep_until(tokio::time::Instant::from(datetime_to_instant(datetime))).await; run_me(); } - if runs.load(Ordering::SeqCst) == 6 { + if runs.load(Ordering::Acquire) == 6 { break; } } } let duration = start.elapsed(); - assert!(runs.load(Ordering::SeqCst) == 6, "Failed to run"); + assert!(runs.load(Ordering::Acquire) == 6, "Failed to run"); assert!(duration.as_secs() > 4, "Failed time"); } } \ No newline at end of file diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index ea130548a..30d1d16fc 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -241,7 +241,7 @@ fn map_playlist_counter(target: &ConfigTarget, playlist: &mut [PlaylistGroup]) { for channel in &mut plg.channels { let provider = ValueProvider { pli: channel }; if counter.filter.filter(&provider) { - let cntval = counter.value.fetch_add(1, core::sync::atomic::Ordering::SeqCst); + let cntval = counter.value.fetch_add(1, core::sync::atomic::Ordering::Acquire); let padded_cntval = if counter.padding > 0 { format!("{:0width$}", cntval, width = counter.padding as usize) } else { diff --git a/backend/src/tools/atomic_once_flag.rs b/backend/src/tools/atomic_once_flag.rs index 5da51c174..cf549da9d 100644 --- a/backend/src/tools/atomic_once_flag.rs +++ b/backend/src/tools/atomic_once_flag.rs @@ -36,13 +36,13 @@ impl AtomicOnceFlag { /// /// This operation is atomic and uses the specified memory ordering. pub fn notify(&self) { - self.enabled.store(false, Ordering::SeqCst); + self.enabled.store(false, Ordering::Release); } /// Checks if the flag is still active. /// /// Returns `true` if the flag is active (initial state). Returns `false` if the flag has been disabled. pub fn is_active(&self) -> bool { - self.enabled.load(Ordering::SeqCst) + self.enabled.load(Ordering::Acquire) } } \ No newline at end of file diff --git a/backend/src/utils/file/file_lock_manager.rs b/backend/src/utils/file/file_lock_manager.rs index a2ec03844..943107bc6 100644 --- a/backend/src/utils/file/file_lock_manager.rs +++ b/backend/src/utils/file/file_lock_manager.rs @@ -8,9 +8,15 @@ use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard}; use shared::error::str_to_io_error; use path_clean::PathClean; +#[derive(Clone, PartialEq, Eq, Hash)] +enum LockKey { + Path(PathBuf), + Str(String), +} + #[derive(Clone)] pub struct FileLockManager { - locks: Arc>>>>, + locks: Arc>>>>, } impl FileLockManager { @@ -22,46 +28,55 @@ impl FileLockManager { // Acquires a read lock for the specified file and returns a FileReadGuard. pub async fn read_lock(&self, path: &Path) -> FileReadGuard { - let file_lock = self.get_or_create_lock(path).await; + let file_lock = self.get_or_create_lock(Self::get_lock_key_for_path(path)).await; let guard = Arc::clone(&file_lock).read_owned().await; FileReadGuard::new(guard) } // Acquires a write lock for the specified file and returns a FileWriteGuard. pub async fn write_lock(&self, path: &Path) -> FileWriteGuard { - let file_lock = self.get_or_create_lock(path).await; + let file_lock = self.get_or_create_lock(Self::get_lock_key_for_path(path)).await; let guard = Arc::clone(&file_lock).write_owned().await; FileWriteGuard::new(guard) } // Tries to acquire a write lock for the specified file and returns a FileWriteGuard. pub async fn try_write_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path).await; + let file_lock = self.get_or_create_lock(Self::get_lock_key_for_path(path)).await; match Arc::clone(&file_lock).try_write_owned() { Ok(lock_guard) => Ok(FileWriteGuard::new(lock_guard)), Err(_) => Err(str_to_io_error("Failed to acquire write lock")) } } + /// Acquires a write lock using a raw string key instead of a normalized `Path`. + /// + /// Unlike the standard path-based locks, this method does **not** perform any + /// path normalization or conversion. The string is used directly as the lock key, + /// which can be useful for non-file-based identifiers or dynamic keys. pub async fn write_lock_str(&self, text: &str) -> FileWriteGuard { - let path = PathBuf::from(text); - let file_lock = self.get_or_create_lock(&path).await; + let lock_key = LockKey::Str(text.to_string()); + let file_lock = self.get_or_create_lock(lock_key).await; let guard = Arc::clone(&file_lock).write_owned().await; FileWriteGuard::new(guard) } - // Helper function: retrieves or creates a lock for a file. - async fn get_or_create_lock(&self, path: &Path) -> Arc> { + fn get_lock_key_for_path(path: &Path) -> LockKey { let normalized_path = normalize_path(path); + LockKey::Path(normalized_path) + } + + // Helper function: retrieves or creates a lock for a file. + async fn get_or_create_lock(&self, lock_key: LockKey) -> Arc> { let mut locks = self.locks.lock().await; - if let Some(lock) = locks.get(&normalized_path) { + if let Some(lock) = locks.get(&lock_key) { return lock.clone(); } let file_lock = Arc::new(RwLock::new(())); - locks.insert(normalized_path, file_lock.clone()); + locks.insert(lock_key, file_lock.clone()); drop(locks); file_lock } @@ -118,3 +133,16 @@ fn normalize_path(path: &Path) -> PathBuf { base.clean() } + + +#[cfg(test)] +mod test { + use super::*; + #[test] + fn test_normalize_path() { + let path = PathBuf::from("https://10.1.41.41"); + let normalized = normalize_path(&path); + assert_eq!(normalized.display().to_string(), "dsd sdf sf d"); + + } +} \ No newline at end of file diff --git a/frontend/src/services/config_service.rs b/frontend/src/services/config_service.rs index 305dbc425..e0dce14c6 100644 --- a/frontend/src/services/config_service.rs +++ b/frontend/src/services/config_service.rs @@ -56,7 +56,7 @@ impl ConfigService { } async fn fetch_server_config(&self) { - if self.is_fetching.swap(true, Ordering::SeqCst) { + if self.is_fetching.swap(true, Ordering::AcqRel) { return; } let result = match request_get::(&self.config_path, None, None).await { @@ -109,7 +109,7 @@ impl ConfigService { }; self.server_config.replace(result.clone()); self.config_channel.set(result); - self.is_fetching.store(false, Ordering::SeqCst); + self.is_fetching.store(false, Ordering::Release); } pub async fn get_ip_info(&self) -> Option { diff --git a/frontend/src/services/event_service.rs b/frontend/src/services/event_service.rs index d466b0937..459cff90b 100644 --- a/frontend/src/services/event_service.rs +++ b/frontend/src/services/event_service.rs @@ -53,7 +53,7 @@ impl EventService { } pub fn subscribe(&self, callback: F) -> usize { - let sub_id = self.subscriber_id.fetch_add(1, Ordering::SeqCst); + let sub_id = self.subscriber_id.fetch_add(1, Ordering::Acquire); self.subscribers.borrow_mut().insert(sub_id, Box::new(callback)); sub_id } diff --git a/frontend/src/services/websocket_service.rs b/frontend/src/services/websocket_service.rs index c94df6414..d0cb15073 100644 --- a/frontend/src/services/websocket_service.rs +++ b/frontend/src/services/websocket_service.rs @@ -79,7 +79,7 @@ impl WebSocketService { } pub fn connect_ws_with_backoff(&self) { - if self.connected.load(Ordering::SeqCst) { + if self.connected.load(Ordering::Acquire) { return; } match WebSocket::new(&self.ws_path) { @@ -115,7 +115,7 @@ impl WebSocketService { // on open is called on a connect attempt, it does not mean it is connected! trace!("WebSocket connection opened."); if Self::try_send_message(ws_open_clone.borrow().as_ref(), ProtocolMessage::Version(PROTOCOL_VERSION)) { - connected_clone.store(true, Ordering::SeqCst); + connected_clone.store(true, Ordering::Release); } })); socket.set_onopen(Some(onopen_callback.as_ref().unchecked_ref())); @@ -158,7 +158,7 @@ impl WebSocketService { // it will be dropped when the service or field is taken elsewhere if desired. *ws_close_rc.borrow_mut() = None; - connected_clone.store(false, Ordering::SeqCst); + connected_clone.store(false, Ordering::Release); event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); // schedule reconnect @@ -177,7 +177,7 @@ impl WebSocketService { let onerror_callback = Closure::::wrap(Box::new(move |e: ErrorEvent| { error!("WebSocket error: {:?}", e); - connected_clone.store(false, Ordering::SeqCst); + connected_clone.store(false, Ordering::Release); event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); // ws_service_reconnect_clone.schedule_reconnect(); })); @@ -190,7 +190,7 @@ impl WebSocketService { fn schedule_reconnect(&self) { // increment attempts atomically and get the previous value - let attempt = self.attempt_counter.fetch_add(1, Ordering::SeqCst) as u32; + let attempt = self.attempt_counter.fetch_add(1, Ordering::Acquire) as u32; if attempt >= WS_RECONNECT_MAX_ATTEMPTS { warn!("WebSocket reconnect attempts exceeded ({attempt}). Giving up."); @@ -230,7 +230,7 @@ impl WebSocketService { } pub async fn get_server_status(&self) { - if self.connected.load(Ordering::SeqCst) { + if self.connected.load(Ordering::Acquire) { if let Some(token) = get_token() { self.send_message(ProtocolMessage::StatusRequest(token)); } @@ -295,7 +295,7 @@ fn handle_socket_protocol_msg(event: MessageEvent, event_service: &Rc { - attempt_counter.store(0, Ordering::SeqCst); + attempt_counter.store(0, Ordering::Release); event_service.broadcast(EventMessage::WebSocketStatus(true)); if let Some(token) = get_token() { return Some(ProtocolMessage::Auth(token)); diff --git a/shared/src/utils/request.rs b/shared/src/utils/request.rs index 465f7bcd3..bdf381f26 100644 --- a/shared/src/utils/request.rs +++ b/shared/src/utils/request.rs @@ -5,10 +5,10 @@ use crate::utils::{CONSTANTS, DASH_EXT, DASH_EXT_FRAGMENT, DASH_EXT_QUERY, HLS_E pub fn set_sanitize_sensitive_info(value: bool) { - CONSTANTS.sanitize.store(value, Ordering::SeqCst); + CONSTANTS.sanitize.store(value, Ordering::Relaxed); } pub fn sanitize_sensitive_info(query: &str) -> Cow<'_, str> { - if !CONSTANTS.sanitize.load(Ordering::SeqCst) { + if !CONSTANTS.sanitize.load(Ordering::Relaxed) { return Cow::Borrowed(query); }