hubcio commented on code in PR #3615:
URL: https://github.com/apache/iggy/pull/3615#discussion_r3535835633
##########
core/shard_allocator/src/lib.rs:
##########
@@ -287,39 +335,66 @@ impl ShardAllocator {
pub fn to_shard_assignments(&self) -> Result<Vec<ShardInfo>,
ShardingError> {
match &self.allocation {
CpuAllocation::All => {
+ // `available_parallelism` already accounts for both the
+ // affinity mask and any cgroup CPU quota, so a
+ // quota-restricted process gets proportionally fewer shards.
let available_cpus = available_parallelism()
.map_err(|err| ShardingError::Other {
msg: format!("Failed to get available_parallelism:
{:?}", err),
})?
.get();
- let shard_assignments: Vec<_> = (0..available_cpus)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
- .collect();
+ if !self.pin_cores {
+ info!("Using all available CPU cores ({available_cpus}
shards, unpinned)");
+ return Ok(unpinned(available_cpus));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments =
+ pinned_from_allowed(&allowed,
available_cpus.min(allowed.len()))?;
info!(
- "Using all available CPU cores ({} shards with affinity)",
- shard_assignments.len()
+ "Using all available CPU cores ({} shards pinned within
allowed set {:?})",
+ shard_assignments.len(),
+ allowed
);
Ok(shard_assignments)
}
CpuAllocation::Count(count) => {
- let shard_assignments = (0..*count)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
+ if !self.pin_cores {
+ info!("Using {count} shard(s), unpinned");
+ return Ok(unpinned(*count));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments = pinned_from_allowed(&allowed, *count)?;
+ let cores: Vec<usize> = shard_assignments
+ .iter()
+ .flat_map(|shard| shard.cpu_set.iter().copied())
.collect();
- info!("Using {count} shards with affinity to cores
0..{count}");
+ info!("Using {count} shard(s) with affinity to cores
{cores:?}");
Ok(shard_assignments)
}
CpuAllocation::Range(start, end) => {
+ if !self.pin_cores {
+ info!(
+ "Using {} shard(s) for range {start}..{end}, unpinned",
+ end - start
+ );
+ return Ok(unpinned(end - start));
Review Comment:
`unpinned(end - start)` underflows when `start > end` - debug build panics
on the subtract, release wraps to a huge value and `unpinned()` then tries to
allocate a vec of ~`usize::MAX` `ShardInfo` and aborts. not reachable in
production today since both boot paths run `ShardingConfig::validate` first and
it rejects `start >= end`, but `to_shard_assignments` is a pub fn in an
internal crate and the server-ng `validate_sharding_runtime_knobs` net doesn't
cover cpu_allocation, so a direct/test/embedded caller hits it. cheap fix -
reject `start >= end` in this arm so the allocator self-guards instead of
trusting the caller.
##########
core/server/src/shard/system/stats.rs:
##########
@@ -88,6 +88,14 @@ impl IggyShard {
stats.written_bytes = disk_usage.total_written_bytes.into();
stats.threads_count = process.tasks().map(|t| t.len() as
u32).unwrap_or(0);
+
+ if let Some(limits) = process
+ .cgroup_limits()
+ .filter(|limits| limits.total_memory < sys.total_memory())
+ {
+ stats.total_memory = limits.total_memory.into();
+ stats.available_memory = limits.free_memory.into();
Review Comment:
under a memory-capped cgroup this sets `available_memory` to `memory.max -
memory.current` (or lower if a parent cgroup is tighter), and `memory.current`
charges reclaimable page cache as used. iggy is cache-heavy, so on a warm
instance `memory.current` climbs toward the cap and `available_memory` trends
toward ~0 even though the kernel reclaims that cache long before OOM - it
under-reports real headroom, which is the opposite of what a per-instance stats
endpoint wants. for an honest number you'd want a cgroup
MemAvailable-equivalent: `memory.max - (memory.current - reclaimable file
cache)`, where reclaimable is `inactive_file + active_file` from `memory.stat`
- sysinfo's `free_memory` doesn't do that, so it needs reading `memory.stat`
directly. also this override only fires when a memory limit is set (the `total
< host total` filter just above), so an instance confined by cpuset alone still
reports host-wide memory - multi-instance setups need a memory cap too.
re-scoping `total
_memory` to the cap is correct.
##########
core/configs/src/server_config/validators.rs:
##########
@@ -488,9 +489,27 @@ impl Validatable<ConfigurationError> for ShardingConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
- if *end > available_cpus {
+ if !self.pin_cores {
+ if *end - *start > available_cpus {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation
range {start}..{end} yields {} shards, exceeding available CPU cores
{available_cpus}",
+ *end - *start
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ return Ok(());
+ }
+ if *end - *start > available_cpus {
Review Comment:
this `*end - *start > available_cpus` check and its `eprintln!` are
identical to the one in the `!self.pin_cores` branch above. hoist it above the
`if !self.pin_cores` split so it runs once, and leave only the allowed-set
membership check gated behind pinning. the `start >= end` guard earlier still
protects the subtraction. saves ~7 lines.
##########
core/shard_allocator/src/lib.rs:
##########
@@ -287,39 +335,66 @@ impl ShardAllocator {
pub fn to_shard_assignments(&self) -> Result<Vec<ShardInfo>,
ShardingError> {
match &self.allocation {
CpuAllocation::All => {
+ // `available_parallelism` already accounts for both the
+ // affinity mask and any cgroup CPU quota, so a
+ // quota-restricted process gets proportionally fewer shards.
let available_cpus = available_parallelism()
.map_err(|err| ShardingError::Other {
msg: format!("Failed to get available_parallelism:
{:?}", err),
})?
.get();
- let shard_assignments: Vec<_> = (0..available_cpus)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
- .collect();
+ if !self.pin_cores {
+ info!("Using all available CPU cores ({available_cpus}
shards, unpinned)");
+ return Ok(unpinned(available_cpus));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments =
+ pinned_from_allowed(&allowed,
available_cpus.min(allowed.len()))?;
info!(
- "Using all available CPU cores ({} shards with affinity)",
- shard_assignments.len()
+ "Using all available CPU cores ({} shards pinned within
allowed set {:?})",
+ shard_assignments.len(),
+ allowed
);
Ok(shard_assignments)
}
CpuAllocation::Count(count) => {
- let shard_assignments = (0..*count)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
+ if !self.pin_cores {
+ info!("Using {count} shard(s), unpinned");
+ return Ok(unpinned(*count));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments = pinned_from_allowed(&allowed, *count)?;
+ let cores: Vec<usize> = shard_assignments
Review Comment:
this `cores` vec is only built to log it, and since `pinned_from_allowed`
takes `allowed.iter().take(count)` in order it's just `&allowed[..*count]`. log
the slice directly and drop the vec + `flat_map`.
##########
core/shard_allocator/src/lib.rs:
##########
@@ -287,39 +335,66 @@ impl ShardAllocator {
pub fn to_shard_assignments(&self) -> Result<Vec<ShardInfo>,
ShardingError> {
match &self.allocation {
CpuAllocation::All => {
+ // `available_parallelism` already accounts for both the
+ // affinity mask and any cgroup CPU quota, so a
+ // quota-restricted process gets proportionally fewer shards.
let available_cpus = available_parallelism()
.map_err(|err| ShardingError::Other {
msg: format!("Failed to get available_parallelism:
{:?}", err),
})?
.get();
- let shard_assignments: Vec<_> = (0..available_cpus)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
- .collect();
+ if !self.pin_cores {
+ info!("Using all available CPU cores ({available_cpus}
shards, unpinned)");
+ return Ok(unpinned(available_cpus));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments =
+ pinned_from_allowed(&allowed,
available_cpus.min(allowed.len()))?;
info!(
- "Using all available CPU cores ({} shards with affinity)",
- shard_assignments.len()
+ "Using all available CPU cores ({} shards pinned within
allowed set {:?})",
+ shard_assignments.len(),
+ allowed
);
Ok(shard_assignments)
}
CpuAllocation::Count(count) => {
- let shard_assignments = (0..*count)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
+ if !self.pin_cores {
+ info!("Using {count} shard(s), unpinned");
+ return Ok(unpinned(*count));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments = pinned_from_allowed(&allowed, *count)?;
+ let cores: Vec<usize> = shard_assignments
+ .iter()
+ .flat_map(|shard| shard.cpu_set.iter().copied())
.collect();
- info!("Using {count} shards with affinity to cores
0..{count}");
+ info!("Using {count} shard(s) with affinity to cores
{cores:?}");
Ok(shard_assignments)
}
CpuAllocation::Range(start, end) => {
+ if !self.pin_cores {
+ info!(
+ "Using {} shard(s) for range {start}..{end}, unpinned",
+ end - start
+ );
+ return Ok(unpinned(end - start));
+ }
+
+ // An explicit range is operator intent for exact cores; honor
Review Comment:
the comment says the range is honored as-is, but the validator already caps
a pinned range by `available_parallelism()` before this runs, so a range wider
than the quota never reaches here. minor, but the comment overstates what this
arm actually sees.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
