mirror-github-bspeice/aeron-rs
1
0
Fork 0
mirror of https://github.com/bspeice/aeron-rs synced 2024-12-22 05:48:10 -05:00
Code Issues Releases Wiki Activity

Formatting

Browse Source
This commit is contained in:
Bradlee Speice 2019-11-02 15:50:33 -04:00
parent 8fac817ba3
commit b235655f71
2 changed files with 13 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
aeron-rs/src/client/concurrent/mod.rs
View File

@ -191,7 +191,9 @@ pub trait AtomicBuffer: Deref<Target = [u8]> + DerefMut<Target = [u8]> {
/// Repeatedly write a value into an atomic buffer. Guaranteed to use `memset`. /// Repeatedly write a value into an atomic buffer. Guaranteed to use `memset`.
fn set_memory(&mut self, offset: IndexT, length: usize, value: u8) -> Result<()> { fn set_memory(&mut self, offset: IndexT, length: usize, value: u8) -> Result<()> {
self.bounds_check(offset, length as IndexT).map(|_| unsafe { self.bounds_check(offset, length as IndexT).map(|_| unsafe {
self.as_mut_ptr().offset(offset as isize).write_bytes(value, length) self.as_mut_ptr()
.offset(offset as isize)
.write_bytes(value, length)
}) })
} }

19
aeron-rs/src/client/concurrent/ringbuffer.rs
View File

@ -199,7 +199,10 @@ where
{ {
// QUESTION: Should I implement the `get_i64` method that C++ uses? // QUESTION: Should I implement the `get_i64` method that C++ uses?
// UNWRAP: Bounds check performed during buffer creation // UNWRAP: Bounds check performed during buffer creation
let head = self.buffer.get_i64_volatile(self.head_position_index).unwrap(); let head = self
.buffer
.get_i64_volatile(self.head_position_index)
.unwrap();
let head_index = (head & i64::from(self.capacity - 1)) as i32; let head_index = (head & i64::from(self.capacity - 1)) as i32;
let contiguous_block_length = self.capacity - head_index; let contiguous_block_length = self.capacity - head_index;
let mut messages_read = 0; let mut messages_read = 0;
@ -379,9 +382,7 @@ where
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::client::concurrent::ringbuffer::{ use crate::client::concurrent::ringbuffer::{record_descriptor, ManyToOneRingBuffer};
record_descriptor, ManyToOneRingBuffer,
};
use crate::client::concurrent::AtomicBuffer; use crate::client::concurrent::AtomicBuffer;
use crate::util::IndexT; use crate::util::IndexT;
use std::mem::size_of; use std::mem::size_of;
@ -449,7 +450,8 @@ mod tests {
#[test] #[test]
fn read_basic() { fn read_basic() {
// Similar to write basic, put something into the buffer // Similar to write basic, put something into the buffer
let mut ring_buffer = ManyToOneRingBuffer::new(vec![0u8; BUFFER_SIZE]).expect("Invalid buffer size"); let mut ring_buffer =
ManyToOneRingBuffer::new(vec![0u8; BUFFER_SIZE]).expect("Invalid buffer size");
let mut source_buffer = &mut [12u8, 0, 0, 0, 0, 0, 0, 0][..]; let mut source_buffer = &mut [12u8, 0, 0, 0, 0, 0, 0, 0][..];
let source_len = source_buffer.len() as IndexT; let source_len = source_buffer.len() as IndexT;
@ -459,9 +461,7 @@ mod tests {
.unwrap(); .unwrap();
// Now we can start the actual read process // Now we can start the actual read process
let c = |_, buf: &Vec<u8>, offset, _| { let c = |_, buf: &Vec<u8>, offset, _| assert_eq!(buf.get_i64_volatile(offset).unwrap(), 12);
assert_eq!(buf.get_i64_volatile(offset).unwrap(), 12)
};
ring_buffer.read(c, 1).unwrap(); ring_buffer.read(c, 1).unwrap();
// Make sure that the buffer was zeroed on finish // Make sure that the buffer was zeroed on finish
@ -472,7 +472,8 @@ mod tests {
#[test] #[test]
fn read_multiple() { fn read_multiple() {
let mut ring_buffer = ManyToOneRingBuffer::new(vec![0u8; BUFFER_SIZE]).expect("Invalid buffer size"); let mut ring_buffer =
ManyToOneRingBuffer::new(vec![0u8; BUFFER_SIZE]).expect("Invalid buffer size");
let mut source_buffer = &mut [12u8, 0, 0, 0, 0, 0, 0, 0][..]; let mut source_buffer = &mut [12u8, 0, 0, 0, 0, 0, 0, 0][..];
let source_len = source_buffer.len() as IndexT; let source_len = source_buffer.len() as IndexT;