Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ARROW-10149: [Rust] Improved support for externally owned memory regions #8316

Closed
wants to merge 3 commits into from
Closed

ARROW-10149: [Rust] Improved support for externally owned memory regions #8316

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
548cf5b
Improved support for externally owned memory regions.
jorgecarleitao Oct 1, 2020
377c908
Added tests, improved code and docs.
jorgecarleitao Oct 1, 2020
b0ff0ec
Improved test
jorgecarleitao Oct 3, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
162 changes: 24 additions & 138 deletions rust/arrow/src/buffer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,14 @@
#[cfg(feature = "simd")]
use packed_simd::u8x64;

use crate::bytes::{Bytes, Deallocation, DropFn};

use std::cmp;
use std::convert::AsRef;
use std::fmt::{Debug, Formatter};
use std::fmt::Debug;
use std::io::{Error as IoError, ErrorKind, Result as IoResult, Write};
use std::mem;
use std::ops::{BitAnd, BitOr, Not};
use std::ptr::NonNull;
use std::slice::{from_raw_parts, from_raw_parts_mut};
use std::sync::Arc;

Expand All @@ -38,92 +39,16 @@ use crate::util::bit_util;
#[cfg(feature = "simd")]
use std::borrow::BorrowMut;

/// Buffer is a contiguous memory region of fixed size and is aligned at a 64-byte
/// boundary. Buffer is immutable.
#[derive(PartialEq, Debug)]
/// [`Buffer`] is an immutable contiguous memory region.
#[derive(PartialEq, Clone, Debug)]
pub struct Buffer {
/// Reference-counted pointer to the internal byte buffer.
data: Arc<BufferData>,
data: Arc<Bytes>,

/// The offset into the buffer.
offset: usize,
}

struct BufferData {
/// The raw pointer into the buffer bytes
ptr: *const u8,

/// The length (num of bytes) of the buffer. The region `[0, len)` of the buffer
/// is occupied with meaningful data, while the rest `[len, capacity)` is the
/// unoccupied region.
len: usize,

/// Whether this piece of memory is owned by this object
owned: bool,

/// The capacity (num of bytes) of the buffer
/// Invariant: len <= capacity
capacity: usize,
}

impl PartialEq for BufferData {
fn eq(&self, other: &BufferData) -> bool {
if self.capacity != other.capacity {
return false;
}

self.data() == other.data()
}
}

/// Release the underlying memory when the current buffer goes out of scope
impl Drop for BufferData {
fn drop(&mut self) {
if self.is_allocated() && self.owned {
unsafe { memory::free_aligned(self.ptr as *mut u8, self.capacity) };
}
}
}

impl Debug for BufferData {
fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
write!(
f,
"BufferData {{ ptr: {:?}, len: {}, capacity: {}, data: ",
self.ptr, self.len, self.capacity
)?;

f.debug_list().entries(self.data().iter()).finish()?;

write!(f, " }}")
}
}

impl BufferData {
fn data(&self) -> &[u8] {
if !self.is_allocated() {
&[]
} else {
unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
}
}

fn is_zst(&self) -> bool {
self.ptr == BUFFER_INIT.as_ptr() as _
}

fn is_allocated(&self) -> bool {
!(self.is_zst() || self.ptr.is_null())
}
}

///
/// SAFETY: (vcq):
/// As you can see this is global and lives as long as the program lives.
/// This is used for lazy allocation in the further steps of buffer allocations.
/// Pointer below is well-aligned, only used for ZSTs and discarded afterwards.
const BUFFER_INIT: NonNull<u8> = NonNull::dangling();

impl Buffer {
/// Creates a buffer from an existing memory region (must already be byte-aligned), this
/// `Buffer` will free this piece of memory when dropped.
Expand All @@ -139,7 +64,7 @@ impl Buffer {
/// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
/// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
pub unsafe fn from_raw_parts(ptr: *const u8, len: usize, capacity: usize) -> Self {
Buffer::build_with_arguments(ptr, len, capacity, true)
Buffer::build_with_arguments(ptr, len, Deallocation::Native(capacity))
}

/// Creates a buffer from an existing memory region (must already be byte-aligned), this
Expand All @@ -149,70 +74,48 @@ impl Buffer {
///
/// * `ptr` - Pointer to raw parts
/// * `len` - Length of raw parts in **bytes**
/// * `capacity` - Total allocated memory for the pointer `ptr`, in **bytes**
/// * `drop` - Function that frees the corresponding memory region. Note that the region may be larger than `len`.
///
/// # Safety
///
/// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
/// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
pub unsafe fn from_unowned(ptr: *const u8, len: usize, capacity: usize) -> Self {
Buffer::build_with_arguments(ptr, len, capacity, false)
/// bytes and that the foreign deallocator frees the region.
pub unsafe fn from_unowned(ptr: *const u8, len: usize, drop: DropFn) -> Self {
Buffer::build_with_arguments(ptr, len, Deallocation::Foreign(drop))
}

/// Creates a buffer from an existing memory region (must already be byte-aligned).
///
/// # Arguments
///
/// * `ptr` - Pointer to raw parts
/// * `len` - Length of raw parts in bytes
/// * `capacity` - Total allocated memory for the pointer `ptr`, in **bytes**
/// * `owned` - Whether the raw parts is owned by this `Buffer`. If true, this `Buffer` will
/// free this memory when dropped, otherwise it will skip freeing the raw parts.
///
/// # Safety
///
/// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
/// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
/// Auxiliary method to create a new Buffer
unsafe fn build_with_arguments(
ptr: *const u8,
len: usize,
capacity: usize,
owned: bool,
deallocation: Deallocation,
) -> Self {
assert!(
memory::is_aligned(ptr, memory::ALIGNMENT),
"memory not aligned"
);
let buf_data = BufferData {
ptr,
len,
capacity,
owned,
};
let bytes = Bytes::new(ptr, len, deallocation);
Buffer {
data: Arc::new(buf_data),
data: Arc::new(bytes),
offset: 0,
}
}

/// Returns the number of bytes in the buffer
pub fn len(&self) -> usize {
self.data.len - self.offset
self.data.len() - self.offset
}

/// Returns the capacity of this buffer
/// Returns the capacity of this buffer.
/// For exernally owned buffers, this returns zero
pub fn capacity(&self) -> usize {
self.data.capacity
self.data.capacity()
}

/// Returns whether the buffer is empty.
pub fn is_empty(&self) -> bool {
self.data.len - self.offset == 0
self.data.len() - self.offset == 0
}

/// Returns the byte slice stored in this buffer
pub fn data(&self) -> &[u8] {
&self.data.data()[self.offset..]
&self.data.as_slice()[self.offset..]
}

/// Returns a slice of this buffer, starting from `offset`.
Expand All @@ -232,7 +135,7 @@ impl Buffer {
/// Note that this should be used cautiously, and the returned pointer should not be
/// stored anywhere, to avoid dangling pointers.
pub fn raw_data(&self) -> *const u8 {
unsafe { self.data.ptr.add(self.offset) }
unsafe { self.data.raw_data().add(self.offset) }
}

/// View buffer as typed slice.
Expand All @@ -253,20 +156,6 @@ impl Buffer {
self.len() / mem::size_of::<T>(),
)
}

/// Returns an empty buffer.
pub fn empty() -> Self {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was not being used, and thus I dropped it.

unsafe { Self::from_raw_parts(BUFFER_INIT.as_ptr() as _, 0, 0) }
}
}

impl Clone for Buffer {
fn clone(&self) -> Buffer {
Buffer {
data: self.data.clone(),
offset: self.offset,
}
}
}

/// Creating a `Buffer` instance by copying the memory from a `AsRef<[u8]>` into a newly
Expand Down Expand Up @@ -667,11 +556,8 @@ impl MutableBuffer {

/// Freezes this buffer and return an immutable version of it.
pub fn freeze(self) -> Buffer {
let buffer_data = BufferData {
ptr: self.data,
len: self.len,
capacity: self.capacity,
owned: true,
let buffer_data = unsafe {
Bytes::new(self.data, self.len, Deallocation::Native(self.capacity))
};
std::mem::forget(self);
Buffer {
Expand Down
Loading