## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/execution/src/memory_pool/mod.rs

    Line
    Count
    Source
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    //! [`MemoryPool`] for memory management during query execution, [`proxy`] for
    19
    //! help with allocation accounting.
    20
    21
    use datafusion_common::{internal_err, Result};
    22
    use std::hash::{Hash, Hasher};
    23
    use std::{cmp::Ordering, sync::atomic, sync::Arc};
    24
    25
    mod pool;
    26
    pub mod proxy {
    27
        pub use datafusion_common::utils::proxy::{HashTableAllocExt, VecAllocExt};
    28
    }
    29
    30
    pub use pool::*;
    31
    32
    /// Tracks and potentially limits memory use across operators during execution.
    33
    ///
    34
    /// # Memory Management Overview
    35
    ///
    36
    /// DataFusion is a streaming query engine, processing most queries without
    37
    /// buffering the entire input. Most operators require a fixed amount of memory
    38
    /// based on the schema and target batch size. However, certain operations such
    39
    /// as sorting and grouping/joining, require buffering intermediate results,
    40
    /// which can require memory proportional to the number of input rows.
    41
    ///
    42
    /// Rather than tracking all allocations, DataFusion takes a pragmatic approach:
    43
    /// Intermediate memory used as data streams through the system is not accounted
    44
    /// (it assumed to be "small") but the large consumers of memory must register
    45
    /// and constrain their use. This design trades off the additional code
    46
    /// complexity of memory tracking with limiting resource usage.
    47
    ///
    48
    /// When limiting memory with a `MemoryPool` you should typically reserve some
    49
    /// overhead (e.g. 10%) for the "small" memory allocations that are not tracked.
    50
    ///
    51
    /// # Memory Management Design
    52
    ///
    53
    /// As explained above, DataFusion's design ONLY limits operators that require
    54
    /// "large" amounts of memory (proportional to number of input rows), such as
    55
    /// `GroupByHashExec`. It does NOT track and limit memory used internally by
    56
    /// other operators such as `DataSourceExec` or the `RecordBatch`es that flow
    57
    /// between operators. Furthermore, operators should not reserve memory for the
    58
    /// batches they produce. Instead, if a consumer operator needs to hold batches
    59
    /// from its producers in memory for an extended period, it is the consumer
    60
    /// operator's responsibility to reserve the necessary memory for those batches.
    61
    ///
    62
    /// In order to avoid allocating memory until the OS or the container system
    63
    /// kills the process, DataFusion `ExecutionPlan`s (operators) that consume
    64
    /// large amounts of memory must first request their desired allocation from a
    65
    /// [`MemoryPool`] before allocating more.  The request is typically managed via
    66
    /// a  [`MemoryReservation`] and [`MemoryConsumer`].
    67
    ///
    68
    /// If the allocation is successful, the operator should proceed and allocate
    69
    /// the desired memory. If the allocation fails, the operator must either first
    70
    /// free memory (e.g. by spilling to local disk) and try again, or error.
    71
    ///
    72
    /// Note that a `MemoryPool` can be shared by concurrently executing plans,
    73
    /// which can be used to control memory usage in a multi-tenant system.
    74
    ///
    75
    /// # How MemoryPool works by example
    76
    ///
    77
    /// Scenario 1:
    78
    /// For `Filter` operator, `RecordBatch`es will stream through it, so it
    79
    /// don't have to keep track of memory usage through [`MemoryPool`].
    80
    ///
    81
    /// Scenario 2:
    82
    /// For `CrossJoin` operator, if the input size gets larger, the intermediate
    83
    /// state will also grow. So `CrossJoin` operator will use [`MemoryPool`] to
    84
    /// limit the memory usage.
    85
    /// 2.1 `CrossJoin` operator has read a new batch, asked memory pool for
    86
    /// additional memory. Memory pool updates the usage and returns success.
    87
    /// 2.2 `CrossJoin` has read another batch, and tries to reserve more memory
    88
    /// again, memory pool does not have enough memory. Since `CrossJoin` operator
    89
    /// has not implemented spilling, it will stop execution and return an error.
    90
    ///
    91
    /// Scenario 3:
    92
    /// For `Aggregate` operator, its intermediate states will also accumulate as
    93
    /// the input size gets larger, but with spilling capability. When it tries to
    94
    /// reserve more memory from the memory pool, and the memory pool has already
    95
    /// reached the memory limit, it will return an error. Then, `Aggregate`
    96
    /// operator will spill the intermediate buffers to disk, and release memory
    97
    /// from the memory pool, and continue to retry memory reservation.
    98
    ///
    99
    /// # Related Structs
    100
    ///
    101
    /// To better understand memory management in DataFusion, here are the key structs
    102
    /// and their relationships:
    103
    ///
    104
    /// - [`MemoryConsumer`]: A named allocation traced by a particular operator. If an
    105
    ///   execution is parallelized, and there are multiple partitions of the same
    106
    ///   operator, each partition will have a separate `MemoryConsumer`.
    107
    /// - `SharedRegistration`: A registration of a `MemoryConsumer` with a `MemoryPool`.
    108
    ///   `SharedRegistration` and `MemoryPool` have a many-to-one relationship. `MemoryPool`
    109
    ///   implementation can decide how to allocate memory based on the registered consumers.
    110
    ///   (e.g. `FairSpillPool` will try to share available memory evenly among all registered
    111
    ///   consumers)
    112
    /// - [`MemoryReservation`]: Each `MemoryConsumer`/operator can have multiple
    113
    ///   `MemoryReservation`s for different internal data structures. The relationship
    114
    ///   between `MemoryConsumer` and `MemoryReservation` is one-to-many. This design
    115
    ///   enables cleaner operator implementations:
    116
    ///   - Different `MemoryReservation`s can be used for different purposes
    117
    ///   - `MemoryReservation` follows RAII principles - to release a reservation,
    118
    ///     simply drop the `MemoryReservation` object. When all `MemoryReservation`s
    119
    ///     for a `SharedRegistration` are dropped, the `SharedRegistration` is dropped
    120
    ///     when its reference count reaches zero, automatically unregistering the
    121
    ///     `MemoryConsumer` from the `MemoryPool`.
    122
    ///
    123
    /// ## Relationship Diagram
    124
    ///
    125
    /// ```text
    126
    /// ┌──────────────────┐     ┌──────────────────┐
    127
    /// │MemoryReservation │     │MemoryReservation │
    128
    /// └───┬──────────────┘     └──────────────────┘ ......
    129
    ///     │belongs to                    │
    130
    ///     │      ┌───────────────────────┘           │  │
    131
    ///     │      │                                   │  │
    132
    ///     ▼      ▼                                   ▼  ▼
    133
    /// ┌────────────────────────┐       ┌────────────────────────┐
    134
    /// │   SharedRegistration   │       │   SharedRegistration   │
    135
    /// │   ┌────────────────┐   │       │   ┌────────────────┐   │
    136
    /// │   │                │   │       │   │                │   │
    137
    /// │   │ MemoryConsumer │   │       │   │ MemoryConsumer │   │
    138
    /// │   │                │   │       │   │                │   │
    139
    /// │   └────────────────┘   │       │   └────────────────┘   │
    140
    /// └────────────┬───────────┘       └────────────┬───────────┘
    141
    ///              │                                │
    142
    ///              │                        register│into
    143
    ///              │                                │
    144
    ///              └─────────────┐   ┌──────────────┘
    145
    ///                            │   │
    146
    ///                            ▼   ▼
    147
    ///    ╔═══════════════════════════════════════════════════╗
    148
    ///    ║                                                   ║
    149
    ///    ║                    MemoryPool                     ║
    150
    ///    ║                                                   ║
    151
    ///    ╚═══════════════════════════════════════════════════╝
    152
    /// ```
    153
    ///
    154
    /// For example, there are two parallel partitions of an operator X: each partition
    155
    /// corresponds to a `MemoryConsumer` in the above diagram. Inside each partition of
    156
    /// operator X, there are typically several `MemoryReservation`s - one for each
    157
    /// internal data structure that needs memory tracking (e.g., 1 reservation for the hash
    158
    /// table, and 1 reservation for buffered input, etc.).
    159
    ///
    160
    /// # Implementing `MemoryPool`
    161
    ///
    162
    /// You can implement a custom allocation policy by implementing the
    163
    /// [`MemoryPool`] trait and configuring a `SessionContext` appropriately.
    164
    /// However, DataFusion comes with the following simple memory pool implementations that
    165
    /// handle many common cases:
    166
    ///
    167
    /// * [`UnboundedMemoryPool`]: no memory limits (the default)
    168
    ///
    169
    /// * [`GreedyMemoryPool`]: Limits memory usage to a fixed size using a "first
    170
    ///   come first served" policy
    171
    ///
    172
    /// * [`FairSpillPool`]: Limits memory usage to a fixed size, allocating memory
    173
    ///   to all spilling operators fairly
    174
    ///
    175
    /// * [`TrackConsumersPool`]: Wraps another [`MemoryPool`] and tracks consumers,
    176
    ///   providing better error messages on the largest memory users.
    177
    pub trait MemoryPool: Send + Sync + std::fmt::Debug {
    178
        /// Registers a new [`MemoryConsumer`]
    179
        ///
    180
        /// Note: Subsequent calls to [`Self::grow`] must be made to reserve memory
    181
    0
        fn register(&self, _consumer: &MemoryConsumer) {}
    182
    183
        /// Records the destruction of a [`MemoryReservation`] with [`MemoryConsumer`]
    184
        ///
    185
        /// Note: Prior calls to [`Self::shrink`] must be made to free any reserved memory
    186
    0
        fn unregister(&self, _consumer: &MemoryConsumer) {}
    187
    188
        /// Infallibly grow the provided `reservation` by `additional` bytes
    189
        ///
    190
        /// This must always succeed
    191
        fn grow(&self, reservation: &MemoryReservation, additional: usize);
    192
    193
        /// Infallibly shrink the provided `reservation` by `shrink` bytes
    194
        fn shrink(&self, reservation: &MemoryReservation, shrink: usize);
    195
    196
        /// Attempt to grow the provided `reservation` by `additional` bytes
    197
        ///
    198
        /// On error the `allocation` will not be increased in size
    199
        fn try_grow(&self, reservation: &MemoryReservation, additional: usize) -> Result<()>;
    200
    201
        /// Return the total amount of memory reserved
    202
        fn reserved(&self) -> usize;
    203
    204
        /// Return the memory limit of the pool
    205
        ///
    206
        /// The default implementation of `MemoryPool::memory_limit`
    207
        /// will return `MemoryLimit::Unknown`.
    208
        /// If you are using your custom memory pool, but have the requirement to
    209
        /// know the memory usage limit of the pool, please implement this method
    210
        /// to return it(`Memory::Finite(limit)`).
    211
    0
        fn memory_limit(&self) -> MemoryLimit {
    212
    0
            MemoryLimit::Unknown
    213
    0
        }
    214
    }
    215
    216
    /// Memory limit of `MemoryPool`
    217
    pub enum MemoryLimit {
    218
        Infinite,
    219
        /// Bounded memory limit in bytes.
    220
        Finite(usize),
    221
        Unknown,
    222
    }
    223
    224
    /// A memory consumer is a named allocation traced by a particular
    225
    /// [`MemoryReservation`] in a [`MemoryPool`]. All allocations are registered to
    226
    /// a particular `MemoryConsumer`;
    227
    ///
    228
    /// Each `MemoryConsumer` is identifiable by a process-unique id, and is therefor not cloneable,
    229
    /// If you want a clone of a `MemoryConsumer`, you should look into [`MemoryConsumer::clone_with_new_id`],
    230
    /// but note that this `MemoryConsumer` may be treated as a separate entity based on the used pool,
    231
    /// and is only guaranteed to share the name and inner properties.
    232
    ///
    233
    /// For help with allocation accounting, see the [`proxy`] module.
    234
    ///
    235
    /// [proxy]: datafusion_common::utils::proxy
    236
    #[derive(Debug)]
    237
    pub struct MemoryConsumer {
    238
        name: String,
    239
        can_spill: bool,
    240
        id: usize,
    241
    }
    242
    243
    impl PartialEq for MemoryConsumer {
    244
    0
        fn eq(&self, other: &Self) -> bool {
    245
    0
            let is_same_id = self.id == other.id;
    246
    247
            #[cfg(debug_assertions)]
    248
    0
            if is_same_id {
    249
    0
                assert_eq!(self.name, other.name);
    250
    0
                assert_eq!(self.can_spill, other.can_spill);
    251
    0
            }
    252
    253
    0
            is_same_id
    254
    0
        }
    255
    }
    256
    257
    impl Eq for MemoryConsumer {}
    258
    259
    impl Hash for MemoryConsumer {
    260
    0
        fn hash<H: Hasher>(&self, state: &mut H) {
    261
    0
            self.id.hash(state);
    262
    0
            self.name.hash(state);
    263
    0
            self.can_spill.hash(state);
    264
    0
        }
    265
    }
    266
    267
    impl MemoryConsumer {
    268
    0
        fn new_unique_id() -> usize {
    269
            static ID: atomic::AtomicUsize = atomic::AtomicUsize::new(0);
    270
    0
            ID.fetch_add(1, atomic::Ordering::Relaxed)
    271
    0
        }
    272
    273
        /// Create a new empty [`MemoryConsumer`] that can be grown using [`MemoryReservation`]
    274
    0
        pub fn new(name: impl Into<String>) -> Self {
    275
    0
            Self {
    276
    0
                name: name.into(),
    277
    0
                can_spill: false,
    278
    0
                id: Self::new_unique_id(),
    279
    0
            }
    280
    0
        }
    281
    282
        /// Returns a clone of this [`MemoryConsumer`] with a new unique id,
    283
        /// which can be registered with a [`MemoryPool`],
    284
        /// This new consumer is separate from the original.
    285
    0
        pub fn clone_with_new_id(&self) -> Self {
    286
    0
            Self {
    287
    0
                name: self.name.clone(),
    288
    0
                can_spill: self.can_spill,
    289
    0
                id: Self::new_unique_id(),
    290
    0
            }
    291
    0
        }
    292
    293
        /// Return the unique id of this [`MemoryConsumer`]
    294
    0
        pub fn id(&self) -> usize {
    295
    0
            self.id
    296
    0
        }
    297
    298
        /// Set whether this allocation can be spilled to disk
    299
    0
        pub fn with_can_spill(self, can_spill: bool) -> Self {
    300
    0
            Self { can_spill, ..self }
    301
    0
        }
    302
    303
        /// Returns true if this allocation can spill to disk
    304
    0
        pub fn can_spill(&self) -> bool {
    305
    0
            self.can_spill
    306
    0
        }
    307
    308
        /// Returns the name associated with this allocation
    309
    0
        pub fn name(&self) -> &str {
    310
    0
            &self.name
    311
    0
        }
    312
    313
        /// Registers this [`MemoryConsumer`] with the provided [`MemoryPool`] returning
    314
        /// a [`MemoryReservation`] that can be used to grow or shrink the memory reservation
    315
    0
        pub fn register(self, pool: &Arc<dyn MemoryPool>) -> MemoryReservation {
    316
    0
            pool.register(&self);
    317
    0
            MemoryReservation {
    318
    0
                registration: Arc::new(SharedRegistration {
    319
    0
                    pool: Arc::clone(pool),
    320
    0
                    consumer: self,
    321
    0
                }),
    322
    0
                size: 0,
    323
    0
            }
    324
    0
        }
    325
    }
    326
    327
    /// A registration of a [`MemoryConsumer`] with a [`MemoryPool`].
    328
    ///
    329
    /// Calls [`MemoryPool::unregister`] on drop to return any memory to
    330
    /// the underlying pool.
    331
    #[derive(Debug)]
    332
    struct SharedRegistration {
    333
        pool: Arc<dyn MemoryPool>,
    334
        consumer: MemoryConsumer,
    335
    }
    336
    337
    impl Drop for SharedRegistration {
    338
    0
        fn drop(&mut self) {
    339
    0
            self.pool.unregister(&self.consumer);
    340
    0
        }
    341
    }
    342
    343
    /// A [`MemoryReservation`] tracks an individual reservation of a
    344
    /// number of bytes of memory in a [`MemoryPool`] that is freed back
    345
    /// to the pool on drop.
    346
    ///
    347
    /// The reservation can be grown or shrunk over time.
    348
    #[derive(Debug)]
    349
    pub struct MemoryReservation {
    350
        registration: Arc<SharedRegistration>,
    351
        size: usize,
    352
    }
    353
    354
    impl MemoryReservation {
    355
        /// Returns the size of this reservation in bytes
    356
    0
        pub fn size(&self) -> usize {
    357
    0
            self.size
    358
    0
        }
    359
    360
        /// Returns [MemoryConsumer] for this [MemoryReservation]
    361
    0
        pub fn consumer(&self) -> &MemoryConsumer {
    362
    0
            &self.registration.consumer
    363
    0
        }
    364
    365
        /// Frees all bytes from this reservation back to the underlying
    366
        /// pool, returning the number of bytes freed.
    367
    0
        pub fn free(&mut self) -> usize {
    368
    0
            let size = self.size;
    369
    0
            if size != 0 {
    370
    0
                self.shrink(size)
    371
    0
            }
    372
    0
            size
    373
    0
        }
    374
    375
        /// Frees `capacity` bytes from this reservation
    376
        ///
    377
        /// # Panics
    378
        ///
    379
        /// Panics if `capacity` exceeds [`Self::size`]
    380
    0
        pub fn shrink(&mut self, capacity: usize) {
    381
    0
            let new_size = self.size.checked_sub(capacity).unwrap();
    382
    0
            self.registration.pool.shrink(self, capacity);
    383
    0
            self.size = new_size
    384
    0
        }
    385
    386
        /// Tries to free `capacity` bytes from this reservation
    387
        /// if `capacity` does not exceed [`Self::size`]
    388
        /// Returns new reservation size
    389
        /// or error if shrinking capacity is more than allocated size
    390
    0
        pub fn try_shrink(&mut self, capacity: usize) -> Result<usize> {
    391
    0
            if let Some(new_size) = self.size.checked_sub(capacity) {
    392
    0
                self.registration.pool.shrink(self, capacity);
    393
    0
                self.size = new_size;
    394
    0
                Ok(new_size)
    395
            } else {
    396
    0
                internal_err!(
    397
    0
                    "Cannot free the capacity {capacity} out of allocated size {}",
    398
                    self.size
    399
                )
    400
            }
    401
    0
        }
    402
    403
        /// Sets the size of this reservation to `capacity`
    404
    0
        pub fn resize(&mut self, capacity: usize) {
    405
    0
            match capacity.cmp(&self.size) {
    406
    0
                Ordering::Greater => self.grow(capacity - self.size),
    407
    0
                Ordering::Less => self.shrink(self.size - capacity),
    408
    0
                _ => {}
    409
            }
    410
    0
        }
    411
    412
        /// Try to set the size of this reservation to `capacity`
    413
    0
        pub fn try_resize(&mut self, capacity: usize) -> Result<()> {
    414
    0
            match capacity.cmp(&self.size) {
    415
    0
                Ordering::Greater => self.try_grow(capacity - self.size)?,
    416
    0
                Ordering::Less => self.shrink(self.size - capacity),
    417
    0
                _ => {}
    418
            };
    419
    0
            Ok(())
    420
    0
        }
    421
    422
        /// Increase the size of this reservation by `capacity` bytes
    423
    0
        pub fn grow(&mut self, capacity: usize) {
    424
    0
            self.registration.pool.grow(self, capacity);
    425
    0
            self.size += capacity;
    426
    0
        }
    427
    428
        /// Try to increase the size of this reservation by `capacity`
    429
        /// bytes, returning error if there is insufficient capacity left
    430
        /// in the pool.
    431
    0
        pub fn try_grow(&mut self, capacity: usize) -> Result<()> {
    432
    0
            self.registration.pool.try_grow(self, capacity)?;
    433
    0
            self.size += capacity;
    434
    0
            Ok(())
    435
    0
        }
    436
    437
        /// Splits off `capacity` bytes from this [`MemoryReservation`]
    438
        /// into a new [`MemoryReservation`] with the same
    439
        /// [`MemoryConsumer`].
    440
        ///
    441
        /// This can be useful to free part of this reservation with RAAI
    442
        /// style dropping
    443
        ///
    444
        /// # Panics
    445
        ///
    446
        /// Panics if `capacity` exceeds [`Self::size`]
    447
    0
        pub fn split(&mut self, capacity: usize) -> MemoryReservation {
    448
    0
            self.size = self.size.checked_sub(capacity).unwrap();
    449
    0
            Self {
    450
    0
                size: capacity,
    451
    0
                registration: Arc::clone(&self.registration),
    452
    0
            }
    453
    0
        }
    454
    455
        /// Returns a new empty [`MemoryReservation`] with the same [`MemoryConsumer`]
    456
    0
        pub fn new_empty(&self) -> Self {
    457
    0
            Self {
    458
    0
                size: 0,
    459
    0
                registration: Arc::clone(&self.registration),
    460
    0
            }
    461
    0
        }
    462
    463
        /// Splits off all the bytes from this [`MemoryReservation`] into
    464
        /// a new [`MemoryReservation`] with the same [`MemoryConsumer`]
    465
    0
        pub fn take(&mut self) -> MemoryReservation {
    466
    0
            self.split(self.size)
    467
    0
        }
    468
    }
    469
    470
    impl Drop for MemoryReservation {
    471
    0
        fn drop(&mut self) {
    472
    0
            self.free();
    473
    0
        }
    474
    }
    475
    476
    pub mod units {
    477
        pub const TB: u64 = 1 << 40;
    478
        pub const GB: u64 = 1 << 30;
    479
        pub const MB: u64 = 1 << 20;
    480
        pub const KB: u64 = 1 << 10;
    481
    }
    482
    483
    /// Present size in human-readable form
    484
    0
    pub fn human_readable_size(size: usize) -> String {
    485
        use units::*;
    486
    487
    0
        let size = size as u64;
    488
    0
        let (value, unit) = {
    489
    0
            if size >= 2 * TB {
    490
    0
                (size as f64 / TB as f64, "TB")
    491
    0
            } else if size >= 2 * GB {
    492
    0
                (size as f64 / GB as f64, "GB")
    493
    0
            } else if size >= 2 * MB {
    494
    0
                (size as f64 / MB as f64, "MB")
    495
    0
            } else if size >= 2 * KB {
    496
    0
                (size as f64 / KB as f64, "KB")
    497
            } else {
    498
    0
                (size as f64, "B")
    499
            }
    500
        };
    501
    0
        format!("{value:.1} {unit}")
    502
    0
    }
    503
    504
    /// Present count in human-readable form with K, M, B, T suffixes
    505
    0
    pub fn human_readable_count(count: usize) -> String {
    506
    0
        let count = count as u64;
    507
    0
        let (value, unit) = {
    508
    0
            if count >= 1_000_000_000_000 {
    509
    0
                (count as f64 / 1_000_000_000_000.0, " T")
    510
    0
            } else if count >= 1_000_000_000 {
    511
    0
                (count as f64 / 1_000_000_000.0, " B")
    512
    0
            } else if count >= 1_000_000 {
    513
    0
                (count as f64 / 1_000_000.0, " M")
    514
    0
            } else if count >= 1_000 {
    515
    0
                (count as f64 / 1_000.0, " K")
    516
            } else {
    517
    0
                return count.to_string();
    518
            }
    519
        };
    520
    521
        // Format with appropriate precision
    522
        // For values >= 100, show 1 decimal place (e.g., 123.4 K)
    523
        // For values < 100, show 2 decimal places (e.g., 10.12 K)
    524
    0
        if value >= 100.0 {
    525
    0
            format!("{value:.1}{unit}")
    526
        } else {
    527
    0
            format!("{value:.2}{unit}")
    528
        }
    529
    0
    }
    530
    531
    /// Present duration in human-readable form with 2 decimal places
    532
    0
    pub fn human_readable_duration(nanos: u64) -> String {
    533
        const NANOS_PER_SEC: f64 = 1_000_000_000.0;
    534
        const NANOS_PER_MILLI: f64 = 1_000_000.0;
    535
        const NANOS_PER_MICRO: f64 = 1_000.0;
    536
    537
    0
        let nanos_f64 = nanos as f64;
    538
    539
    0
        if nanos >= 1_000_000_000 {
    540
            // >= 1 second: show in seconds
    541
    0
            format!("{:.2}s", nanos_f64 / NANOS_PER_SEC)
    542
    0
        } else if nanos >= 1_000_000 {
    543
            // >= 1 millisecond: show in milliseconds
    544
    0
            format!("{:.2}ms", nanos_f64 / NANOS_PER_MILLI)
    545
    0
        } else if nanos >= 1_000 {
    546
            // >= 1 microsecond: show in microseconds
    547
    0
            format!("{:.2}µs", nanos_f64 / NANOS_PER_MICRO)
    548
        } else {
    549
            // < 1 microsecond: show in nanoseconds
    550
    0
            format!("{nanos}ns")
    551
        }
    552
    0
    }
    553
    554
    #[cfg(test)]
    555
    mod tests {
    556
        use super::*;
    557
    558
        #[test]
    559
        fn test_id_uniqueness() {
    560
            let mut ids = std::collections::HashSet::new();
    561
            for _ in 0..100 {
    562
                let consumer = MemoryConsumer::new("test");
    563
                assert!(ids.insert(consumer.id())); // Ensures unique insertion
    564
            }
    565
        }
    566
    567
        #[test]
    568
        fn test_memory_pool_underflow() {
    569
            let pool = Arc::new(GreedyMemoryPool::new(50)) as _;
    570
            let mut a1 = MemoryConsumer::new("a1").register(&pool);
    571
            assert_eq!(pool.reserved(), 0);
    572
    573
            a1.grow(100);
    574
            assert_eq!(pool.reserved(), 100);
    575
    576
            assert_eq!(a1.free(), 100);
    577
            assert_eq!(pool.reserved(), 0);
    578
    579
            a1.try_grow(100).unwrap_err();
    580
            assert_eq!(pool.reserved(), 0);
    581
    582
            a1.try_grow(30).unwrap();
    583
            assert_eq!(pool.reserved(), 30);
    584
    585
            let mut a2 = MemoryConsumer::new("a2").register(&pool);
    586
            a2.try_grow(25).unwrap_err();
    587
            assert_eq!(pool.reserved(), 30);
    588
    589
            drop(a1);
    590
            assert_eq!(pool.reserved(), 0);
    591
    592
            a2.try_grow(25).unwrap();
    593
            assert_eq!(pool.reserved(), 25);
    594
        }
    595
    596
        #[test]
    597
        fn test_split() {
    598
            let pool = Arc::new(GreedyMemoryPool::new(50)) as _;
    599
            let mut r1 = MemoryConsumer::new("r1").register(&pool);
    600
    601
            r1.try_grow(20).unwrap();
    602
            assert_eq!(r1.size(), 20);
    603
            assert_eq!(pool.reserved(), 20);
    604
    605
            // take 5 from r1, should still have same reservation split
    606
            let r2 = r1.split(5);
    607
            assert_eq!(r1.size(), 15);
    608
            assert_eq!(r2.size(), 5);
    609
            assert_eq!(pool.reserved(), 20);
    610
    611
            // dropping r1 frees 15 but retains 5 as they have the same consumer
    612
            drop(r1);
    613
            assert_eq!(r2.size(), 5);
    614
            assert_eq!(pool.reserved(), 5);
    615
        }
    616
    617
        #[test]
    618
        fn test_new_empty() {
    619
            let pool = Arc::new(GreedyMemoryPool::new(50)) as _;
    620
            let mut r1 = MemoryConsumer::new("r1").register(&pool);
    621
    622
            r1.try_grow(20).unwrap();
    623
            let mut r2 = r1.new_empty();
    624
            r2.try_grow(5).unwrap();
    625
    626
            assert_eq!(r1.size(), 20);
    627
            assert_eq!(r2.size(), 5);
    628
            assert_eq!(pool.reserved(), 25);
    629
        }
    630
    631
        #[test]
    632
        fn test_take() {
    633
            let pool = Arc::new(GreedyMemoryPool::new(50)) as _;
    634
            let mut r1 = MemoryConsumer::new("r1").register(&pool);
    635
    636
            r1.try_grow(20).unwrap();
    637
            let mut r2 = r1.take();
    638
            r2.try_grow(5).unwrap();
    639
    640
            assert_eq!(r1.size(), 0);
    641
            assert_eq!(r2.size(), 25);
    642
            assert_eq!(pool.reserved(), 25);
    643
    644
            // r1 can still grow again
    645
            r1.try_grow(3).unwrap();
    646
            assert_eq!(r1.size(), 3);
    647
            assert_eq!(r2.size(), 25);
    648
            assert_eq!(pool.reserved(), 28);
    649
        }
    650
    651
        #[test]
    652
        fn test_human_readable_count() {
    653
            // Test small numbers (< 1000) - should display as-is
    654
            assert_eq!(human_readable_count(0), "0");
    655
            assert_eq!(human_readable_count(1), "1");
    656
            assert_eq!(human_readable_count(999), "999");
    657
    658
            // Test thousands (K)
    659
            assert_eq!(human_readable_count(1_000), "1.00 K");
    660
            assert_eq!(human_readable_count(10_100), "10.10 K");
    661
            assert_eq!(human_readable_count(1_532), "1.53 K");
    662
            assert_eq!(human_readable_count(99_999), "100.00 K");
    663
    664
            // Test millions (M)
    665
            assert_eq!(human_readable_count(1_000_000), "1.00 M");
    666
            assert_eq!(human_readable_count(1_532_000), "1.53 M");
    667
            assert_eq!(human_readable_count(99_000_000), "99.00 M");
    668
            assert_eq!(human_readable_count(123_456_789), "123.5 M");
    669
    670
            // Test billions (B)
    671
            assert_eq!(human_readable_count(1_000_000_000), "1.00 B");
    672
            assert_eq!(human_readable_count(1_532_000_000), "1.53 B");
    673
            assert_eq!(human_readable_count(999_999_999_999), "1000.0 B");
    674
    675
            // Test trillions (T)
    676
            assert_eq!(human_readable_count(1_000_000_000_000), "1.00 T");
    677
            assert_eq!(human_readable_count(42_000_000_000_000), "42.00 T");
    678
        }
    679
    680
        #[test]
    681
        fn test_human_readable_duration() {
    682
            // Test nanoseconds (< 1µs)
    683
            assert_eq!(human_readable_duration(0), "0ns");
    684
            assert_eq!(human_readable_duration(1), "1ns");
    685
            assert_eq!(human_readable_duration(999), "999ns");
    686
    687
            // Test microseconds (1µs to < 1ms)
    688
            assert_eq!(human_readable_duration(1_000), "1.00µs");
    689
            assert_eq!(human_readable_duration(1_234), "1.23µs");
    690
            assert_eq!(human_readable_duration(999_999), "1000.00µs");
    691
    692
            // Test milliseconds (1ms to < 1s)
    693
            assert_eq!(human_readable_duration(1_000_000), "1.00ms");
    694
            assert_eq!(human_readable_duration(11_295_377), "11.30ms");
    695
            assert_eq!(human_readable_duration(1_234_567), "1.23ms");
    696
            assert_eq!(human_readable_duration(999_999_999), "1000.00ms");
    697
    698
            // Test seconds (>= 1s)
    699
            assert_eq!(human_readable_duration(1_000_000_000), "1.00s");
    700
            assert_eq!(human_readable_duration(1_234_567_890), "1.23s");
    701
            assert_eq!(human_readable_duration(42_000_000_000), "42.00s");
    702
        }
    703
    }