@@ -4,10 +4,15 @@ use tokio::task::AbortHandle;
44
55use crate :: servers:: udp:: UDP_TRACKER_LOG_TARGET ;
66
7- /// Ring-Buffer of Active Requests
7+ /// A ring buffer for managing active UDP request abort handles.
8+ ///
9+ /// The `ActiveRequests` struct maintains a fixed-size ring buffer of abort
10+ /// handles for UDP request processor tasks. It ensures that at most 50 requests
11+ /// are handled concurrently, and provides mechanisms to handle buffer overflow
12+ /// by removing finished or oldest unfinished tasks.
813#[ derive( Default ) ]
914pub struct ActiveRequests {
10- rb : StaticRb < AbortHandle , 50 > , // the number of requests we handle at the same time .
15+ rb : StaticRb < AbortHandle , 50 > , // The number of requests handled simultaneously .
1116}
1217
1318impl std:: fmt:: Debug for ActiveRequests {
@@ -29,67 +34,107 @@ impl Drop for ActiveRequests {
2934}
3035
3136impl ActiveRequests {
32- /// It inserts the abort handle for the UDP request processor tasks .
37+ /// Inserts an abort handle for a UDP request processor task .
3338///
34- /// If there is no room for the new task, it tries to make place :
39+ /// If the buffer is full, this method attempts to make space by :
3540///
36- /// - Firstly, removing finished tasks.
37- /// - Secondly, removing the oldest unfinished tasks.
41+ /// 1. Removing finished tasks.
42+ /// 2. Removing the oldest unfinished task if no finished tasks are found .
3843///
3944/// # Panics
4045///
41- /// Will panics if it can't make space for the new handle.
42- pub async fn force_push ( & mut self , abort_handle : AbortHandle , local_addr : & str ) {
43- // fill buffer with requests
44- let Err ( abort_handle) = self . rb . try_push ( abort_handle) else {
45- return ;
46- } ;
47-
48- let mut finished: u64 = 0 ;
49- let mut unfinished_task = None ;
50-
51- // buffer is full.. lets make some space.
52- for h in self . rb . pop_iter ( ) {
53- // remove some finished tasks
54- if h. is_finished ( ) {
55- finished += 1 ;
56- continue ;
46+ /// This method will panic if it cannot make space for adding a new handle.
47+ ///
48+ /// # Arguments
49+ ///
50+ /// * `abort_handle` - The `AbortHandle` for the UDP request processor task.
51+ /// * `local_addr` - A string slice representing the local address for logging.
52+ pub async fn force_push ( & mut self , new_task : AbortHandle , local_addr : & str ) {
53+ // Attempt to add the new handle to the buffer.
54+ match self . rb . try_push ( new_task) {
55+ Ok ( ( ) ) => {
56+ // Successfully added the task, no further action needed.
5757 }
58-
59- // task is unfinished.. give it another chance.
60- tokio:: task:: yield_now ( ) . await ;
61-
62- // if now finished, we continue.
63- if h. is_finished ( ) {
64- finished += 1 ;
65- continue ;
58+ Err ( new_task) => {
59+ // Buffer is full, attempt to make space.
60+
61+ let mut finished: u64 = 0 ;
62+ let mut unfinished_task = None ;
63+
64+ for old_task in self . rb . pop_iter ( ) {
65+ // We found a finished tasks ... increase the counter and
66+ // continue searching for more and ...
67+ if old_task. is_finished ( ) {
68+ finished += 1 ;
69+ continue ;
70+ }
71+
72+ // The current removed tasks is not finished.
73+
74+ // Give it a second chance to finish.
75+ tokio:: task:: yield_now ( ) . await ;
76+
77+ // Recheck if it finished ... increase the counter and
78+ // continue searching for more and ...
79+ if old_task. is_finished ( ) {
80+ finished += 1 ;
81+ continue ;
82+ }
83+
84+ // At this point we found a "definitive" unfinished task.
85+
86+ // Log unfinished task.
87+ tracing:: debug!(
88+ target: UDP_TRACKER_LOG_TARGET ,
89+ local_addr,
90+ removed_count = finished,
91+ "Udp::run_udp_server::loop (got unfinished task)"
92+ ) ;
93+
94+ // If no finished tasks were found, abort the current
95+ // unfinished task.
96+ if finished == 0 {
97+ // We make place aborting this task.
98+ old_task. abort ( ) ;
99+
100+ tracing:: warn!(
101+ target: UDP_TRACKER_LOG_TARGET ,
102+ local_addr,
103+ "Udp::run_udp_server::loop aborting request: (no finished tasks)"
104+ ) ;
105+
106+ break ;
107+ }
108+
109+ // At this point we found at least one finished task, but the
110+ // current one is not finished and it was removed from the
111+ // buffer, so we need to re-insert in in the buffer.
112+
113+ // Save the unfinished task for re-entry.
114+ unfinished_task = Some ( old_task) ;
115+ }
116+
117+ // After this point there can't be a race condition because only
118+ // one thread owns the active buffer. There is no way for the
119+ // buffer to be full again. That means the "expects" should
120+ // never happen.
121+
122+ // Reinsert the unfinished task if any.
123+ if let Some ( h) = unfinished_task {
124+ self . rb . try_push ( h) . expect ( "it was previously inserted" ) ;
125+ }
126+
127+ // Insert the new task.
128+ //
129+ // Notice that space has already been made for this new task in
130+ // the buffer. One or many old task have already been finished
131+ // or yielded, freeing space in the buffer. Or a single
132+ // unfinished task has been aborted to make space for this new
133+ // task.
134+ if !new_task. is_finished ( ) {
135+ self . rb . try_push ( new_task) . expect ( "it should have space for this new task." ) ;
136+ }
66137 }
67-
68- tracing:: debug!( target: UDP_TRACKER_LOG_TARGET , local_addr, removed_count = finished, "Udp::run_udp_server::loop (got unfinished task)" ) ;
69-
70- if finished == 0 {
71- // we have _no_ finished tasks.. will abort the unfinished task to make space...
72- h. abort ( ) ;
73-
74- tracing:: warn!( target: UDP_TRACKER_LOG_TARGET , local_addr, "Udp::run_udp_server::loop aborting request: (no finished tasks)" ) ;
75-
76- break ;
77- }
78-
79- // we have space, return unfinished task for re-entry.
80- unfinished_task = Some ( h) ;
81- }
82-
83- // re-insert the previous unfinished task.
84- if let Some ( h) = unfinished_task {
85- self . rb . try_push ( h) . expect ( "it was previously inserted" ) ;
86- }
87-
88- // insert the new task.
89- if !abort_handle. is_finished ( ) {
90- self . rb
91- . try_push ( abort_handle)
92- . expect ( "it should remove at least one element." ) ;
93- }
138+ } ;
94139 }
95140}
0 commit comments