diff --git a/config/config.yml b/config/config.yml index bb4c59638..929d2fb02 100644 --- a/config/config.yml +++ b/config/config.yml @@ -6,6 +6,7 @@ update_on_boot: false custom_stream_response: channel_unavailable: /home/m3u-filter/channel_unavailable.ts user_connections_exhausted: /home/m3u-filter/user_connections_exhausted.ts + provider_connections_exhausted: /home/m3u-filter/provider_connections_exhausted.ts # sec min hour day of month month day of week year schedules: diff --git a/src/api/model/active_provider_manager.rs b/src/api/model/active_provider_manager.rs index a775da15f..c561b3c77 100644 --- a/src/api/model/active_provider_manager.rs +++ b/src/api/model/active_provider_manager.rs @@ -65,12 +65,12 @@ impl ProviderConfig { #[inline] pub fn is_exhausted(&self) -> bool { - self.max_connections > 0 && self.current_connections.load(Ordering::Acquire) >= self.max_connections + self.max_connections > 0 && self.current_connections.load(Ordering::SeqCst) >= self.max_connections } #[inline] pub fn is_over_limit(&self) -> bool { - self.max_connections > 0 && self.current_connections.load(Ordering::Acquire) > self.max_connections + self.max_connections > 0 && self.current_connections.load(Ordering::SeqCst) > self.max_connections } // @@ -80,26 +80,26 @@ impl ProviderConfig { // } pub fn try_allocate(&self, grace: bool) -> ProviderAllocation { - let connections = self.current_connections.load(Ordering::Acquire); + let connections = self.current_connections.load(Ordering::SeqCst); if self.max_connections == 0 { return ProviderAllocation::Available(self); } if (!grace && connections < self.max_connections) || (grace && connections <= self.max_connections) { - self.current_connections.fetch_add(1, Ordering::AcqRel); + self.current_connections.fetch_add(1, Ordering::SeqCst); return if connections < self.max_connections { ProviderAllocation::Available(self) } else { ProviderAllocation::GracePeriod(self) }; } ProviderAllocation::Exhausted } pub fn release(&self) { - let connections = self.current_connections.load(Ordering::Acquire); + let connections = self.current_connections.load(Ordering::SeqCst); if connections > 0 { - self.current_connections.fetch_sub(1, Ordering::AcqRel); + self.current_connections.fetch_sub(1, Ordering::SeqCst); } } pub fn get_connection(&self) -> u16 { - self.current_connections.load(Ordering::Acquire) + self.current_connections.load(Ordering::SeqCst) } } @@ -182,6 +182,7 @@ impl ProviderPriorityGroup { /// Manages multiple providers, ensuring that connections are allocated in a round-robin manner based on priority. +#[repr(align(64))] #[derive(Debug)] struct MultiProviderLineup { providers: Vec, @@ -256,7 +257,7 @@ impl MultiProviderLineup { } } ProviderPriorityGroup::MultiProviderGroup(index, pg) => { - let mut idx = index.load(Ordering::Acquire); + let mut idx = index.load(Ordering::SeqCst); let provider_count = pg.len(); for _ in idx..provider_count { let p = pg.get(idx).unwrap(); @@ -265,12 +266,12 @@ impl MultiProviderLineup { match result { ProviderAllocation::Exhausted => {} ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => { - index.store(idx, Ordering::Release); + index.store(idx, Ordering::SeqCst); return result; } } } - index.store(idx, Ordering::Release); + index.store(idx, Ordering::SeqCst); } } ProviderAllocation::Exhausted @@ -302,7 +303,7 @@ impl MultiProviderLineup { /// } /// ``` fn acquire(&self) -> ProviderAllocation { - let main_idx = self.index.load(Ordering::Acquire); + let main_idx = self.index.load(Ordering::SeqCst); let provider_count = self.providers.len(); for index in main_idx..provider_count { @@ -320,7 +321,7 @@ impl MultiProviderLineup { ProviderAllocation::Available(_) | ProviderAllocation::GracePeriod(_) => { if priority_group.is_exhausted() { - self.index.store((index + 1) % provider_count, Ordering::Release); + self.index.store((index + 1) % provider_count, Ordering::SeqCst); } return allocation; } @@ -329,13 +330,13 @@ impl MultiProviderLineup { ProviderAllocation::Exhausted // let provider = &self.providers[main_idx]; - // self.index.store((main_idx + 1) % provider_count, Ordering::Release); + // self.index.store((main_idx + 1) % provider_count, Ordering::SeqCst); // // match provider { // ProviderPriorityGroup::SingleProviderGroup(p) => ProviderAllocation::Available(p), // ProviderPriorityGroup::MultiProviderGroup(gindex, group) => { - // let idx = gindex.load(Ordering::Acquire); - // gindex.store((idx + 1) % group.len(), Ordering::Release); + // let idx = gindex.load(Ordering::SeqCst); + // gindex.store((idx + 1) % group.len(), Ordering::SeqCst); // match group.get(idx) { // None => ProviderAllocation::Exhausted, // Some(p) => ProviderAllocation::Available(p) @@ -439,7 +440,7 @@ impl ActiveProviderManager { pub fn active_connections(&self) -> Option> { let result = RefCell::new(HashMap::::new()); let add_provider = |provider: &ProviderConfig| { - let count = provider.current_connections.load(Ordering::Acquire); + let count = provider.current_connections.load(Ordering::SeqCst); if count > 0 { result.borrow_mut().insert(provider.name.to_string(), count); } @@ -742,9 +743,9 @@ mod tests { let handle = thread::spawn(move || { // Each thread tries to acquire a connection match lineup_clone.acquire() { - ProviderAllocation::Exhausted => exhausted.fetch_sub(1, Ordering::Release), - ProviderAllocation::Available(_) => available.fetch_sub(1, Ordering::Release), - ProviderAllocation::GracePeriod(_) => grace_period.fetch_sub(1, Ordering::Release), + 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), } }); @@ -756,9 +757,9 @@ mod tests { handle.join().unwrap(); } - 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); + 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); } } diff --git a/src/api/model/active_user_manager.rs b/src/api/model/active_user_manager.rs index 181d452a0..0a54bdea5 100644 --- a/src/api/model/active_user_manager.rs +++ b/src/api/model/active_user_manager.rs @@ -21,7 +21,7 @@ impl ActiveUserManager { pub async fn user_connections(&self, username: &str) -> u32 { if let Some(counter) = self.user.read().await.get(username) { - return counter.load(Ordering::Acquire); + return counter.load(Ordering::SeqCst); } 0 } @@ -31,13 +31,13 @@ impl ActiveUserManager { } pub async fn active_connections(&self) -> usize { - self.user.read().await.values().map(|c| c.load(Ordering::Acquire) as usize).sum() + self.user.read().await.values().map(|c| c.load(Ordering::SeqCst) as usize).sum() } pub async fn add_connection(&self, username: &str) -> (usize, usize) { let mut lock = self.user.write().await; if let Some(counter) = lock.get(username) { - counter.fetch_add(1, Ordering::AcqRel); + counter.fetch_add(1, Ordering::SeqCst); } else { lock.insert(username.to_string(), AtomicU32::new(1)); } @@ -48,11 +48,53 @@ impl ActiveUserManager { pub async fn remove_connection(&self, username: &str) -> (usize, usize) { let mut lock = self.user.write().await; if let Some(counter) = lock.get(username) { - if counter.fetch_sub(1, Ordering::AcqRel) == 1 { + if counter.fetch_sub(1, Ordering::SeqCst) == 1 { lock.remove(username); } } drop(lock); (self.active_users().await, self.active_connections().await) } -} \ No newline at end of file +} + +// +// mod tests { +// use std::sync::{Arc, atomic::{AtomicUsize, Ordering}}; +// use std::time::Instant; +// use std::thread; +// +// fn benchmark(ordering: Ordering, iterations: usize) -> u128 { +// let counter = Arc::new(AtomicUsize::new(0)); +// let start = Instant::now(); +// +// let handles: Vec<_> = (0..32) +// .map(|_| { +// let counter_ref = Arc::clone(&counter); +// thread::spawn(move || { +// for _ in 0..iterations { +// counter_ref.fetch_add(1, ordering); +// } +// }) +// }) +// .collect(); +// +// for handle in handles { +// handle.join().unwrap(); +// } +// +// let duration = start.elapsed(); +// duration.as_millis() +// } +// +// #[test] +// fn test_ordering() { +// let iterations = 1_000_000; +// +// let time_acqrel = benchmark(Ordering::SeqCst, iterations); +// println!("AcqRel: {} ms", time_acqrel); +// +// let time_seqcst = benchmark(Ordering::SeqCst, iterations); +// println!("SeqCst: {} ms", time_seqcst); +// } +// +// } diff --git a/src/api/model/hls_cache.rs b/src/api/model/hls_cache.rs index 2414dcb2b..78ee136ce 100644 --- a/src/api/model/hls_cache.rs +++ b/src/api/model/hls_cache.rs @@ -33,6 +33,7 @@ fn start_garbage_collector(cache: &Arc) { }); } +#[repr(align(64))] pub struct HlsCache { pub entries: RwLock>, counter: AtomicU32, @@ -64,9 +65,9 @@ impl HlsCache { } pub fn new_token(&self) -> u32 { - let token = self.counter.fetch_add(1, std::sync::atomic::Ordering::AcqRel); + let token = self.counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst); if token > TOKEN_MAX { - self.counter.store(1, std::sync::atomic::Ordering::Release); + self.counter.store(1, std::sync::atomic::Ordering::SeqCst); return 1; } return token; diff --git a/src/api/model/streams/active_client_stream.rs b/src/api/model/streams/active_client_stream.rs index 16f78c078..985e16a89 100644 --- a/src/api/model/streams/active_client_stream.rs +++ b/src/api/model/streams/active_client_stream.rs @@ -15,6 +15,7 @@ use crate::api::model::streams::chunked_buffer::ChunkedBuffer; const GRACE_PERIOD_SECONDS: u64 = 2; +#[repr(align(64))] pub(in crate::api) struct ActiveClientStream { inner: BoxedProviderStream, username: String, @@ -62,7 +63,7 @@ impl ActiveClientStream { tokio::time::sleep(tokio::time::Duration::from_secs(GRACE_PERIOD_SECONDS)).await; if provider_manager.is_over_limit(&provider_name) { info!("is over limit for active clients: {provider_name}"); - stop_stream_flag.store(true, std::sync::atomic::Ordering::Release); + stop_stream_flag.store(true, std::sync::atomic::Ordering::SeqCst); if let Some(connect_flag) = reconnect_flag { info!("stopped reconnect"); connect_flag.notify(); @@ -80,7 +81,7 @@ impl Stream for ActiveClientStream { fn poll_next(mut self: Pin<&mut Self>,cx: &mut std::task::Context<'_>,) -> Poll> { if let Some(send_custom_stream_flag) = &self.send_custom_stream_flag { - if send_custom_stream_flag.load(std::sync::atomic::Ordering::Acquire) { + if send_custom_stream_flag.load(std::sync::atomic::Ordering::SeqCst) { return match self.custom_video.as_mut() { None => { Poll::Ready(None) diff --git a/src/api/model/streams/client_stream.rs b/src/api/model/streams/client_stream.rs index 0923a202d..33bf67c1b 100644 --- a/src/api/model/streams/client_stream.rs +++ b/src/api/model/streams/client_stream.rs @@ -13,6 +13,7 @@ use crate::utils::network::request::sanitize_sensitive_info; /// This stream counts the send bytes for reconnecting to the actual position and /// sets the `close_signal` if the client drops the connection. +#[repr(align(64))] pub(in crate::api::model) struct ClientStream { inner: BoxedProviderStream, close_signal: Arc, @@ -42,7 +43,7 @@ impl Stream for ClientStream { } if let Some(counter) = self.total_bytes.as_ref() { - counter.fetch_add(bytes.len(), Ordering::AcqRel); + counter.fetch_add(bytes.len(), Ordering::SeqCst); } return Poll::Ready(Some(Ok(bytes))); diff --git a/src/api/model/streams/persist_pipe_stream.rs b/src/api/model/streams/persist_pipe_stream.rs index 6fd078895..f518b4f2d 100644 --- a/src/api/model/streams/persist_pipe_stream.rs +++ b/src/api/model/streams/persist_pipe_stream.rs @@ -53,7 +53,7 @@ where fn on_complete(&mut self) { if !self.completed { self.completed = true; - let size = self.size.load(Ordering::Acquire); + let size = self.size.load(Ordering::SeqCst); if self.writer.flush().is_ok() { (self.callback)(size); } @@ -62,7 +62,7 @@ where fn on_data(&mut self, data: &Result) { if let Ok(bytes) = data { - self.size.fetch_add(bytes.len(), Ordering::AcqRel); + self.size.fetch_add(bytes.len(), Ordering::SeqCst); let bytes_to_write = bytes.clone(); if let Err(e) = self.writer.write_all(&bytes_to_write) { error!("Error writing to resource file: {e}"); diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs index 11dc83e3b..17a11ab44 100644 --- a/src/api/model/streams/provider_stream_factory.rs +++ b/src/api/model/streams/provider_stream_factory.rs @@ -140,7 +140,7 @@ impl ProviderStreamOptions { #[inline] pub fn get_total_bytes_send(&self) -> Option { - self.range_bytes.as_ref().as_ref().map(|atomic| atomic.load(Ordering::Acquire)) + self.range_bytes.as_ref().as_ref().map(|atomic| atomic.load(Ordering::SeqCst)) } // pub fn get_range_bytes(&self) -> &Arc> { diff --git a/src/api/scheduler.rs b/src/api/scheduler.rs index 2969ddc8b..bac5c69ae 100644 --- a/src/api/scheduler.rs +++ b/src/api/scheduler.rs @@ -54,7 +54,7 @@ mod tests { let expression = "0/1 * * * * * *"; // every second let runs = AtomicU8::new(0); - let run_me = || runs.fetch_add(1, Ordering::AcqRel); + let run_me = || runs.fetch_add(1, Ordering::SeqCst); let start = std::time::Instant::now(); match Schedule::from_str(expression) { @@ -66,7 +66,7 @@ mod tests { tokio::time::sleep_until(tokio::time::Instant::from(datetime_to_instant(datetime))).await; run_me(); } - if runs.load(Ordering::Acquire) == 6 { + if runs.load(Ordering::SeqCst) == 6 { break; } } @@ -75,7 +75,7 @@ mod tests { }; let duration = start.elapsed(); - assert!(runs.load(Ordering::Acquire) == 6, "Failed to run"); + assert!(runs.load(Ordering::SeqCst) == 6, "Failed to run"); assert!(duration.as_secs() > 4, "Failed time"); } } \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 5895a0168..cffe64444 100644 --- a/src/main.rs +++ b/src/main.rs @@ -91,11 +91,19 @@ const BUILD_TIMESTAMP:&str = env!("VERGEN_BUILD_TIMESTAMP"); fn main() { let args = Args::parse(); - let env_log_level = std::env::var("M3U_FILTER_LOG"); + + if args.genpwd { + match generate_password() { + Ok(pwd) => println!("{pwd}"), + Err(err) => eprintln!("{err}"), + } + return; + } let config_path: String = args.config_path.unwrap_or_else(file_utils::get_default_config_path); let config_file: String = args.config_file.unwrap_or_else(|| file_utils::get_default_config_file_path(&config_path)); + let env_log_level = std::env::var("M3U_FILTER_LOG"); init_logger(args.log_level.as_ref(), env_log_level.ok(), config_file.as_str()); if args.healthcheck { @@ -107,14 +115,6 @@ fn main() { set_sanitize_sensitive_info(cfg.log.as_ref().is_none_or(|l| l.sanitize_sensitive_info)); - if args.genpwd { - match generate_password() { - Ok(pwd) => println!("{pwd}"), - Err(err) => error!("{err}"), - } - return; - } - let temp_path = PathBuf::from(&cfg.working_dir).join("tmp"); create_directories(&cfg, &temp_path); let _ = tempfile::env::override_temp_dir(&temp_path); diff --git a/src/processing/processor/playlist.rs b/src/processing/processor/playlist.rs index d2d2ef425..d0700fe9b 100644 --- a/src/processing/processor/playlist.rs +++ b/src/processing/processor/playlist.rs @@ -265,7 +265,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, &mut mock_processor) { - let cntval = counter.value.load(core::sync::atomic::Ordering::Acquire); + let cntval = counter.value.load(core::sync::atomic::Ordering::SeqCst); let new_value = if counter.modifier == CounterModifier::Assign { cntval.to_string() } else { @@ -277,7 +277,7 @@ fn map_playlist_counter(target: &ConfigTarget, playlist: &mut [PlaylistGroup]) { } }; channel.header.set_field(&counter.field, new_value.as_str()); - counter.value.fetch_add(1, core::sync::atomic::Ordering::AcqRel); + counter.value.fetch_add(1, core::sync::atomic::Ordering::SeqCst); } } } diff --git a/src/tools/atomic_once_flag.rs b/src/tools/atomic_once_flag.rs index cf549da9d..5da51c174 100644 --- a/src/tools/atomic_once_flag.rs +++ b/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::Release); + self.enabled.store(false, Ordering::SeqCst); } /// 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::Acquire) + self.enabled.load(Ordering::SeqCst) } } \ No newline at end of file diff --git a/src/utils/file/config_reader.rs b/src/utils/file/config_reader.rs index 5fd9d43f1..d0180797d 100644 --- a/src/utils/file/config_reader.rs +++ b/src/utils/file/config_reader.rs @@ -13,7 +13,7 @@ use std::fs::File; use std::io::{self, BufRead}; use std::path::PathBuf; use std::sync::LazyLock; -use url::Url; +use url::{ParseError, Url}; pub fn read_mappings(args_mapping: Option, cfg: &mut Config) -> Result, M3uFilterError> { let mappings_file: String = args_mapping.unwrap_or_else(|| file_utils::get_default_mappings_path(cfg.t_config_path.as_str())); @@ -164,12 +164,30 @@ pub fn resolve_env_var(value: &str) -> String { // // } -const CSV_SEPARATOR: char = ','; +const CSV_SEPARATOR: char = ';'; const HEADER_PREFIX: char = '#'; - const FIELD_MAX_CON: &str = "max_connections"; const FIELD_PRIO: &str = "priority"; const FIELD_URL: &str = "url"; +const FIELD_NAME: &str = "name"; +const FIELD_UNKNOWN: &str = "?"; + +fn csv_assign_mandatory_fields(alias: &mut ConfigInputAlias) { + if alias.name.is_empty() && !alias.url.is_empty() { + match Url::parse(alias.url.as_str()) { + Ok(url) => { + let username = alias.username.as_ref().map(|s|s.as_str()).unwrap_or_default(); + let domain: Vec<&str> = url.domain().unwrap_or_default().split('.').collect(); + if domain.len() > 1 { + alias.name = format!("{}_{username}", domain[domain.len() - 2]); + } else { + alias.name = format!("{username}"); + } + } + Err(err) => {} + } + } +} fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, input_type: &InputType, header: &str, value: &str) -> Result<(), io::Error> { match header { @@ -202,6 +220,9 @@ fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, input_typ let priority = value.parse::().unwrap_or(0); config_input.priority = priority; }, + FIELD_NAME => { + config_input.name = value.to_string(); + }, _ => {} } Ok(()) @@ -211,11 +232,25 @@ pub fn csv_read_inputs(input_type: InputType, file_path: &str) -> Result FIELD_URL, + FIELD_MAX_CON => FIELD_MAX_CON, + FIELD_PRIO => FIELD_PRIO, + FIELD_NAME => FIELD_NAME, + _ => { + error!("Field {s} is unsupported for csv input"); + FIELD_UNKNOWN + } + } + }).collect(); + println!("{default_columns:?}"); continue; } @@ -236,6 +271,7 @@ pub fn csv_read_inputs(input_type: InputType, file_path: &str) -> Result = LazyLock::new(|| Regex::new(r"(.*://) static SANITIZE_SENSITIVE_INFO: LazyLock = LazyLock::new(|| AtomicBool::new(true)); pub fn set_sanitize_sensitive_info(value: bool) { - SANITIZE_SENSITIVE_INFO.store(value, Ordering::Release); + SANITIZE_SENSITIVE_INFO.store(value, Ordering::SeqCst); } pub fn sanitize_sensitive_info(query: &str) -> String { - if SANITIZE_SENSITIVE_INFO.load(Ordering::Acquire) { + if SANITIZE_SENSITIVE_INFO.load(Ordering::SeqCst) { // Replace with "***" let masked_query = USERNAME_REGEX.replace_all(query, "$1***"); let masked_query = PASSWORD_REGEX.replace_all(&masked_query, "$1***");