mirror of
https://github.com/ceph/ceph
synced 2026-08-02 07:03:18 +00:00
Merge 22f59bc2e0 into eed82023c5
This commit is contained in:
Kyle Bader
GitHub
commit
d60a300489
@ -7,7 +7,15 @@
|
||||
* RADOS: PG autoscaler allows overlapping roots. Each root receives a PG target based
|
||||
on its OSDs, with OSDs shared across multiple roots contributing proportionally
|
||||
less to each root's allocation.
|
||||
|
||||
|
||||
* RADOS: A new experimental OSD op queue, ``bfq``, is available via
|
||||
``osd_op_queue = bfq``. Modeled on the Linux BFQ elevator as exposed through
|
||||
the cgroups v2 io controller, it shares OSD throughput purely proportionally
|
||||
between a background group and client workload-type streams (block, object,
|
||||
file -- classified by pool application metadata), using weighted budget fair
|
||||
queueing with no device capacity estimate. Weights and budgets are tunable at
|
||||
runtime via the ``osd_bfq_*`` options.
|
||||
|
||||
* CephFS: CephFS snapshots metadata is now mutable. It is now possible to add,
|
||||
update and remove existing key-value pairs that are part of a snapshot
|
||||
metadata via libcephfs API ceph_do_snap_md_op().
|
||||
|
||||
123
doc/dev/osd_internals/bfq_scheduler.rst
Normal file
123
doc/dev/osd_internals/bfq_scheduler.rst
Normal file
@ -0,0 +1,123 @@
|
||||
=================
|
||||
BFQ Op Scheduler
|
||||
=================
|
||||
|
||||
The ``bfq`` op queue (``osd_op_queue = bfq``, experimental) is modeled on
|
||||
the Linux BFQ (Budget Fair Queueing) I/O scheduler as exposed through the
|
||||
cgroups v2 io controller. It is an alternative to ``mclock_scheduler``
|
||||
and ``wpq`` built around a different first principle: **purely
|
||||
proportional sharing**. Where mclock's reservation and limit tags are
|
||||
denominated in absolute IOPS -- which is why it needs the boot-time OSD
|
||||
bench and the ``osd_mclock_max_capacity_iops_*`` estimates -- bfq
|
||||
distributes whatever throughput the OSD actually delivers according to
|
||||
weights, and needs no estimate of device capacity at all.
|
||||
|
||||
Hierarchy
|
||||
=========
|
||||
|
||||
Service (in scaled bytes) is shared by a two-level B-WF2Q+ hierarchy of
|
||||
weighted groups, in the spirit of ``io.bfq.weight``::
|
||||
|
||||
root
|
||||
|-- client group (osd_bfq_client_group_weight)
|
||||
| |-- block (rbd pools) (osd_bfq_client_block_weight)
|
||||
| |-- object (rgw pools) (osd_bfq_client_object_weight)
|
||||
| |-- file (cephfs pools) (osd_bfq_client_file_weight)
|
||||
| `-- other (osd_bfq_client_other_weight)
|
||||
`-- background group (osd_bfq_background_group_weight)
|
||||
|-- recovery (osd_bfq_background_recovery_weight)
|
||||
`-- best_effort (osd_bfq_background_best_effort_weight)
|
||||
|
||||
Client ops are classified into workload streams by the application
|
||||
metadata of their pool (``ceph osd pool application enable``): pools
|
||||
tagged ``rbd`` map to the block stream, ``rgw`` to object, ``cephfs`` to
|
||||
file; untagged or ambiguously tagged pools fall into the other stream.
|
||||
The per-shard pool map refreshes on every OSDMap the shard consumes, so
|
||||
tag changes take effect without a restart. Background ops are split by
|
||||
their existing scheduler class (``background_recovery`` vs
|
||||
``background_best_effort``).
|
||||
|
||||
Weights follow the cgroups v2 convention: 1..1000, default 100, purely
|
||||
relative, adjustable at runtime.
|
||||
|
||||
Algorithm
|
||||
=========
|
||||
|
||||
Each backlogged leaf stream is granted a byte *budget*
|
||||
(``osd_bfq_max_budget`` caps it) and is served exclusively until the
|
||||
budget is exhausted, the stream empties, or the round exceeds
|
||||
``osd_bfq_budget_timeout``. Streams and groups are scheduled by WF2Q+:
|
||||
an entity is *eligible* once its virtual start tag is at or behind the
|
||||
tree's virtual time, and the eligible entity with the smallest virtual
|
||||
finish tag runs next. Finish tags are computed over the *assigned*
|
||||
budget at activation and **back-shifted** to the *consumed* service at
|
||||
expiration, so unused budget carries no penalty. The next budget adapts
|
||||
to observed demand: it doubles (up to the cap) when a round ends by
|
||||
exhaustion and shrinks toward actual usage when the stream empties.
|
||||
|
||||
Deliberate exclusions relative to the kernel implementation:
|
||||
|
||||
* no weight raising (the interactive/soft-realtime low-latency
|
||||
heuristics), and
|
||||
* no device idling/anticipation -- an emptied stream expires
|
||||
immediately, and ``dequeue()`` never asks the shard worker to wait.
|
||||
|
||||
Accounting model
|
||||
================
|
||||
|
||||
Service is charged at *dequeue* (dispatch to a shard worker), not at
|
||||
device completion; the kernel charges at completion and knows the device
|
||||
is exclusively occupied, while the OSD does not. Fairness is therefore
|
||||
over dispatched scaled bytes -- the same accounting model mclock already
|
||||
uses -- and budget exclusivity means exclusive *selection*, not
|
||||
exclusive device occupancy, when ``osd_op_num_threads_per_shard > 1``.
|
||||
Each item is charged ``max(item cost, osd_bfq_min_cost)`` bytes, the
|
||||
floor standing in for fixed per-IO overhead.
|
||||
|
||||
As with mclock, ops of the ``immediate`` class (peering and similar) and
|
||||
ops at or above ``osd_op_queue_cut_off`` bypass the hierarchy through a
|
||||
strict priority queue, and requeued ops (``enqueue_front``) re-enter via
|
||||
that queue rather than through the fair hierarchy.
|
||||
|
||||
Each OSD shard runs an independent scheduler instance, so weights
|
||||
describe per-shard shares; fairness across shards is statistical, via
|
||||
PG-to-shard hashing (again matching mclock).
|
||||
|
||||
Observability
|
||||
=============
|
||||
|
||||
``ceph daemon osd.N dump_op_pq_state`` dumps, per shard, the virtual
|
||||
times and tags of both tree levels, per-stream queue depths and next
|
||||
budgets, and the strict queue.
|
||||
|
||||
Microbenchmarks
|
||||
===============
|
||||
|
||||
``ceph_bench_op_scheduler`` (built from ``src/test/osd``) runs identical
|
||||
synthetic workloads through wpq, mclock, and bfq in-process, pacing
|
||||
dequeues at a simulated device rate so saturation dynamics are
|
||||
meaningful; mclock's capacity model is calibrated to the same rate. The
|
||||
scenarios: a backlogged victim vs an aggressor fanning out over 1..16
|
||||
client sessions (share isolation), a paced victim under a saturating
|
||||
aggressor (latency isolation), and client-vs-recovery (class weights).
|
||||
``unittest_scheduler_isolation`` asserts loose bounds on the bfq
|
||||
isolation properties; because the harness paces in wall time it skips
|
||||
itself unless ``CEPH_TEST_SCHEDULER_ISOLATION`` is set in the
|
||||
environment (shared CI executors are too noisy for it).
|
||||
|
||||
The headline result: with an rbd-pool victim and an rgw-pool aggressor,
|
||||
wpq and mclock dilute the victim's share as 1/(sessions+1) -- wpq
|
||||
round-robins owners, and mclock maps *all* client ops to a single
|
||||
dmclock client -- while bfq holds the configured weight ratio flat
|
||||
regardless of session count, and bounds a paced victim's p99 queueing
|
||||
delay by budget rotation rather than by the aggressor's backlog depth.
|
||||
|
||||
Follow-up candidates
|
||||
====================
|
||||
|
||||
* Auto-tuning ``osd_bfq_max_budget`` from the observed per-shard service
|
||||
rate (the analogue of the kernel's peak-rate estimator -- measured
|
||||
online, never configured).
|
||||
* Perf counters mirroring the mclock ones.
|
||||
* Completion-based charging, which would recover more of BFQ's
|
||||
device-time fairness but requires feedback from the op pipeline.
|
||||
5
qa/suites/rados/perf/scheduler/bfq_default_shards.yaml
Normal file
5
qa/suites/rados/perf/scheduler/bfq_default_shards.yaml
Normal file
@ -0,0 +1,5 @@
|
||||
overrides:
|
||||
ceph:
|
||||
conf:
|
||||
osd:
|
||||
osd op queue: bfq
|
||||
@ -1031,11 +1031,15 @@ options:
|
||||
desc: Do not store full-object checksums if the backend (bluestore) does its own
|
||||
checksums. Only usable with all BlueStore OSDs.
|
||||
default: false
|
||||
# Weighted Priority Queue (wpq), mClock Scheduler (mclock_scheduler: default)
|
||||
# or debug_random. "mclock_scheduler" is based on the mClock/dmClock
|
||||
# Weighted Priority Queue (wpq), mClock Scheduler (mclock_scheduler: default),
|
||||
# BFQ (bfq) or debug_random. "mclock_scheduler" is based on the mClock/dmClock
|
||||
# algorithm (Gulati, et al. 2010). "mclock_scheduler" prioritizes based on
|
||||
# the class the operation belongs to. "wpq" dequeues ops based on their
|
||||
# priorities. "debug_random" chooses among the two with equal probability.
|
||||
# priorities. "bfq" is modeled on the Linux BFQ elevator as exposed via the
|
||||
# cgroups v2 io controller: purely proportional weighted sharing between
|
||||
# background work and client workload types, with no device capacity
|
||||
# estimate. "debug_random" chooses between wpq and mclock_scheduler with
|
||||
# equal probability.
|
||||
# Note: PrioritzedQueue (prio) implementation is not used for scheduling ops
|
||||
# within OSDs and is therefore not listed.
|
||||
- name: osd_op_queue
|
||||
@ -1044,7 +1048,7 @@ options:
|
||||
desc: which operation priority queue algorithm to use
|
||||
long_desc: which operation priority queue algorithm to use
|
||||
fmt_desc: This sets the type of queue to be used for prioritizing ops
|
||||
within each OSD. Both queues feature a strict sub-queue which is
|
||||
within each OSD. All queues feature a strict sub-queue which is
|
||||
dequeued before the normal queue. The normal queue is different
|
||||
between implementations. The WeightedPriorityQueue (``wpq``)
|
||||
dequeues operations in relation to their priorities to prevent
|
||||
@ -1052,13 +1056,17 @@ options:
|
||||
are more overloaded than others. The mClockQueue
|
||||
(``mclock_scheduler``) prioritizes operations based on which class
|
||||
they belong to (recovery, scrub, snaptrim, client op, osd subop).
|
||||
See `QoS Based on mClock`_. Requires a restart.
|
||||
See `QoS Based on mClock`_. The bfq scheduler (``bfq``,
|
||||
experimental) provides weighted proportional sharing of OSD
|
||||
throughput between background work and client workload types,
|
||||
modeled on the Linux BFQ elevator. Requires a restart.
|
||||
default: mclock_scheduler
|
||||
see_also:
|
||||
- osd_op_queue_cut_off
|
||||
enum_values:
|
||||
- wpq
|
||||
- mclock_scheduler
|
||||
- bfq
|
||||
- debug_random
|
||||
with_legacy: true
|
||||
# Min priority to go to strict queue. (low, high)
|
||||
@ -1086,6 +1094,169 @@ options:
|
||||
- high
|
||||
- debug_random
|
||||
with_legacy: true
|
||||
# BFQ (Budget Fair Queueing) op scheduler options. Modeled on the Linux
|
||||
# BFQ elevator as exposed via the cgroups v2 io controller: OSD service
|
||||
# is shared between weighted groups purely proportionally, with no
|
||||
# estimate of the device's absolute capacity. Weights follow the
|
||||
# cgroups v2 io.bfq.weight convention (1..1000, default 100).
|
||||
- name: osd_bfq_client_group_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of the client op group for the bfq op queue
|
||||
fmt_desc: Weight of the client op group relative to the background op
|
||||
group when ``osd_op_queue`` is ``bfq``. Client ops receive a share of
|
||||
OSD throughput proportional to this weight whenever both groups are
|
||||
backlogged.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_background_group_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of the background op group for the bfq op queue
|
||||
fmt_desc: Weight of the background op group (recovery, scrub, snaptrim,
|
||||
PG deletion) relative to the client op group when ``osd_op_queue`` is
|
||||
``bfq``.
|
||||
default: 25
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_client_block_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of block (rbd pool) client ops within the bfq client group
|
||||
fmt_desc: Weight of ops from pools tagged with the ``rbd`` application,
|
||||
relative to the other client streams, when ``osd_op_queue`` is ``bfq``.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_client_object_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of object (rgw pool) client ops within the bfq client group
|
||||
fmt_desc: Weight of ops from pools tagged with the ``rgw`` application,
|
||||
relative to the other client streams, when ``osd_op_queue`` is ``bfq``.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_client_file_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of file (cephfs pool) client ops within the bfq client group
|
||||
fmt_desc: Weight of ops from pools tagged with the ``cephfs`` application,
|
||||
relative to the other client streams, when ``osd_op_queue`` is ``bfq``.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_client_other_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of client ops from untagged pools within the bfq client group
|
||||
fmt_desc: Weight of ops from pools without a recognized application tag
|
||||
(or with ambiguous tags), relative to the other client streams, when
|
||||
``osd_op_queue`` is ``bfq``.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_background_recovery_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of recovery ops within the bfq background group
|
||||
fmt_desc: Weight of recovery and backfill ops relative to best-effort
|
||||
background ops when ``osd_op_queue`` is ``bfq``.
|
||||
default: 100
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_background_best_effort_weight
|
||||
type: uint
|
||||
level: advanced
|
||||
desc: Weight of best-effort ops within the bfq background group
|
||||
fmt_desc: Weight of best-effort background ops (scrub, snaptrim, PG
|
||||
deletion) relative to recovery ops when ``osd_op_queue`` is ``bfq``.
|
||||
default: 50
|
||||
min: 1
|
||||
max: 1000
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_max_budget
|
||||
type: size
|
||||
level: advanced
|
||||
desc: Maximum service budget, in bytes, granted to a bfq stream per
|
||||
scheduling round
|
||||
fmt_desc: The maximum number of (scaled) bytes a stream may be served
|
||||
within one scheduling round when ``osd_op_queue`` is ``bfq``. Larger
|
||||
budgets favor throughput (longer exclusive rounds); smaller budgets
|
||||
favor latency (faster rotation between streams). Matches the spirit
|
||||
of the kernel BFQ default of 8 MiB.
|
||||
default: 8_M
|
||||
min: 4_K
|
||||
max: 4_G
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- osd_bfq_budget_timeout
|
||||
- name: osd_bfq_min_cost
|
||||
type: size
|
||||
level: advanced
|
||||
desc: Minimum scaled cost, in bytes, charged for any op by the bfq op
|
||||
queue
|
||||
fmt_desc: A floor on the per-op service charge accounting for fixed
|
||||
per-IO overhead when ``osd_op_queue`` is ``bfq``. Unlike mclock this
|
||||
is a plain configuration value, not derived from a device capacity
|
||||
estimate.
|
||||
default: 64_K
|
||||
min: 1
|
||||
max: 4_G
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- name: osd_bfq_budget_timeout
|
||||
type: millisecs
|
||||
level: advanced
|
||||
desc: Maximum wall-clock duration of a bfq scheduling round
|
||||
fmt_desc: A stream that has not consumed its budget within this time is
|
||||
rotated out (charged only for what it actually used) when
|
||||
``osd_op_queue`` is ``bfq``. Matches the kernel BFQ default of 125
|
||||
milliseconds. A very small timeout degenerates into per-op stream
|
||||
rotation, defeating budget batching.
|
||||
default: 125
|
||||
min: 1
|
||||
flags:
|
||||
- runtime
|
||||
see_also:
|
||||
- osd_op_queue
|
||||
- osd_bfq_max_budget
|
||||
- name: osd_mclock_scheduler_client_res
|
||||
type: float
|
||||
level: advanced
|
||||
|
||||
@ -33,6 +33,7 @@ set(osd_srcs
|
||||
Session.cc
|
||||
SnapMapper.cc
|
||||
osd_types.cc
|
||||
scheduler/BfqScheduler.cc
|
||||
scheduler/OpScheduler.cc
|
||||
scheduler/OpSchedulerItem.cc
|
||||
scheduler/mClockScheduler.cc
|
||||
|
||||
@ -295,7 +295,7 @@ uint64_t ECSubRead::cost(CephContext *cct, std::pair<int, int>& subchunk_info)
|
||||
* running with mClock scheduler will interpret this and set
|
||||
* the cost to 1 accordingly.
|
||||
*/
|
||||
if (cct->_conf->osd_op_queue != "mclock_scheduler") {
|
||||
if (!op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
return total_cost; // Legacy behavior for WPQ scheduler
|
||||
}
|
||||
|
||||
|
||||
@ -1729,7 +1729,7 @@ void OSDService::queue_recovery_context(
|
||||
epoch_t e = get_osdmap_epoch();
|
||||
|
||||
uint64_t cost_for_queue = [this, cost] {
|
||||
if (op_queue_type_t::mClockScheduler == osd->osd_op_queue_type()) {
|
||||
if (op_queue_type_uses_qos_cost(osd->osd_op_queue_type())) {
|
||||
return cost;
|
||||
} else {
|
||||
/* We retain this legacy behavior for WeightedPriorityQueue. It seems to
|
||||
@ -1756,7 +1756,7 @@ void OSDService::queue_for_snap_trim(PG *pg, uint64_t cost_per_object)
|
||||
{
|
||||
dout(10) << "queueing " << *pg << " for snaptrim" << dendl;
|
||||
uint64_t cost_for_queue = [this, cost_per_object] {
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* The cost calculation is valid for most snap trim iterations except
|
||||
* for the following cases:
|
||||
* 1) The penultimate iteration which may return 1 object to trim, in
|
||||
@ -1925,7 +1925,7 @@ void OSDService::queue_for_pg_delete(spg_t pgid, epoch_t e, int64_t num_objects)
|
||||
{
|
||||
dout(10) << __func__ << " on " << pgid << " e " << e << dendl;
|
||||
uint64_t cost_for_queue = [this, num_objects] {
|
||||
if (op_queue_type_t::mClockScheduler == osd->osd_op_queue_type()) {
|
||||
if (op_queue_type_uses_qos_cost(osd->osd_op_queue_type())) {
|
||||
return num_objects * cct->_conf->osd_pg_delete_cost;
|
||||
} else {
|
||||
return cct->_conf->osd_pg_delete_cost;
|
||||
@ -2088,7 +2088,7 @@ void OSDService::_queue_for_recovery(
|
||||
ceph_assert(ceph_mutex_is_locked_by_me(recovery_lock));
|
||||
|
||||
uint64_t cost_for_queue = [this, &reserved_pushes, &p] {
|
||||
if (op_queue_type_t::mClockScheduler == osd->osd_op_queue_type()) {
|
||||
if (op_queue_type_uses_qos_cost(osd->osd_op_queue_type())) {
|
||||
return p.cost_per_object * reserved_pushes;
|
||||
} else {
|
||||
/* We retain this legacy behavior for WeightedPriorityQueue. It seems to
|
||||
@ -10756,6 +10756,7 @@ void OSDShard::consume_map(
|
||||
dout(10) << new_osdmap->get_epoch()
|
||||
<< " (was " << (old_osdmap ? old_osdmap->get_epoch() : 0) << ")"
|
||||
<< dendl;
|
||||
scheduler->update_from_osdmap(*new_osdmap);
|
||||
int queued = 0;
|
||||
|
||||
// check slots
|
||||
|
||||
@ -1676,9 +1676,10 @@ public:
|
||||
|
||||
/// get message priority for recovery messages
|
||||
int get_recovery_op_priority() const {
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
/* For mclock, we use special priority values which will be
|
||||
* translated into op classes within PGRecoveryMsg::get_scheduler_class
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* For the QoS schedulers (mclock, bfq), we use special priority
|
||||
* values which will be translated into op classes within
|
||||
* PGRecoveryMsg::get_scheduler_class
|
||||
*/
|
||||
if (is_forced_recovery_or_backfill()) {
|
||||
return recovery_msg_priority_t::FORCED;
|
||||
|
||||
@ -14239,7 +14239,7 @@ uint64_t PrimaryLogPG::recover_backfill(
|
||||
spg_t(info.pgid.pgid, bt.shard),
|
||||
pbi.end, hobject_t());
|
||||
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* This guard preserves legacy WeightedPriorityQueue behavior for
|
||||
* now, but should be removed after Reef */
|
||||
m->set_priority(recovery_state.get_recovery_op_priority());
|
||||
@ -14425,7 +14425,7 @@ uint64_t PrimaryLogPG::recover_backfill(
|
||||
m = reqs[peer] = new MOSDPGBackfillRemove(
|
||||
spg_t(info.pgid.pgid, peer.shard),
|
||||
get_osdmap_epoch());
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* This guard preserves legacy WeightedPriorityQueue behavior for
|
||||
* now, but should be removed after Reef */
|
||||
m->set_priority(recovery_state.get_recovery_op_priority());
|
||||
@ -14515,7 +14515,7 @@ uint64_t PrimaryLogPG::recover_backfill(
|
||||
m->last_backfill = pinfo.last_backfill;
|
||||
m->stats = pinfo.stats;
|
||||
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* This guard preserves legacy WeightedPriorityQueue behavior for
|
||||
* now, but should be removed after Reef */
|
||||
m->set_priority(recovery_state.get_recovery_op_priority());
|
||||
|
||||
@ -7027,7 +7027,7 @@ ostream& operator<<(ostream& out, const PushReplyOp &op)
|
||||
|
||||
uint64_t PushReplyOp::cost(CephContext *cct) const
|
||||
{
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
/* In general, we really never want to throttle PushReplyOp messages.
|
||||
* As long as the object is smaller than osd_recovery_max_chunk (8M at
|
||||
* time of writing this comment, so this is basically always true),
|
||||
@ -7120,7 +7120,7 @@ ostream& operator<<(ostream& out, const PullOp &op)
|
||||
|
||||
uint64_t PullOp::cost(CephContext *cct) const
|
||||
{
|
||||
if (cct->_conf->osd_op_queue == "mclock_scheduler") {
|
||||
if (op_queue_type_uses_qos_cost(cct->_conf->osd_op_queue)) {
|
||||
return std::clamp<uint64_t>(
|
||||
recovery_progress.estimate_remaining_data_to_recover(recovery_info),
|
||||
1,
|
||||
@ -7731,6 +7731,8 @@ std::string_view get_op_queue_type_name(const op_queue_type_t &q)
|
||||
return "mclock_scheduler";
|
||||
case op_queue_type_t::PrioritizedQueue:
|
||||
return "PrioritizedQueue";
|
||||
case op_queue_type_t::BfqScheduler:
|
||||
return "bfq";
|
||||
default:
|
||||
return "unknown";
|
||||
}
|
||||
@ -7745,7 +7747,20 @@ std::optional<op_queue_type_t> get_op_queue_type_by_name(
|
||||
return op_queue_type_t::mClockScheduler;
|
||||
} else if (s == "PrioritizedQueue") {
|
||||
return op_queue_type_t::PrioritizedQueue;
|
||||
} else if (s == "bfq") {
|
||||
return op_queue_type_t::BfqScheduler;
|
||||
} else {
|
||||
return std::nullopt;
|
||||
}
|
||||
}
|
||||
|
||||
bool op_queue_type_uses_qos_cost(const op_queue_type_t &q)
|
||||
{
|
||||
return q == op_queue_type_t::mClockScheduler ||
|
||||
q == op_queue_type_t::BfqScheduler;
|
||||
}
|
||||
|
||||
bool op_queue_type_uses_qos_cost(const std::string_view &s)
|
||||
{
|
||||
return s == "mclock_scheduler" || s == "bfq";
|
||||
}
|
||||
|
||||
@ -7192,10 +7192,24 @@ using missing_map_t = std::map<hobject_t,
|
||||
enum class op_queue_type_t : uint8_t {
|
||||
WeightedPriorityQueue = 0,
|
||||
mClockScheduler,
|
||||
PrioritizedQueue
|
||||
PrioritizedQueue,
|
||||
BfqScheduler
|
||||
};
|
||||
std::string_view get_op_queue_type_name(const op_queue_type_t &q);
|
||||
std::optional<op_queue_type_t> get_op_queue_type_by_name(
|
||||
const std::string_view &s);
|
||||
|
||||
/**
|
||||
* True if the given op queue distributes QoS costs: byte-denominated
|
||||
* item costs and recovery priority values that encode a scheduler
|
||||
* class (see PGRecoveryMsg::get_scheduler_class). False for the
|
||||
* legacy WeightedPriorityQueue cost conventions.
|
||||
*
|
||||
* The string overload matches an osd_op_queue config value; note that
|
||||
* an unresolved "debug_random" intentionally reads as non-QoS, exactly
|
||||
* like the string comparisons it replaces.
|
||||
*/
|
||||
bool op_queue_type_uses_qos_cost(const op_queue_type_t &q);
|
||||
bool op_queue_type_uses_qos_cost(const std::string_view &s);
|
||||
|
||||
#endif
|
||||
|
||||
556
src/osd/scheduler/BfqScheduler.cc
Normal file
556
src/osd/scheduler/BfqScheduler.cc
Normal file
@ -0,0 +1,556 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
/*
|
||||
* Ceph - scalable distributed file system
|
||||
*
|
||||
* Copyright (C) 2026 IBM, Inc.
|
||||
*
|
||||
* This is free software; you can redistribute it and/or
|
||||
* modify it under the terms of the GNU Lesser General Public
|
||||
* License version 2.1, as published by the Free Software
|
||||
* Foundation. See file COPYING.
|
||||
*
|
||||
*/
|
||||
|
||||
#include <algorithm>
|
||||
|
||||
#include "osd/scheduler/BfqScheduler.h"
|
||||
|
||||
#include "common/debug.h"
|
||||
#include "osd/OSDMap.h"
|
||||
|
||||
#define dout_context cct
|
||||
#define dout_subsys ceph_subsys_osd
|
||||
#undef dout_prefix
|
||||
#define dout_prefix *_dout << "BfqScheduler(" << shard_id << "): "
|
||||
|
||||
namespace ceph::osd::scheduler {
|
||||
|
||||
std::string_view bfq_stream_name(bfq_stream_t s)
|
||||
{
|
||||
switch (s) {
|
||||
case bfq_stream_t::client_block:
|
||||
return "client_block";
|
||||
case bfq_stream_t::client_object:
|
||||
return "client_object";
|
||||
case bfq_stream_t::client_file:
|
||||
return "client_file";
|
||||
case bfq_stream_t::client_other:
|
||||
return "client_other";
|
||||
case bfq_stream_t::background_recovery:
|
||||
return "background_recovery";
|
||||
case bfq_stream_t::background_best_effort:
|
||||
return "background_best_effort";
|
||||
default:
|
||||
return "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
std::string_view bfq_group_name(bfq_group_t g)
|
||||
{
|
||||
switch (g) {
|
||||
case bfq_group_t::client:
|
||||
return "client";
|
||||
case bfq_group_t::background:
|
||||
return "background";
|
||||
default:
|
||||
return "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
namespace bfq_detail {
|
||||
|
||||
void BfqServiceTree::activate(unsigned idx, uint64_t allotted,
|
||||
uint32_t weight)
|
||||
{
|
||||
BfqEntity &e = entities[idx];
|
||||
ceph_assert(!e.active);
|
||||
ceph_assert(weight > 0);
|
||||
e.weight = weight;
|
||||
e.start = std::max(vtime, e.finish);
|
||||
e.finish = e.start + static_cast<vtime_t>(allotted) / e.weight;
|
||||
e.allotted = allotted;
|
||||
e.active = true;
|
||||
total_weight += e.weight;
|
||||
}
|
||||
|
||||
void BfqServiceTree::charge(uint64_t served)
|
||||
{
|
||||
if (total_weight > 0) {
|
||||
vtime += static_cast<vtime_t>(served) / total_weight;
|
||||
}
|
||||
}
|
||||
|
||||
void BfqServiceTree::expire(unsigned idx, uint64_t served,
|
||||
uint64_t next_allotted, uint32_t next_weight,
|
||||
bool still_backlogged)
|
||||
{
|
||||
BfqEntity &e = entities[idx];
|
||||
ceph_assert(e.active);
|
||||
ceph_assert(next_weight > 0);
|
||||
// back-shift: charge the entity for the service it actually
|
||||
// received, not the estimate it was tagged with at activation
|
||||
e.finish = e.start + static_cast<vtime_t>(served) / e.weight;
|
||||
if (still_backlogged) {
|
||||
e.start = e.finish;
|
||||
total_weight -= e.weight;
|
||||
e.weight = next_weight;
|
||||
total_weight += e.weight;
|
||||
e.finish = e.start + static_cast<vtime_t>(next_allotted) / e.weight;
|
||||
e.allotted = next_allotted;
|
||||
} else {
|
||||
e.active = false;
|
||||
total_weight -= e.weight;
|
||||
if (total_weight == 0) {
|
||||
// fully idle: the ideal fluid server has no backlog, so relative
|
||||
// finish memories no longer matter; renormalize to keep the
|
||||
// virtual time magnitude (and double precision) bounded
|
||||
vtime = 0.0;
|
||||
for (auto &ent : entities) {
|
||||
ent.start = 0.0;
|
||||
ent.finish = 0.0;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::optional<unsigned> BfqServiceTree::select()
|
||||
{
|
||||
int best = -1;
|
||||
bool any_active = false;
|
||||
vtime_t min_start = 0.0;
|
||||
for (unsigned i = 0; i < entities.size(); ++i) {
|
||||
const BfqEntity &e = entities[i];
|
||||
if (!e.active) {
|
||||
continue;
|
||||
}
|
||||
if (!any_active || e.start < min_start) {
|
||||
min_start = e.start;
|
||||
any_active = true;
|
||||
}
|
||||
if (e.start <= vtime &&
|
||||
(best < 0 || e.finish < entities[best].finish)) {
|
||||
best = static_cast<int>(i);
|
||||
}
|
||||
}
|
||||
if (best < 0 && any_active) {
|
||||
// active entities exist but none is eligible: the ideal server
|
||||
// idled while V lagged; jump V forward to the earliest pending
|
||||
// start (WF2Q+ virtual time rule) and pick again
|
||||
vtime = std::max(vtime, min_start);
|
||||
for (unsigned i = 0; i < entities.size(); ++i) {
|
||||
const BfqEntity &e = entities[i];
|
||||
if (e.active && e.start <= vtime &&
|
||||
(best < 0 || e.finish < entities[best].finish)) {
|
||||
best = static_cast<int>(i);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (best < 0) {
|
||||
return std::nullopt;
|
||||
}
|
||||
return static_cast<unsigned>(best);
|
||||
}
|
||||
|
||||
void BfqServiceTree::dump(ceph::Formatter &f) const
|
||||
{
|
||||
f.dump_float("vtime", vtime);
|
||||
f.dump_unsigned("total_active_weight", total_weight);
|
||||
f.open_array_section("entities");
|
||||
for (const auto &e : entities) {
|
||||
f.open_object_section("entity");
|
||||
f.dump_bool("active", e.active);
|
||||
f.dump_unsigned("weight", e.weight);
|
||||
f.dump_float("vstart", e.start);
|
||||
f.dump_float("vfinish", e.finish);
|
||||
f.dump_unsigned("allotted", e.allotted);
|
||||
f.close_section();
|
||||
}
|
||||
f.close_section();
|
||||
}
|
||||
|
||||
} // namespace bfq_detail
|
||||
|
||||
BfqScheduler::BfqScheduler(CephContext *cct, int whoami, uint32_t num_shards,
|
||||
int shard_id, bool is_rotational,
|
||||
unsigned cutoff_priority)
|
||||
: cct(cct),
|
||||
cutoff_priority(cutoff_priority),
|
||||
shard_id(shard_id),
|
||||
root_tree(bfq_num_groups),
|
||||
group_trees{bfq_detail::BfqServiceTree(bfq_num_client_streams),
|
||||
bfq_detail::BfqServiceTree(bfq_num_background_streams)}
|
||||
{
|
||||
ceph_assert(num_shards > 0);
|
||||
refresh_config();
|
||||
for (auto &s : streams) {
|
||||
s.next_budget = std::max(min_budget, max_budget / 2);
|
||||
}
|
||||
cct->_conf.add_observer(this);
|
||||
}
|
||||
|
||||
BfqScheduler::~BfqScheduler()
|
||||
{
|
||||
cct->_conf.remove_observer(this);
|
||||
}
|
||||
|
||||
void BfqScheduler::refresh_config()
|
||||
{
|
||||
const auto &conf = cct->_conf;
|
||||
max_budget = conf.get_val<Option::size_t>("osd_bfq_max_budget");
|
||||
min_cost = conf.get_val<Option::size_t>("osd_bfq_min_cost");
|
||||
// floor budgets so a round always covers at least one scaled item
|
||||
// and rounds cannot degenerate into per-item switching
|
||||
min_budget = std::max<uint64_t>(
|
||||
{1, min_cost, max_budget / 32});
|
||||
max_budget = std::max(max_budget, min_budget);
|
||||
budget_timeout =
|
||||
conf.get_val<std::chrono::milliseconds>("osd_bfq_budget_timeout");
|
||||
|
||||
group_weights[idx(bfq_group_t::client)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_client_group_weight");
|
||||
group_weights[idx(bfq_group_t::background)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_background_group_weight");
|
||||
stream_weights[idx(bfq_stream_t::client_block)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_client_block_weight");
|
||||
stream_weights[idx(bfq_stream_t::client_object)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_client_object_weight");
|
||||
stream_weights[idx(bfq_stream_t::client_file)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_client_file_weight");
|
||||
stream_weights[idx(bfq_stream_t::client_other)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_client_other_weight");
|
||||
stream_weights[idx(bfq_stream_t::background_recovery)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_background_recovery_weight");
|
||||
stream_weights[idx(bfq_stream_t::background_best_effort)] =
|
||||
conf.get_val<uint64_t>("osd_bfq_background_best_effort_weight");
|
||||
|
||||
dout(10) << __func__ << " max_budget " << max_budget
|
||||
<< " min_budget " << min_budget
|
||||
<< " min_cost " << min_cost
|
||||
<< " budget_timeout " << budget_timeout.count() << "ms"
|
||||
<< dendl;
|
||||
}
|
||||
|
||||
std::vector<std::string> BfqScheduler::get_tracked_keys() const noexcept
|
||||
{
|
||||
using namespace std::literals;
|
||||
return {
|
||||
"osd_bfq_client_group_weight"s,
|
||||
"osd_bfq_background_group_weight"s,
|
||||
"osd_bfq_client_block_weight"s,
|
||||
"osd_bfq_client_object_weight"s,
|
||||
"osd_bfq_client_file_weight"s,
|
||||
"osd_bfq_client_other_weight"s,
|
||||
"osd_bfq_background_recovery_weight"s,
|
||||
"osd_bfq_background_best_effort_weight"s,
|
||||
"osd_bfq_max_budget"s,
|
||||
"osd_bfq_min_cost"s,
|
||||
"osd_bfq_budget_timeout"s
|
||||
};
|
||||
}
|
||||
|
||||
void BfqScheduler::handle_conf_change(const ConfigProxy &conf,
|
||||
const std::set<std::string> &changed)
|
||||
{
|
||||
// runs on the config thread; fold the change in under the shard
|
||||
// lock on the next enqueue/dequeue. Weights of currently active
|
||||
// entities take effect at their next (re)activation.
|
||||
config_dirty = true;
|
||||
}
|
||||
|
||||
uint64_t BfqScheduler::calc_scaled_cost(int item_cost) const
|
||||
{
|
||||
// a pure config floor covering per-IO overhead; unlike mclock this
|
||||
// is not derived from a device capacity estimate
|
||||
return std::max(static_cast<uint64_t>(std::max(1, item_cost)), min_cost);
|
||||
}
|
||||
|
||||
bfq_stream_t BfqScheduler::classify(const OpSchedulerItem &item) const
|
||||
{
|
||||
switch (item.get_scheduler_class()) {
|
||||
case SchedulerClass::background_recovery:
|
||||
return bfq_stream_t::background_recovery;
|
||||
case SchedulerClass::background_best_effort:
|
||||
return bfq_stream_t::background_best_effort;
|
||||
default:
|
||||
// immediate never reaches the fair hierarchy (handled in enqueue)
|
||||
ceph_assert(item.get_scheduler_class() == SchedulerClass::client);
|
||||
[[fallthrough]];
|
||||
case SchedulerClass::client:
|
||||
if (auto p = pool_streams.find(item.get_ordering_token().pool());
|
||||
p != pool_streams.end()) {
|
||||
return p->second;
|
||||
}
|
||||
return bfq_stream_t::client_other;
|
||||
}
|
||||
}
|
||||
|
||||
void BfqScheduler::update_from_osdmap(const OSDMap &map)
|
||||
{
|
||||
std::unordered_map<int64_t, bfq_stream_t> next;
|
||||
for (const auto &[id, pool] : map.get_pools()) {
|
||||
std::optional<bfq_stream_t> stream;
|
||||
bool ambiguous = false;
|
||||
for (const auto &[app, app_md] : pool.application_metadata) {
|
||||
std::optional<bfq_stream_t> tagged;
|
||||
if (app == pg_pool_t::APPLICATION_NAME_RBD) {
|
||||
tagged = bfq_stream_t::client_block;
|
||||
} else if (app == pg_pool_t::APPLICATION_NAME_RGW) {
|
||||
tagged = bfq_stream_t::client_object;
|
||||
} else if (app == pg_pool_t::APPLICATION_NAME_CEPHFS) {
|
||||
tagged = bfq_stream_t::client_file;
|
||||
}
|
||||
if (tagged) {
|
||||
if (stream && *stream != *tagged) {
|
||||
ambiguous = true;
|
||||
}
|
||||
stream = tagged;
|
||||
}
|
||||
}
|
||||
if (stream && !ambiguous) {
|
||||
next[id] = *stream;
|
||||
}
|
||||
// untagged and ambiguously tagged pools fall to client_other via
|
||||
// lookup miss
|
||||
}
|
||||
pool_streams = std::move(next);
|
||||
dout(20) << __func__ << " classified " << pool_streams.size()
|
||||
<< " of " << map.get_pools().size() << " pools" << dendl;
|
||||
}
|
||||
|
||||
void BfqScheduler::set_pool_streams(
|
||||
std::unordered_map<int64_t, bfq_stream_t> &&map)
|
||||
{
|
||||
pool_streams = std::move(map);
|
||||
}
|
||||
|
||||
void BfqScheduler::activate_stream(bfq_stream_t s)
|
||||
{
|
||||
const auto g = bfq_group_of(s);
|
||||
auto &group_tree = group_trees[idx(g)];
|
||||
const bool group_was_idle = !group_tree.has_active();
|
||||
auto &stream = streams[idx(s)];
|
||||
stream.next_budget = std::clamp(stream.next_budget, min_budget, max_budget);
|
||||
group_tree.activate(bfq_index_in_group(s), stream.next_budget,
|
||||
stream_weights[idx(s)]);
|
||||
if (group_was_idle) {
|
||||
// the group's expected service is one full budget; the estimate
|
||||
// is corrected to actual service when the round expires
|
||||
root_tree.activate(idx(g), max_budget, group_weights[idx(g)]);
|
||||
}
|
||||
}
|
||||
|
||||
void BfqScheduler::begin_round(bfq_stream_t s)
|
||||
{
|
||||
auto &stream = streams[idx(s)];
|
||||
in_service = s;
|
||||
stream.budget_remaining = static_cast<int64_t>(stream.next_budget);
|
||||
stream.served_round = 0;
|
||||
group_served[idx(bfq_group_of(s))] = 0;
|
||||
round_start = ceph::coarse_mono_clock::now();
|
||||
dout(20) << __func__ << " " << bfq_stream_name(s)
|
||||
<< " budget " << stream.next_budget << dendl;
|
||||
}
|
||||
|
||||
void BfqScheduler::end_round(expire_reason reason)
|
||||
{
|
||||
const bfq_stream_t s = *in_service;
|
||||
auto &stream = streams[idx(s)];
|
||||
const auto g = bfq_group_of(s);
|
||||
auto &group_tree = group_trees[idx(g)];
|
||||
const uint64_t served = stream.served_round;
|
||||
|
||||
// budget feedback (simplified from the kernel): grow when the
|
||||
// stream exhausted its budget while still backlogged, adapt down to
|
||||
// observed demand when it emptied, pin to actual progress on
|
||||
// timeout
|
||||
switch (reason) {
|
||||
case expire_reason::exhausted:
|
||||
stream.next_budget = std::min(max_budget, stream.next_budget * 2);
|
||||
break;
|
||||
case expire_reason::emptied:
|
||||
case expire_reason::timed_out:
|
||||
stream.next_budget = std::clamp(served, min_budget, max_budget);
|
||||
break;
|
||||
}
|
||||
|
||||
const bool backlogged = !stream.items.empty();
|
||||
group_tree.expire(bfq_index_in_group(s), served, stream.next_budget,
|
||||
stream_weights[idx(s)], backlogged);
|
||||
|
||||
root_tree.expire(idx(g), group_served[idx(g)], max_budget,
|
||||
group_weights[idx(g)], group_tree.has_active());
|
||||
group_served[idx(g)] = 0;
|
||||
in_service.reset();
|
||||
|
||||
dout(20) << __func__ << " " << bfq_stream_name(s)
|
||||
<< " reason " << static_cast<int>(reason)
|
||||
<< " served " << served
|
||||
<< " next_budget " << stream.next_budget
|
||||
<< dendl;
|
||||
}
|
||||
|
||||
void BfqScheduler::enqueue_high(unsigned priority, OpSchedulerItem &&item,
|
||||
bool front)
|
||||
{
|
||||
if (front) {
|
||||
high_priority[priority].push_back(std::move(item));
|
||||
} else {
|
||||
high_priority[priority].push_front(std::move(item));
|
||||
}
|
||||
}
|
||||
|
||||
void BfqScheduler::enqueue(OpSchedulerItem &&item)
|
||||
{
|
||||
maybe_refresh_config();
|
||||
const unsigned priority = item.get_priority();
|
||||
|
||||
if (SchedulerClass::immediate == item.get_scheduler_class()) {
|
||||
enqueue_high(immediate_class_priority, std::move(item));
|
||||
} else if (priority >= cutoff_priority) {
|
||||
enqueue_high(priority, std::move(item));
|
||||
} else {
|
||||
const bfq_stream_t s = classify(item);
|
||||
const uint64_t scaled = calc_scaled_cost(item.get_cost());
|
||||
item.set_qos_cost(static_cast<uint32_t>(
|
||||
std::min<uint64_t>(scaled, std::numeric_limits<uint32_t>::max())));
|
||||
dout(20) << __func__ << " " << bfq_stream_name(s)
|
||||
<< " cost " << item.get_cost()
|
||||
<< " scaled_cost " << scaled
|
||||
<< dendl;
|
||||
auto &stream = streams[idx(s)];
|
||||
const bool was_idle = stream.items.empty();
|
||||
stream.items.push_back(std::move(item));
|
||||
++fair_queued;
|
||||
if (was_idle && in_service != s) {
|
||||
// an empty stream that is not in service is never on its tree;
|
||||
// an in-service stream keeps its round until it expires
|
||||
activate_stream(s);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void BfqScheduler::enqueue_front(OpSchedulerItem &&item)
|
||||
{
|
||||
maybe_refresh_config();
|
||||
const unsigned priority = item.get_priority();
|
||||
if (SchedulerClass::immediate == item.get_scheduler_class()) {
|
||||
enqueue_high(immediate_class_priority, std::move(item), true);
|
||||
} else if (priority >= cutoff_priority) {
|
||||
enqueue_high(priority, std::move(item), true);
|
||||
} else {
|
||||
// like mclock: the fair hierarchy cannot insert at the front, so
|
||||
// requeued items bypass it via the strict queue at priority 0
|
||||
enqueue_high(0, std::move(item), true);
|
||||
}
|
||||
}
|
||||
|
||||
WorkItem BfqScheduler::dequeue()
|
||||
{
|
||||
maybe_refresh_config();
|
||||
if (!high_priority.empty()) {
|
||||
auto iter = high_priority.begin();
|
||||
// invariant: high_priority entries are never empty
|
||||
ceph_assert(!iter->second.empty());
|
||||
WorkItem ret{std::move(iter->second.back())};
|
||||
iter->second.pop_back();
|
||||
if (iter->second.empty()) {
|
||||
high_priority.erase(iter);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
ceph_assert(fair_queued > 0 &&
|
||||
"Impossible, must have checked empty() first");
|
||||
return dequeue_fair();
|
||||
}
|
||||
|
||||
OpSchedulerItem BfqScheduler::dequeue_fair()
|
||||
{
|
||||
if (in_service) {
|
||||
auto &stream = streams[idx(*in_service)];
|
||||
if (stream.items.empty()) {
|
||||
end_round(expire_reason::emptied);
|
||||
} else if (stream.budget_remaining <= 0) {
|
||||
end_round(expire_reason::exhausted);
|
||||
} else if (ceph::coarse_mono_clock::now() - round_start >
|
||||
budget_timeout) {
|
||||
end_round(expire_reason::timed_out);
|
||||
}
|
||||
}
|
||||
if (!in_service) {
|
||||
// two-level B-WF2Q+ selection: group, then leaf within the group.
|
||||
// Invariant: every non-empty stream is on its group tree (or in
|
||||
// service) and every group with active leaves is on the root
|
||||
// tree, so fair_queued > 0 guarantees both selections succeed.
|
||||
auto gsel = root_tree.select();
|
||||
ceph_assert(gsel);
|
||||
auto lsel = group_trees[*gsel].select();
|
||||
ceph_assert(lsel);
|
||||
begin_round(bfq_stream_at(*gsel, *lsel));
|
||||
}
|
||||
|
||||
const bfq_stream_t s = *in_service;
|
||||
auto &stream = streams[idx(s)];
|
||||
ceph_assert(!stream.items.empty());
|
||||
OpSchedulerItem item = std::move(stream.items.front());
|
||||
stream.items.pop_front();
|
||||
--fair_queued;
|
||||
|
||||
const uint64_t served = calc_scaled_cost(item.get_cost());
|
||||
stream.budget_remaining -= static_cast<int64_t>(served);
|
||||
stream.served_round += served;
|
||||
group_served[idx(bfq_group_of(s))] += served;
|
||||
group_trees[idx(bfq_group_of(s))].charge(served);
|
||||
root_tree.charge(served);
|
||||
return item;
|
||||
}
|
||||
|
||||
void BfqScheduler::dump(ceph::Formatter &f) const
|
||||
{
|
||||
f.open_object_section("queue_sizes");
|
||||
f.dump_unsigned("high_priority_queue", high_priority.size());
|
||||
f.dump_unsigned("fair_queued", fair_queued);
|
||||
f.close_section();
|
||||
|
||||
f.open_object_section("bfq");
|
||||
f.dump_string("in_service",
|
||||
in_service ? std::string(bfq_stream_name(*in_service))
|
||||
: "none");
|
||||
f.open_object_section("root_tree");
|
||||
root_tree.dump(f);
|
||||
f.close_section();
|
||||
f.open_array_section("groups");
|
||||
for (size_t g = 0; g < bfq_num_groups; ++g) {
|
||||
f.open_object_section("group");
|
||||
f.dump_string("name", std::string(
|
||||
bfq_group_name(static_cast<bfq_group_t>(g))));
|
||||
f.dump_unsigned("weight", group_weights[g]);
|
||||
f.open_object_section("tree");
|
||||
group_trees[g].dump(f);
|
||||
f.close_section();
|
||||
f.close_section();
|
||||
}
|
||||
f.close_section();
|
||||
f.open_array_section("streams");
|
||||
for (size_t s = 0; s < bfq_num_streams; ++s) {
|
||||
f.open_object_section("stream");
|
||||
f.dump_string("name", std::string(
|
||||
bfq_stream_name(static_cast<bfq_stream_t>(s))));
|
||||
f.dump_unsigned("weight", stream_weights[s]);
|
||||
f.dump_unsigned("queue_size", streams[s].items.size());
|
||||
f.dump_unsigned("next_budget", streams[s].next_budget);
|
||||
f.close_section();
|
||||
}
|
||||
f.close_section();
|
||||
f.close_section();
|
||||
|
||||
f.open_object_section("HighPriorityQueue");
|
||||
for (auto it = high_priority.begin(); it != high_priority.end(); it++) {
|
||||
f.dump_int("priority", it->first);
|
||||
f.dump_int("queue_size", it->second.size());
|
||||
}
|
||||
f.close_section();
|
||||
}
|
||||
|
||||
}
|
||||
312
src/osd/scheduler/BfqScheduler.h
Normal file
312
src/osd/scheduler/BfqScheduler.h
Normal file
@ -0,0 +1,312 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
/*
|
||||
* Ceph - scalable distributed file system
|
||||
*
|
||||
* Copyright (C) 2026 IBM, Inc.
|
||||
*
|
||||
* This is free software; you can redistribute it and/or
|
||||
* modify it under the terms of the GNU Lesser General Public
|
||||
* License version 2.1, as published by the Free Software
|
||||
* Foundation. See file COPYING.
|
||||
*
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <array>
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <deque>
|
||||
#include <limits>
|
||||
#include <list>
|
||||
#include <map>
|
||||
#include <optional>
|
||||
#include <ostream>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
#include "common/ceph_context.h"
|
||||
#include "common/ceph_time.h"
|
||||
#include "common/config.h"
|
||||
#include "osd/scheduler/OpScheduler.h"
|
||||
#include "osd/scheduler/OpSchedulerItem.h"
|
||||
|
||||
class OSDMap;
|
||||
|
||||
namespace ceph::osd::scheduler {
|
||||
|
||||
/**
|
||||
* bfq_stream_t
|
||||
*
|
||||
* The leaf queues of the bfq hierarchy. Client ops are subdivided by
|
||||
* the workload type of their pool (derived from the pool application
|
||||
* metadata); background ops by their scheduler class.
|
||||
*/
|
||||
enum class bfq_stream_t : uint8_t {
|
||||
client_block = 0, ///< pools tagged with the rbd application
|
||||
client_object, ///< pools tagged with the rgw application
|
||||
client_file, ///< pools tagged with the cephfs application
|
||||
client_other, ///< untagged or ambiguously tagged pools
|
||||
background_recovery, ///< SchedulerClass::background_recovery
|
||||
background_best_effort, ///< SchedulerClass::background_best_effort
|
||||
stream_count
|
||||
};
|
||||
|
||||
enum class bfq_group_t : uint8_t {
|
||||
client = 0,
|
||||
background,
|
||||
group_count
|
||||
};
|
||||
|
||||
inline constexpr size_t bfq_num_streams =
|
||||
static_cast<size_t>(bfq_stream_t::stream_count);
|
||||
inline constexpr size_t bfq_num_groups =
|
||||
static_cast<size_t>(bfq_group_t::group_count);
|
||||
inline constexpr size_t bfq_num_client_streams = 4;
|
||||
inline constexpr size_t bfq_num_background_streams = 2;
|
||||
|
||||
constexpr bfq_group_t bfq_group_of(bfq_stream_t s) {
|
||||
return s < bfq_stream_t::background_recovery ?
|
||||
bfq_group_t::client : bfq_group_t::background;
|
||||
}
|
||||
|
||||
/// index of a stream's entity within its group's service tree
|
||||
constexpr unsigned bfq_index_in_group(bfq_stream_t s) {
|
||||
return bfq_group_of(s) == bfq_group_t::client ?
|
||||
static_cast<unsigned>(s) :
|
||||
static_cast<unsigned>(s) -
|
||||
static_cast<unsigned>(bfq_stream_t::background_recovery);
|
||||
}
|
||||
|
||||
constexpr bfq_stream_t bfq_stream_at(unsigned group, unsigned index_in_group) {
|
||||
return static_cast<bfq_stream_t>(
|
||||
group == static_cast<unsigned>(bfq_group_t::client) ?
|
||||
index_in_group :
|
||||
static_cast<unsigned>(bfq_stream_t::background_recovery) +
|
||||
index_in_group);
|
||||
}
|
||||
|
||||
std::string_view bfq_stream_name(bfq_stream_t s);
|
||||
std::string_view bfq_group_name(bfq_group_t g);
|
||||
|
||||
namespace bfq_detail {
|
||||
|
||||
using vtime_t = double;
|
||||
|
||||
struct BfqEntity {
|
||||
uint32_t weight = 1; ///< share while active, fixed at (re)activation
|
||||
vtime_t start = 0.0; ///< S: virtual start of the current allotment
|
||||
vtime_t finish = 0.0; ///< F while active; finish memory while idle
|
||||
uint64_t allotted = 0; ///< service allotted at activation (scaled bytes)
|
||||
bool active = false;
|
||||
};
|
||||
|
||||
/**
|
||||
* BfqServiceTree
|
||||
*
|
||||
* WF2Q+ scheduling over a small fixed set of entities addressed by
|
||||
* index. The kernel maintains augmented rb-trees because a cgroup
|
||||
* hierarchy may hold thousands of entities; with at most a handful
|
||||
* per tree we scan linearly instead.
|
||||
*
|
||||
* Not thread safe: callers rely on the OSD shard lock.
|
||||
*/
|
||||
class BfqServiceTree {
|
||||
public:
|
||||
explicit BfqServiceTree(unsigned num_entities) : entities(num_entities) {}
|
||||
|
||||
/**
|
||||
* Make an idle entity backlogged. S = max(V, previous F) -- the
|
||||
* finish memory prevents a queue from banking service while idle --
|
||||
* and F = S + allotted/weight.
|
||||
*/
|
||||
void activate(unsigned idx, uint64_t allotted, uint32_t weight);
|
||||
|
||||
/// account served (scaled) bytes to the tree's virtual time
|
||||
void charge(uint64_t served);
|
||||
|
||||
/**
|
||||
* Close out an entity's allotment: back-shift F to the service it
|
||||
* actually received (so unused budget carries no penalty), then
|
||||
* either re-tag it for another allotment or remove it from the tree.
|
||||
*/
|
||||
void expire(unsigned idx, uint64_t served, uint64_t next_allotted,
|
||||
uint32_t next_weight, bool still_backlogged);
|
||||
|
||||
/**
|
||||
* WF2Q+ selection: the eligible (S <= V) active entity with the
|
||||
* minimum F. If active entities exist but none is eligible, V
|
||||
* jumps forward to the earliest pending S first.
|
||||
*/
|
||||
std::optional<unsigned> select();
|
||||
|
||||
bool has_active() const {
|
||||
return total_weight > 0;
|
||||
}
|
||||
const BfqEntity &entity(unsigned idx) const {
|
||||
return entities[idx];
|
||||
}
|
||||
vtime_t get_vtime() const {
|
||||
return vtime;
|
||||
}
|
||||
void dump(ceph::Formatter &f) const;
|
||||
|
||||
private:
|
||||
std::vector<BfqEntity> entities;
|
||||
vtime_t vtime = 0.0;
|
||||
uint64_t total_weight = 0; ///< sum of active entity weights
|
||||
};
|
||||
|
||||
} // namespace bfq_detail
|
||||
|
||||
/**
|
||||
* BfqScheduler
|
||||
*
|
||||
* OpScheduler implementation modeled on the Linux BFQ (Budget Fair
|
||||
* Queueing) I/O scheduler as exposed through the cgroups v2 io
|
||||
* controller. Service (scaled bytes) is distributed by a two-level
|
||||
* B-WF2Q+ hierarchy of weighted groups:
|
||||
*
|
||||
* root
|
||||
* |-- client group (osd_bfq_client_group_weight)
|
||||
* | |-- block (rbd pools) (osd_bfq_client_block_weight)
|
||||
* | |-- object (rgw pools) (osd_bfq_client_object_weight)
|
||||
* | |-- file (cephfs pools) (osd_bfq_client_file_weight)
|
||||
* | `-- other (osd_bfq_client_other_weight)
|
||||
* `-- background group (osd_bfq_background_group_weight)
|
||||
* |-- recovery (osd_bfq_background_recovery_weight)
|
||||
* `-- best_effort (osd_bfq_background_best_effort_weight)
|
||||
*
|
||||
* Unlike mclock, bfq is purely proportional-share: it needs no
|
||||
* estimate of the device's absolute capacity (IOPS or bandwidth).
|
||||
* Each backlogged leaf is granted a byte budget and served
|
||||
* exclusively until the budget is exhausted, its timeout elapses, or
|
||||
* it empties; virtual finish times computed over the assigned budget
|
||||
* are back-shifted to the consumed service on expiration, and the
|
||||
* next budget adapts to observed demand.
|
||||
*
|
||||
* Deliberate scope exclusions relative to the kernel: no weight
|
||||
* raising (low-latency heuristics), no device idling/anticipation.
|
||||
* Service is charged at dequeue (dispatch to a shard worker), not at
|
||||
* device completion, so fairness is over dispatched scaled bytes and
|
||||
* budget exclusivity is selection-exclusivity; this mirrors the
|
||||
* accounting model mclock already uses.
|
||||
*
|
||||
* Like mClockScheduler, items of SchedulerClass::immediate and items
|
||||
* at or above the priority cutoff bypass the fair hierarchy through a
|
||||
* strict high-priority queue, and enqueue_front lands requeued items
|
||||
* in that queue at priority 0.
|
||||
*/
|
||||
class BfqScheduler final : public OpScheduler, public md_config_obs_t {
|
||||
public:
|
||||
BfqScheduler(CephContext *cct, int whoami, uint32_t num_shards,
|
||||
int shard_id, bool is_rotational, unsigned cutoff_priority);
|
||||
~BfqScheduler() final;
|
||||
|
||||
void enqueue(OpSchedulerItem &&item) final;
|
||||
void enqueue_front(OpSchedulerItem &&item) final;
|
||||
|
||||
bool empty() const final {
|
||||
return fair_queued == 0 && high_priority.empty();
|
||||
}
|
||||
|
||||
// Never returns the future-time (double) alternative: if empty() is
|
||||
// false an item is always returned.
|
||||
WorkItem dequeue() final;
|
||||
|
||||
void dump(ceph::Formatter &f) const final;
|
||||
|
||||
void print(std::ostream &out) const final {
|
||||
out << get_op_queue_type_name(get_type())
|
||||
<< ", cutoff=" << cutoff_priority;
|
||||
}
|
||||
|
||||
op_queue_type_t get_type() const final {
|
||||
return op_queue_type_t::BfqScheduler;
|
||||
}
|
||||
|
||||
/// rebuild the pool -> client stream map from pool application tags
|
||||
void update_from_osdmap(const OSDMap &map) final;
|
||||
|
||||
// md_config_obs_t
|
||||
std::vector<std::string> get_tracked_keys() const noexcept final;
|
||||
void handle_conf_change(const ConfigProxy &conf,
|
||||
const std::set<std::string> &changed) final;
|
||||
|
||||
// exposed for unit tests
|
||||
uint64_t calc_scaled_cost(int item_cost) const;
|
||||
bfq_stream_t classify(const OpSchedulerItem &item) const;
|
||||
void set_pool_streams(std::unordered_map<int64_t, bfq_stream_t> &&map);
|
||||
|
||||
private:
|
||||
enum class expire_reason { emptied, exhausted, timed_out };
|
||||
|
||||
static constexpr size_t idx(bfq_stream_t s) {
|
||||
return static_cast<size_t>(s);
|
||||
}
|
||||
static constexpr size_t idx(bfq_group_t g) {
|
||||
return static_cast<size_t>(g);
|
||||
}
|
||||
|
||||
void refresh_config();
|
||||
void maybe_refresh_config() {
|
||||
if (config_dirty.exchange(false)) {
|
||||
refresh_config();
|
||||
}
|
||||
}
|
||||
void enqueue_high(unsigned prio, OpSchedulerItem &&item, bool front = false);
|
||||
void activate_stream(bfq_stream_t s);
|
||||
void begin_round(bfq_stream_t s);
|
||||
void end_round(expire_reason reason);
|
||||
OpSchedulerItem dequeue_fair();
|
||||
|
||||
CephContext *cct;
|
||||
const unsigned cutoff_priority;
|
||||
const int shard_id;
|
||||
|
||||
// config cache; refreshed under the shard lock when config_dirty
|
||||
uint64_t max_budget = 0;
|
||||
uint64_t min_cost = 0;
|
||||
uint64_t min_budget = 0;
|
||||
std::chrono::milliseconds budget_timeout{125};
|
||||
std::array<uint32_t, bfq_num_streams> stream_weights;
|
||||
std::array<uint32_t, bfq_num_groups> group_weights;
|
||||
std::atomic<bool> config_dirty = true;
|
||||
|
||||
struct Stream {
|
||||
std::deque<OpSchedulerItem> items;
|
||||
uint64_t next_budget = 0; ///< adapted by budget feedback
|
||||
uint64_t served_round = 0; ///< scaled bytes served this round
|
||||
int64_t budget_remaining = 0; ///< may go negative for oversized items
|
||||
};
|
||||
std::array<Stream, bfq_num_streams> streams;
|
||||
std::array<uint64_t, bfq_num_groups> group_served = {0, 0};
|
||||
size_t fair_queued = 0;
|
||||
|
||||
bfq_detail::BfqServiceTree root_tree;
|
||||
std::array<bfq_detail::BfqServiceTree, bfq_num_groups> group_trees;
|
||||
|
||||
std::optional<bfq_stream_t> in_service;
|
||||
ceph::coarse_mono_clock::time_point round_start;
|
||||
|
||||
/// pool id -> client stream, rebuilt on every OSDMap the shard consumes
|
||||
std::unordered_map<int64_t, bfq_stream_t> pool_streams;
|
||||
|
||||
using priority_t = unsigned;
|
||||
using SubQueue = std::map<priority_t,
|
||||
std::list<OpSchedulerItem>,
|
||||
std::greater<priority_t>>;
|
||||
/**
|
||||
* high_priority
|
||||
*
|
||||
* Holds entries to be dequeued in strict order ahead of the fair
|
||||
* hierarchy. Invariant: entries are never empty.
|
||||
*/
|
||||
SubQueue high_priority;
|
||||
static constexpr priority_t immediate_class_priority =
|
||||
std::numeric_limits<priority_t>::max();
|
||||
};
|
||||
|
||||
}
|
||||
@ -18,6 +18,7 @@
|
||||
#include "osd/scheduler/OpScheduler.h"
|
||||
|
||||
#include "common/WeightedPriorityQueue.h"
|
||||
#include "osd/scheduler/BfqScheduler.h"
|
||||
#include "osd/scheduler/mClockScheduler.h"
|
||||
|
||||
namespace ceph::osd::scheduler {
|
||||
@ -43,6 +44,10 @@ OpSchedulerRef make_scheduler(
|
||||
return std::make_unique<
|
||||
mClockScheduler>(cct, whoami, num_shards, shard_id, is_rotational,
|
||||
op_queue_cut_off);
|
||||
} else if (op_queue_type_t::BfqScheduler == osd_scheduler) {
|
||||
return std::make_unique<
|
||||
BfqScheduler>(cct, whoami, num_shards, shard_id, is_rotational,
|
||||
op_queue_cut_off);
|
||||
} else {
|
||||
ceph_abort_msg("Invalid choice of wq");
|
||||
}
|
||||
|
||||
@ -25,6 +25,8 @@
|
||||
|
||||
#include "include/ceph_assert.h"
|
||||
|
||||
class OSDMap;
|
||||
|
||||
namespace ceph::osd::scheduler {
|
||||
|
||||
using client = uint64_t;
|
||||
@ -63,6 +65,11 @@ public:
|
||||
return 0.0;
|
||||
}
|
||||
|
||||
// Notify the scheduler of a new OSDMap. Called under the shard
|
||||
// lock whenever the shard's map advances; schedulers that classify
|
||||
// ops by pool properties override this.
|
||||
virtual void update_from_osdmap(const OSDMap &map) {}
|
||||
|
||||
// Destructor
|
||||
virtual ~OpScheduler() {};
|
||||
};
|
||||
|
||||
@ -96,14 +96,14 @@ private:
|
||||
/**
|
||||
* qos_cost
|
||||
*
|
||||
* Set by mClockScheduler iff queued into mclock proper and not the
|
||||
* high/immediate queues. Represents mClockScheduler's adjusted
|
||||
* cost value.
|
||||
* Set by the QoS schedulers (mclock, bfq) iff queued into the fair
|
||||
* queue proper and not the high/immediate queues. Represents the
|
||||
* scheduler's adjusted cost value.
|
||||
*/
|
||||
uint32_t qos_cost = 0;
|
||||
|
||||
/// True iff queued via mclock proper, not the high/immediate queues
|
||||
bool was_queued_via_mclock() const {
|
||||
/// True iff queued via a fair queue proper, not the high/immediate queues
|
||||
bool was_queued_via_qos() const {
|
||||
return qos_cost > 0;
|
||||
}
|
||||
|
||||
@ -181,7 +181,7 @@ public:
|
||||
|
||||
out << " prio " << item.get_priority();
|
||||
|
||||
if (item.was_queued_via_mclock()) {
|
||||
if (item.was_queued_via_qos()) {
|
||||
out << " qos_cost " << item.qos_cost;
|
||||
}
|
||||
|
||||
@ -679,7 +679,7 @@ struct fmt::formatter<ceph::osd::scheduler::OpSchedulerItem> {
|
||||
// matching existing op_scheduler_item_t::operator<<() format
|
||||
using class_t =
|
||||
std::underlying_type_t<SchedulerClass>;
|
||||
const auto qos_cost = opsi.was_queued_via_mclock()
|
||||
const auto qos_cost = opsi.was_queued_via_qos()
|
||||
? fmt::format(" qos_cost {}", opsi.qos_cost)
|
||||
: "";
|
||||
const auto pushes =
|
||||
|
||||
@ -1086,8 +1086,9 @@ void PgScrubber::select_range_n_notify()
|
||||
uint64_t PgScrubber::get_scrub_cost(uint64_t num_chunk_objects)
|
||||
{
|
||||
const auto& conf = m_pg->get_cct()->_conf;
|
||||
if (op_queue_type_t::WeightedPriorityQueue == m_osds->osd->osd_op_queue_type()) {
|
||||
// if the osd_op_queue is WPQ, we will use the default osd_scrub_cost value
|
||||
if (op_queue_type_t::mClockScheduler != m_osds->osd->osd_op_queue_type()) {
|
||||
// only mclock derives a per-IO metadata cost from its device capacity
|
||||
// model; wpq and bfq use the flat osd_scrub_cost value
|
||||
return conf->osd_scrub_cost;
|
||||
}
|
||||
uint64_t cost = 0;
|
||||
|
||||
@ -215,6 +215,33 @@ target_link_libraries(unittest_mclock_scheduler
|
||||
global osd dmclock os
|
||||
)
|
||||
|
||||
# unittest_bfq_scheduler
|
||||
add_executable(unittest_bfq_scheduler
|
||||
TestBfqScheduler.cc
|
||||
)
|
||||
add_ceph_unittest(unittest_bfq_scheduler)
|
||||
target_link_libraries(unittest_bfq_scheduler
|
||||
global osd os
|
||||
)
|
||||
|
||||
# unittest_scheduler_isolation
|
||||
add_executable(unittest_scheduler_isolation
|
||||
TestSchedulerIsolation.cc
|
||||
)
|
||||
add_ceph_unittest(unittest_scheduler_isolation)
|
||||
target_link_libraries(unittest_scheduler_isolation
|
||||
global osd dmclock os
|
||||
)
|
||||
|
||||
# ceph_bench_op_scheduler: isolation micro benchmarks across the op
|
||||
# schedulers; a manual tool, not part of the unit test suite
|
||||
add_executable(ceph_bench_op_scheduler
|
||||
bench_op_scheduler.cc
|
||||
)
|
||||
target_link_libraries(ceph_bench_op_scheduler
|
||||
global osd dmclock os
|
||||
)
|
||||
|
||||
# unittest ECOmapJournal
|
||||
add_executable(unittest_ec_omap_journal
|
||||
test_ec_omap_journal.cc
|
||||
@ -226,6 +253,7 @@ target_link_libraries(unittest_ec_omap_journal osd global ${BLKID_LIBRARIES})
|
||||
# Not including unittest_osdmap, as it is slow. It is tested elsewhere.
|
||||
set(OSD_UNITTESTS
|
||||
unittest_backend_basics
|
||||
unittest_bfq_scheduler
|
||||
unittest_ec_omap_journal
|
||||
unittest_ec_transaction
|
||||
unittest_ec_transaction_l
|
||||
@ -241,6 +269,7 @@ set(OSD_UNITTESTS
|
||||
unittest_osd_types
|
||||
unittest_osdscrub
|
||||
unittest_peeringstate
|
||||
unittest_scheduler_isolation
|
||||
unittest_pg_transaction
|
||||
unittest_pglog
|
||||
unittest_scrubber_be
|
||||
|
||||
531
src/test/osd/TestBfqScheduler.cc
Normal file
531
src/test/osd/TestBfqScheduler.cc
Normal file
@ -0,0 +1,531 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
#include <chrono>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <thread>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
|
||||
#include "global/global_context.h"
|
||||
#include "global/global_init.h"
|
||||
#include "common/common_init.h"
|
||||
|
||||
#include "osd/scheduler/BfqScheduler.h"
|
||||
#include "osd/scheduler/OpSchedulerItem.h"
|
||||
|
||||
using namespace ceph::osd::scheduler;
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
std::vector<const char*> args(argv, argv+argc);
|
||||
auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_OSD,
|
||||
CODE_ENVIRONMENT_UTILITY,
|
||||
CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
|
||||
common_init_finish(g_ceph_context);
|
||||
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
const std::vector<std::string> bfq_conf_keys = {
|
||||
"osd_bfq_client_group_weight",
|
||||
"osd_bfq_background_group_weight",
|
||||
"osd_bfq_client_block_weight",
|
||||
"osd_bfq_client_object_weight",
|
||||
"osd_bfq_client_file_weight",
|
||||
"osd_bfq_client_other_weight",
|
||||
"osd_bfq_background_recovery_weight",
|
||||
"osd_bfq_background_best_effort_weight",
|
||||
"osd_bfq_max_budget",
|
||||
"osd_bfq_min_cost",
|
||||
"osd_bfq_budget_timeout"
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
class BfqSchedulerTest : public testing::Test {
|
||||
public:
|
||||
int whoami = 0;
|
||||
uint32_t num_shards = 1;
|
||||
int shard_id = 0;
|
||||
bool is_rotational = false;
|
||||
unsigned cutoff_priority = 12;
|
||||
std::unique_ptr<BfqScheduler> q;
|
||||
|
||||
void set_conf(const std::string &key, const std::string &val) {
|
||||
ASSERT_EQ(0, g_ceph_context->_conf.set_val(key, val));
|
||||
}
|
||||
|
||||
// construct the scheduler after any config overrides so the
|
||||
// constructor picks them up
|
||||
void create_queue() {
|
||||
q = std::make_unique<BfqScheduler>(
|
||||
g_ceph_context, whoami, num_shards, shard_id, is_rotational,
|
||||
cutoff_priority);
|
||||
}
|
||||
|
||||
void TearDown() override {
|
||||
q.reset();
|
||||
for (const auto &key : bfq_conf_keys) {
|
||||
g_ceph_context->_conf.rm_val(key);
|
||||
}
|
||||
}
|
||||
|
||||
struct MockBfqItem : public PGOpQueueable {
|
||||
SchedulerClass scheduler_class;
|
||||
|
||||
MockBfqItem(spg_t pgid, SchedulerClass _scheduler_class)
|
||||
: PGOpQueueable(pgid),
|
||||
scheduler_class(_scheduler_class) {}
|
||||
|
||||
std::ostream &print(std::ostream &rhs) const final { return rhs; }
|
||||
|
||||
std::string print() const final {
|
||||
return std::string();
|
||||
}
|
||||
|
||||
std::optional<OpRequestRef> maybe_get_op() const final {
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
SchedulerClass get_scheduler_class() const final {
|
||||
return scheduler_class;
|
||||
}
|
||||
|
||||
void run(OSD *osd, OSDShard *sdata, PGRef& pg,
|
||||
ThreadPool::TPHandle &handle) final {}
|
||||
};
|
||||
};
|
||||
|
||||
namespace {
|
||||
|
||||
spg_t make_pgid(int64_t pool) {
|
||||
return spg_t(pg_t(0, pool));
|
||||
}
|
||||
|
||||
OpSchedulerItem create_item(
|
||||
epoch_t e, uint64_t owner, SchedulerClass klass,
|
||||
int64_t pool = 1, int cost = 4096, unsigned priority = 1)
|
||||
{
|
||||
return OpSchedulerItem(
|
||||
std::make_unique<BfqSchedulerTest::MockBfqItem>(make_pgid(pool), klass),
|
||||
cost, priority, utime_t(), owner, e);
|
||||
}
|
||||
|
||||
OpSchedulerItem create_high_prio_item(
|
||||
unsigned priority, epoch_t e, uint64_t owner, SchedulerClass klass)
|
||||
{
|
||||
return OpSchedulerItem(
|
||||
std::make_unique<BfqSchedulerTest::MockBfqItem>(make_pgid(1), klass),
|
||||
4096, priority, utime_t(), owner, e);
|
||||
}
|
||||
|
||||
OpSchedulerItem get_item(WorkItem item)
|
||||
{
|
||||
// throws (fails the test) if the variant holds monostate or a
|
||||
// future time -- bfq must always return a real item
|
||||
return std::move(std::get<OpSchedulerItem>(item));
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestEmpty) {
|
||||
create_queue();
|
||||
ASSERT_TRUE(q->empty());
|
||||
|
||||
for (unsigned i = 100; i < 105; i += 2) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client));
|
||||
}
|
||||
ASSERT_FALSE(q->empty());
|
||||
|
||||
std::list<OpSchedulerItem> reqs;
|
||||
reqs.push_back(get_item(q->dequeue()));
|
||||
reqs.push_back(get_item(q->dequeue()));
|
||||
ASSERT_EQ(2u, reqs.size());
|
||||
ASSERT_FALSE(q->empty());
|
||||
|
||||
for (auto &&i : reqs) {
|
||||
q->enqueue_front(std::move(i));
|
||||
}
|
||||
reqs.clear();
|
||||
ASSERT_FALSE(q->empty());
|
||||
|
||||
for (int i = 0; i < 3; ++i) {
|
||||
ASSERT_FALSE(q->empty());
|
||||
get_item(q->dequeue());
|
||||
}
|
||||
ASSERT_TRUE(q->empty());
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestSingleStreamFifo) {
|
||||
create_queue();
|
||||
for (unsigned i = 100; i < 105; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client));
|
||||
}
|
||||
for (unsigned i = 100; i < 105; ++i) {
|
||||
auto r = get_item(q->dequeue());
|
||||
ASSERT_EQ(i, r.get_map_epoch());
|
||||
}
|
||||
ASSERT_TRUE(q->empty());
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestImmediateAndCutoffOrdering) {
|
||||
create_queue();
|
||||
ASSERT_TRUE(q->empty());
|
||||
|
||||
// fair-queue items
|
||||
for (unsigned i = 100; i < 102; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client));
|
||||
}
|
||||
// immediate items
|
||||
for (unsigned i = 103; i < 105; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::immediate));
|
||||
}
|
||||
// strict high-priority items (priority >= cutoff of 12)
|
||||
for (unsigned i = 200; i < 202; ++i) {
|
||||
q->enqueue(create_high_prio_item(i - 180, i, 1, SchedulerClass::client));
|
||||
}
|
||||
|
||||
// immediate class first
|
||||
ASSERT_EQ(103u, get_item(q->dequeue()).get_map_epoch());
|
||||
ASSERT_EQ(104u, get_item(q->dequeue()).get_map_epoch());
|
||||
// then the strict queue, higher priority first
|
||||
ASSERT_EQ(201u, get_item(q->dequeue()).get_map_epoch());
|
||||
ASSERT_EQ(200u, get_item(q->dequeue()).get_map_epoch());
|
||||
// the fair hierarchy drains last
|
||||
ASSERT_EQ(100u, get_item(q->dequeue()).get_map_epoch());
|
||||
ASSERT_EQ(101u, get_item(q->dequeue()).get_map_epoch());
|
||||
ASSERT_TRUE(q->empty());
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestEnqueueFront) {
|
||||
create_queue();
|
||||
for (unsigned i = 100; i < 104; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client));
|
||||
}
|
||||
auto r = get_item(q->dequeue());
|
||||
ASSERT_EQ(100u, r.get_map_epoch());
|
||||
q->enqueue_front(std::move(r));
|
||||
// the requeued item bypasses the fair hierarchy and comes back first
|
||||
ASSERT_EQ(100u, get_item(q->dequeue()).get_map_epoch());
|
||||
ASSERT_EQ(101u, get_item(q->dequeue()).get_map_epoch());
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestClientStreamWeights) {
|
||||
set_conf("osd_bfq_max_budget", "16384");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "100000");
|
||||
set_conf("osd_bfq_client_block_weight", "300");
|
||||
set_conf("osd_bfq_client_object_weight", "100");
|
||||
create_queue();
|
||||
q->set_pool_streams({
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object}
|
||||
});
|
||||
|
||||
constexpr unsigned per_stream = 600;
|
||||
for (unsigned i = 0; i < per_stream; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 1));
|
||||
q->enqueue(create_item(i, 2, SchedulerClass::client, 2));
|
||||
}
|
||||
|
||||
// both streams stay backlogged over the measurement window
|
||||
std::map<uint64_t, unsigned> count;
|
||||
for (unsigned i = 0; i < 400; ++i) {
|
||||
ASSERT_FALSE(q->empty());
|
||||
++count[get_item(q->dequeue()).get_owner()];
|
||||
}
|
||||
// weights 300:100 -> expect a 3:1 split of the 400 dequeues
|
||||
ASSERT_NEAR(300, count[1], 30);
|
||||
ASSERT_NEAR(100, count[2], 30);
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestGroupWeights) {
|
||||
set_conf("osd_bfq_max_budget", "16384");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "100000");
|
||||
set_conf("osd_bfq_client_group_weight", "100");
|
||||
set_conf("osd_bfq_background_group_weight", "25");
|
||||
create_queue();
|
||||
|
||||
constexpr unsigned per_stream = 600;
|
||||
for (unsigned i = 0; i < per_stream; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 5));
|
||||
q->enqueue(create_item(i, 2, SchedulerClass::background_recovery, 5));
|
||||
}
|
||||
|
||||
std::map<uint64_t, unsigned> count;
|
||||
for (unsigned i = 0; i < 500; ++i) {
|
||||
ASSERT_FALSE(q->empty());
|
||||
++count[get_item(q->dequeue()).get_owner()];
|
||||
}
|
||||
// group weights 100:25 -> expect a 4:1 split of the 500 dequeues
|
||||
ASSERT_NEAR(400, count[1], 40);
|
||||
ASSERT_NEAR(100, count[2], 40);
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestBudgetRotation) {
|
||||
set_conf("osd_bfq_max_budget", "16384");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "100000");
|
||||
create_queue();
|
||||
q->set_pool_streams({
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object}
|
||||
});
|
||||
|
||||
constexpr unsigned per_stream = 64;
|
||||
for (unsigned i = 0; i < per_stream; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 1));
|
||||
q->enqueue(create_item(i, 2, SchedulerClass::client, 2));
|
||||
}
|
||||
|
||||
// with equal weights and 4-item budgets, service alternates in
|
||||
// budget-sized runs rather than per-op interleave
|
||||
uint64_t prev_owner = 0;
|
||||
unsigned switches = 0;
|
||||
for (unsigned i = 0; i < 2 * per_stream; ++i) {
|
||||
auto owner = get_item(q->dequeue()).get_owner();
|
||||
if (prev_owner && owner != prev_owner) {
|
||||
++switches;
|
||||
}
|
||||
prev_owner = owner;
|
||||
}
|
||||
ASSERT_TRUE(q->empty());
|
||||
// strict alternation would switch 127 times; 4-item rounds yield
|
||||
// roughly 2*64/4 = 32 switches (a few more during budget ramp-up)
|
||||
ASSERT_GE(switches, 2u);
|
||||
ASSERT_LT(switches, 64u);
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestPoolClassification) {
|
||||
create_queue();
|
||||
|
||||
// no pool map yet: every client op classifies as client_other
|
||||
auto unmapped = create_item(1, 1, SchedulerClass::client, 42);
|
||||
ASSERT_EQ(bfq_stream_t::client_other, q->classify(unmapped));
|
||||
|
||||
q->set_pool_streams({
|
||||
{42, bfq_stream_t::client_block},
|
||||
{7, bfq_stream_t::client_file}
|
||||
});
|
||||
auto block = create_item(1, 1, SchedulerClass::client, 42);
|
||||
auto file = create_item(1, 1, SchedulerClass::client, 7);
|
||||
auto other = create_item(1, 1, SchedulerClass::client, 9);
|
||||
ASSERT_EQ(bfq_stream_t::client_block, q->classify(block));
|
||||
ASSERT_EQ(bfq_stream_t::client_file, q->classify(file));
|
||||
ASSERT_EQ(bfq_stream_t::client_other, q->classify(other));
|
||||
|
||||
// background classes ignore the pool
|
||||
auto rec = create_item(1, 1, SchedulerClass::background_recovery, 42);
|
||||
auto be = create_item(1, 1, SchedulerClass::background_best_effort, 42);
|
||||
ASSERT_EQ(bfq_stream_t::background_recovery, q->classify(rec));
|
||||
ASSERT_EQ(bfq_stream_t::background_best_effort, q->classify(be));
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestDrain) {
|
||||
create_queue();
|
||||
q->set_pool_streams({
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object},
|
||||
{3, bfq_stream_t::client_file}
|
||||
});
|
||||
|
||||
unsigned enqueued = 0;
|
||||
for (unsigned i = 0; i < 20; ++i) {
|
||||
for (int64_t pool = 1; pool <= 4; ++pool) {
|
||||
q->enqueue(create_item(i, pool, SchedulerClass::client, pool));
|
||||
++enqueued;
|
||||
}
|
||||
q->enqueue(create_item(i, 5, SchedulerClass::background_recovery));
|
||||
q->enqueue(create_item(i, 6, SchedulerClass::background_best_effort));
|
||||
q->enqueue(create_item(i, 7, SchedulerClass::immediate));
|
||||
enqueued += 3;
|
||||
}
|
||||
q->enqueue(create_high_prio_item(20, 1000, 8, SchedulerClass::client));
|
||||
++enqueued;
|
||||
|
||||
// the fast-shutdown pattern: dequeue until empty must terminate
|
||||
unsigned drained = 0;
|
||||
while (!q->empty()) {
|
||||
get_item(q->dequeue());
|
||||
++drained;
|
||||
}
|
||||
ASSERT_EQ(enqueued, drained);
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestOversizedItemRotation) {
|
||||
set_conf("osd_bfq_max_budget", "16384");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "100000");
|
||||
create_queue();
|
||||
q->set_pool_streams({
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object}
|
||||
});
|
||||
|
||||
// stream A leads with an item 4x the entire budget
|
||||
q->enqueue(create_item(100, 1, SchedulerClass::client, 1, 65536));
|
||||
for (unsigned i = 101; i < 104; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 1));
|
||||
}
|
||||
for (unsigned i = 200; i < 204; ++i) {
|
||||
q->enqueue(create_item(i, 2, SchedulerClass::client, 2));
|
||||
}
|
||||
|
||||
// the oversized item must dispatch (budget goes negative, no hang)...
|
||||
ASSERT_EQ(100u, get_item(q->dequeue()).get_map_epoch());
|
||||
// ...and the overrun charge makes stream A ineligible, rotating to B
|
||||
ASSERT_EQ(200u, get_item(q->dequeue()).get_map_epoch());
|
||||
|
||||
unsigned drained = 2;
|
||||
while (!q->empty()) {
|
||||
get_item(q->dequeue());
|
||||
++drained;
|
||||
}
|
||||
ASSERT_EQ(8u, drained);
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestBudgetTimeoutExpiry) {
|
||||
set_conf("osd_bfq_max_budget", "1048576");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "1");
|
||||
create_queue();
|
||||
|
||||
// a single backlogged stream: every timeout expiry must re-tag the
|
||||
// stream and continue in FIFO order with adapted budgets
|
||||
for (unsigned i = 100; i < 112; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 1));
|
||||
}
|
||||
for (unsigned i = 100; i < 112; ++i) {
|
||||
if (i % 3 == 0) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(2));
|
||||
}
|
||||
auto r = get_item(q->dequeue());
|
||||
ASSERT_EQ(i, r.get_map_epoch());
|
||||
}
|
||||
ASSERT_TRUE(q->empty());
|
||||
}
|
||||
|
||||
TEST_F(BfqSchedulerTest, TestRuntimeWeightChange) {
|
||||
set_conf("osd_bfq_max_budget", "16384");
|
||||
set_conf("osd_bfq_min_cost", "4096");
|
||||
set_conf("osd_bfq_budget_timeout", "100000");
|
||||
create_queue();
|
||||
q->set_pool_streams({
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object}
|
||||
});
|
||||
|
||||
constexpr unsigned per_stream = 800;
|
||||
for (unsigned i = 0; i < per_stream; ++i) {
|
||||
q->enqueue(create_item(i, 1, SchedulerClass::client, 1));
|
||||
q->enqueue(create_item(i, 2, SchedulerClass::client, 2));
|
||||
}
|
||||
|
||||
// equal weights: roughly even split
|
||||
std::map<uint64_t, unsigned> before;
|
||||
for (unsigned i = 0; i < 100; ++i) {
|
||||
++before[get_item(q->dequeue()).get_owner()];
|
||||
}
|
||||
ASSERT_NEAR(50, before[1], 15);
|
||||
|
||||
// raise block weight at runtime; the observer marks the config
|
||||
// dirty and the next dequeue folds it in at round re-tag time
|
||||
set_conf("osd_bfq_client_block_weight", "300");
|
||||
g_ceph_context->_conf.apply_changes(nullptr);
|
||||
|
||||
std::map<uint64_t, unsigned> after;
|
||||
for (unsigned i = 0; i < 400; ++i) {
|
||||
ASSERT_FALSE(q->empty());
|
||||
++after[get_item(q->dequeue()).get_owner()];
|
||||
}
|
||||
ASSERT_NEAR(300, after[1], 40);
|
||||
ASSERT_NEAR(100, after[2], 40);
|
||||
}
|
||||
|
||||
// ---- bfq_detail::BfqServiceTree unit tests ----
|
||||
|
||||
using bfq_detail::BfqServiceTree;
|
||||
|
||||
TEST(BfqServiceTree, ActivationAndBackShift) {
|
||||
BfqServiceTree t(2);
|
||||
ASSERT_FALSE(t.has_active());
|
||||
ASSERT_EQ(std::nullopt, t.select());
|
||||
|
||||
t.activate(0, 1000, 100); // S=0, F=10
|
||||
t.activate(1, 1000, 100); // S=0, F=10
|
||||
ASSERT_TRUE(t.has_active());
|
||||
ASSERT_DOUBLE_EQ(0.0, t.entity(0).start);
|
||||
ASSERT_DOUBLE_EQ(10.0, t.entity(0).finish);
|
||||
|
||||
// tie on F: first index wins
|
||||
auto sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
ASSERT_EQ(0u, *sel);
|
||||
|
||||
t.charge(500); // V = 500/200 = 2.5
|
||||
|
||||
// entity 0 consumed only half its allotment: back-shift F to 5,
|
||||
// re-tag S=5, F=5+10=15
|
||||
t.expire(0, 500, 1000, 100, true);
|
||||
ASSERT_DOUBLE_EQ(5.0, t.entity(0).start);
|
||||
ASSERT_DOUBLE_EQ(15.0, t.entity(0).finish);
|
||||
|
||||
// entity 1 (F=10, eligible) now beats entity 0 (F=15)
|
||||
sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
ASSERT_EQ(1u, *sel);
|
||||
}
|
||||
|
||||
TEST(BfqServiceTree, EligibilityGate) {
|
||||
BfqServiceTree t(2);
|
||||
t.activate(0, 1000, 10); // S=0, F=100 (light weight)
|
||||
t.activate(1, 100, 100); // S=0, F=1
|
||||
|
||||
auto sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
ASSERT_EQ(1u, *sel);
|
||||
|
||||
t.charge(100); // V = 100/110 ~ 0.909
|
||||
t.expire(1, 100, 100, 100, true); // F1=1 -> S1=1, F1=2
|
||||
|
||||
// entity 1 has the smaller F (2 < 100) but S1=1 > V, so it is not
|
||||
// eligible; WF2Q+ must pick entity 0 instead of letting 1 run ahead
|
||||
sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
ASSERT_EQ(0u, *sel);
|
||||
|
||||
// when only ineligible entities remain, V jumps forward
|
||||
t.expire(0, 1000, 1000, 10, false); // deactivate 0
|
||||
sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
ASSERT_EQ(1u, *sel);
|
||||
|
||||
// full idle renormalizes the tree
|
||||
t.expire(1, 100, 100, 100, false);
|
||||
ASSERT_FALSE(t.has_active());
|
||||
ASSERT_DOUBLE_EQ(0.0, t.get_vtime());
|
||||
ASSERT_DOUBLE_EQ(0.0, t.entity(0).finish);
|
||||
}
|
||||
|
||||
TEST(BfqServiceTree, WeightedShare) {
|
||||
BfqServiceTree t(2);
|
||||
t.activate(0, 1000, 200);
|
||||
t.activate(1, 1000, 100);
|
||||
|
||||
std::array<unsigned, 2> rounds = {0, 0};
|
||||
for (unsigned i = 0; i < 300; ++i) {
|
||||
auto sel = t.select();
|
||||
ASSERT_TRUE(sel);
|
||||
++rounds[*sel];
|
||||
t.charge(1000);
|
||||
t.expire(*sel, 1000, 1000, *sel == 0 ? 200 : 100, true);
|
||||
}
|
||||
// weights 200:100 with equal allotments -> 2:1 round share
|
||||
ASSERT_NEAR(200, rounds[0], 10);
|
||||
ASSERT_NEAR(100, rounds[1], 10);
|
||||
}
|
||||
90
src/test/osd/TestSchedulerIsolation.cc
Normal file
90
src/test/osd/TestSchedulerIsolation.cc
Normal file
@ -0,0 +1,90 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
/*
|
||||
* Isolation assertions for the bfq op scheduler, built on the
|
||||
* scheduler_bench.h harness. The harness paces in wall time, which
|
||||
* is unfit for shared CI executors, so these tests SKIP unless
|
||||
* CEPH_TEST_SCHEDULER_ISOLATION is set in the environment. Bounds
|
||||
* are deliberately loose even then. The full comparative study lives
|
||||
* in ceph_bench_op_scheduler.
|
||||
*/
|
||||
|
||||
#include <cstdlib>
|
||||
|
||||
#include "gtest/gtest.h"
|
||||
|
||||
#include "global/global_context.h"
|
||||
#include "global/global_init.h"
|
||||
#include "common/common_init.h"
|
||||
|
||||
#include "scheduler_bench.h"
|
||||
|
||||
using namespace scheduler_bench;
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
std::vector<const char*> args(argv, argv+argc);
|
||||
auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_OSD,
|
||||
CODE_ENVIRONMENT_UTILITY,
|
||||
CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
|
||||
common_init_finish(g_ceph_context);
|
||||
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr double kRate = 100e6; // simulated device, bytes/sec
|
||||
const std::unordered_map<int64_t, bfq_stream_t> kPoolMap = {
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object},
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
// wall-clock pacing is unreliable on shared CI executors; opt in
|
||||
#define REQUIRE_ISOLATION_ENV() \
|
||||
do { \
|
||||
if (!std::getenv("CEPH_TEST_SCHEDULER_ISOLATION")) { \
|
||||
GTEST_SKIP() \
|
||||
<< "set CEPH_TEST_SCHEDULER_ISOLATION=1 to run " \
|
||||
"wall-clock isolation checks"; \
|
||||
} \
|
||||
} while (0)
|
||||
|
||||
// A backlogged rbd-pool victim must hold ~its weight share (50% at
|
||||
// equal weights) even when the rgw-pool aggressor fans out over 16
|
||||
// client sessions. This is the property wpq (per-owner round robin)
|
||||
// and mclock (single dmclock client for all client ops) structurally
|
||||
// dilute with session count.
|
||||
TEST(BfqIsolation, ShareUnderSessionScaling) {
|
||||
REQUIRE_ISOLATION_ENV();
|
||||
std::vector<StreamSpec> specs = {
|
||||
{.name = "victim", .pool = 1, .first_owner = 1, .num_owners = 1},
|
||||
{.name = "aggr", .pool = 2, .first_owner = 100, .num_owners = 16},
|
||||
};
|
||||
auto r = run_cell(g_ceph_context, op_queue_type_t::BfqScheduler,
|
||||
specs, kPoolMap, kRate, 2.0, 0.5);
|
||||
const double share = r.streams.at("victim").share;
|
||||
EXPECT_GT(share, 0.35) << "victim diluted by aggressor sessions";
|
||||
EXPECT_LT(share, 0.65) << "victim over-served";
|
||||
}
|
||||
|
||||
// A paced victim (20% of device rate, under its fair share) must keep
|
||||
// its offered throughput with queueing delay bounded by budget
|
||||
// rotation, despite a saturating multi-session aggressor.
|
||||
TEST(BfqIsolation, PacedVictimServiced) {
|
||||
REQUIRE_ISOLATION_ENV();
|
||||
std::vector<StreamSpec> specs = {
|
||||
{.name = "victim", .pool = 1, .first_owner = 1, .num_owners = 1,
|
||||
.offered_ops_per_sec = kRate * 0.2 / 65536.0},
|
||||
{.name = "aggr", .pool = 2, .first_owner = 100, .num_owners = 8},
|
||||
};
|
||||
auto r = run_cell(g_ceph_context, op_queue_type_t::BfqScheduler,
|
||||
specs, kPoolMap, kRate, 2.0, 0.5);
|
||||
const auto &victim = r.streams.at("victim");
|
||||
// offered 20 MB/s; allow generous scheduling/pacing slack
|
||||
EXPECT_GT(victim.mbps, 12.0) << "victim throughput collapsed";
|
||||
EXPECT_LT(victim.p99_ms, 500.0) << "victim latency unbounded";
|
||||
}
|
||||
186
src/test/osd/bench_op_scheduler.cc
Normal file
186
src/test/osd/bench_op_scheduler.cc
Normal file
@ -0,0 +1,186 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
/*
|
||||
* ceph_bench_op_scheduler
|
||||
*
|
||||
* Isolation micro benchmarks comparing the OSD op schedulers (wpq,
|
||||
* mclock_scheduler, bfq) on identical synthetic workloads with a
|
||||
* simulated device drain rate. See scheduler_bench.h for the
|
||||
* harness; doc/dev/osd_internals/bfq_scheduler.rst for context.
|
||||
*
|
||||
* Usage:
|
||||
* ceph_bench_op_scheduler [--rate-mb N] [--secs S] [--csv FILE]
|
||||
* [--scenario share|latency|recovery|all]
|
||||
*/
|
||||
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <fstream>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
#include "global/global_context.h"
|
||||
#include "global/global_init.h"
|
||||
#include "common/common_init.h"
|
||||
|
||||
#include "scheduler_bench.h"
|
||||
|
||||
using namespace scheduler_bench;
|
||||
|
||||
namespace {
|
||||
|
||||
struct Options {
|
||||
double rate_mb = 100.0;
|
||||
double secs = 3.0;
|
||||
double warmup = 0.5;
|
||||
std::string scenario = "all";
|
||||
std::string csv_path;
|
||||
};
|
||||
|
||||
const std::vector<op_queue_type_t> kTypes = {
|
||||
op_queue_type_t::WeightedPriorityQueue,
|
||||
op_queue_type_t::mClockScheduler,
|
||||
op_queue_type_t::BfqScheduler,
|
||||
};
|
||||
|
||||
const std::unordered_map<int64_t, bfq_stream_t> kPoolMap = {
|
||||
{1, bfq_stream_t::client_block},
|
||||
{2, bfq_stream_t::client_object},
|
||||
};
|
||||
|
||||
std::ofstream csv;
|
||||
|
||||
void report_cell(const std::string &scenario, const std::string &cell,
|
||||
op_queue_type_t type,
|
||||
const std::vector<StreamSpec> &specs,
|
||||
const CellResult &r)
|
||||
{
|
||||
for (const auto &s : specs) {
|
||||
const auto &sr = r.streams.at(s.name);
|
||||
std::printf(" %-16s %-14s %8.1f %7.3f %8.1f %9.1f\n",
|
||||
std::string(get_op_queue_type_name(type)).c_str(),
|
||||
s.name.c_str(), sr.mbps, sr.share, sr.p50_ms, sr.p99_ms);
|
||||
if (csv.is_open()) {
|
||||
csv << scenario << ',' << cell << ','
|
||||
<< get_op_queue_type_name(type) << ',' << s.name << ','
|
||||
<< sr.ops << ',' << sr.bytes << ',' << sr.share << ','
|
||||
<< sr.mbps << ',' << sr.p50_ms << ',' << sr.p99_ms << '\n';
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void run_matrix(const Options &opt, const std::string &scenario,
|
||||
const std::string &cell,
|
||||
const std::vector<StreamSpec> &specs)
|
||||
{
|
||||
std::printf("--- %s / %s (device %.0f MB/s, %.1fs)\n",
|
||||
scenario.c_str(), cell.c_str(), opt.rate_mb, opt.secs);
|
||||
std::printf(" %-16s %-14s %8s %7s %8s %9s\n",
|
||||
"scheduler", "stream", "MB/s", "share", "p50ms", "p99ms");
|
||||
for (auto type : kTypes) {
|
||||
auto r = run_cell(g_ceph_context, type, specs, kPoolMap,
|
||||
opt.rate_mb * 1e6, opt.secs, opt.warmup);
|
||||
report_cell(scenario, cell, type, specs, r);
|
||||
}
|
||||
std::printf("\n");
|
||||
}
|
||||
|
||||
// A1: both streams saturating; the aggressor fans out over K client
|
||||
// sessions. Isolation == the victim's share is insensitive to K.
|
||||
void scenario_share(const Options &opt)
|
||||
{
|
||||
for (unsigned k : {1u, 4u, 16u}) {
|
||||
std::vector<StreamSpec> specs = {
|
||||
{.name = "victim_rbd", .pool = 1, .first_owner = 1, .num_owners = 1},
|
||||
{.name = "aggr_rgw", .pool = 2, .first_owner = 100, .num_owners = k},
|
||||
};
|
||||
run_matrix(opt, "share", "sessions=" + std::to_string(k), specs);
|
||||
}
|
||||
}
|
||||
|
||||
// A2: paced victim (20% of device rate) vs a saturating 8-session
|
||||
// aggressor. Isolation == the victim keeps its offered throughput
|
||||
// with bounded queueing delay.
|
||||
void scenario_latency(const Options &opt)
|
||||
{
|
||||
const double victim_ops = opt.rate_mb * 1e6 * 0.2 / 65536.0;
|
||||
std::vector<StreamSpec> specs = {
|
||||
{.name = "victim_rbd", .pool = 1, .first_owner = 1, .num_owners = 1,
|
||||
.offered_ops_per_sec = victim_ops},
|
||||
{.name = "aggr_rgw", .pool = 2, .first_owner = 100, .num_owners = 8},
|
||||
};
|
||||
run_matrix(opt, "latency", "victim=20%,sessions=8", specs);
|
||||
}
|
||||
|
||||
// B: saturating client stream vs saturating recovery stream (1M
|
||||
// chunks, DEGRADED priority encoding).
|
||||
void scenario_recovery(const Options &opt)
|
||||
{
|
||||
std::vector<StreamSpec> specs = {
|
||||
{.name = "client_rbd", .pool = 1, .first_owner = 1, .num_owners = 2},
|
||||
{.name = "recovery",
|
||||
.klass = SchedulerClass::background_recovery,
|
||||
.pool = 9, .first_owner = 500, .num_owners = 1,
|
||||
.op_size = 1048576, .priority = 10, .backlog_per_owner = 16},
|
||||
};
|
||||
run_matrix(opt, "recovery", "client-vs-recovery", specs);
|
||||
}
|
||||
|
||||
} // anonymous namespace
|
||||
|
||||
int main(int argc, char **argv)
|
||||
{
|
||||
std::vector<const char*> args(argv, argv + argc);
|
||||
auto cct = global_init(nullptr, args, CEPH_ENTITY_TYPE_OSD,
|
||||
CODE_ENVIRONMENT_UTILITY,
|
||||
CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
|
||||
common_init_finish(g_ceph_context);
|
||||
|
||||
Options opt;
|
||||
for (size_t i = 1; i < args.size(); ++i) {
|
||||
auto need_val = [&](const char *flag) -> const char * {
|
||||
if (i + 1 >= args.size()) {
|
||||
std::fprintf(stderr, "%s requires a value\n", flag);
|
||||
exit(1);
|
||||
}
|
||||
return args[++i];
|
||||
};
|
||||
if (!std::strcmp(args[i], "--rate-mb")) {
|
||||
opt.rate_mb = std::stod(need_val("--rate-mb"));
|
||||
} else if (!std::strcmp(args[i], "--secs")) {
|
||||
opt.secs = std::stod(need_val("--secs"));
|
||||
} else if (!std::strcmp(args[i], "--csv")) {
|
||||
opt.csv_path = need_val("--csv");
|
||||
} else if (!std::strcmp(args[i], "--scenario")) {
|
||||
opt.scenario = need_val("--scenario");
|
||||
}
|
||||
}
|
||||
opt.warmup = std::min(opt.warmup, opt.secs / 4);
|
||||
|
||||
// calibrate mclock's capacity model to the simulated device so we
|
||||
// benchmark mclock, not a misconfigured mclock
|
||||
g_ceph_context->_conf.set_val(
|
||||
"osd_mclock_max_sequential_bandwidth_ssd",
|
||||
std::to_string(static_cast<uint64_t>(opt.rate_mb * 1e6)));
|
||||
g_ceph_context->_conf.set_val(
|
||||
"osd_mclock_max_capacity_iops_ssd",
|
||||
std::to_string(opt.rate_mb * 1e6 / 65536.0));
|
||||
|
||||
if (!opt.csv_path.empty()) {
|
||||
csv.open(opt.csv_path);
|
||||
csv << "scenario,cell,scheduler,stream,ops,bytes,share,mbps,"
|
||||
"p50_ms,p99_ms\n";
|
||||
}
|
||||
|
||||
if (opt.scenario == "all" || opt.scenario == "share") {
|
||||
scenario_share(opt);
|
||||
}
|
||||
if (opt.scenario == "all" || opt.scenario == "latency") {
|
||||
scenario_latency(opt);
|
||||
}
|
||||
if (opt.scenario == "all" || opt.scenario == "recovery") {
|
||||
scenario_recovery(opt);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
222
src/test/osd/scheduler_bench.h
Normal file
222
src/test/osd/scheduler_bench.h
Normal file
@ -0,0 +1,222 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 sts=2 expandtab
|
||||
|
||||
/*
|
||||
* Shared harness for op-scheduler isolation micro benchmarks.
|
||||
*
|
||||
* Drives synthetic per-stream workloads through any OpScheduler
|
||||
* implementation in-process, pacing dequeues at a simulated device
|
||||
* rate so that saturation dynamics (and therefore isolation) are
|
||||
* meaningful. Runs in real time because mclock's dmclock tags are
|
||||
* wall-clock based; bfq and wpq are indifferent.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <algorithm>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
#include "common/Clock.h"
|
||||
#include "common/ceph_context.h"
|
||||
#include "osd/scheduler/BfqScheduler.h"
|
||||
#include "osd/scheduler/OpScheduler.h"
|
||||
#include "osd/scheduler/OpSchedulerItem.h"
|
||||
|
||||
namespace scheduler_bench {
|
||||
|
||||
using namespace ceph::osd::scheduler;
|
||||
|
||||
struct MockItem : public PGOpQueueable {
|
||||
SchedulerClass klass;
|
||||
|
||||
MockItem(spg_t pgid, SchedulerClass k)
|
||||
: PGOpQueueable(pgid), klass(k) {}
|
||||
|
||||
std::ostream &print(std::ostream &rhs) const final { return rhs; }
|
||||
std::string print() const final { return std::string(); }
|
||||
std::optional<OpRequestRef> maybe_get_op() const final {
|
||||
return std::nullopt;
|
||||
}
|
||||
SchedulerClass get_scheduler_class() const final { return klass; }
|
||||
void run(OSD *osd, OSDShard *sdata, PGRef &pg,
|
||||
ThreadPool::TPHandle &handle) final {}
|
||||
};
|
||||
|
||||
struct StreamSpec {
|
||||
std::string name;
|
||||
SchedulerClass klass = SchedulerClass::client;
|
||||
int64_t pool = 1;
|
||||
uint64_t first_owner = 1; ///< owners model client sessions
|
||||
unsigned num_owners = 1;
|
||||
uint64_t op_size = 65536; ///< item cost in bytes
|
||||
unsigned priority = 63; ///< osd_client_op_priority default
|
||||
double offered_ops_per_sec = 0; ///< 0 => saturating
|
||||
unsigned backlog_per_owner = 64; ///< target queue depth when saturating
|
||||
};
|
||||
|
||||
struct StreamResult {
|
||||
uint64_t ops = 0;
|
||||
uint64_t bytes = 0;
|
||||
double share = 0; ///< of post-warmup dequeued bytes
|
||||
double mbps = 0;
|
||||
double p50_ms = 0;
|
||||
double p99_ms = 0;
|
||||
};
|
||||
|
||||
struct CellResult {
|
||||
std::map<std::string, StreamResult> streams;
|
||||
double total_mbps = 0;
|
||||
};
|
||||
|
||||
inline double percentile(std::vector<double> &v, double p)
|
||||
{
|
||||
if (v.empty()) {
|
||||
return 0;
|
||||
}
|
||||
std::sort(v.begin(), v.end());
|
||||
return v[std::min(v.size() - 1, static_cast<size_t>(p * v.size()))];
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one benchmark cell: a fresh scheduler of the given type, the
|
||||
* given streams, a simulated device draining rate_bytes_per_sec, for
|
||||
* duration_sec of wall time. Samples from the first warmup_sec are
|
||||
* discarded.
|
||||
*/
|
||||
inline CellResult run_cell(
|
||||
CephContext *cct,
|
||||
op_queue_type_t type,
|
||||
const std::vector<StreamSpec> &specs,
|
||||
const std::unordered_map<int64_t, bfq_stream_t> &pool_map,
|
||||
double rate_bytes_per_sec,
|
||||
double duration_sec,
|
||||
double warmup_sec)
|
||||
{
|
||||
using steady = std::chrono::steady_clock;
|
||||
|
||||
auto sched = make_scheduler(cct, 0, 1, 0, false /*is_rotational*/,
|
||||
"bluestore", type, 196 /*cutoff*/);
|
||||
if (type == op_queue_type_t::BfqScheduler) {
|
||||
auto pm = pool_map;
|
||||
static_cast<BfqScheduler*>(sched.get())->set_pool_streams(std::move(pm));
|
||||
}
|
||||
|
||||
std::unordered_map<uint64_t, size_t> owner_stream;
|
||||
std::unordered_map<uint64_t, unsigned> owner_queued;
|
||||
for (size_t i = 0; i < specs.size(); ++i) {
|
||||
for (unsigned o = 0; o < specs[i].num_owners; ++o) {
|
||||
owner_stream[specs[i].first_owner + o] = i;
|
||||
owner_queued[specs[i].first_owner + o] = 0;
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<uint64_t> bytes(specs.size(), 0), ops(specs.size(), 0);
|
||||
std::vector<std::vector<double>> lat(specs.size());
|
||||
std::vector<double> credit(specs.size(), 0.0);
|
||||
std::vector<unsigned> paced_rr(specs.size(), 0);
|
||||
epoch_t epoch = 1;
|
||||
|
||||
auto enqueue_one = [&](size_t si, uint64_t owner) {
|
||||
const auto &s = specs[si];
|
||||
sched->enqueue(OpSchedulerItem(
|
||||
std::make_unique<MockItem>(spg_t(pg_t(0, s.pool)), s.klass),
|
||||
static_cast<int>(s.op_size), s.priority, ceph_clock_now(), owner,
|
||||
epoch++));
|
||||
++owner_queued[owner];
|
||||
};
|
||||
|
||||
const auto t0 = steady::now();
|
||||
const auto t_warm =
|
||||
t0 + std::chrono::duration_cast<steady::duration>(
|
||||
std::chrono::duration<double>(warmup_sec));
|
||||
const auto t_end =
|
||||
t0 + std::chrono::duration_cast<steady::duration>(
|
||||
std::chrono::duration<double>(duration_sec));
|
||||
auto last = t0;
|
||||
double tokens = 0;
|
||||
const double burst_cap = rate_bytes_per_sec / 4;
|
||||
|
||||
while (true) {
|
||||
const auto now = steady::now();
|
||||
if (now >= t_end) {
|
||||
break;
|
||||
}
|
||||
const double dt = std::chrono::duration<double>(now - last).count();
|
||||
last = now;
|
||||
tokens = std::min(tokens + rate_bytes_per_sec * dt, burst_cap);
|
||||
|
||||
// load generators
|
||||
for (size_t i = 0; i < specs.size(); ++i) {
|
||||
const auto &s = specs[i];
|
||||
if (s.offered_ops_per_sec > 0) {
|
||||
credit[i] = std::min(credit[i] + s.offered_ops_per_sec * dt, 1000.0);
|
||||
while (credit[i] >= 1.0) {
|
||||
enqueue_one(i, s.first_owner + (paced_rr[i]++ % s.num_owners));
|
||||
credit[i] -= 1.0;
|
||||
}
|
||||
} else {
|
||||
for (unsigned o = 0; o < s.num_owners; ++o) {
|
||||
const uint64_t owner = s.first_owner + o;
|
||||
while (owner_queued[owner] < s.backlog_per_owner) {
|
||||
enqueue_one(i, owner);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// simulated device drain
|
||||
bool progressed = false;
|
||||
while (tokens > 0 && !sched->empty()) {
|
||||
WorkItem wi = sched->dequeue();
|
||||
if (std::holds_alternative<double>(wi)) {
|
||||
// mclock says "not ready until later"; let wall time advance
|
||||
break;
|
||||
}
|
||||
auto &item = std::get<OpSchedulerItem>(wi);
|
||||
const uint64_t owner = item.get_owner();
|
||||
const size_t si = owner_stream.at(owner);
|
||||
if (owner_queued[owner] > 0) {
|
||||
--owner_queued[owner];
|
||||
}
|
||||
tokens -= static_cast<double>(item.get_cost());
|
||||
progressed = true;
|
||||
if (now >= t_warm) {
|
||||
bytes[si] += item.get_cost();
|
||||
++ops[si];
|
||||
const utime_t sojourn = ceph_clock_now() - item.get_start_time();
|
||||
lat[si].push_back(sojourn.to_nsec() / 1e6);
|
||||
}
|
||||
}
|
||||
if (!progressed) {
|
||||
std::this_thread::sleep_for(std::chrono::microseconds(200));
|
||||
}
|
||||
}
|
||||
|
||||
CellResult result;
|
||||
const double measured_sec = duration_sec - warmup_sec;
|
||||
uint64_t total_bytes = 0;
|
||||
for (auto b : bytes) {
|
||||
total_bytes += b;
|
||||
}
|
||||
for (size_t i = 0; i < specs.size(); ++i) {
|
||||
StreamResult r;
|
||||
r.ops = ops[i];
|
||||
r.bytes = bytes[i];
|
||||
r.share = total_bytes ? static_cast<double>(bytes[i]) / total_bytes : 0;
|
||||
r.mbps = bytes[i] / 1e6 / measured_sec;
|
||||
r.p50_ms = percentile(lat[i], 0.50);
|
||||
r.p99_ms = percentile(lat[i], 0.99);
|
||||
result.streams[specs[i].name] = r;
|
||||
}
|
||||
result.total_mbps = total_bytes / 1e6 / measured_sec;
|
||||
return result;
|
||||
}
|
||||
|
||||
} // namespace scheduler_bench
|
||||
183
src/test/osd/vstart_bench_scheduler.sh
Executable file
183
src/test/osd/vstart_bench_scheduler.sh
Executable file
@ -0,0 +1,183 @@
|
||||
#!/usr/bin/env bash
|
||||
#
|
||||
# Tier-2 op scheduler isolation benchmark: drives real rados bench
|
||||
# workloads through a single-OSD vstart cluster, cycling osd_op_queue
|
||||
# across schedulers. The tier-1 in-process companion is
|
||||
# ceph_bench_op_scheduler; this script validates that dispatch-level
|
||||
# isolation survives the full OSD pipeline with BlueStore underneath.
|
||||
#
|
||||
# Run from a built build directory:
|
||||
# ../src/test/osd/vstart_bench_scheduler.sh
|
||||
#
|
||||
# Tunables (env):
|
||||
# SCHEDULERS default "wpq mclock_scheduler bfq"
|
||||
# SESSIONS aggressor session counts, default "1 4 16"
|
||||
# SECS victim measurement window per cell, default 12
|
||||
# BLOCK op size in bytes, default 65536
|
||||
# AGGR_QD queue depth per aggressor session, default 8
|
||||
# VICTIM_QD victim queue depth in share cells, default 16
|
||||
# THROTTLE_BYTES bluestore throttle, default 4194304; 0 disables
|
||||
#
|
||||
# THROTTLE_BYTES bounds BlueStore's in-flight window so that, on fast
|
||||
# devices, contention surfaces at the op queue where the scheduler can
|
||||
# act on it. Without it a fast NVMe absorbs the offered load below
|
||||
# the scheduler and every scheduler degenerates to FIFO passthrough --
|
||||
# the same reason the mclock tuning guide recommends constraining
|
||||
# bluestore_throttle_bytes.
|
||||
#
|
||||
# Scenarios per scheduler:
|
||||
# share_k<K> victim (rbd pool) saturating vs K saturating rgw
|
||||
# sessions; isolation == victim MB/s insensitive to K
|
||||
# latency_qd1 victim at queue depth 1 (latency probe) vs 8
|
||||
# saturating sessions; isolation == bounded avg/max lat
|
||||
#
|
||||
# Results: CSV + per-cell dump_op_pq_state snapshots under
|
||||
# ./scheduler-isolation-results/
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
BUILD_DIR=${BUILD_DIR:-$PWD}
|
||||
SCHEDULERS=${SCHEDULERS:-"wpq mclock_scheduler bfq"}
|
||||
SESSIONS=${SESSIONS:-"1 4 16"}
|
||||
SECS=${SECS:-12}
|
||||
BLOCK=${BLOCK:-65536}
|
||||
AGGR_QD=${AGGR_QD:-8}
|
||||
# The victim must be able to consume its fair share: by Little's law
|
||||
# it needs QD >= share * op_rate * latency, or it self-caps below any
|
||||
# scheduler's allocation and every scheduler looks identical.
|
||||
VICTIM_QD=${VICTIM_QD:-64}
|
||||
THROTTLE_BYTES=${THROTTLE_BYTES:-4194304}
|
||||
|
||||
cd "$BUILD_DIR"
|
||||
CEPH=./bin/ceph
|
||||
RADOS=./bin/rados
|
||||
if [ ! -x "$CEPH" ] || [ ! -x "$RADOS" ]; then
|
||||
echo "run from a built build directory (./bin/ceph missing)" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
RESULTS_DIR=$BUILD_DIR/scheduler-isolation-results
|
||||
CSV=$RESULTS_DIR/results.csv
|
||||
mkdir -p "$RESULTS_DIR"
|
||||
echo "scheduler,cell,victim_mbps,victim_avg_lat_s,victim_max_lat_s,aggr_mbps_total,victim_share" > "$CSV"
|
||||
|
||||
THROTTLE_OPTS=()
|
||||
if [ "$THROTTLE_BYTES" -gt 0 ]; then
|
||||
THROTTLE_OPTS=(-o "bluestore_throttle_bytes = $THROTTLE_BYTES"
|
||||
-o "bluestore_throttle_deferred_bytes = $THROTTLE_BYTES")
|
||||
fi
|
||||
|
||||
start_cluster() {
|
||||
local sched=$1
|
||||
../src/stop.sh >/dev/null 2>&1 || true
|
||||
if ! MON=1 OSD=1 MGR=1 MDS=0 RGW=0 ../src/vstart.sh -n --without-dashboard \
|
||||
-o "osd_op_queue = $sched" \
|
||||
-o "osd_op_num_shards = 1" \
|
||||
-o "osd_pool_default_size = 1" \
|
||||
-o "osd_pool_default_min_size = 1" \
|
||||
-o "mon_allow_pool_size_one = true" \
|
||||
-o "mon_allow_pool_delete = true" \
|
||||
"${THROTTLE_OPTS[@]}" \
|
||||
> "$RESULTS_DIR/vstart_$sched.log" 2>&1; then
|
||||
echo "vstart failed for $sched, see $RESULTS_DIR/vstart_$sched.log" >&2
|
||||
exit 1
|
||||
fi
|
||||
$CEPH osd set noscrub >/dev/null 2>&1
|
||||
$CEPH osd set nodeep-scrub >/dev/null 2>&1
|
||||
local active
|
||||
active=$($CEPH daemon osd.0 config get osd_op_queue 2>/dev/null |
|
||||
python3 -c 'import json,sys; print(json.load(sys.stdin)["osd_op_queue"])')
|
||||
if [ "$active" != "$sched" ]; then
|
||||
echo "expected osd_op_queue=$sched, got '$active'" >&2
|
||||
exit 1
|
||||
fi
|
||||
}
|
||||
|
||||
wait_clean() {
|
||||
for _ in $(seq 90); do
|
||||
if $CEPH pg stat --format json 2>/dev/null | python3 -c '
|
||||
import json, sys
|
||||
states = json.load(sys.stdin)["pg_summary"]["num_pg_by_state"]
|
||||
total = sum(s["num"] for s in states)
|
||||
clean = sum(s["num"] for s in states if s["name"] == "active+clean")
|
||||
sys.exit(0 if total > 0 and total == clean else 1)'; then
|
||||
return 0
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
echo "timed out waiting for active+clean pgs" >&2
|
||||
return 1
|
||||
}
|
||||
|
||||
make_pools() {
|
||||
$CEPH osd pool create victim 32 >/dev/null 2>&1
|
||||
$CEPH osd pool create aggr 32 >/dev/null 2>&1
|
||||
$CEPH osd pool application enable victim rbd >/dev/null 2>&1
|
||||
$CEPH osd pool application enable aggr rgw >/dev/null 2>&1
|
||||
wait_clean
|
||||
}
|
||||
|
||||
destroy_pools() {
|
||||
$CEPH osd pool rm victim victim --yes-i-really-really-mean-it >/dev/null 2>&1
|
||||
$CEPH osd pool rm aggr aggr --yes-i-really-really-mean-it >/dev/null 2>&1
|
||||
}
|
||||
|
||||
# run_cell <scheduler> <aggr_sessions> <victim_qd> <label>
|
||||
run_cell() {
|
||||
local sched=$1 k=$2 victim_qd=$3 label=$4
|
||||
local tmp="$RESULTS_DIR/${sched}_${label}"
|
||||
mkdir -p "$tmp"
|
||||
make_pools
|
||||
|
||||
local pids=()
|
||||
for i in $(seq 1 "$k"); do
|
||||
$RADOS -p aggr bench $((SECS + 25)) write -b "$BLOCK" -t "$AGGR_QD" \
|
||||
--no-cleanup > "$tmp/aggr_$i.log" 2>&1 &
|
||||
pids+=($!)
|
||||
done
|
||||
sleep 4 # let the aggressor backlog build
|
||||
|
||||
$RADOS -p victim bench "$SECS" write -b "$BLOCK" -t "$victim_qd" \
|
||||
--no-cleanup > "$tmp/victim.log" 2>&1
|
||||
|
||||
# snapshot scheduler state while the aggressors are still running
|
||||
$CEPH daemon osd.0 dump_op_pq_state > "$tmp/pq_state.json" 2>/dev/null || true
|
||||
|
||||
kill "${pids[@]}" >/dev/null 2>&1 || true
|
||||
wait >/dev/null 2>&1 || true
|
||||
|
||||
local vbw vavg vmax abw share
|
||||
vbw=$(awk '/^Bandwidth \(MB\/sec\)/ {print $3}' "$tmp/victim.log")
|
||||
vavg=$(awk '/^Average Latency\(s\)/ {print $3}' "$tmp/victim.log")
|
||||
vmax=$(awk '/^Max latency\(s\)/ {print $3}' "$tmp/victim.log")
|
||||
# killed benches never print a final summary; use each aggressor's
|
||||
# last per-second progress line ("avg MB/s" column). Approximate:
|
||||
# their window is slightly wider than the victim's.
|
||||
abw=$(for f in "$tmp"/aggr_*.log; do
|
||||
awk '$1 ~ /^[0-9]+$/ && $5 ~ /^[0-9.]+$/ {last=$5}
|
||||
END {if (last) print last}' "$f"
|
||||
done | awk '{s += $1} END {printf "%.1f", s}')
|
||||
share=$(awk -v v="${vbw:-0}" -v a="${abw:-0}" \
|
||||
'BEGIN {t = v + a; printf "%.3f", (t > 0 ? v / t : 0)}')
|
||||
|
||||
echo "$sched,$label,${vbw:-0},${vavg:-0},${vmax:-0},${abw:-0},$share" >> "$CSV"
|
||||
printf ' %-16s %-12s victim %8s MB/s share %-6s avg %8ss max %8ss (aggr ~%s MB/s)\n' \
|
||||
"$sched" "$label" "${vbw:-?}" "$share" "${vavg:-?}" "${vmax:-?}" "${abw:-?}"
|
||||
|
||||
destroy_pools
|
||||
}
|
||||
|
||||
echo "device: single vstart OSD, 1 op shard, ${BLOCK}B writes"
|
||||
for sched in $SCHEDULERS; do
|
||||
echo "=== $sched ==="
|
||||
start_cluster "$sched"
|
||||
for k in $SESSIONS; do
|
||||
run_cell "$sched" "$k" "$VICTIM_QD" "share_k$k"
|
||||
done
|
||||
run_cell "$sched" 8 1 "latency_qd1"
|
||||
done
|
||||
../src/stop.sh >/dev/null 2>&1 || true
|
||||
|
||||
echo
|
||||
echo "results: $CSV"
|
||||
column -s, -t < "$CSV"
|
||||
Loading…
Reference in New Issue
Block a user
