From 9cfd7189887daa0626db04366fe4f59fb1d52f55 Mon Sep 17 00:00:00 2001
From: Bryan McCoid <bryan.mccoid@gmail.com>
Date: Mon, 28 Jun 2021 19:24:38 -0700
Subject: [PATCH 1/3] Begins fixing #179: Add 'tracing' dependency + example
tracepoints
Added the very popular 'tracing' library, as well as a basic subscriber for use
in tests, benchmarks, etc. Only includes a few tracepoints as an example and
doesn't begin to pretend it's instrumented the entire system (FAR from it). I
think with the dependency included people will be more inclined to help out
adding tracepoints. Also fixed compiler errors due to inability to infer types,
potentially caused by the addition of this new crate and it's traits, or were
part of a new compiler behavior. All it does is make them more verbose and
precise about the type so it's included in this commit as well. Only tests are
affected by this.
---
glommio/Cargo.toml | 2 ++
glommio/src/executor/mod.rs | 49 ++++++++++++++++++++++++++-
glommio/src/executor/multitask.rs | 1 +
glommio/src/executor/placement/mod.rs | 40 +++++++++++-----------
glommio/src/sys/sysfs.rs | 20 +++++++----
5 files changed, 84 insertions(+), 28 deletions(-)
diff --git a/glommio/Cargo.toml b/glommio/Cargo.toml
index 018cbe724..e1696b61c 100644
--- a/glommio/Cargo.toml
+++ b/glommio/Cargo.toml
@@ -36,12 +36,14 @@ intrusive-collections = "0.9.0"
lockfree = "0.5"
membarrier = "0.2.2"
itertools = "0.10.0"
+tracing = "0.1"
[dev-dependencies]
futures = "0.3.5"
fastrand = "1.4.0"
tokio = { version = "0.3.5", default-features = false, features = ["rt", "macros", "rt-multi-thread", "net", "io-util", "time"] }
rand = "0.8.0"
+tracing-subscriber = "0.2"
[build-dependencies]
cc = "1.0.47"
diff --git a/glommio/src/executor/mod.rs b/glommio/src/executor/mod.rs
index 703aab6d4..be77c4c6b 100644
--- a/glommio/src/executor/mod.rs
+++ b/glommio/src/executor/mod.rs
@@ -213,6 +213,7 @@ macro_rules! to_io_error {
fn bind_to_cpu_set(cpus: impl IntoIterator<Item = usize>) -> Result<()> {
let mut cpuset = nix::sched::CpuSet::new();
for cpu in cpus {
+ tracing::trace!("Attempting to bind to cpu: {}", cpu);
to_io_error!(&cpuset.set(cpu))?;
}
let pid = nix::unistd::Pid::from_raw(0);
@@ -938,6 +939,14 @@ impl LocalExecutor {
};
let io_requirements = IoRequirements::new(latency, index);
+ tracing::trace!(
+ shares = ?shares.clone(),
+ ?latency,
+ index,
+ ?io_requirements,
+ "Creating task queue"
+ );
+
let tq = TaskQueue::new(TaskQueueHandle { index }, name, shares, io_requirements);
self.queues
@@ -1130,6 +1139,22 @@ impl LocalExecutor {
/// assert_eq!(res, 6);
/// ```
pub fn run<T>(&self, future: impl Future<Output = T>) -> T {
+ let queues = self
+ .queues
+ .borrow()
+ .available_executors
+ .iter()
+ .map(|x| {
+ (
+ *x.0,
+ x.1.borrow().name.clone(),
+ x.1.borrow().shares.clone(),
+ x.1.borrow().io_requirements,
+ )
+ })
+ .collect::<Vec<_>>();
+ tracing::trace!(id = self.id, ?queues, "LocalExecutor running");
+
// this waker is never exposed in the public interface and is only used to check
// whether the task's `JoinHandle` is `Ready`
let waker = dummy_waker();
@@ -1486,6 +1511,7 @@ impl<T> Task<T> {
/// [`Shares`]: enum.Shares.html
/// [`Latency`]: enum.Latency.html
pub fn create_task_queue(shares: Shares, latency: Latency, name: &str) -> TaskQueueHandle {
+ tracing::trace!(?shares, ?latency, name, "Creating task queue");
LOCAL_EX.with(|local_ex| local_ex.create_task_queue(shares, latency, name))
}
@@ -1990,12 +2016,27 @@ mod test {
},
task::Waker,
};
+ use tracing_subscriber::EnvFilter;
#[test]
+ #[allow(unused_must_use)]
fn create_and_destroy_executor() {
+ tracing_subscriber::fmt::fmt()
+ .with_env_filter(EnvFilter::from_env("GLOMMIO_TRACE"))
+ .try_init();
+
+ tracing::info!("Started tracing..");
+ tracing::debug!("Started tracing..");
+ tracing::warn!("Started tracing..");
+ tracing::trace!("Started tracing..");
+ tracing::error!("Started tracing..");
let mut var = Rc::new(RefCell::new(0));
let local_ex = LocalExecutor::default();
-
+ let _handle = local_ex.create_task_queue(
+ Shares::Static(700),
+ Latency::Matters(Duration::from_millis(10)),
+ "latency-matters-bigly",
+ );
let varclone = var.clone();
local_ex.run(async move {
let mut m = varclone.borrow_mut();
@@ -2021,7 +2062,13 @@ mod test {
}
#[test]
+ #[allow(unused_must_use)]
fn bind_to_cpu_set_range() {
+ tracing_subscriber::fmt::fmt()
+ .with_env_filter(EnvFilter::from_env("GLOMMIO_TRACE"))
+ .try_init();
+ tracing::info!("Started tracing..");
+
// libc supports cpu ids up to 1023 and will use the intersection of values
// specified by the cpu mask and those present on the system
// https://man7.org/linux/man-pages/man2/sched_setaffinity.2.html#NOTES
diff --git a/glommio/src/executor/multitask.rs b/glommio/src/executor/multitask.rs
index c1995d639..9a6c450ea 100644
--- a/glommio/src/executor/multitask.rs
+++ b/glommio/src/executor/multitask.rs
@@ -139,6 +139,7 @@ impl LocalExecutor {
tq: Rc<RefCell<TaskQueue>>,
future: impl Future<Output = T>,
) -> Task<T> {
+ tracing::trace!(executor_id, task_queue = ?tq, "Spawning future");
let tq = Rc::downgrade(&tq);
// The function that schedules a runnable task when it gets woken up.
diff --git a/glommio/src/executor/placement/mod.rs b/glommio/src/executor/placement/mod.rs
index aa35608c2..a81c52f94 100644
--- a/glommio/src/executor/placement/mod.rs
+++ b/glommio/src/executor/placement/mod.rs
@@ -595,7 +595,7 @@ mod test {
let cpu_location = max_spreader.next().unwrap();
counts[cpu_location.cpu] += 1;
}
- assert_eq!(1, counts[..3].iter().sum());
+ assert_eq!(1, counts[..3].iter().sum::<i32>());
assert_eq!(1, counts[3]);
for _ in 2..4 {
@@ -609,7 +609,7 @@ mod test {
let cpu_location = max_spreader.next().unwrap();
counts[cpu_location.cpu] += 1;
}
- assert_eq!(4, counts[..3].iter().sum());
+ assert_eq!(4, counts[..3].iter().sum::<i32>());
assert_eq!(2, counts[3]);
for _ in 6..8 {
@@ -665,7 +665,7 @@ mod test {
}
counts
.chunks(nr_cpu / selected)
- .for_each(|c| assert_eq!(1, c.iter().sum()));
+ .for_each(|c| assert_eq!(1, c.iter().sum::<i32>()));
assert_eq!(selected, max_spreader.tree.nr_slots_selected());
selected_prev = selected;
}
@@ -689,35 +689,35 @@ mod test {
let mut max_packer = MaxPacker::from_topology(topology);
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
- assert_eq!(1, counts[..4].iter().sum());
- assert_eq!(0, counts[5..].iter().sum());
+ assert_eq!(1, counts[..4].iter().sum::<i32>());
+ assert_eq!(0, counts[5..].iter().sum::<i32>());
for _ in 1..4 {
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
}
- assert_eq!(4, counts[..4].iter().sum());
- assert_eq!(0, counts[5..].iter().sum());
+ assert_eq!(4, counts[..4].iter().sum::<i32>());
+ assert_eq!(0, counts[5..].iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
assert_eq!(1, counts[6]);
- assert_eq!(5, counts.iter().sum());
+ assert_eq!(5, counts.iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
assert_eq!(1, counts[5]);
- assert_eq!(6, counts.iter().sum());
+ assert_eq!(6, counts.iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
assert_eq!(1, counts[4]);
- assert_eq!(7, counts.iter().sum());
+ assert_eq!(7, counts.iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
- assert_eq!(5, counts[..4].iter().sum());
- assert_eq!(8, counts.iter().sum());
+ assert_eq!(5, counts[..4].iter().sum::<i32>());
+ assert_eq!(8, counts.iter().sum::<i32>());
for _ in 8..10 * counts.len() {
let cpu_location = max_packer.next().unwrap();
@@ -746,27 +746,27 @@ mod test {
let mut max_packer = MaxPacker::from_topology(topology);
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
- assert_eq!(1, counts[4..].iter().sum());
- assert_eq!(0, counts[..4].iter().sum());
+ assert_eq!(1, counts[4..].iter().sum::<i32>());
+ assert_eq!(0, counts[..4].iter().sum::<i32>());
for _ in 1..3 {
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
}
- assert_eq!(3, counts[4..].iter().sum());
- assert_eq!(0, counts[..4].iter().sum());
+ assert_eq!(3, counts[4..].iter().sum::<i32>());
+ assert_eq!(0, counts[..4].iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
- assert_eq!(4, counts[3..].iter().sum());
- assert_eq!(0, counts[..3].iter().sum());
+ assert_eq!(4, counts[3..].iter().sum::<i32>());
+ assert_eq!(0, counts[..3].iter().sum::<i32>());
for _ in 4..6 {
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
}
- assert_eq!(6, counts[1..].iter().sum());
- assert_eq!(0, counts[..1].iter().sum());
+ assert_eq!(6, counts[1..].iter().sum::<i32>());
+ assert_eq!(0, counts[..1].iter().sum::<i32>());
let cpu_location = max_packer.next().unwrap();
counts[cpu_location.cpu] += 1;
diff --git a/glommio/src/sys/sysfs.rs b/glommio/src/sys/sysfs.rs
index 5d1bedaad..fb0456b9e 100644
--- a/glommio/src/sys/sysfs.rs
+++ b/glommio/src/sys/sysfs.rs
@@ -537,15 +537,21 @@ mod test {
#[test]
fn hex_bit_iterator() -> io::Result<()> {
- assert_eq!(HexBitIterator::from_str("")?.collect::<Vec<_>>(), vec![]);
- assert_eq!(HexBitIterator::from_str("\n")?.collect::<Vec<_>>(), vec![]);
+ assert_eq!(
+ HexBitIterator::from_str("")?.collect::<Vec<_>>(),
+ Vec::<usize>::new()
+ );
+ assert_eq!(
+ HexBitIterator::from_str("\n")?.collect::<Vec<_>>(),
+ Vec::<usize>::new()
+ );
assert_eq!(
HexBitIterator::from_str("00\n")?.collect::<Vec<_>>(),
- vec![]
+ Vec::<usize>::new()
);
assert_eq!(
HexBitIterator::from_str("0,0\n")?.collect::<Vec<_>>(),
- vec![]
+ Vec::<usize>::new()
);
assert_eq!(
HexBitIterator::from_str("03")?.collect::<Vec<_>>(),
@@ -622,7 +628,7 @@ mod test {
.set_used_sized(0)
.check()?
.collect::<Vec<_>>();
- assert_eq!(it, vec![]);
+ assert_eq!(it, Vec::<usize>::new());
let it = RangeIter::new(5)
.set_end(25)
@@ -683,13 +689,13 @@ mod test {
ListIterator::from_str("\0")?
.collect_ok::<Vec<_>>()
.unwrap(),
- vec![]
+ Vec::<usize>::new()
);
assert_eq!(
ListIterator::from_str("\n")?
.collect_ok::<Vec<_>>()
.unwrap(),
- vec![]
+ Vec::<usize>::new()
);
assert_eq!(
ListIterator::from_str("0,8\n")?
From 8b6af3cd545dfb4b324cc2310c3103c2a20cda4b Mon Sep 17 00:00:00 2001
From: Bryan McCoid <bryan.mccoid@gmail.com>
Date: Tue, 29 Jun 2021 19:13:18 -0700
Subject: [PATCH 2/3] Removed needless tracepoints + explicit use for macros
Removed uneccessary tracepoints and explicitly import macros for easier future
transitions to other macro implementations / backends.
---
glommio/src/executor/mod.rs | 50 ++-----------------------------
glommio/src/executor/multitask.rs | 4 ++-
glommio/src/lib.rs | 16 ++++++++++
3 files changed, 22 insertions(+), 48 deletions(-)
diff --git a/glommio/src/executor/mod.rs b/glommio/src/executor/mod.rs
index be77c4c6b..4592c8df0 100644
--- a/glommio/src/executor/mod.rs
+++ b/glommio/src/executor/mod.rs
@@ -35,6 +35,7 @@ mod placement;
use latch::{Latch, LatchState};
pub use placement::{CpuSet, Placement};
+use tracing::trace;
use std::{
cell::RefCell,
@@ -213,7 +214,7 @@ macro_rules! to_io_error {
fn bind_to_cpu_set(cpus: impl IntoIterator<Item = usize>) -> Result<()> {
let mut cpuset = nix::sched::CpuSet::new();
for cpu in cpus {
- tracing::trace!("Attempting to bind to cpu: {}", cpu);
+ trace!("Attempting to bind to cpu: {}", cpu);
to_io_error!(&cpuset.set(cpu))?;
}
let pid = nix::unistd::Pid::from_raw(0);
@@ -939,14 +940,6 @@ impl LocalExecutor {
};
let io_requirements = IoRequirements::new(latency, index);
- tracing::trace!(
- shares = ?shares.clone(),
- ?latency,
- index,
- ?io_requirements,
- "Creating task queue"
- );
-
let tq = TaskQueue::new(TaskQueueHandle { index }, name, shares, io_requirements);
self.queues
@@ -1139,22 +1132,6 @@ impl LocalExecutor {
/// assert_eq!(res, 6);
/// ```
pub fn run<T>(&self, future: impl Future<Output = T>) -> T {
- let queues = self
- .queues
- .borrow()
- .available_executors
- .iter()
- .map(|x| {
- (
- *x.0,
- x.1.borrow().name.clone(),
- x.1.borrow().shares.clone(),
- x.1.borrow().io_requirements,
- )
- })
- .collect::<Vec<_>>();
- tracing::trace!(id = self.id, ?queues, "LocalExecutor running");
-
// this waker is never exposed in the public interface and is only used to check
// whether the task's `JoinHandle` is `Ready`
let waker = dummy_waker();
@@ -1511,7 +1488,7 @@ impl<T> Task<T> {
/// [`Shares`]: enum.Shares.html
/// [`Latency`]: enum.Latency.html
pub fn create_task_queue(shares: Shares, latency: Latency, name: &str) -> TaskQueueHandle {
- tracing::trace!(?shares, ?latency, name, "Creating task queue");
+ trace!(?shares, ?latency, name, "Creating task queue");
LOCAL_EX.with(|local_ex| local_ex.create_task_queue(shares, latency, name))
}
@@ -2019,24 +1996,9 @@ mod test {
use tracing_subscriber::EnvFilter;
#[test]
- #[allow(unused_must_use)]
fn create_and_destroy_executor() {
- tracing_subscriber::fmt::fmt()
- .with_env_filter(EnvFilter::from_env("GLOMMIO_TRACE"))
- .try_init();
-
- tracing::info!("Started tracing..");
- tracing::debug!("Started tracing..");
- tracing::warn!("Started tracing..");
- tracing::trace!("Started tracing..");
- tracing::error!("Started tracing..");
let mut var = Rc::new(RefCell::new(0));
let local_ex = LocalExecutor::default();
- let _handle = local_ex.create_task_queue(
- Shares::Static(700),
- Latency::Matters(Duration::from_millis(10)),
- "latency-matters-bigly",
- );
let varclone = var.clone();
local_ex.run(async move {
let mut m = varclone.borrow_mut();
@@ -2062,13 +2024,7 @@ mod test {
}
#[test]
- #[allow(unused_must_use)]
fn bind_to_cpu_set_range() {
- tracing_subscriber::fmt::fmt()
- .with_env_filter(EnvFilter::from_env("GLOMMIO_TRACE"))
- .try_init();
- tracing::info!("Started tracing..");
-
// libc supports cpu ids up to 1023 and will use the intersection of values
// specified by the cpu mask and those present on the system
// https://man7.org/linux/man-pages/man2/sched_setaffinity.2.html#NOTES
diff --git a/glommio/src/executor/multitask.rs b/glommio/src/executor/multitask.rs
index 9a6c450ea..029b8f24f 100644
--- a/glommio/src/executor/multitask.rs
+++ b/glommio/src/executor/multitask.rs
@@ -8,6 +8,8 @@
#![forbid(unsafe_code)]
#![warn(missing_docs, missing_debug_implementations)]
+use tracing::trace;
+
use crate::{
executor::{maybe_activate, TaskQueue},
task::{task_impl, JoinHandle},
@@ -139,7 +141,7 @@ impl LocalExecutor {
tq: Rc<RefCell<TaskQueue>>,
future: impl Future<Output = T>,
) -> Task<T> {
- tracing::trace!(executor_id, task_queue = ?tq, "Spawning future");
+ trace!(executor_id, task_queue = ?tq, "Spawning future");
let tq = Rc::downgrade(&tq);
// The function that schedules a runnable task when it gets woken up.
diff --git a/glommio/src/lib.rs b/glommio/src/lib.rs
index e95f4b07c..38a7ab93a 100644
--- a/glommio/src/lib.rs
+++ b/glommio/src/lib.rs
@@ -679,6 +679,8 @@ pub(crate) mod test_utils {
use super::*;
use nix::sys::statfs::*;
use std::path::{Path, PathBuf};
+ use tracing::{debug, error, info, trace, warn};
+ use tracing_subscriber::EnvFilter;
#[derive(Copy, Clone)]
pub(crate) enum TestDirectoryKind {
@@ -730,4 +732,18 @@ pub(crate) mod test_utils {
};
TestDirectory { path: dir, kind }
}
+
+ #[test]
+ #[allow(unused_must_use)]
+ fn test_tracing_init() {
+ tracing_subscriber::fmt::fmt()
+ .with_env_filter(EnvFilter::from_env("GLOMMIO_TRACE"))
+ .try_init();
+
+ info!("Started tracing..");
+ debug!("Started tracing..");
+ warn!("Started tracing..");
+ trace!("Started tracing..");
+ error!("Started tracing..");
+ }
}
From 38c48e2fc6a78eae4d010e761700d67a28f80126 Mon Sep 17 00:00:00 2001
From: Bryan McCoid <bryan.mccoid@gmail.com>
Date: Wed, 30 Jun 2021 21:44:27 -0700
Subject: [PATCH 3/3] Removed more of the poor examples
Left only the executor creation tracepoints as the single example.
---
glommio/src/executor/mod.rs | 3 +--
glommio/src/executor/multitask.rs | 3 ---
2 files changed, 1 insertion(+), 5 deletions(-)
diff --git a/glommio/src/executor/mod.rs b/glommio/src/executor/mod.rs
index 4592c8df0..526a25d7a 100644
--- a/glommio/src/executor/mod.rs
+++ b/glommio/src/executor/mod.rs
@@ -214,7 +214,6 @@ macro_rules! to_io_error {
fn bind_to_cpu_set(cpus: impl IntoIterator<Item = usize>) -> Result<()> {
let mut cpuset = nix::sched::CpuSet::new();
for cpu in cpus {
- trace!("Attempting to bind to cpu: {}", cpu);
to_io_error!(&cpuset.set(cpu))?;
}
let pid = nix::unistd::Pid::from_raw(0);
@@ -907,6 +906,7 @@ impl LocalExecutor {
}
let p = parking::Parker::new();
let queues = ExecutorQueues::new(preempt_timer, spin_before_park);
+ trace!(id = notifier.id(), "Creating executor");
Ok(LocalExecutor {
queues: Rc::new(RefCell::new(queues)),
parker: p,
@@ -1488,7 +1488,6 @@ impl<T> Task<T> {
/// [`Shares`]: enum.Shares.html
/// [`Latency`]: enum.Latency.html
pub fn create_task_queue(shares: Shares, latency: Latency, name: &str) -> TaskQueueHandle {
- trace!(?shares, ?latency, name, "Creating task queue");
LOCAL_EX.with(|local_ex| local_ex.create_task_queue(shares, latency, name))
}
diff --git a/glommio/src/executor/multitask.rs b/glommio/src/executor/multitask.rs
index 029b8f24f..c1995d639 100644
--- a/glommio/src/executor/multitask.rs
+++ b/glommio/src/executor/multitask.rs
@@ -8,8 +8,6 @@
#![forbid(unsafe_code)]
#![warn(missing_docs, missing_debug_implementations)]
-use tracing::trace;
-
use crate::{
executor::{maybe_activate, TaskQueue},
task::{task_impl, JoinHandle},
@@ -141,7 +139,6 @@ impl LocalExecutor {
tq: Rc<RefCell<TaskQueue>>,
future: impl Future<Output = T>,
) -> Task<T> {
- trace!(executor_id, task_queue = ?tq, "Spawning future");
let tq = Rc::downgrade(&tq);
// The function that schedules a runnable task when it gets woken up.