1use crate::AeronErrorType::Unknown;
2#[cfg(feature = "backtrace")]
3use std::backtrace::Backtrace;
4use std::cell::UnsafeCell;
5use std::fmt::Formatter;
6use std::mem::MaybeUninit;
7use std::ops::{Deref, DerefMut};
8pub enum CResource<T> {
9 OwnedOnHeap(std::rc::Rc<ManagedCResource<T>>),
10 OwnedOnStack(std::mem::MaybeUninit<T>),
13 Borrowed(*mut T),
14}
15
16impl<T: Clone> Clone for CResource<T> {
17 fn clone(&self) -> Self {
18 unsafe {
23 match self {
24 CResource::OwnedOnHeap(r) => CResource::OwnedOnHeap(r.clone()),
25 CResource::OwnedOnStack(r) => {
26 CResource::OwnedOnStack(MaybeUninit::new(r.assume_init_ref().clone()))
27 }
28 CResource::Borrowed(r) => CResource::Borrowed(r.clone()),
29 }
30 }
31 }
32}
33
34impl<T> CResource<T> {
35 #[inline]
36 pub fn get(&self) -> *mut T {
37 match self {
38 CResource::OwnedOnHeap(r) => r.get(),
39 CResource::OwnedOnStack(r) => r.as_ptr() as *mut T,
40 CResource::Borrowed(r) => *r,
41 }
42 }
43
44 #[inline]
45 pub fn add_dependency<D: std::any::Any>(&self, dep: D) {
47 match self {
48 CResource::OwnedOnHeap(r) => r.add_dependency(dep),
49 CResource::OwnedOnStack(_) | CResource::Borrowed(_) => {
50 unreachable!("only owned on heap")
51 }
52 }
53 }
54 #[inline]
55 pub fn get_dependency<V: Clone + 'static>(&self) -> Option<V> {
56 match self {
57 CResource::OwnedOnHeap(r) => r.get_dependency(),
58 CResource::OwnedOnStack(_) | CResource::Borrowed(_) => None,
59 }
60 }
61
62 #[inline]
63 pub fn as_owned(&self) -> Option<&std::rc::Rc<ManagedCResource<T>>> {
64 match self {
65 CResource::OwnedOnHeap(r) => Some(r),
66 CResource::OwnedOnStack(_) | CResource::Borrowed(_) => None,
67 }
68 }
69}
70
71impl<T> std::fmt::Debug for CResource<T> {
72 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
73 let name = std::any::type_name::<T>();
74
75 match self {
76 CResource::OwnedOnHeap(r) => {
77 write!(f, "{name} heap({:?})", r)
78 }
79 CResource::OwnedOnStack(r) => {
80 write!(f, "{name} stack({:?})", *r)
81 }
82 CResource::Borrowed(r) => {
83 write!(f, "{name} borrowed ({:?})", r)
84 }
85 }
86 }
87}
88
89#[allow(dead_code)]
94pub struct ManagedCResource<T> {
95 resource: *mut T,
96 cleanup: Option<Box<dyn FnMut(*mut *mut T) -> i32>>,
97 cleanup_struct: bool,
98 close_already_called: std::cell::Cell<bool>,
100 check_for_is_closed: Option<fn(*mut T) -> bool>,
102 auto_close: std::cell::Cell<bool>,
104 resource_released: std::cell::Cell<bool>,
106 dependencies: UnsafeCell<Vec<std::rc::Rc<dyn std::any::Any>>>,
110}
111
112impl<T> std::fmt::Debug for ManagedCResource<T> {
113 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114 let mut debug_struct = f.debug_struct("ManagedCResource");
115
116 if !self.close_already_called.get()
117 && !self.resource.is_null()
118 && !self
119 .check_for_is_closed
120 .as_ref()
121 .map_or(false, |f| f(self.resource))
122 {
123 debug_struct.field("resource", &self.resource);
124 }
125
126 debug_struct
127 .field("type", &std::any::type_name::<T>())
128 .finish()
129 }
130}
131
132impl<T> ManagedCResource<T> {
133 pub fn new(
140 init: impl FnOnce(*mut *mut T) -> i32,
141 cleanup: Option<Box<dyn FnMut(*mut *mut T) -> i32>>,
142 cleanup_struct: bool,
143 check_for_is_closed: Option<fn(*mut T) -> bool>,
144 ) -> Result<Self, AeronCError> {
145 let resource = Self::initialise(init)?;
146
147 let result = Self {
148 resource,
149 cleanup,
150 cleanup_struct,
151 close_already_called: std::cell::Cell::new(false),
152 check_for_is_closed,
153 auto_close: std::cell::Cell::new(false),
154 resource_released: std::cell::Cell::new(false),
155 dependencies: UnsafeCell::new(vec![]),
156 };
157 #[cfg(feature = "extra-logging")]
158 log::info!("created c resource: {:?}", result);
159 Ok(result)
160 }
161
162 pub fn initialise(
163 init: impl FnOnce(*mut *mut T) -> i32 + Sized,
164 ) -> Result<*mut T, AeronCError> {
165 let mut resource: *mut T = std::ptr::null_mut();
166 let result = init(&mut resource);
167 if result < 0 || resource.is_null() {
168 return Err(AeronCError::from_code(result));
169 }
170 Ok(resource)
171 }
172
173 pub fn is_closed_already_called(&self) -> bool {
174 self.close_already_called.get()
175 || self.resource.is_null()
176 || self
177 .check_for_is_closed
178 .as_ref()
179 .map_or(false, |f| f(self.resource))
180 }
181
182 #[inline(always)]
184 pub fn get(&self) -> *mut T {
185 self.resource
186 }
187
188 #[inline(always)]
189 pub fn get_mut(&self) -> &mut T {
190 unsafe { &mut *self.resource }
191 }
192
193 #[inline]
194 pub fn add_dependency<D: std::any::Any>(&self, dep: D) {
196 if let Some(dep) =
197 (&dep as &dyn std::any::Any).downcast_ref::<std::rc::Rc<dyn std::any::Any>>()
198 {
199 unsafe {
200 (*self.dependencies.get()).push(dep.clone());
201 }
202 } else {
203 unsafe {
204 (*self.dependencies.get()).push(std::rc::Rc::new(dep));
205 }
206 }
207 }
208
209 #[inline]
210 pub fn get_dependency<V: Clone + 'static>(&self) -> Option<V> {
211 unsafe {
212 (*self.dependencies.get())
213 .iter()
214 .filter_map(|x| x.as_ref().downcast_ref::<V>().cloned())
215 .next()
216 }
217 }
218
219 #[inline]
220 pub fn is_resource_released(&self) -> bool {
221 self.resource_released.get()
222 }
223
224 #[inline]
225 pub fn mark_resource_released(&self) {
226 self.resource_released.set(true);
227 }
228
229 pub fn close(&mut self) -> Result<(), AeronCError> {
233 if self.close_already_called.get() {
234 return Ok(());
235 }
236 self.close_already_called.set(true);
237
238 let already_closed = self
239 .check_for_is_closed
240 .as_ref()
241 .map_or(false, |f| f(self.resource));
242
243 if let Some(mut cleanup) = self.cleanup.take() {
244 if !self.resource.is_null() {
245 if !already_closed {
246 let result = cleanup(&mut self.resource);
247 if result < 0 {
248 return Err(AeronCError::from_code(result));
249 }
250 }
251 self.resource = std::ptr::null_mut();
252 }
253 }
254
255 Ok(())
256 }
257}
258
259impl<T> Drop for ManagedCResource<T> {
260 fn drop(&mut self) {
261 if !self.resource.is_null() {
262 let already_closed = self.close_already_called.get()
263 || self
264 .check_for_is_closed
265 .as_ref()
266 .map_or(false, |f| f(self.resource));
267
268 let resource = if already_closed {
269 self.resource
270 } else {
271 self.resource.clone()
272 };
273
274 if !already_closed {
275 #[cfg(feature = "extra-logging")]
277 log::info!("closing c resource: {:?}", self);
278 let _ = self.close(); }
280 self.close_already_called.set(true);
281
282 if self.cleanup_struct {
283 #[cfg(feature = "extra-logging")]
284 log::info!("closing rust struct resource: {:?}", resource);
285 unsafe {
286 let _ = Box::from_raw(resource);
287 }
288 }
289 }
290 }
291}
292
293#[derive(Debug, PartialOrd, Eq, PartialEq, Clone)]
294pub enum AeronErrorType {
295 GenericError,
296 ClientErrorDriverTimeout,
297 ClientErrorClientTimeout,
298 ClientErrorConductorServiceTimeout,
299 ClientErrorBufferFull,
300 PublicationBackPressured,
301 PublicationAdminAction,
302 PublicationClosed,
303 PublicationMaxPositionExceeded,
304 PublicationError,
305 TimedOut,
306 Unknown(i32),
307}
308
309impl From<AeronErrorType> for AeronCError {
310 fn from(value: AeronErrorType) -> Self {
311 AeronCError::from_code(value.code())
312 }
313}
314
315impl AeronErrorType {
316 pub fn code(&self) -> i32 {
317 match self {
318 AeronErrorType::GenericError => -1,
319 AeronErrorType::ClientErrorDriverTimeout => -1000,
320 AeronErrorType::ClientErrorClientTimeout => -1001,
321 AeronErrorType::ClientErrorConductorServiceTimeout => -1002,
322 AeronErrorType::ClientErrorBufferFull => -1003,
323 AeronErrorType::PublicationBackPressured => -2,
324 AeronErrorType::PublicationAdminAction => -3,
325 AeronErrorType::PublicationClosed => -4,
326 AeronErrorType::PublicationMaxPositionExceeded => -5,
327 AeronErrorType::PublicationError => -6,
328 AeronErrorType::TimedOut => -234324,
329 AeronErrorType::Unknown(code) => *code,
330 }
331 }
332
333 pub fn is_back_pressured(&self) -> bool {
334 self == &AeronErrorType::PublicationBackPressured
335 }
336
337 pub fn is_admin_action(&self) -> bool {
338 self == &AeronErrorType::PublicationAdminAction
339 }
340
341 pub fn is_back_pressured_or_admin_action(&self) -> bool {
342 self.is_back_pressured() || self.is_admin_action()
343 }
344
345 pub fn from_code(code: i32) -> Self {
346 match code {
347 -1 => AeronErrorType::GenericError,
348 -1000 => AeronErrorType::ClientErrorDriverTimeout,
349 -1001 => AeronErrorType::ClientErrorClientTimeout,
350 -1002 => AeronErrorType::ClientErrorConductorServiceTimeout,
351 -1003 => AeronErrorType::ClientErrorBufferFull,
352 -2 => AeronErrorType::PublicationBackPressured,
353 -3 => AeronErrorType::PublicationAdminAction,
354 -4 => AeronErrorType::PublicationClosed,
355 -5 => AeronErrorType::PublicationMaxPositionExceeded,
356 -6 => AeronErrorType::PublicationError,
357 -234324 => AeronErrorType::TimedOut,
358 _ => Unknown(code),
359 }
360 }
361
362 pub fn to_string(&self) -> &'static str {
363 match self {
364 AeronErrorType::GenericError => "Generic Error",
365 AeronErrorType::ClientErrorDriverTimeout => "Client Error Driver Timeout",
366 AeronErrorType::ClientErrorClientTimeout => "Client Error Client Timeout",
367 AeronErrorType::ClientErrorConductorServiceTimeout => {
368 "Client Error Conductor Service Timeout"
369 }
370 AeronErrorType::ClientErrorBufferFull => "Client Error Buffer Full",
371 AeronErrorType::PublicationBackPressured => "Publication Back Pressured",
372 AeronErrorType::PublicationAdminAction => "Publication Admin Action",
373 AeronErrorType::PublicationClosed => "Publication Closed",
374 AeronErrorType::PublicationMaxPositionExceeded => "Publication Max Position Exceeded",
375 AeronErrorType::PublicationError => "Publication Error",
376 AeronErrorType::TimedOut => "Timed Out",
377 AeronErrorType::Unknown(_) => "Unknown Error",
378 }
379 }
380}
381
382#[derive(Eq, PartialEq, Clone)]
387pub struct AeronCError {
388 pub code: i32,
389}
390
391impl AeronCError {
392 pub fn from_code(code: i32) -> Self {
396 #[cfg(feature = "backtrace")]
397 {
398 if code < 0 {
399 let backtrace = Backtrace::capture();
400 let backtrace = format!("{:?}", backtrace);
401
402 static BACKTRACE_RE: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
406 let re = BACKTRACE_RE.get_or_init(|| {
407 regex::Regex::new(r#"fn: "([^"]+)", file: "([^"]+)", line: (\d+)"#).unwrap()
408 });
409 let mut lines = String::new();
410 re.captures_iter(&backtrace).for_each(|cap| {
411 let function = &cap[1];
412 let mut file = cap[2].to_string();
413 let line = &cap[3];
414 if file.starts_with("./") {
415 file = format!("{}/{}", env!("CARGO_MANIFEST_DIR"), &file[2..]);
416 } else if file.starts_with("/rustc/") {
417 file = file.split("/").last().unwrap().to_string();
418 }
419 lines.push_str(&format!(" {file}:{line} in {function}\n"));
421 });
422
423 log::error!(
424 "Aeron C error code: {}, kind: '{:?}'\n{}",
425 code,
426 AeronErrorType::from_code(code),
427 lines
428 );
429 }
430 }
431 AeronCError { code }
432 }
433
434 pub fn kind(&self) -> AeronErrorType {
435 AeronErrorType::from_code(self.code)
436 }
437
438 pub fn is_back_pressured(&self) -> bool {
439 self.kind().is_back_pressured()
440 }
441
442 pub fn is_admin_action(&self) -> bool {
443 self.kind().is_admin_action()
444 }
445
446 pub fn is_back_pressured_or_admin_action(&self) -> bool {
447 self.kind().is_back_pressured_or_admin_action()
448 }
449}
450
451pub struct Handler<T> {
469 raw_ptr: *mut T,
470 should_drop: bool,
471}
472
473unsafe impl<T> Send for Handler<T> {}
474unsafe impl<T> Sync for Handler<T> {}
475
476pub struct Handlers;
478
479impl<T> Handler<T> {
480 pub fn leak(handler: T) -> Self {
481 let raw_ptr = Box::into_raw(Box::new(handler)) as *mut _;
482 #[cfg(feature = "extra-logging")]
483 log::info!("creating handler {:?}", raw_ptr);
484 Self {
485 raw_ptr,
486 should_drop: true,
487 }
488 }
489
490 pub fn is_none(&self) -> bool {
491 self.raw_ptr.is_null()
492 }
493
494 pub fn as_raw(&self) -> *mut std::os::raw::c_void {
495 self.raw_ptr as *mut std::os::raw::c_void
496 }
497
498 pub fn release(&mut self) {
499 if self.should_drop && !self.raw_ptr.is_null() {
500 unsafe {
501 #[cfg(feature = "extra-logging")]
502 log::info!("dropping handler {:?}", self.raw_ptr);
503 let _ = Box::from_raw(self.raw_ptr as *mut T);
504 self.should_drop = false;
505 self.raw_ptr = std::ptr::null_mut();
511 }
512 }
513 }
514
515 pub unsafe fn new(raw_ptr: *mut T, should_drop: bool) -> Self {
516 Self {
517 raw_ptr,
518 should_drop,
519 }
520 }
521}
522
523impl<T> Drop for Handler<T> {
524 fn drop(&mut self) {
525 if self.should_drop && !self.raw_ptr.is_null() {
526 log::error!(
527 "Handler<{}> at {:?} is being dropped but release() was never called — \
528 memory leak: {} bytes. Call release() explicitly when the C side no longer holds the pointer.",
529 std::any::type_name::<T>(),
530 self.raw_ptr,
531 std::mem::size_of::<T>(),
532 );
533 unsafe {
535 let _ = Box::from_raw(self.raw_ptr as *mut T);
536 }
537 }
538 }
539}
540
541impl<T> Deref for Handler<T> {
542 type Target = T;
543
544 fn deref(&self) -> &Self::Target {
545 unsafe { &*self.raw_ptr as &T }
546 }
547}
548
549impl<T> DerefMut for Handler<T> {
550 fn deref_mut(&mut self) -> &mut Self::Target {
551 unsafe { &mut *self.raw_ptr as &mut T }
552 }
553}
554
555pub fn find_unused_udp_port(start_port: u16) -> Option<u16> {
556 let end_port = u16::MAX;
557
558 for port in start_port..=end_port {
559 if is_udp_port_available(port) {
560 return Some(port);
561 }
562 }
563
564 None
565}
566
567pub fn is_udp_port_available(port: u16) -> bool {
568 std::net::UdpSocket::bind(("127.0.0.1", port)).is_ok()
569}
570
571pub struct ChannelUri {}
573
574impl ChannelUri {
575 pub const AERON_SCHEME: &'static str = "aeron";
576 pub const SPY_QUALIFIER: &'static str = "aeron-spy";
577 pub const MAX_URI_LENGTH: usize = 4095;
578}
579
580pub const DRIVER_TIMEOUT_MS_DEFAULT: u64 = 10_000;
581pub const AERON_DIR_PROP_NAME: &str = "aeron.dir";
582pub const AERON_IPC_MEDIA: &str = "aeron:ipc";
583pub const AERON_UDP_MEDIA: &str = "aeron:udp";
584pub const SPY_PREFIX: &str = "aeron-spy:";
585pub const TAG_PREFIX: &str = "tag:";
586
587#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
589pub enum Media {
590 Ipc,
591 Udp,
592}
593
594impl Media {
595 pub fn as_str(&self) -> &'static str {
596 match self {
597 Media::Ipc => "ipc",
598 Media::Udp => "udp",
599 }
600 }
601}
602
603#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
605pub enum ControlMode {
606 Manual,
607 Dynamic,
608 Response,
610}
611
612impl ControlMode {
613 pub fn as_str(&self) -> &'static str {
614 match self {
615 ControlMode::Manual => "manual",
616 ControlMode::Dynamic => "dynamic",
617 ControlMode::Response => "response",
618 }
619 }
620}
621
622#[cfg(test)]
623#[allow(dead_code)]
624pub(crate) mod test_alloc {
625 use std::alloc::{GlobalAlloc, Layout, System};
626 use std::env;
627 use std::fs::OpenOptions;
628 #[allow(unused_imports)]
629 use std::os::unix::fs::OpenOptionsExt;
630 use std::sync::atomic::{AtomicIsize, Ordering};
631
632 pub struct TrackingAllocator {
635 allocs: AtomicIsize,
636 }
637
638 impl TrackingAllocator {
639 pub const fn new() -> Self {
640 Self {
641 allocs: AtomicIsize::new(0),
642 }
643 }
644 pub fn current(&self) -> isize {
645 self.allocs.load(Ordering::SeqCst)
646 }
647 }
648
649 unsafe impl GlobalAlloc for TrackingAllocator {
650 unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
651 self.allocs.fetch_add(1, Ordering::SeqCst);
652 System.alloc(layout)
653 }
654 unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
655 self.allocs.fetch_sub(1, Ordering::SeqCst);
656 System.dealloc(ptr, layout)
657 }
658 unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
659 self.allocs.fetch_add(1, Ordering::SeqCst);
660 System.alloc_zeroed(layout)
661 }
662 unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
663 System.realloc(ptr, layout, new_size)
664 }
665 }
666
667 #[global_allocator]
668 static GLOBAL: TrackingAllocator = TrackingAllocator::new();
669
670 pub fn current_allocs() -> isize {
672 GLOBAL.current()
673 }
674
675 pub fn assert_no_allocation<F: FnOnce()>(f: F) {
678 let tmp = env::temp_dir().join("rusteron_allocation.lck");
679
680 #[cfg(unix)]
681 let file = {
682 OpenOptions::new()
683 .read(true)
684 .write(true)
685 .create(true)
686 .mode(0o600)
687 .open(&tmp)
688 .expect("Failed to open allocation lock file")
689 };
690 #[cfg(not(unix))]
691 let file = {
692 OpenOptions::new()
693 .read(true)
694 .write(true)
695 .create(true)
696 .open(&tmp)
697 .expect("Failed to open allocation lock file")
698 };
699
700 let mut lock = fd_lock::RwLock::new(file);
701 let lock = lock.write().expect("Failed to acquire file lock");
702
703 let before = current_allocs();
704 f();
705 let after = current_allocs();
706 let diff = (after - before).abs();
707 assert!(
708 diff < 50,
709 "Expected no allocation leak, but alloc count changed from {} to {} (diff {})",
710 before,
711 after,
712 diff
713 );
714
715 drop(lock)
716 }
717}
718
719pub trait IntoCString {
720 fn into_c_string(self) -> std::ffi::CString;
721}
722
723impl IntoCString for std::ffi::CString {
724 fn into_c_string(self) -> std::ffi::CString {
725 self
726 }
727}
728
729impl IntoCString for &str {
730 fn into_c_string(self) -> std::ffi::CString {
731 #[cfg(feature = "extra-logging")]
732 log::info!("created c string on heap: {:?}", self);
733
734 std::ffi::CString::new(self).expect("failed to create CString")
735 }
736}
737
738impl IntoCString for String {
739 fn into_c_string(self) -> std::ffi::CString {
740 #[cfg(feature = "extra-logging")]
741 log::info!("created c string on heap: {:?}", self);
742
743 std::ffi::CString::new(self).expect("failed to create CString")
744 }
745}
746
747#[cfg(test)]
748mod handler_tests {
749 use super::*;
750
751 #[test]
752 fn release_nulls_pointer_so_is_none_is_true() {
753 let mut handler = Handler::leak(42u32);
754 assert!(
756 !handler.is_none(),
757 "freshly leaked handler must be non-null"
758 );
759
760 handler.release();
761
762 assert!(handler.is_none(), "release() must null raw_ptr (was a UAF)");
765 }
766
767 #[test]
768 fn release_is_idempotent_no_double_free() {
769 let mut handler = Handler::leak(99u64);
770 handler.release();
771 handler.release();
774 assert!(handler.is_none());
775 }
776
777 #[test]
778 fn release_then_drop_is_silent() {
779 let mut handler = Handler::leak(7u16);
782 handler.release();
783 drop(handler);
784 }
785}