Struct SyncPool 

pub struct SyncPool<T> { /* private fields */ }

Implementations

impl<T: Default> SyncPool<T>

pub fn new() -> Self

Create a pool with default size of 64 pre-allocated elements in it.

pub fn with_size(size: usize) -> Self

Create a SyncPool with pre-defined number of elements. Note that we will round-up the size such that the total number of elements in the pool will mod to 8.

Examples found in repository

examples/basic.rs (line 35)

unsafe fn pool_setup() -> (
    Pin<&'static mut SyncPool<ComplexStruct>>,
    Pin<&'static mut SyncPool<ComplexStruct>>,
) {
    POOL.as_mut_ptr().write(SyncPool::with_size(COUNT / 2));

    (
        Pin::new(&mut *POOL.as_mut_ptr()),
        Pin::new(&mut *POOL.as_mut_ptr()),
    )
}

examples/complex_bench.rs (line 99)

fn pool_setup() {
    unsafe {
        let mut pool = SyncPool::with_size(128);

        // clean up the underlying buffer, this handle can also be used to shrink the underlying
        // buffer to save for space, though at a cost of extra overhead for doing that.
        pool.reset_handle(sanitizer);

        /*
        // Alternatively, use an anonymous function for the same purpose. Closure can't be used as
        // a handle, though.
        pool.reset_handle(|slice: &mut [u8; BUF_CAP]| {
            for i in 0..slice.len() {
                slice[i] = 0;
            }

            println!("Byte slice cleared...");
        });
        */

        POOL.replace(pool);
    }
}

impl<T> SyncPool<T>

pub fn with_builder(builder: fn() -> T) -> Self

Create a pool with default size of 64 pre-allocated elements in it, which will use the builder handler to obtain the initialized instance of the struct, and then place the object into the syncpool for later use.

Note that the handler shall be responsible for creating and initializing the struct object with all fields being valid. After all, they will be the same objects provided to the caller when invoking the get call.

Examples

use syncpool::*;
use std::vec;

struct BigStruct {
    a: u32,
    b: u32,
    c: Vec<u8>,
}

let mut pool = SyncPool::with_builder(|| {
    BigStruct {
        a: 1,
        b: 42,
        c: vec::from_elem(0u8, 0x1_000_000),
    }
});

let big_box: Box<BigStruct> = pool.get();

assert_eq!(big_box.a, 1);
assert_eq!(big_box.b, 42);
assert_eq!(big_box.c.len(), 0x1_000_000);

pool.put(big_box);

Examples found in repository

examples/with_tools.rs (line 36)

fn call_builder() {
    let mut pool = SyncPool::with_builder(BigStruct::new);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

pub fn with_builder_and_size(size: usize, builder: fn() -> T) -> Self

Create a SyncPool with pre-defined number of elements and a packer handler. The builder handler shall essentially function the same way as in the with_builder, that it shall take the responsibility to create and initialize the element, and return the instance at the end of the builder closure. Note that we will round-up the size such that the total number of elements in the pool will mod to 8.

pub fn with_packer(packer: fn(Box<T>) -> Box<T>) -> Self

Create a pool with default size of 64 pre-allocated elements in it, which will use the packer handler to initialize the element that’s being provided by the pool.

Note that the handler shall take a boxed instance of the element that only contains placeholder fields, and it is the caller/handler’s job to initialize the fields and pack it with valid and meaningful values. If the struct is valid with all-zero values, the handler can just return the input element.

Examples

use syncpool::*;
use std::vec;

struct BigStruct {
    a: u32,
    b: u32,
    c: Vec<u8>,
}

let mut pool = SyncPool::with_packer(|mut src: Box<BigStruct>| {
    src.a = 1;
    src.b = 42;
    src.c = vec::from_elem(0u8, 0x1_000_000);
    src
});

let big_box: Box<BigStruct> = pool.get();

assert_eq!(big_box.a, 1);
assert_eq!(big_box.b, 42);
assert_eq!(big_box.c.len(), 0x1_000_000);

pool.put(big_box);

Examples found in repository

examples/with_tools.rs (line 50)

fn call_packer() {
    let mut pool = SyncPool::with_packer(BigStruct::initializer);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

pub fn with_packer_and_size(size: usize, packer: fn(Box<T>) -> Box<T>) -> Self

Create a SyncPool with pre-defined number of elements and a packer handler. The packer handler shall essentially function the same way as in with_packer, that it shall take the responsibility to initialize all the fields of a placeholder struct on the heap, otherwise the element returned by the pool will be essentially undefined, unless all the struct’s fields can be represented by a 0 value. In addition, we will round-up the size such that the total number of elements in the pool will mod to 8.

pub fn get(&mut self) -> Box<T>

Try to obtain a pre-allocated element from the pool. This method will always succeed even if the pool is empty or not available for anyone to access, and in this case, a new boxed-element will be created.

Examples found in repository

examples/with_tools.rs (line 40)

fn call_builder() {
    let mut pool = SyncPool::with_builder(BigStruct::new);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

fn call_packer() {
    let mut pool = SyncPool::with_packer(BigStruct::initializer);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

examples/basic.rs (line 79)

fn run(pool: &mut SyncPool<ComplexStruct>, chan: &SyncSender<Box<ComplexStruct>>, id: usize) {
    // take a pre-init struct from the pool
    let mut content = pool.get();
    content.id = id;

    // assuming we're doing some stuff in this period
    thread::sleep(Duration::from_nanos(32));

    // done with the stuff, send the result out.
    chan.send(content).unwrap_or_default();
}

examples/complex_bench.rs (line 140)

fn run(alloc: bool) -> u128 {
    let (tx, rx) = mpsc::sync_channel(32);
    let tx_clone = tx.clone();

    let now = Instant::now();

    let send_one = thread::spawn(move || {
        for i in 0..TEST_SIZE {
            if i % DENOMINATOR == 0 {
                thread::sleep(Duration::from_nanos(BUSY_PERIOD));
            }

            let mut data = if alloc {
                Default::default()
            } else {
                unsafe { POOL.as_mut().unwrap().get() }
            };

            assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
            assert_ne!(data.id, 42);
            data.id = 42;

            /*            assert_eq!(arr.len(), BUF_CAP);
            assert_eq!(arr.0[42], 42);*/

            tx_clone.try_send(data).unwrap_or_default();
        }

        //        println!("Send one done...");
    });

    let send_two = thread::spawn(move || {
        for i in 0..TEST_SIZE {
            if i % DENOMINATOR == 0 {
                thread::sleep(Duration::from_nanos(BUSY_PERIOD));
            }

            let mut data = if alloc {
                Default::default()
            } else {
                unsafe { POOL.as_mut().unwrap().get() }
            };

            assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
            assert_ne!(data.id, 42);
            data.id = 42;

            /*            assert_eq!(arr.len(), BUF_CAP);
            assert_eq!(arr.0[42], 42);*/

            tx.try_send(data).unwrap_or_default();
        }

        //        println!("Send two done...");
    });

    let recv_one = thread::spawn(move || {
        thread::sleep(Duration::from_micros(5));

        while let Ok(arr) = rx.recv() {
            //            assert_eq!(arr.len(), BUF_CAP);

            if !alloc {
                unsafe {
                    POOL.as_mut().unwrap().put(arr);
                }
            }
        }

        //        println!("Recv one done...");
    });

    for i in 0..TEST_SIZE {
        // sleep a bit to create some concurrent actions
        if i % DENOMINATOR == 1 {
            thread::sleep(Duration::from_nanos(BUSY_PERIOD));
        }

        let mut data = if alloc {
            Default::default()
        } else {
            unsafe { POOL.as_mut().unwrap().get() }
        };

        //        assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
        assert_ne!(data.id, 42);
        data.id = 42;

        /*        assert_eq!(arr.len(), BUF_CAP);
        assert_eq!(arr.0[42], 42);*/

        if !alloc {
            // when done using the object, make sure to put it back so the pool won't dry up
            unsafe { POOL.as_mut().unwrap().put(data) };
        }
    }

    //    println!("Main loop done...");

    send_one.join().unwrap_or_default();
    send_two.join().unwrap_or_default();
    recv_one.join().unwrap_or_default();

    now.elapsed().as_micros()
}

pub fn put(&mut self, val: Box<T>) -> Option<Box<T>>

Try to return an element to the SyncPool. If succeed, we will return None to indicate that the value has been placed in an empty slot; otherwise, we will return Option<Box<T>> such that the caller can decide if the element shall be just discarded, or try put it back again.

Examples found in repository

examples/with_tools.rs (line 46)

fn call_builder() {
    let mut pool = SyncPool::with_builder(BigStruct::new);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

fn call_packer() {
    let mut pool = SyncPool::with_packer(BigStruct::initializer);

    println!("Pool created...");

    let big_box = pool.get();

    assert_eq!(big_box.a, 1);
    assert_eq!(big_box.b, 42);
    assert_eq!(big_box.c.len(), 0x1_000_000);

    pool.put(big_box);
}

examples/basic.rs (line 67)

fn main() {
    // let's make the pool slightly smaller than the demand, this will simulate a service under pressure
    // such that the pool can't completely meet the demand without dynamically expand the pool.
    let (pinned_producer, pinned_consumer) = unsafe { pool_setup() };

    // make the channel that establish a concurrent pipeline.
    let (tx, rx) = mpsc::sync_channel(64);

    // data producer loop
    thread::spawn(move || {
        let producer = pinned_producer.get_mut();

        for i in 0..COUNT {
            run(producer, &tx, i);
        }
    });

    // data consumer logic
    let handler = thread::spawn(move || {
        let consumer = pinned_consumer.get_mut();

        for content in rx {
            println!("Receiving struct with id: {}", content.id);
            consumer.put(content);
        }
    });

    // wait for the receiver to finish and print the result.
    handler.join().unwrap_or_default();

    println!("All done...");
}

examples/complex_bench.rs (line 189)

fn run(alloc: bool) -> u128 {
    let (tx, rx) = mpsc::sync_channel(32);
    let tx_clone = tx.clone();

    let now = Instant::now();

    let send_one = thread::spawn(move || {
        for i in 0..TEST_SIZE {
            if i % DENOMINATOR == 0 {
                thread::sleep(Duration::from_nanos(BUSY_PERIOD));
            }

            let mut data = if alloc {
                Default::default()
            } else {
                unsafe { POOL.as_mut().unwrap().get() }
            };

            assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
            assert_ne!(data.id, 42);
            data.id = 42;

            /*            assert_eq!(arr.len(), BUF_CAP);
            assert_eq!(arr.0[42], 42);*/

            tx_clone.try_send(data).unwrap_or_default();
        }

        //        println!("Send one done...");
    });

    let send_two = thread::spawn(move || {
        for i in 0..TEST_SIZE {
            if i % DENOMINATOR == 0 {
                thread::sleep(Duration::from_nanos(BUSY_PERIOD));
            }

            let mut data = if alloc {
                Default::default()
            } else {
                unsafe { POOL.as_mut().unwrap().get() }
            };

            assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
            assert_ne!(data.id, 42);
            data.id = 42;

            /*            assert_eq!(arr.len(), BUF_CAP);
            assert_eq!(arr.0[42], 42);*/

            tx.try_send(data).unwrap_or_default();
        }

        //        println!("Send two done...");
    });

    let recv_one = thread::spawn(move || {
        thread::sleep(Duration::from_micros(5));

        while let Ok(arr) = rx.recv() {
            //            assert_eq!(arr.len(), BUF_CAP);

            if !alloc {
                unsafe {
                    POOL.as_mut().unwrap().put(arr);
                }
            }
        }

        //        println!("Recv one done...");
    });

    for i in 0..TEST_SIZE {
        // sleep a bit to create some concurrent actions
        if i % DENOMINATOR == 1 {
            thread::sleep(Duration::from_nanos(BUSY_PERIOD));
        }

        let mut data = if alloc {
            Default::default()
        } else {
            unsafe { POOL.as_mut().unwrap().get() }
        };

        //        assert!(data.id == 21 || data.id == 0, "Wrong id: {}", data.id);
        assert_ne!(data.id, 42);
        data.id = 42;

        /*        assert_eq!(arr.len(), BUF_CAP);
        assert_eq!(arr.0[42], 42);*/

        if !alloc {
            // when done using the object, make sure to put it back so the pool won't dry up
            unsafe { POOL.as_mut().unwrap().put(data) };
        }
    }

    //    println!("Main loop done...");

    send_one.join().unwrap_or_default();
    send_two.join().unwrap_or_default();
    recv_one.join().unwrap_or_default();

    now.elapsed().as_micros()
}

Trait Implementations

impl<T> Default for SyncPool<T>

where T: Default,

fn default() -> Self

Returns the “default value” for a type.

impl<T> Drop for SyncPool<T>

fn drop(&mut self)

Executes the destructor for this type.

impl<T> PoolManager<T> for SyncPool<T>

The pool manager that provide many useful utilities to keep the SyncPool close to the needs of the caller program.

fn reset_handle(&mut self, handle: fn(&mut T)) -> &mut Self

Set or update the reset handle. If set, the reset handle will be invoked every time an element has been returned back to the pool (i.e. calling the put method), regardless of if the element is created by the pool or not.

fn allow_expansion(&mut self, allow: bool) -> &mut Self

Set or update the settings that if we will allow the SyncPool to be expanded.

fn expand(&mut self, additional: usize, block: bool) -> bool

Try to expand the SyncPool and add more elements to it. Usually invoke this API only when the caller is certain that the pool is under pressure, and that a short block to the access of the pool won’t cause serious issues, since the function will block the current caller’s thread until it’s finished (i.e. get the opportunity to raise the writer’s barrier and wait everyone to leave).

If we’re unable to expand the pool, it’s due to one of the following reasons: 1) someone has already raised the writer’s barrier and is likely modifying the pool, we will leave immediately, and it’s up to the caller if they want to try again; 2) we’ve waited too long but still couldn’t obtain an exclusive access to the pool, and similar to reason 1), we will quit now.

fn refill(&mut self, additional: usize) -> usize

Due to contentious access to the pool, sometimes the put action could not finish and return the element to the pool successfully. Overtime, this could cause the number of elements in the pool to dwell. This would only happen slowly if we’re running a very contentious multithreading program, but it surely could happen. If the caller detects such situation, they can invoke the refill API and try to refill the pool with elements.

We will try to refill as many elements as requested

impl<T> PoolState for SyncPool<T>

fn expansion_enabled(&self) -> bool

fn miss_count(&self) -> usize

fn capacity(&self) -> usize

fn len(&self) -> usize

fn is_empty(&self) -> bool