mirror of https://github.com/aya-rs/aya
bpf, perf_map: make maps usable from multiple threads
Change PerfMap API so that individual buffers can be read from multiple threads. Change the way maps are stored in the `Bpf` struct from RefCell to a custom RwLock.pull/1/head
Alessandro Decina
4 years ago
parent
d7c91efb2d
commit
d4e282535b
@ -0,0 +1,79 @@
|
||||
use parking_lot::{RwLock, RwLockReadGuard, RwLockWriteGuard};
|
||||
use std::{
|
||||
mem,
|
||||
ops::{Deref, DerefMut},
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use crate::maps::Map;
|
||||
|
||||
pub(crate) struct MapLockError;
|
||||
|
||||
/* FIXME: write a full RwLock implementation that doesn't use borrowing guards
|
||||
* so that try_read() and try_write() don't have to use the ugly lifetime
|
||||
* extension hack */
|
||||
|
||||
#[derive(Debug)]
|
||||
pub(crate) struct MapLock {
|
||||
inner: Arc<RwLock<Map>>,
|
||||
}
|
||||
|
||||
impl MapLock {
|
||||
pub(crate) fn new(map: Map) -> MapLock {
|
||||
MapLock {
|
||||
inner: Arc::new(RwLock::new(map)),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn try_read(&self) -> Result<MapRef, MapLockError> {
|
||||
let lock: Option<RwLockReadGuard<'static, Map>> =
|
||||
unsafe { mem::transmute(self.inner.try_read()) };
|
||||
lock.map(|guard| MapRef {
|
||||
_lock: self.inner.clone(),
|
||||
guard,
|
||||
})
|
||||
.ok_or(MapLockError)
|
||||
}
|
||||
|
||||
pub(crate) fn try_write(&self) -> Result<MapRefMut, MapLockError> {
|
||||
let lock: Option<RwLockWriteGuard<'static, Map>> =
|
||||
unsafe { mem::transmute(self.inner.try_write()) };
|
||||
lock.map(|guard| MapRefMut {
|
||||
_lock: self.inner.clone(),
|
||||
guard,
|
||||
})
|
||||
.ok_or(MapLockError)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct MapRef {
|
||||
_lock: Arc<RwLock<Map>>,
|
||||
guard: RwLockReadGuard<'static, Map>,
|
||||
}
|
||||
|
||||
pub struct MapRefMut {
|
||||
_lock: Arc<RwLock<Map>>,
|
||||
guard: RwLockWriteGuard<'static, Map>,
|
||||
}
|
||||
|
||||
impl Deref for MapRef {
|
||||
type Target = Map;
|
||||
|
||||
fn deref(&self) -> &Map {
|
||||
&*self.guard
|
||||
}
|
||||
}
|
||||
|
||||
impl Deref for MapRefMut {
|
||||
type Target = Map;
|
||||
|
||||
fn deref(&self) -> &Map {
|
||||
&*self.guard
|
||||
}
|
||||
}
|
||||
|
||||
impl DerefMut for MapRefMut {
|
||||
fn deref_mut(&mut self) -> &mut Map {
|
||||
&mut *self.guard
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,117 @@
|
||||
use std::{
|
||||
convert::TryFrom,
|
||||
io,
|
||||
ops::DerefMut,
|
||||
os::unix::io::{AsRawFd, RawFd},
|
||||
sync::Arc,
|
||||
};
|
||||
|
||||
use bytes::BytesMut;
|
||||
use libc::{sysconf, _SC_PAGESIZE};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::{
|
||||
generated::bpf_map_type::BPF_MAP_TYPE_PERF_EVENT_ARRAY,
|
||||
maps::{
|
||||
perf_map::{Events, PerfBuffer, PerfBufferError},
|
||||
Map, MapError, MapRefMut,
|
||||
},
|
||||
sys::bpf_map_update_elem,
|
||||
};
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum PerfMapError {
|
||||
#[error("error parsing /sys/devices/system/cpu/online")]
|
||||
InvalidOnlineCpuFile,
|
||||
|
||||
#[error("no CPUs specified")]
|
||||
NoCpus,
|
||||
|
||||
#[error("invalid cpu {cpu_id}")]
|
||||
InvalidCpu { cpu_id: u32 },
|
||||
|
||||
#[error("map error: {0}")]
|
||||
MapError(#[from] MapError),
|
||||
|
||||
#[error("perf buffer error: {0}")]
|
||||
PerfBufferError(#[from] PerfBufferError),
|
||||
|
||||
#[error(transparent)]
|
||||
IOError(#[from] io::Error),
|
||||
|
||||
#[error("bpf_map_update_elem failed: {io_error}")]
|
||||
UpdateElementError {
|
||||
#[source]
|
||||
io_error: io::Error,
|
||||
},
|
||||
}
|
||||
|
||||
pub struct PerfMapBuffer<T: DerefMut<Target = Map>> {
|
||||
_map: Arc<T>,
|
||||
buf: PerfBuffer,
|
||||
}
|
||||
|
||||
impl<T: DerefMut<Target = Map>> PerfMapBuffer<T> {
|
||||
pub fn readable(&self) -> bool {
|
||||
self.buf.readable()
|
||||
}
|
||||
|
||||
pub fn read_events(&mut self, buffers: &mut [BytesMut]) -> Result<Events, PerfBufferError> {
|
||||
self.buf.read_events(buffers)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: DerefMut<Target = Map>> AsRawFd for PerfMapBuffer<T> {
|
||||
fn as_raw_fd(&self) -> RawFd {
|
||||
self.buf.as_raw_fd()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct PerfMap<T: DerefMut<Target = Map>> {
|
||||
map: Arc<T>,
|
||||
page_size: usize,
|
||||
}
|
||||
|
||||
impl<T: DerefMut<Target = Map>> PerfMap<T> {
|
||||
pub fn new(map: T) -> Result<PerfMap<T>, PerfMapError> {
|
||||
let map_type = map.obj.def.map_type;
|
||||
if map_type != BPF_MAP_TYPE_PERF_EVENT_ARRAY as u32 {
|
||||
return Err(MapError::InvalidMapType {
|
||||
map_type: map_type as u32,
|
||||
})?;
|
||||
}
|
||||
let _fd = map.fd_or_err()?;
|
||||
|
||||
Ok(PerfMap {
|
||||
map: Arc::new(map),
|
||||
// Safety: libc
|
||||
page_size: unsafe { sysconf(_SC_PAGESIZE) } as usize,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn open(
|
||||
&mut self,
|
||||
index: u32,
|
||||
page_count: Option<usize>,
|
||||
) -> Result<PerfMapBuffer<T>, PerfMapError> {
|
||||
// FIXME: keep track of open buffers
|
||||
|
||||
let map_fd = self.map.fd_or_err()?;
|
||||
let buf = PerfBuffer::open(index, self.page_size, page_count.unwrap_or(2))?;
|
||||
bpf_map_update_elem(map_fd, &index, &buf.as_raw_fd(), 0)
|
||||
.map_err(|(_, io_error)| PerfMapError::UpdateElementError { io_error })?;
|
||||
|
||||
Ok(PerfMapBuffer {
|
||||
buf,
|
||||
_map: self.map.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<MapRefMut> for PerfMap<MapRefMut> {
|
||||
type Error = PerfMapError;
|
||||
|
||||
fn try_from(a: MapRefMut) -> Result<PerfMap<MapRefMut>, PerfMapError> {
|
||||
PerfMap::new(a)
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
use std::{fs, io, str::FromStr};
|
||||
|
||||
const ONLINE_CPUS: &str = "/sys/devices/system/cpu/online";
|
||||
|
||||
pub fn online_cpus() -> Result<Vec<u32>, io::Error> {
|
||||
let data = fs::read_to_string(ONLINE_CPUS)?;
|
||||
parse_online_cpus(data.trim()).map_err(|_| {
|
||||
io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
format!("unexpected {} format", ONLINE_CPUS),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_online_cpus(data: &str) -> Result<Vec<u32>, ()> {
|
||||
let mut cpus = Vec::new();
|
||||
for range in data.split(',') {
|
||||
cpus.extend({
|
||||
match range
|
||||
.splitn(2, '-')
|
||||
.map(u32::from_str)
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|_| ())?
|
||||
.as_slice()
|
||||
{
|
||||
&[] | &[_, _, _, ..] => return Err(()),
|
||||
&[start] => start..=start,
|
||||
&[start, end] => start..=end,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Ok(cpus)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::iter::FromIterator;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_parse_online_cpus() {
|
||||
assert_eq!(parse_online_cpus("0").unwrap(), vec![0]);
|
||||
assert_eq!(parse_online_cpus("0,1").unwrap(), vec![0, 1]);
|
||||
assert_eq!(parse_online_cpus("0,1,2").unwrap(), vec![0, 1, 2]);
|
||||
assert_eq!(parse_online_cpus("0-7").unwrap(), Vec::from_iter(0..=7));
|
||||
assert_eq!(parse_online_cpus("0-3,4-7").unwrap(), Vec::from_iter(0..=7));
|
||||
assert_eq!(parse_online_cpus("0-5,6,7").unwrap(), Vec::from_iter(0..=7));
|
||||
assert!(parse_online_cpus("").is_err());
|
||||
assert!(parse_online_cpus("0-1,2-").is_err());
|
||||
assert!(parse_online_cpus("foo").is_err());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue