mirror of
https://github.com/ceph/ceph
synced 2026-08-02 07:03:18 +00:00
Merge a225eac39f into eed82023c5
This commit is contained in:
Kyle Bader
GitHub
commit
8d52167227
@ -209,6 +209,25 @@ if(WITH_RDMA)
|
||||
set(HAVE_RDMA TRUE)
|
||||
endif()
|
||||
|
||||
# PSP Security Protocol (kernel inline AEAD for messenger zero-copy
|
||||
# secure mode, kernel v6.18+ on CX-7+/mlx5). For now this ships
|
||||
# header detection plus a no-op stub backend only.
|
||||
option(WITH_PSP "Enable PSP Security Protocol in async messenger" ON)
|
||||
if(WITH_PSP AND LINUX)
|
||||
include(CheckIncludeFile)
|
||||
# PSP_HEADER_FOUND is the cached probe result; HAVE_PSP is a plain
|
||||
# variable derived from it every configure, so that turning WITH_PSP
|
||||
# off actually turns the feature off instead of leaving a stale
|
||||
# cache entry asserting the headers are present.
|
||||
check_include_file("linux/psp.h" PSP_HEADER_FOUND)
|
||||
if(PSP_HEADER_FOUND)
|
||||
set(HAVE_PSP TRUE)
|
||||
else()
|
||||
message(STATUS "WITH_PSP=ON but linux/psp.h not found; "
|
||||
"PSP support will not be built")
|
||||
endif()
|
||||
endif()
|
||||
|
||||
find_package(Backtrace)
|
||||
|
||||
option(WITH_RBD "Enable RADOS Block Device related targets" ON)
|
||||
|
||||
@ -1,3 +1,19 @@
|
||||
* Messenger: new opt-in msgr2 connection mode ``secure-psp``, accepted by
|
||||
the ``ms_*_mode`` options alongside ``crc`` and ``secure``. It is absent
|
||||
from all defaults and currently behaves identically to ``secure``.
|
||||
NOTE: daemons older than this release do not understand the name and
|
||||
will parse a mode list containing it as *empty*, which makes them
|
||||
refuse all msgr2 connections. Because the ``ms_*_mode`` options are
|
||||
read at startup, do not set ``secure-psp`` in the cluster config until
|
||||
every daemon has been upgraded, and always keep a known mode such as
|
||||
``secure`` or ``crc`` in the list.
|
||||
|
||||
* Messenger: new ``ms_tcp_zerocopy`` option (default ``false``) makes the
|
||||
async Posix stack send large plaintext payloads with ``MSG_ZEROCOPY``,
|
||||
eliminating the kernel send copy. ``ms_tcp_zerocopy_min_size``
|
||||
(default 16K) sets the per-send threshold. Only crc-mode connections
|
||||
are eligible; secure mode already copies during encryption.
|
||||
|
||||
* CephFS: The ``client_force_lazyio`` configuration option is now correctly marked
|
||||
as not supporting runtime updates. Previously, the configuration schema indicated
|
||||
this option could be changed at runtime, but changes had no effect on opened file
|
||||
|
||||
@ -93,7 +93,7 @@ Similarly, two options control whether IPv4 and IPv6 addresses are used:
|
||||
Connection modes
|
||||
----------------
|
||||
|
||||
The v2 protocol supports two connection modes:
|
||||
The v2 protocol supports the following connection modes:
|
||||
|
||||
* *crc* mode provides:
|
||||
|
||||
@ -124,6 +124,24 @@ The v2 protocol supports two connection modes:
|
||||
which is generally very fast on modern processors (e.g., faster than
|
||||
a SHA-256 cryptographic hash).
|
||||
|
||||
* *secure-psp* mode provides the same guarantees as *secure*, and is
|
||||
intended to carry them with the encryption offloaded to the NIC via
|
||||
the Linux PSP Security Protocol. It is opt-in, is not present in any
|
||||
default mode list, and currently behaves identically to *secure*
|
||||
(the AES-GCM work is still done in process); the kernel offload
|
||||
lands in later work.
|
||||
|
||||
.. warning::
|
||||
|
||||
Daemons older than the release that introduced *secure-psp* do not
|
||||
recognize the name. Such a daemon parses a mode list containing it
|
||||
as an *empty* list and will then refuse every msgr2 connection.
|
||||
Because the mode options are read at startup and may live in the
|
||||
monitor configuration database, this can lock old monitors out of
|
||||
quorum. Do not configure *secure-psp* until every daemon is
|
||||
upgraded, and always leave a mode such as *secure* or *crc* in the
|
||||
list.
|
||||
|
||||
Connection mode configuration options
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
|
||||
@ -130,8 +130,30 @@ struct AuthConnectionMeta {
|
||||
bool is_mode_crc() const {
|
||||
return con_mode == CEPH_CON_MODE_CRC;
|
||||
}
|
||||
// Is this connection confidential? Governs policy that cares only
|
||||
// that the payload is encrypted (e.g. compression-vs-secure, and
|
||||
// AuthRegistry::is_secure_mode()). secure-psp qualifies.
|
||||
bool is_mode_secure() const {
|
||||
return con_mode == CEPH_CON_MODE_SECURE;
|
||||
return con_mode == CEPH_CON_MODE_SECURE ||
|
||||
con_mode == CEPH_CON_MODE_SECURE_PSP;
|
||||
}
|
||||
bool is_mode_secure_psp() const {
|
||||
return con_mode == CEPH_CON_MODE_SECURE_PSP;
|
||||
}
|
||||
// Does *this process* encrypt each frame itself? Governs the data
|
||||
// path: crypto_onwire handler setup and the MSG_ZEROCOPY exclusion
|
||||
// (in-process AEAD already copies the payload, so there is no copy
|
||||
// left to elide).
|
||||
//
|
||||
// secure-psp answers true here for now because it still runs the
|
||||
// aesgcm path. When the netlink attach lands it stays confidential
|
||||
// but stops doing in-process AEAD - i.e. ONLY this predicate flips,
|
||||
// and its callers are exactly the ones that must change. Keeping it
|
||||
// separate from is_mode_secure() is what makes that a safe edit
|
||||
// rather than a silent downgrade of the compression policy.
|
||||
bool is_mode_in_process_aead() const {
|
||||
return con_mode == CEPH_CON_MODE_SECURE ||
|
||||
con_mode == CEPH_CON_MODE_SECURE_PSP;
|
||||
}
|
||||
|
||||
CryptoKey session_key; ///< per-ticket key
|
||||
@ -141,6 +163,7 @@ struct AuthConnectionMeta {
|
||||
case CEPH_CON_MODE_CRC:
|
||||
return 0;
|
||||
case CEPH_CON_MODE_SECURE:
|
||||
case CEPH_CON_MODE_SECURE_PSP:
|
||||
return 16 * 4;
|
||||
}
|
||||
return 0;
|
||||
|
||||
@ -101,6 +101,8 @@ void AuthRegistry::_parse_mode_list(const string& s,
|
||||
v->push_back(CEPH_CON_MODE_CRC);
|
||||
} else if (i == "secure") {
|
||||
v->push_back(CEPH_CON_MODE_SECURE);
|
||||
} else if (i == "secure-psp") {
|
||||
v->push_back(CEPH_CON_MODE_SECURE_PSP);
|
||||
} else {
|
||||
lderr(cct) << "WARNING: unknown connection mode " << i << dendl;
|
||||
}
|
||||
|
||||
@ -66,7 +66,8 @@ public:
|
||||
}
|
||||
|
||||
static bool is_secure_mode(uint32_t mode) {
|
||||
return (mode == CEPH_CON_MODE_SECURE);
|
||||
return (mode == CEPH_CON_MODE_SECURE ||
|
||||
mode == CEPH_CON_MODE_SECURE_PSP);
|
||||
}
|
||||
|
||||
AuthAuthorizeHandler *get_handler(int peer_type, int method);
|
||||
|
||||
@ -25,6 +25,7 @@ const char *ceph_con_mode_name(int con_mode)
|
||||
case CEPH_CON_MODE_UNKNOWN: return "unknown";
|
||||
case CEPH_CON_MODE_CRC: return "crc";
|
||||
case CEPH_CON_MODE_SECURE: return "secure";
|
||||
case CEPH_CON_MODE_SECURE_PSP: return "secure-psp";
|
||||
default: return "???";
|
||||
}
|
||||
}
|
||||
|
||||
@ -959,7 +959,7 @@ options:
|
||||
- name: ms_mon_cluster_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Connection modes (crc, secure) for intra-mon connections in order of preference
|
||||
desc: Connection modes (crc, secure, secure-psp) for intra-mon connections in order of preference
|
||||
fmt_desc: the connection mode (or permitted modes) to use between monitors.
|
||||
default: secure crc
|
||||
see_also:
|
||||
@ -973,7 +973,7 @@ options:
|
||||
- name: ms_mon_service_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Allowed connection modes (crc, secure) for connections to mons
|
||||
desc: Allowed connection modes (crc, secure, secure-psp) for connections to mons
|
||||
fmt_desc: A list of permitted modes for clients or
|
||||
other Ceph daemons to use when connecting to monitors.
|
||||
default: secure crc
|
||||
@ -988,7 +988,7 @@ options:
|
||||
- name: ms_mon_client_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Connection modes (crc, secure) for connections from clients to monitors in
|
||||
desc: Connection modes (crc, secure, secure-psp) for connections from clients to monitors in
|
||||
order of preference
|
||||
fmt_desc: A list of connection modes, in order of
|
||||
preference, for clients or non-monitor daemons to use when
|
||||
@ -1005,7 +1005,7 @@ options:
|
||||
- name: ms_cluster_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Connection modes (crc, secure) for intra-cluster connections in order of preference
|
||||
desc: Connection modes (crc, secure, secure-psp) for intra-cluster connections in order of preference
|
||||
fmt_desc: connection mode (or permitted modes) used
|
||||
for intra-cluster communication between Ceph daemons. If multiple
|
||||
modes are listed, the modes listed first are preferred.
|
||||
@ -1018,7 +1018,7 @@ options:
|
||||
- name: ms_service_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Allowed connection modes (crc, secure) for connections to daemons
|
||||
desc: Allowed connection modes (crc, secure, secure-psp) for connections to daemons
|
||||
fmt_desc: a list of permitted modes for clients to use
|
||||
when connecting to the cluster.
|
||||
default: crc secure
|
||||
@ -1030,7 +1030,7 @@ options:
|
||||
- name: ms_client_mode
|
||||
type: str
|
||||
level: basic
|
||||
desc: Connection modes (crc, secure) for connections from clients in order of preference
|
||||
desc: Connection modes (crc, secure, secure-psp) for connections from clients in order of preference
|
||||
fmt_desc: A list of connection modes, in order of
|
||||
preference, for clients to use (or allow) when talking to a Ceph
|
||||
cluster.
|
||||
@ -1126,6 +1126,38 @@ options:
|
||||
desc: Maximum amount of data to prefetch out of the socket receive buffer
|
||||
default: 64_K
|
||||
with_legacy: true
|
||||
- name: ms_tcp_zerocopy
|
||||
type: bool
|
||||
level: advanced
|
||||
desc: Use MSG_ZEROCOPY for large plaintext sends on the Posix messenger stack
|
||||
long_desc: When enabled, the Posix stack opts each TCP socket into
|
||||
SO_ZEROCOPY at connect/accept and sends data segments at or above
|
||||
ms_tcp_zerocopy_min_size with MSG_ZEROCOPY, eliminating the kernel
|
||||
send copy. Only applies to crc-mode connections (secure mode already
|
||||
copies during encryption) and only to connections established after
|
||||
the option is set. Transparent fallback to a normal copying send on
|
||||
ENOBUFS or when the kernel/socket does not support it. Disabled by
|
||||
default; the benefit is only realized over a real NIC.
|
||||
default: false
|
||||
see_also:
|
||||
- ms_tcp_zerocopy_min_size
|
||||
with_legacy: true
|
||||
- name: ms_tcp_zerocopy_min_size
|
||||
type: size
|
||||
level: advanced
|
||||
desc: Minimum per-send size to use MSG_ZEROCOPY (below this, send copies)
|
||||
long_desc: MSG_ZEROCOPY pins user pages and incurs a per-send
|
||||
completion notification; below this threshold the copy is cheaper
|
||||
than the pinning plus notification overhead, so small sends (control
|
||||
frames, small replies) always use the normal copying path. Only
|
||||
consulted when ms_tcp_zerocopy is enabled. Very low values are
|
||||
counterproductive - they pin control frames and handshake traffic
|
||||
for no bandwidth gain.
|
||||
default: 16_K
|
||||
min: 1_K
|
||||
see_also:
|
||||
- ms_tcp_zerocopy
|
||||
with_legacy: true
|
||||
- name: ms_initial_backoff
|
||||
type: float
|
||||
level: advanced
|
||||
|
||||
@ -103,9 +103,10 @@ struct ceph_dir_layout {
|
||||
#define CEPH_AUTH_CEPHX 0x2
|
||||
|
||||
/* msgr2 protocol modes */
|
||||
#define CEPH_CON_MODE_UNKNOWN 0x0
|
||||
#define CEPH_CON_MODE_CRC 0x1
|
||||
#define CEPH_CON_MODE_SECURE 0x2
|
||||
#define CEPH_CON_MODE_UNKNOWN 0x0
|
||||
#define CEPH_CON_MODE_CRC 0x1
|
||||
#define CEPH_CON_MODE_SECURE 0x2
|
||||
#define CEPH_CON_MODE_SECURE_PSP 0x3
|
||||
|
||||
extern const char *ceph_con_mode_name(int con_mode);
|
||||
|
||||
|
||||
@ -121,6 +121,9 @@
|
||||
/* ibverbs experimental conditional compilation */
|
||||
#cmakedefine HAVE_IBV_EXP
|
||||
|
||||
/* AsyncMessenger PSP Security Protocol conditional compilation */
|
||||
#cmakedefine HAVE_PSP
|
||||
|
||||
/* define if bluestore enabled */
|
||||
#cmakedefine WITH_BLUESTORE
|
||||
|
||||
|
||||
@ -21,6 +21,14 @@ list(APPEND msg_srcs
|
||||
async/frames_v2.cc
|
||||
async/net_handler.cc)
|
||||
|
||||
# PSPNetlinkMock.cc is deliberately NOT here: it is a test double and
|
||||
# belongs only to the unit test that uses it, not to ceph-common (and
|
||||
# therefore to every daemon and every librados client).
|
||||
if(HAVE_PSP)
|
||||
list(APPEND msg_srcs
|
||||
async/PSPNetlink.cc)
|
||||
endif()
|
||||
|
||||
if(LINUX)
|
||||
list(APPEND msg_srcs
|
||||
async/EventEpoll.cc)
|
||||
|
||||
@ -336,11 +336,21 @@ ssize_t AsyncConnection::_try_send(bool more)
|
||||
}
|
||||
|
||||
ceph_assert(center->in_thread());
|
||||
|
||||
// Drive MSG_ERRQUEUE completion processing so pinned buffers
|
||||
// retire (the socket self-credits the zerocopy perfcounters at
|
||||
// the event site; no-op unless zero-copy engaged).
|
||||
if (cs)
|
||||
cs.drain_zerocopy_completions();
|
||||
|
||||
ldout(async_msgr->cct, 25) << __func__ << " cs.send " << outgoing_bl.length()
|
||||
<< " bytes" << dendl;
|
||||
// network block would make ::send return EAGAIN, that would make here looks
|
||||
// like do not call cs.send() and r = 0
|
||||
ssize_t r = 0;
|
||||
// Capture the scatter/gather vector size before cs.send(), which
|
||||
// splices the sent prefix off outgoing_bl.
|
||||
const unsigned iov_segs = outgoing_bl.get_num_buffers();
|
||||
if (likely(!inject_network_congestion())) {
|
||||
r = cs.send(outgoing_bl, more);
|
||||
}
|
||||
@ -348,6 +358,16 @@ ssize_t AsyncConnection::_try_send(bool more)
|
||||
ldout(async_msgr->cct, 1) << __func__ << " send error: " << cpp_strerror(r) << dendl;
|
||||
return r;
|
||||
}
|
||||
if (r > 0) {
|
||||
// The socket pinned `zc` bytes for MSG_ZEROCOPY (and counted
|
||||
// submitted/zerocopy/pinned itself); the remainder went through
|
||||
// the kernel copy and is accounted here.
|
||||
const size_t zc = cs.last_send_zerocopy_bytes();
|
||||
const size_t cp = (size_t)r > zc ? (size_t)r - zc : 0;
|
||||
if (cp)
|
||||
logger->inc(l_msgr_send_bytes_copied, cp);
|
||||
logger->hinc(l_msgr_send_iov_segments, iov_segs, r);
|
||||
}
|
||||
|
||||
ldout(async_msgr->cct, 10) << __func__ << " sent bytes " << r
|
||||
<< " remaining bytes " << outgoing_bl.length() << dendl;
|
||||
@ -390,6 +410,14 @@ void AsyncConnection::process() {
|
||||
|
||||
ldout(async_msgr->cct, 20) << __func__ << dendl;
|
||||
|
||||
// Retire zero-copy pins here as well as in _try_send: a connection
|
||||
// that has stopped sending would otherwise hold its last payload
|
||||
// until it closes. Pending error-queue data also raises EPOLLERR,
|
||||
// which lands us here, so this is where that wakeup gets consumed.
|
||||
// No-op (no syscall) unless something is actually pinned.
|
||||
if (cs)
|
||||
cs.drain_zerocopy_completions();
|
||||
|
||||
switch (state) {
|
||||
case STATE_NONE: {
|
||||
ldout(async_msgr->cct, 20) << __func__ << " enter none state" << dendl;
|
||||
@ -656,6 +684,10 @@ void AsyncConnection::shutdown_socket() {
|
||||
}
|
||||
if (cs) {
|
||||
center->delete_file_event(cs.fd(), EVENT_READABLE | EVENT_WRITABLE);
|
||||
// cs.close() drains, then credits any residual pinned
|
||||
// sends as completed and zeroes the pinned gauge itself (it holds
|
||||
// the Worker logger), so submitted == completed and pinned == 0
|
||||
// hold exactly on every close path - no reconciliation needed here.
|
||||
cs.shutdown();
|
||||
cs.close();
|
||||
}
|
||||
|
||||
44
src/msg/async/PSPNetlink.cc
Normal file
44
src/msg/async/PSPNetlink.cc
Normal file
@ -0,0 +1,44 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 smarttab
|
||||
//
|
||||
// No-op stub. Always reports psp_supported=false.
|
||||
// Follow-on work replaces make_real_backend() with a libmnl/libnl-genl
|
||||
// implementation gated on HAVE_PSP.
|
||||
|
||||
#include "PSPNetlink.h"
|
||||
|
||||
#include <cerrno>
|
||||
|
||||
namespace ceph::msgr::psp {
|
||||
|
||||
namespace {
|
||||
|
||||
class StubBackend : public NetlinkBackend {
|
||||
public:
|
||||
std::optional<DeviceCaps> get_dev_caps(int /*sock_fd*/) override {
|
||||
return DeviceCaps{}; // psp_supported=false (the field default)
|
||||
}
|
||||
std::optional<std::vector<uint8_t>>
|
||||
alloc_rx_assoc(int /*sock_fd*/) override {
|
||||
return std::nullopt;
|
||||
}
|
||||
int install_tx_assoc(int /*sock_fd*/,
|
||||
std::span<const uint8_t> /*wrapped*/) override {
|
||||
return -ENOSYS;
|
||||
}
|
||||
int teardown(int /*sock_fd*/) override {
|
||||
return 0;
|
||||
}
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
std::unique_ptr<NetlinkBackend> make_real_backend() {
|
||||
return std::make_unique<StubBackend>();
|
||||
}
|
||||
|
||||
// make_mock_backend() is intentionally undefined here; it lands
|
||||
// in PSPNetlinkMock.cc. Linking a unit test that references it
|
||||
// will fail until then.
|
||||
|
||||
} // namespace ceph::msgr::psp
|
||||
89
src/msg/async/PSPNetlink.h
Normal file
89
src/msg/async/PSPNetlink.h
Normal file
@ -0,0 +1,89 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 smarttab
|
||||
//
|
||||
// Linux PSP Security Protocol netlink wrapper.
|
||||
//
|
||||
// This ships the interface + a no-op default backend that always
|
||||
// reports psp_supported=false. The mock backend for CI tests lands
|
||||
// in the next commit; the real netlink-backed implementation (gated
|
||||
// on HAVE_PSP) comes in follow-on work.
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <cstdint>
|
||||
#include <memory>
|
||||
#include <optional>
|
||||
#include <span>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
namespace ceph::msgr::psp {
|
||||
|
||||
struct DeviceCaps {
|
||||
std::string ifname;
|
||||
bool psp_supported = false;
|
||||
uint32_t versions_mask = 0; // AEAD-version bitmask (NIC-defined)
|
||||
uint32_t free_tx_assoc_slots = 0; // capacity hint; 0 if not exposed
|
||||
uint32_t free_rx_assoc_slots = 0;
|
||||
};
|
||||
|
||||
// Backend abstraction. CI swaps in a mock; production will swap in
|
||||
// a real libmnl/libnl-genl impl. Today only the stub exists, which
|
||||
// reports psp_supported=false from every call so the rest of the
|
||||
// messenger can develop against the interface without committing
|
||||
// to behavior.
|
||||
class NetlinkBackend {
|
||||
public:
|
||||
virtual ~NetlinkBackend() = default;
|
||||
|
||||
// Query PSP capability for the interface backing this socket.
|
||||
// A backend that does not support PSP returns DeviceCaps with
|
||||
// psp_supported=false (NOT std::nullopt). std::nullopt is
|
||||
// reserved for hard error.
|
||||
virtual std::optional<DeviceCaps> get_dev_caps(int sock_fd) = 0;
|
||||
|
||||
// Allocate a rx-direction PSP key for sock_fd; returns the
|
||||
// wrapped-key blob the peer will install via install_tx_assoc.
|
||||
// The blob is opaque to Ceph (kernel/NIC-defined size + layout)
|
||||
// and must round-trip verbatim through the in-band msgr2
|
||||
// handshake.
|
||||
virtual std::optional<std::vector<uint8_t>>
|
||||
alloc_rx_assoc(int sock_fd) = 0;
|
||||
|
||||
// Install a peer-supplied wrapped-key blob as the tx-direction
|
||||
// PSP key for sock_fd. Returns 0 on success, -errno on failure;
|
||||
// -ENOSYS indicates the backend does not support PSP.
|
||||
virtual int install_tx_assoc(int sock_fd,
|
||||
std::span<const uint8_t> wrapped) = 0;
|
||||
|
||||
// Best-effort cleanup of PSP state bound to sock_fd. Always
|
||||
// succeeds at the API level (the kernel may still need to GC).
|
||||
virtual int teardown(int sock_fd) = 0;
|
||||
};
|
||||
|
||||
// Production backend factory. Currently returns a stub that reports
|
||||
// unsupported; the real implementation will make netlink calls when
|
||||
// HAVE_PSP is defined.
|
||||
std::unique_ptr<NetlinkBackend> make_real_backend();
|
||||
|
||||
// Knobs for the CI-only mock backend. The mock is deterministic
|
||||
// and stateful per-instance: each MockBackend acts as one "host"
|
||||
// with its own tx/rx slot accounting. Failure-injection counters
|
||||
// decrement on use so tests can script "next N calls fail."
|
||||
struct MockConfig {
|
||||
std::string ifname = "mock0";
|
||||
bool psp_supported = true;
|
||||
uint32_t versions_mask = 0x1; // AES-GCM-128 by default
|
||||
uint32_t tx_capacity = 64;
|
||||
uint32_t rx_capacity = 64;
|
||||
|
||||
// Failure injection (decrement on use). 0 = no failures.
|
||||
uint32_t fail_next_alloc_rx_assoc = 0;
|
||||
uint32_t fail_next_install_tx_assoc = 0;
|
||||
bool fail_get_dev_caps = false;
|
||||
};
|
||||
|
||||
|
||||
std::unique_ptr<NetlinkBackend> make_mock_backend(const MockConfig&);
|
||||
|
||||
} // namespace ceph::msgr::psp
|
||||
137
src/msg/async/PSPNetlinkMock.cc
Normal file
137
src/msg/async/PSPNetlinkMock.cc
Normal file
@ -0,0 +1,137 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 smarttab
|
||||
//
|
||||
// Deterministic mock backend for CI.
|
||||
//
|
||||
// Each MockBackend instance maintains its own slot accounting and
|
||||
// is NOT thread-safe (single-threaded unit-test consumption only).
|
||||
// Two MockBackend instances act as independent peers in the
|
||||
// two-peer rendezvous test: peer A's alloc_rx_assoc blob is
|
||||
// passed by the test harness to peer B's install_tx_assoc.
|
||||
|
||||
#include "PSPNetlink.h"
|
||||
|
||||
#include <cerrno>
|
||||
#include <cstring>
|
||||
#include <map>
|
||||
#include <set>
|
||||
|
||||
namespace ceph::msgr::psp {
|
||||
|
||||
namespace {
|
||||
|
||||
// Wrapped-key blob layout the mock emits and validates.
|
||||
// 16 bytes, fully deterministic: magic + version + seq + fd + pad.
|
||||
// Real NICs emit opaque blobs of vendor-defined size; this is
|
||||
// pure bookkeeping for the mock.
|
||||
constexpr size_t kBlobSize = 16;
|
||||
|
||||
class MockBackend : public NetlinkBackend {
|
||||
public:
|
||||
explicit MockBackend(MockConfig cfg) : cfg_(std::move(cfg)) {}
|
||||
|
||||
std::optional<DeviceCaps> get_dev_caps(int /*sock_fd*/) override {
|
||||
if (cfg_.fail_get_dev_caps) return std::nullopt;
|
||||
return DeviceCaps{
|
||||
.ifname = cfg_.ifname,
|
||||
.psp_supported = cfg_.psp_supported,
|
||||
.versions_mask = cfg_.versions_mask,
|
||||
.free_tx_assoc_slots = cfg_.tx_capacity - tx_used(),
|
||||
.free_rx_assoc_slots = cfg_.rx_capacity - rx_used(),
|
||||
};
|
||||
}
|
||||
|
||||
std::optional<std::vector<uint8_t>>
|
||||
alloc_rx_assoc(int sock_fd) override {
|
||||
if (!cfg_.psp_supported) return std::nullopt;
|
||||
if (cfg_.fail_next_alloc_rx_assoc) {
|
||||
--cfg_.fail_next_alloc_rx_assoc;
|
||||
return std::nullopt;
|
||||
}
|
||||
if (rx_used() >= cfg_.rx_capacity) return std::nullopt;
|
||||
rx_fds_.insert(sock_fd);
|
||||
return make_blob(sock_fd);
|
||||
}
|
||||
|
||||
int install_tx_assoc(int sock_fd,
|
||||
std::span<const uint8_t> wrapped) override {
|
||||
if (!cfg_.psp_supported) return -ENOSYS;
|
||||
if (cfg_.fail_next_install_tx_assoc) {
|
||||
--cfg_.fail_next_install_tx_assoc;
|
||||
// Distinct from -ENOSPC so tests can tell an injected kernel
|
||||
// rejection from genuine capacity exhaustion.
|
||||
return -EIO;
|
||||
}
|
||||
if (!validate_blob(wrapped)) return -EINVAL;
|
||||
// A peer must install the *other* side's rx key. Installing a blob
|
||||
// this instance issued means the handshake crossed its wires, so
|
||||
// reject it rather than silently modelling a working association.
|
||||
if (blob_issuer(wrapped) == instance_id_) return -EINVAL;
|
||||
if (tx_used() >= cfg_.tx_capacity) return -ENOSPC;
|
||||
installed_tx_[sock_fd] =
|
||||
std::vector<uint8_t>(wrapped.begin(), wrapped.end());
|
||||
return 0;
|
||||
}
|
||||
|
||||
int teardown(int sock_fd) override {
|
||||
installed_tx_.erase(sock_fd);
|
||||
rx_fds_.erase(sock_fd);
|
||||
return 0;
|
||||
}
|
||||
|
||||
private:
|
||||
uint32_t tx_used() const {
|
||||
return static_cast<uint32_t>(installed_tx_.size());
|
||||
}
|
||||
uint32_t rx_used() const {
|
||||
return static_cast<uint32_t>(rx_fds_.size());
|
||||
}
|
||||
|
||||
std::vector<uint8_t> make_blob(int sock_fd) {
|
||||
// magic(4) version(2) issuer(2) seq(4) fd(4) == 16
|
||||
std::vector<uint8_t> b(kBlobSize, 0);
|
||||
b[0] = 'P'; b[1] = 'S'; b[2] = 'P'; b[3] = 'M';
|
||||
const uint16_t ver = 1;
|
||||
const uint16_t issuer = instance_id_;
|
||||
const uint32_t seq = ++seq_;
|
||||
const uint32_t fd = static_cast<uint32_t>(sock_fd);
|
||||
std::memcpy(&b[4], &ver, sizeof(ver));
|
||||
std::memcpy(&b[6], &issuer, sizeof(issuer));
|
||||
std::memcpy(&b[8], &seq, sizeof(seq));
|
||||
std::memcpy(&b[12], &fd, sizeof(fd));
|
||||
return b;
|
||||
}
|
||||
static uint16_t blob_issuer(std::span<const uint8_t> w) {
|
||||
uint16_t issuer = 0;
|
||||
std::memcpy(&issuer, w.data() + 6, sizeof(issuer));
|
||||
return issuer;
|
||||
}
|
||||
static bool validate_blob(std::span<const uint8_t> w) {
|
||||
if (w.size() != kBlobSize)
|
||||
return false;
|
||||
if (!(w[0] == 'P' && w[1] == 'S' && w[2] == 'P' && w[3] == 'M'))
|
||||
return false;
|
||||
uint16_t ver = 0;
|
||||
std::memcpy(&ver, w.data() + 4, sizeof(ver));
|
||||
return ver == 1;
|
||||
}
|
||||
|
||||
MockConfig cfg_;
|
||||
// Identifies which mock instance issued a blob, so a peer can tell
|
||||
// its own rx key from the far side's. Deterministic: instances are
|
||||
// numbered in construction order (single-threaded test use only).
|
||||
static inline uint16_t next_instance_id_ = 0;
|
||||
const uint16_t instance_id_ = ++next_instance_id_;
|
||||
uint32_t seq_ = 0;
|
||||
std::set<int> rx_fds_;
|
||||
std::map<int, std::vector<uint8_t>> installed_tx_;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
std::unique_ptr<NetlinkBackend>
|
||||
make_mock_backend(const MockConfig& cfg) {
|
||||
return std::make_unique<MockBackend>(cfg);
|
||||
}
|
||||
|
||||
} // namespace ceph::msgr::psp
|
||||
@ -22,6 +22,16 @@
|
||||
#include <errno.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <deque>
|
||||
#include <map>
|
||||
|
||||
// SO_ZEROCOPY comes from the kernel uapi headers while MSG_ZEROCOPY /
|
||||
// MSG_ERRQUEUE come from libc, so a toolchain can define one without
|
||||
// the other. Require both before compiling any of the send path.
|
||||
#if defined(SO_ZEROCOPY) && defined(MSG_ZEROCOPY)
|
||||
#define CEPH_HAVE_MSG_ZEROCOPY
|
||||
#include <linux/errqueue.h>
|
||||
#endif
|
||||
|
||||
#include "PosixStack.h"
|
||||
|
||||
@ -43,11 +53,89 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
int _fd;
|
||||
entity_addr_t sa;
|
||||
bool connected;
|
||||
CephContext *cct;
|
||||
PerfCounters *zc_logger; // Worker l_msgr_* logger (may be null)
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
// MSG_ZEROCOPY send state. Touched only on the owning worker
|
||||
// thread (send / drain / close), so no lock is needed. The zerocopy
|
||||
// perfcounters are driven HERE, at each event site, so they are
|
||||
// exact regardless of which teardown path closes the socket.
|
||||
//
|
||||
// Ids are tracked in a 64-bit space whose low 32 bits mirror the
|
||||
// kernel's per-socket counter; widen() maps a reported id back into
|
||||
// it. That keeps the wrap handling in one place and lets completed
|
||||
// ranges be ordered/merged with plain comparisons.
|
||||
bool zc_socket_enabled = false; // SO_ZEROCOPY active on _fd
|
||||
bool zc_eligible = true; // cleared for secure connections
|
||||
uint64_t zc_min_size = 0; // ms_tcp_zerocopy_min_size
|
||||
uint64_t zc_next_id = 0; // next id we will submit
|
||||
uint64_t zc_retire_id = 0; // oldest id not yet known complete
|
||||
// Completed-but-not-yet-applied ranges (lo -> hi, inclusive, merged).
|
||||
// The kernel may deliver ranges out of order - a locally delivered
|
||||
// (looped-back) skb completes via skb_copy_ubufs() while an earlier
|
||||
// id is still unacked - so a pin may only retire once every id up to
|
||||
// and including its own has been reported.
|
||||
std::map<uint64_t, uint64_t> zc_done_ranges;
|
||||
uint64_t zc_pinned = 0; // bytes awaiting completion
|
||||
size_t zc_last_bytes = 0; // bytes of the last send() sent zero-copy
|
||||
unsigned zc_last_submitted = 0; // zc sendmsg()s of the last send()
|
||||
struct ZeroCopyPin {
|
||||
uint64_t last_id; // highest id covering these bytes
|
||||
uint64_t bytes;
|
||||
ceph::buffer::list held; // refcounted: keeps the raws alive
|
||||
};
|
||||
std::deque<ZeroCopyPin> zc_pending;
|
||||
|
||||
// Map a kernel-reported 32-bit id into the 64-bit space, resolving
|
||||
// wrap against the oldest outstanding id. Correct while fewer than
|
||||
// 2^32 ids are outstanding, which the pin queue bounds far below.
|
||||
uint64_t widen_zc_id(uint32_t id) const {
|
||||
uint64_t v = (zc_retire_id & ~0xffffffffULL) | id;
|
||||
if (v < zc_retire_id)
|
||||
v += 0x100000000ULL;
|
||||
return v;
|
||||
}
|
||||
// Merge [lo,hi] into zc_done_ranges, coalescing with neighbours.
|
||||
void record_zc_range(uint64_t lo, uint64_t hi) {
|
||||
auto [it, inserted] = zc_done_ranges.emplace(lo, hi);
|
||||
if (!inserted)
|
||||
it->second = std::max(it->second, hi);
|
||||
auto nx = std::next(it);
|
||||
while (nx != zc_done_ranges.end() && nx->first <= it->second + 1) {
|
||||
it->second = std::max(it->second, nx->second);
|
||||
nx = zc_done_ranges.erase(nx);
|
||||
}
|
||||
if (it != zc_done_ranges.begin()) {
|
||||
auto pv = std::prev(it);
|
||||
if (pv->second + 1 >= it->first) {
|
||||
pv->second = std::max(pv->second, it->second);
|
||||
zc_done_ranges.erase(it);
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
public:
|
||||
explicit PosixConnectedSocketImpl(ceph::NetHandler &h, const entity_addr_t &sa,
|
||||
int f, bool connected)
|
||||
: handler(h), _fd(f), sa(sa), connected(connected) {}
|
||||
int f, bool connected, CephContext *c,
|
||||
PerfCounters *plog)
|
||||
: handler(h), _fd(f), sa(sa), connected(connected), cct(c),
|
||||
zc_logger(plog) {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
// Authoritative: did SO_ZEROCOPY actually take on this fd?
|
||||
// set_socket_options sets it only when ms_tcp_zerocopy is on.
|
||||
int zc = 0;
|
||||
socklen_t zl = sizeof(zc);
|
||||
if (cct &&
|
||||
::getsockopt(_fd, SOL_SOCKET, SO_ZEROCOPY, &zc, &zl) == 0 && zc) {
|
||||
zc_socket_enabled = true;
|
||||
// type:size option: read via the with_legacy member (the
|
||||
// established ms_tcp_* idiom, e.g. ms_tcp_prefetch_max_size);
|
||||
// get_val<uint64_t> on a size option throws bad_variant_access.
|
||||
zc_min_size = cct->_conf->ms_tcp_zerocopy_min_size;
|
||||
}
|
||||
#endif
|
||||
}
|
||||
|
||||
int is_connected() override {
|
||||
if (connected)
|
||||
@ -78,23 +166,68 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
// return the sent length
|
||||
// < 0 means error occurred
|
||||
#ifndef _WIN32
|
||||
static ssize_t do_sendmsg(int fd, struct msghdr &msg, unsigned len, bool more)
|
||||
// Per-send zero-copy accounting filled in by do_sendmsg().
|
||||
struct zc_send_stats {
|
||||
size_t bytes = 0; // bytes handed to the kernel with MSG_ZEROCOPY
|
||||
unsigned submitted = 0; // MSG_ZEROCOPY sendmsg()s == kernel ids consumed
|
||||
unsigned fallback = 0; // ENOBUFS degradations to a copying send
|
||||
};
|
||||
|
||||
static ssize_t do_sendmsg(int fd, struct msghdr &msg, unsigned len, bool more,
|
||||
bool zerocopy, zc_send_stats *zc_)
|
||||
{
|
||||
size_t sent = 0;
|
||||
while (1) {
|
||||
MSGR_SIGPIPE_STOPPER;
|
||||
ssize_t r;
|
||||
r = ::sendmsg(fd, &msg, MSG_NOSIGNAL | (more ? MSG_MORE : 0));
|
||||
int flags = MSG_NOSIGNAL | (more ? MSG_MORE : 0);
|
||||
bool zc = false;
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
if (zerocopy) {
|
||||
flags |= MSG_ZEROCOPY;
|
||||
zc = true;
|
||||
}
|
||||
#endif
|
||||
r = ::sendmsg(fd, &msg, flags);
|
||||
if (r < 0) {
|
||||
int err = ceph_sock_errno();
|
||||
if (err == EINTR) {
|
||||
continue;
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
} else if (zc && err == ENOBUFS) {
|
||||
// optmem_max exhausted: transparent fallback to a copying
|
||||
// send for this syscall (not counted as a zero-copy submit,
|
||||
// and no kernel id is consumed - msg_zerocopy_alloc() fails
|
||||
// before sk_zckey is bumped).
|
||||
r = ::sendmsg(fd, &msg, flags & ~MSG_ZEROCOPY);
|
||||
zc = false;
|
||||
if (zc_)
|
||||
zc_->fallback++;
|
||||
if (r < 0) {
|
||||
int e2 = ceph_sock_errno();
|
||||
if (e2 == EINTR) continue;
|
||||
if (e2 == EAGAIN) break;
|
||||
// Report the bytes the kernel already accepted; the caller
|
||||
// must still pin them. The error resurfaces on the next
|
||||
// send() once there is no progress left to report.
|
||||
if (sent) break;
|
||||
return -e2;
|
||||
}
|
||||
#endif
|
||||
} else if (err == EAGAIN) {
|
||||
break;
|
||||
} else {
|
||||
if (sent) break;
|
||||
return -err;
|
||||
}
|
||||
return -err;
|
||||
}
|
||||
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
if (zc && zc_) {
|
||||
zc_->submitted++;
|
||||
zc_->bytes += r;
|
||||
}
|
||||
#endif
|
||||
sent += r;
|
||||
if (len == sent) break;
|
||||
|
||||
@ -116,6 +249,12 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
|
||||
ssize_t send(ceph::buffer::list &bl, bool more) override {
|
||||
size_t sent_bytes = 0;
|
||||
zc_send_stats zc_stats;
|
||||
int send_err = 0;
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
zc_last_bytes = 0;
|
||||
zc_last_submitted = 0;
|
||||
#endif
|
||||
auto pb = std::cbegin(bl.buffers());
|
||||
uint64_t left_pbrs = bl.get_num_buffers();
|
||||
while (left_pbrs) {
|
||||
@ -134,9 +273,28 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
msglen += pb->length();
|
||||
++pb;
|
||||
}
|
||||
ssize_t r = do_sendmsg(_fd, msg, msglen, left_pbrs || more);
|
||||
if (r < 0)
|
||||
return r;
|
||||
bool zerocopy = false;
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
// Per-chunk: only large plaintext segments on a zero-copy-capable,
|
||||
// non-secure socket. Sub-threshold framing/control stays plain.
|
||||
// A zero-length chunk must never take this path: the kernel
|
||||
// consumes no id for it, which would desync our id mirror.
|
||||
zerocopy = zc_socket_enabled && zc_eligible &&
|
||||
msglen && msglen >= zc_min_size;
|
||||
#endif
|
||||
ssize_t r = do_sendmsg(_fd, msg, msglen, left_pbrs || more,
|
||||
zerocopy, &zc_stats);
|
||||
if (r < 0) {
|
||||
// Nothing was accepted anywhere in this send(): report the
|
||||
// error. If an earlier chunk did make progress we must fall
|
||||
// through to pin it - the kernel owns those pages until it
|
||||
// says otherwise, even though the connection is about to fault.
|
||||
if (!sent_bytes) {
|
||||
send_err = r;
|
||||
break;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
// "r" is the remaining length
|
||||
sent_bytes += r;
|
||||
@ -146,15 +304,46 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
}
|
||||
|
||||
if (sent_bytes) {
|
||||
ceph::buffer::list swapped;
|
||||
ceph::buffer::list sent_prefix;
|
||||
if (sent_bytes < bl.length()) {
|
||||
bl.splice(sent_bytes, bl.length()-sent_bytes, &swapped);
|
||||
bl.swap(swapped);
|
||||
bl.splice(0, sent_bytes, &sent_prefix);
|
||||
} else {
|
||||
bl.clear();
|
||||
sent_prefix.swap(bl);
|
||||
}
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
if (zc_stats.submitted) {
|
||||
// The kernel still references these pages until it posts a
|
||||
// MSG_ERRQUEUE completion; keep the (refcounted) buffers
|
||||
// pinned instead of releasing them here. The pin covers the
|
||||
// whole sent prefix even if some sub-threshold framing bytes
|
||||
// rode along - conservative for lifetime, never premature.
|
||||
// Byte *accounting* is not conservative in the same direction,
|
||||
// so it uses the measured zero-copy subset, not the prefix.
|
||||
const uint64_t last = zc_next_id + zc_stats.submitted - 1;
|
||||
zc_next_id += zc_stats.submitted;
|
||||
zc_pinned += sent_bytes;
|
||||
zc_pending.push_back(
|
||||
ZeroCopyPin{last, (uint64_t)sent_bytes, std::move(sent_prefix)});
|
||||
zc_last_bytes = zc_stats.bytes;
|
||||
zc_last_submitted = zc_stats.submitted;
|
||||
if (zc_logger) {
|
||||
// One pin == one zero-copy send operation. submitted counts
|
||||
// pins (the lifecycle unit), matching completed/force_dropped
|
||||
// which are also per-pin; the kernel-id stride may be >1 for
|
||||
// a partial/multi-chunk send and must NOT be the unit here.
|
||||
zc_logger->inc(l_msgr_zerocopy_submitted, 1);
|
||||
zc_logger->inc(l_msgr_send_bytes_zerocopy, zc_stats.bytes);
|
||||
zc_logger->inc(l_msgr_zerocopy_pinned_bytes, sent_bytes);
|
||||
}
|
||||
}
|
||||
// else sent_prefix is released here - identical to the old path.
|
||||
if (zc_stats.fallback && zc_logger)
|
||||
zc_logger->inc(l_msgr_zerocopy_fallback, zc_stats.fallback);
|
||||
#endif
|
||||
}
|
||||
|
||||
if (!sent_bytes && send_err)
|
||||
return send_err;
|
||||
return static_cast<ssize_t>(sent_bytes);
|
||||
}
|
||||
#else
|
||||
@ -205,7 +394,40 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
::shutdown(_fd, SHUT_RDWR);
|
||||
}
|
||||
void close() override {
|
||||
compat_closesocket(_fd);
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
// Best-effort final retire of real completions first.
|
||||
drain_zerocopy_completions();
|
||||
if (!zc_pending.empty()) {
|
||||
// Pins are still outstanding, so the kernel holds page
|
||||
// references for data it has not finished sending. A graceful
|
||||
// close does NOT drop those: the socket lives on in the
|
||||
// background retransmitting from exactly the pages we are about
|
||||
// to hand back to the allocator. Force an abortive close - the
|
||||
// RST discards the retransmit queue - before releasing them.
|
||||
struct linger lg = {1, 0};
|
||||
::setsockopt(_fd, SOL_SOCKET, SO_LINGER,
|
||||
(SOCKOPT_VAL_TYPE)&lg, sizeof(lg));
|
||||
// Counted separately from real completions: force_dropped != 0
|
||||
// is the signal that pins are leaking or the peer stalled, and
|
||||
// is what makes submitted == completed + force_dropped a real
|
||||
// invariant instead of one that holds by construction.
|
||||
if (zc_logger)
|
||||
zc_logger->inc(l_msgr_zerocopy_force_dropped, zc_pending.size());
|
||||
}
|
||||
if (zc_pinned && zc_logger)
|
||||
zc_logger->dec(l_msgr_zerocopy_pinned_bytes, zc_pinned);
|
||||
#endif
|
||||
if (_fd >= 0) {
|
||||
compat_closesocket(_fd);
|
||||
_fd = -1;
|
||||
}
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
// Only after the fd is gone: the kernel can no longer reference
|
||||
// these pages once the RST has torn the connection down.
|
||||
zc_pending.clear();
|
||||
zc_pinned = 0;
|
||||
zc_done_ranges.clear();
|
||||
#endif
|
||||
}
|
||||
void set_priority(int sd, int prio, int domain) override {
|
||||
handler.set_priority(sd, prio, domain);
|
||||
@ -213,6 +435,120 @@ class PosixConnectedSocketImpl final : public ConnectedSocketImpl {
|
||||
int fd() const override {
|
||||
return _fd;
|
||||
}
|
||||
|
||||
void set_zerocopy_eligible(bool e) override {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
zc_eligible = e;
|
||||
#endif
|
||||
}
|
||||
size_t last_send_zerocopy_bytes() const override {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
return zc_last_bytes;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
unsigned last_send_zerocopy_submitted() const override {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
return zc_last_submitted;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
uint64_t pinned_zerocopy_bytes() const override {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
return zc_pinned;
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
unsigned pending_zerocopy_count() const override {
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
return zc_pending.size();
|
||||
#else
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
ConnectedSocketImpl::zerocopy_drain drain_zerocopy_completions() override {
|
||||
ConnectedSocketImpl::zerocopy_drain out;
|
||||
#ifdef CEPH_HAVE_MSG_ZEROCOPY
|
||||
if (zc_pending.empty() || _fd < 0)
|
||||
return out; // nothing pinned: skip the recvmsg syscall
|
||||
char ctrl[512];
|
||||
while (true) {
|
||||
struct msghdr msg;
|
||||
// FIPS zeroization audit: not security related.
|
||||
memset(&msg, 0, sizeof(msg));
|
||||
msg.msg_control = ctrl;
|
||||
msg.msg_controllen = sizeof(ctrl);
|
||||
ssize_t r = ::recvmsg(_fd, &msg, MSG_ERRQUEUE);
|
||||
if (r < 0) {
|
||||
if (ceph_sock_errno() == EINTR)
|
||||
continue; // not "drained" - retry
|
||||
break; // EAGAIN: error queue drained
|
||||
}
|
||||
if (msg.msg_flags & MSG_CTRUNC) {
|
||||
// Completions were dropped on the floor; without them the
|
||||
// matching pins can never retire. Nothing to do here but keep
|
||||
// going - the buffer is sized for far more than the kernel
|
||||
// coalesces in practice, and close() force-drops the residue.
|
||||
ldout(cct, 1) << __func__
|
||||
<< " MSG_ERRQUEUE control data truncated" << dendl;
|
||||
}
|
||||
for (cmsghdr *cm = CMSG_FIRSTHDR(&msg); cm;
|
||||
cm = CMSG_NXTHDR(&msg, cm)) {
|
||||
if (!((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) ||
|
||||
(cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR)))
|
||||
continue;
|
||||
auto *serr =
|
||||
reinterpret_cast<struct sock_extended_err *>(CMSG_DATA(cm));
|
||||
if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)
|
||||
continue;
|
||||
// A notification covers the inclusive id range [ee_info,
|
||||
// ee_data]. Both ends matter: ranges can arrive out of order,
|
||||
// so the high end alone must not be taken as "everything below
|
||||
// is done" - that would retire pins the kernel still owns.
|
||||
uint64_t lo = widen_zc_id(serr->ee_info);
|
||||
uint64_t hi = widen_zc_id(serr->ee_data);
|
||||
if (hi < lo)
|
||||
continue; // malformed; ignore
|
||||
if (serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED) {
|
||||
// The kernel copied after all (e.g. the egress path cannot
|
||||
// do scatter/gather). Count every id in the range, not the
|
||||
// notification, so this stays on the same unit as submitted.
|
||||
const uint64_t n = hi - lo + 1;
|
||||
out.fallback += n;
|
||||
if (zc_logger)
|
||||
zc_logger->inc(l_msgr_zerocopy_fallback, n);
|
||||
}
|
||||
if (hi < zc_retire_id)
|
||||
continue; // wholly retired already
|
||||
record_zc_range(std::max(lo, zc_retire_id), hi);
|
||||
|
||||
// Apply only a run that is contiguous from the oldest
|
||||
// outstanding id; a range past a gap waits in the map.
|
||||
auto it = zc_done_ranges.begin();
|
||||
if (it != zc_done_ranges.end() && it->first <= zc_retire_id) {
|
||||
zc_retire_id = it->second + 1;
|
||||
zc_done_ranges.erase(it);
|
||||
}
|
||||
while (!zc_pending.empty() &&
|
||||
zc_pending.front().last_id < zc_retire_id) {
|
||||
const uint64_t b = zc_pending.front().bytes;
|
||||
out.completed++;
|
||||
out.retired_bytes += b;
|
||||
zc_pinned -= b;
|
||||
zc_pending.pop_front();
|
||||
if (zc_logger) {
|
||||
zc_logger->inc(l_msgr_zerocopy_completed, 1);
|
||||
zc_logger->dec(l_msgr_zerocopy_pinned_bytes, b);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
return out;
|
||||
}
|
||||
friend class PosixServerSocketImpl;
|
||||
friend class PosixNetworkStack;
|
||||
};
|
||||
@ -263,7 +599,7 @@ int PosixServerSocketImpl::accept(ConnectedSocket *sock, const SocketOptions &op
|
||||
out->set_sockaddr((sockaddr*)&ss);
|
||||
handler.set_priority(sd, opt.priority, out->get_family());
|
||||
|
||||
std::unique_ptr<PosixConnectedSocketImpl> csi(new PosixConnectedSocketImpl(handler, *out, sd, true));
|
||||
std::unique_ptr<PosixConnectedSocketImpl> csi(new PosixConnectedSocketImpl(handler, *out, sd, true, w->cct, w->get_perf_counter()));
|
||||
*sock = ConnectedSocket(std::move(csi));
|
||||
return 0;
|
||||
}
|
||||
@ -332,7 +668,7 @@ int PosixWorker::connect(const entity_addr_t &addr, const SocketOptions &opts, C
|
||||
|
||||
net.set_priority(sd, opts.priority, addr.get_family());
|
||||
*socket = ConnectedSocket(
|
||||
std::unique_ptr<PosixConnectedSocketImpl>(new PosixConnectedSocketImpl(net, addr, sd, !opts.nonblock)));
|
||||
std::unique_ptr<PosixConnectedSocketImpl>(new PosixConnectedSocketImpl(net, addr, sd, !opts.nonblock, cct, get_perf_counter())));
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
@ -1344,6 +1344,22 @@ CtPtr ProtocolV2::ready() {
|
||||
{
|
||||
std::lock_guard<std::mutex> l(connection->write_lock);
|
||||
can_write = true;
|
||||
// MSG_ZEROCOPY is pointless and adds pin-lifetime risk for
|
||||
// secure connections - the payload is already copied during
|
||||
// encryption (crypto_onwire), so there is no copy left to elide.
|
||||
// Keyed on in-process AEAD, not confidentiality: when secure-psp
|
||||
// moves the AEAD into the NIC there is a copy left to elide again,
|
||||
// and this exclusion must stop applying to it.
|
||||
if (auth_meta->is_mode_in_process_aead()) {
|
||||
connection->cs.set_zerocopy_eligible(false);
|
||||
}
|
||||
// Telemetry: count secure-psp negotiations (still aesgcm under
|
||||
// the hood until the kernel attach engages). ready() runs again on
|
||||
// every reconnect and on session replacement, so this counts
|
||||
// negotiations, not distinct connections - hence the name.
|
||||
if (auth_meta->is_mode_secure_psp()) {
|
||||
connection->logger->inc(l_msgr_psp_negotiations);
|
||||
}
|
||||
if (!out_queue.empty()) {
|
||||
connection->center->dispatch_event_external(connection->write_handler);
|
||||
}
|
||||
|
||||
@ -47,6 +47,28 @@ class ConnectedSocketImpl {
|
||||
virtual void close() = 0;
|
||||
virtual int fd() const = 0;
|
||||
virtual void set_priority(int sd, int prio, int domain) = 0;
|
||||
|
||||
// MSG_ZEROCOPY send support. The default implementations make
|
||||
// this a no-op for stacks/paths that do not implement it (RDMA,
|
||||
// DPDK, the Windows Posix path); only PosixConnectedSocketImpl
|
||||
// overrides them.
|
||||
struct zerocopy_drain {
|
||||
uint64_t completed = 0; // completion notifications retired
|
||||
uint64_t fallback = 0; // kernel degraded to a copy
|
||||
uint64_t retired_bytes = 0; // pinned bytes released
|
||||
};
|
||||
// Disable MSG_ZEROCOPY for this connection (e.g. secure mode).
|
||||
virtual void set_zerocopy_eligible(bool) {}
|
||||
// Payload bytes of the most recent send() pinned for MSG_ZEROCOPY
|
||||
// (awaiting kernel completion) rather than copied.
|
||||
virtual size_t last_send_zerocopy_bytes() const { return 0; }
|
||||
// MSG_ZEROCOPY sendmsg() calls the most recent send() submitted.
|
||||
virtual unsigned last_send_zerocopy_submitted() const { return 0; }
|
||||
// Bytes / count still pinned awaiting kernel completion.
|
||||
virtual uint64_t pinned_zerocopy_bytes() const { return 0; }
|
||||
virtual unsigned pending_zerocopy_count() const { return 0; }
|
||||
// Drain MSG_ERRQUEUE completions and retire pinned buffers.
|
||||
virtual zerocopy_drain drain_zerocopy_completions() { return {}; }
|
||||
};
|
||||
|
||||
class ConnectedSocket;
|
||||
@ -142,6 +164,27 @@ class ConnectedSocket {
|
||||
_csi->set_priority(sd, prio, domain);
|
||||
}
|
||||
|
||||
// MSG_ZEROCOPY forwards (safe when _csi is null).
|
||||
void set_zerocopy_eligible(bool e) {
|
||||
if (_csi) _csi->set_zerocopy_eligible(e);
|
||||
}
|
||||
size_t last_send_zerocopy_bytes() const {
|
||||
return _csi ? _csi->last_send_zerocopy_bytes() : 0;
|
||||
}
|
||||
unsigned last_send_zerocopy_submitted() const {
|
||||
return _csi ? _csi->last_send_zerocopy_submitted() : 0;
|
||||
}
|
||||
uint64_t pinned_zerocopy_bytes() const {
|
||||
return _csi ? _csi->pinned_zerocopy_bytes() : 0;
|
||||
}
|
||||
unsigned pending_zerocopy_count() const {
|
||||
return _csi ? _csi->pending_zerocopy_count() : 0;
|
||||
}
|
||||
ConnectedSocketImpl::zerocopy_drain drain_zerocopy_completions() {
|
||||
return _csi ? _csi->drain_zerocopy_completions()
|
||||
: ConnectedSocketImpl::zerocopy_drain{};
|
||||
}
|
||||
|
||||
explicit operator bool() const {
|
||||
return _csi.get();
|
||||
}
|
||||
@ -225,6 +268,29 @@ enum {
|
||||
l_msgr_recv_encrypted_bytes,
|
||||
l_msgr_send_encrypted_bytes,
|
||||
|
||||
// copy / zero-copy accounting on the send path.
|
||||
// _copied and _iov_segments are driven by the copying send path;
|
||||
// the zerocopy_* counters are driven at the socket event sites of
|
||||
// the MSG_ZEROCOPY send path (PosixStack).
|
||||
l_msgr_send_bytes_copied,
|
||||
l_msgr_send_bytes_zerocopy,
|
||||
l_msgr_send_iov_segments,
|
||||
l_msgr_zerocopy_submitted,
|
||||
l_msgr_zerocopy_completed,
|
||||
l_msgr_zerocopy_force_dropped,
|
||||
l_msgr_zerocopy_fallback,
|
||||
l_msgr_zerocopy_pinned_bytes,
|
||||
|
||||
// PSP Security Protocol negotiation telemetry.
|
||||
// Increments per completed handshake that landed on secure-psp, not
|
||||
// per connection: a reconnect or a session replacement re-runs the
|
||||
// negotiation on the same connection and counts again. Independent
|
||||
// of whether the kernel offload is engaged (today secure-psp still
|
||||
// does aesgcm AEAD under the hood; the kernel-offload follow-up
|
||||
// lights up the netlink attach and will add _attach_failed,
|
||||
// _fallback_to_aesgcm, _handshake_bytes, _active_assocs).
|
||||
l_msgr_psp_negotiations,
|
||||
|
||||
l_msgr_last,
|
||||
};
|
||||
|
||||
@ -283,6 +349,50 @@ class Worker {
|
||||
plb.add_u64_counter(l_msgr_recv_encrypted_bytes, "msgr_recv_encrypted_bytes", "Network received encrypted bytes", NULL, 0, unit_t(UNIT_BYTES));
|
||||
plb.add_u64_counter(l_msgr_send_encrypted_bytes, "msgr_send_encrypted_bytes", "Network sent encrypted bytes", NULL, 0, unit_t(UNIT_BYTES));
|
||||
|
||||
plb.add_u64_counter(l_msgr_send_bytes_copied, "msgr_send_bytes_copied",
|
||||
"Network sent bytes that passed through a kernel copy",
|
||||
NULL, 0, unit_t(UNIT_BYTES));
|
||||
plb.add_u64_counter(l_msgr_send_bytes_zerocopy, "msgr_send_bytes_zerocopy",
|
||||
"Network sent bytes via zero-copy",
|
||||
NULL, 0, unit_t(UNIT_BYTES));
|
||||
PerfHistogramCommon::axis_config_d msgr_send_iov_segs_x_axis{
|
||||
"iovec segments", ///< scatter/gather segment count
|
||||
PerfHistogramCommon::SCALE_LOG2, ///< covers 1 .. >IOV_MAX
|
||||
0, ///< start at 0
|
||||
1, ///< quantization unit
|
||||
14, ///< top bucket starts at 2048
|
||||
};
|
||||
PerfHistogramCommon::axis_config_d msgr_send_iov_bytes_y_axis{
|
||||
"sent bytes (bytes)",
|
||||
PerfHistogramCommon::SCALE_LOG2, ///< logarithmic byte scale
|
||||
0, ///< start at 0
|
||||
4096, ///< quantization unit
|
||||
13, ///< enough to cover 4+M sends
|
||||
};
|
||||
plb.add_u64_counter_histogram(
|
||||
l_msgr_send_iov_segments, "msgr_send_iov_segments",
|
||||
msgr_send_iov_segs_x_axis, msgr_send_iov_bytes_y_axis,
|
||||
"Histogram of pending iovec segment count vs. bytes drained per socket write");
|
||||
plb.add_u64_counter(l_msgr_zerocopy_submitted, "msgr_zerocopy_submitted",
|
||||
"Zero-copy sends submitted (pins created)");
|
||||
plb.add_u64_counter(l_msgr_zerocopy_completed, "msgr_zerocopy_completed",
|
||||
"Zero-copy pins retired by a kernel completion");
|
||||
plb.add_u64_counter(l_msgr_zerocopy_force_dropped,
|
||||
"msgr_zerocopy_force_dropped",
|
||||
"Zero-copy pins abandoned at close (abortive); "
|
||||
"submitted == completed + force_dropped, and a "
|
||||
"non-zero value means completions were not arriving");
|
||||
plb.add_u64_counter(l_msgr_zerocopy_fallback, "msgr_zerocopy_fallback",
|
||||
"Zero-copy requested but degraded to a copy "
|
||||
"(ENOBUFS or kernel-side copy)");
|
||||
plb.add_u64(l_msgr_zerocopy_pinned_bytes, "msgr_zerocopy_pinned_bytes",
|
||||
"Bytes currently pinned awaiting zero-copy completion");
|
||||
|
||||
plb.add_u64_counter(l_msgr_psp_negotiations, "msgr_psp_negotiations",
|
||||
"Handshakes that negotiated secure-psp mode "
|
||||
"(counts reconnects and session replacements, "
|
||||
"so it exceeds the live connection count)");
|
||||
|
||||
perf_logger = plb.create_perf_counters();
|
||||
cct->get_perfcounters_collection()->add(perf_logger);
|
||||
|
||||
|
||||
@ -283,7 +283,9 @@ ceph::crypto::onwire::rxtx_t ceph::crypto::onwire::rxtx_t::create_handler_pair(
|
||||
bool new_nonce_format,
|
||||
bool crossed)
|
||||
{
|
||||
if (auth_meta.is_mode_secure()) {
|
||||
// In-process AEAD, not merely "confidential": a kernel-offloaded
|
||||
// mode is confidential but installs no handler pair here.
|
||||
if (auth_meta.is_mode_in_process_aead()) {
|
||||
ceph_assert_always(auth_meta.connection_secret.length() >= \
|
||||
sizeof(key_t) + 2 * sizeof(nonce_t));
|
||||
const char* secbuf = auth_meta.connection_secret.c_str();
|
||||
|
||||
@ -134,6 +134,25 @@ int NetHandler::set_socket_options(int sd, bool nodelay, int size)
|
||||
}
|
||||
}
|
||||
|
||||
// MSG_ZEROCOPY send opt-in (default off). Setting SO_ZEROCOPY
|
||||
// alone changes nothing on the wire; it only takes effect once a
|
||||
// send passes MSG_ZEROCOPY.
|
||||
// Failure is non-fatal: the connection proceeds in the normal
|
||||
// copying mode. Linux >= 4.14 only; absent from Ceph compat headers.
|
||||
#ifdef SO_ZEROCOPY
|
||||
if (cct->_conf.get_val<bool>("ms_tcp_zerocopy")) {
|
||||
int zc = 1;
|
||||
if (::setsockopt(sd, SOL_SOCKET, SO_ZEROCOPY,
|
||||
(SOCKOPT_VAL_TYPE)&zc, sizeof(zc)) < 0) {
|
||||
int e = ceph_sock_errno();
|
||||
ldout(cct, 1) << "couldn't set SO_ZEROCOPY: " << cpp_strerror(e)
|
||||
<< " (continuing without zero-copy send)" << dendl;
|
||||
} else {
|
||||
ldout(cct, 10) << "SO_ZEROCOPY enabled on socket " << sd << dendl;
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
// block ESIGPIPE
|
||||
#ifdef CEPH_USE_SO_NOSIGPIPE
|
||||
int val = 1;
|
||||
|
||||
@ -114,3 +114,143 @@ see ceph-messenger.conf and ceph-messenger.fio for details.
|
||||
To run:
|
||||
|
||||
./fio ./ceph-messenger.fio
|
||||
|
||||
### Copy-accounting instrumentation and payload modes
|
||||
|
||||
The engine carries copy-accounting instrumentation and several payload
|
||||
modes used by the messenger zero-copy investigation. Counters are
|
||||
emitted in the engine teardown dump (look for the `PERFCOUNTERS`,
|
||||
`PERF HISTOGRAMS`, and `BUFFER CRC CACHE` marker blocks):
|
||||
|
||||
- `l_msgr_send_bytes_copied` — payload bytes that went through the
|
||||
kernel copy on send (today: all of them).
|
||||
- `l_msgr_send_iov_segments` — histogram of iovec elements per
|
||||
`send()`; this is the headline send-shape counter.
|
||||
- `buffer_cached_crc` / `buffer_missed_crc` — bufferlist CRC-cache
|
||||
hits/misses (`buffer_track_crc` is enabled by the engine).
|
||||
|
||||
When `ms_tcp_zerocopy` is on, also watch:
|
||||
|
||||
- `msgr_zerocopy_submitted` / `_completed` / `_force_dropped` —
|
||||
`submitted == completed + force_dropped` always holds; a non-zero
|
||||
`force_dropped` means completions were not arriving and pins were
|
||||
abandoned at close, which is a problem worth chasing.
|
||||
- `msgr_zerocopy_fallback` — kernel ids the kernel copied anyway.
|
||||
**On loopback this is essentially every send**: the loopback path
|
||||
has no scatter/gather DMA, so `MSG_ZEROCOPY` degrades to a copy and
|
||||
the kernel reports `SO_EE_CODE_ZEROCOPY_COPIED`. Loopback runs
|
||||
therefore validate the pin/retire *mechanism*, never the bandwidth
|
||||
benefit — that needs two hosts and a real NIC.
|
||||
|
||||
Payload-shape options (default off == single contiguous non-owning
|
||||
ptr, original behavior, no allocation or memcpy in the harness):
|
||||
|
||||
- `payload_frags=N` — split each send payload into N non-owning
|
||||
bufferptrs carved from fio's I/O buffer, to mimic the
|
||||
RGW-multipart / EC-scatter shape that drives multi-segment iovec
|
||||
assembly. Shifts the `l_msgr_send_iov_segments` histogram.
|
||||
- `payload_frag_unalign=1` — skew the fragment boundaries off page
|
||||
alignment.
|
||||
|
||||
EC read send-shape models (companions; run back to back and diff the
|
||||
`l_msgr_send_iov_segments` histogram — that delta is the EC
|
||||
horizontal-gather reply-fragmentation cost):
|
||||
|
||||
./fio ./ceph-messenger-ec-r1.fio # single-shard / client-scatter
|
||||
./fio ./ceph-messenger-ec-r2.fio # full-stripe horizontal gather
|
||||
|
||||
### Data-integrity gate (verify_echo)
|
||||
|
||||
`verify_echo=1` round-trips the payload and the **engine self-checks**
|
||||
it: the receiver echoes the bytes it actually received, and the sender
|
||||
compares crc32c + length of the echo against a snapshot taken just
|
||||
before send. A mismatch fails the io_u (`EILSEQ`) so fio aborts. This
|
||||
catches corrupted/stale sends — iovec mis-assembly, fragment-boundary
|
||||
bugs, truncation/partial send, and a buffer mutated before the kernel
|
||||
finished sending.
|
||||
|
||||
Note what it does **not** cover: the engine completes an io_u only
|
||||
when the reply arrives, and a reply implies the kernel already read
|
||||
the payload, so fio can never recycle a buffer out from under an
|
||||
in-flight send. The "recycled by another request" hazard therefore
|
||||
needs a different harness; this gate covers mutation, not reuse.
|
||||
|
||||
./fio ./ceph-messenger-verify.fio
|
||||
|
||||
Why not fio's `verify=crc32c`: fio's verify is medium/offset-keyed
|
||||
(write a pattern to offset X, later read offset X back from the
|
||||
medium), but this engine is a storageless request/reply pipe keyed by
|
||||
an io_u pointer abused as the object name — it has no offset→data map,
|
||||
and `FIO_UNIDIR` disables fio's post-write verify phase. fio-native
|
||||
verify fails even on a correct build ("bad magic header 0"). The
|
||||
engine-side crc compare needs no medium and is the actual
|
||||
buffer-lifetime gate for zero-copy sends.
|
||||
|
||||
`verify_echo` adds a full reply payload and a crc scan, so it **must
|
||||
not be used for throughput/bandwidth runs** — the memory-traffic
|
||||
numbers would be polluted. Bandwidth baselines use `ceph-messenger.fio`
|
||||
and the EC R1/R2 variants (all with `verify_echo` off).
|
||||
|
||||
#### The wire CRC comes first (why the job sets `ms_crc_data=false`)
|
||||
|
||||
msgr2 computes each segment's crc32c when the frame is assembled, i.e.
|
||||
*before* `cs.send()`. So a payload that changes after that point — the
|
||||
zero-copy defect class — is caught by the **receiver** as a frame
|
||||
error: the connection resets and no reply is ever produced, so the
|
||||
echo comparison never runs and the job simply stops making progress.
|
||||
|
||||
That is a detection, but a useless one for this purpose: it reports as
|
||||
a connection reset with no indication of what differed, and a lossless
|
||||
sender then re-sends the same bad frame in a loop. Running the job with
|
||||
`ms_crc_data=false` (via `ceph_conf_file=ceph-messenger.conf`, which
|
||||
`ceph-messenger-verify.fio` now sets) removes the wire CRC so the
|
||||
mutated bytes reach the receiver, get echoed, and the engine reports a
|
||||
legible `verify_echo MISMATCH`.
|
||||
|
||||
#### Broken-build gate procedure (env-gated, no rebuild)
|
||||
|
||||
The gate proves the job actually catches a bad send. It is built into
|
||||
the engine and toggled by env vars — no code edit or rebuild:
|
||||
|
||||
- `CEPH_MSGR_VERIFY_ECHO_CORRUPT=1` — mutate one payload byte after
|
||||
the integrity snapshot but before the send (and invalidate the CRC
|
||||
cache so it round-trips; see the finding below). Note this models
|
||||
the *detection* path, not the defect faithfully: a real zero-copy
|
||||
bug mutates the buffer without invalidating anything, which is why
|
||||
`ms_crc_data=false` matters above.
|
||||
- `CEPH_MSGR_VERIFY_ECHO_DEBUG=1` — print `sent` vs `echoed`
|
||||
len/crc32c for every reply (diagnostic).
|
||||
|
||||
Pass:
|
||||
|
||||
./fio ./ceph-messenger-verify.fio
|
||||
# runs to completion, err=0, no "verify_echo MISMATCH"
|
||||
|
||||
Fail (gate fires):
|
||||
|
||||
CEPH_MSGR_VERIFY_ECHO_CORRUPT=1 ./fio ./ceph-messenger-verify.fio
|
||||
# engine prints "fio: verify_echo MISMATCH io=... sent crc=A
|
||||
# echoed crc=B" and the client job aborts with err=84 (EILSEQ)
|
||||
|
||||
Note the *server* job does not exit on its own when the client aborts —
|
||||
it is still waiting for requests that will never arrive — so the run has
|
||||
to be interrupted (or given a `runtime=`). The gate evidence is the
|
||||
client's `MISMATCH` lines plus `err=84`.
|
||||
|
||||
Record both outcomes as the gate evidence.
|
||||
|
||||
Finding (why `invalidate_crc()` is in the corrupt path): a naive
|
||||
in-place mutation of a pending send buffer is **masked by the
|
||||
bufferlist per-raw CRC cache**. `buflist.crc32c()` (the integrity
|
||||
snapshot) caches the crc on the non-owning `raw` wrapping the fio
|
||||
buffer; an external pointer write does *not* invalidate it, so the
|
||||
messenger ships a *stale* data-CRC while the wire carries the mutated
|
||||
bytes, and the receiver rejects the frame — the corruption is caught,
|
||||
but as a connection CRC error with no round trip, not a legible
|
||||
content mismatch. The self-test calls
|
||||
`req_msg->get_data().invalidate_crc()` so the messenger re-CRCs the
|
||||
mutated bytes, the frame round-trips, and the echo self-check reports
|
||||
it cleanly. This is a concrete instance of the fragile CRC cache:
|
||||
**a zero-copy send path MUST invalidate/guard the CRC cache on any
|
||||
buffer it mutates while a send is still pending**, or corruption
|
||||
surfaces as opaque connection resets.
|
||||
|
||||
47
src/test/fio/ceph-messenger-ec-r1.fio
Normal file
47
src/test/fio/ceph-messenger-ec-r1.fio
Normal file
@ -0,0 +1,47 @@
|
||||
# Messenger send-shape model — EC read, R1 "single-shard / client-side
|
||||
# scatter" (companion to ceph-messenger-ec-r2.fio).
|
||||
#
|
||||
# Models the MOSDOpReply the primary OSD sends back to RGW for an EC
|
||||
# object read, under a client-side-scatter config:
|
||||
#
|
||||
# profile 6+2, stripe_unit 1M => stripe_width = k*stripe_unit = 6M
|
||||
# rgw_obj_stripe_size = rgw_max_chunk_size = 6M (full-stripe writes)
|
||||
# rgw_get_obj_max_req_size = 1M (= stripe_unit = chunk_size)
|
||||
#
|
||||
# With rgw_get_obj_max_req_size = stripe_unit, RGW issues independent
|
||||
# 1M read ops; each maps to ONE data shard (Fast EC single-shard path,
|
||||
# no primary horizontal gather). The reply to RGW is therefore a single
|
||||
# ~1M contiguous payload — a near-single-ptr send. This is the BASELINE
|
||||
# the horizontal-gather case (R2) is compared against.
|
||||
#
|
||||
# What this measures: the l_msgr_send_iov_segments histogram on the
|
||||
# sender Worker should sit in the low (single-ptr) bucket. The
|
||||
# client->server send is used as the data-bearing carrier since that is
|
||||
# the only payload-bearing direction in this engine; the iovec shape is
|
||||
# what is being modeled.
|
||||
#
|
||||
# Not a verify job and not an EC cluster — a send-path shape proxy.
|
||||
|
||||
[global]
|
||||
bs=1m
|
||||
size=4g
|
||||
iodepth=128
|
||||
|
||||
ioengine=libfio_ceph_messenger.so
|
||||
#ceph_conf_file=ceph-messenger.conf
|
||||
|
||||
hostname=127.0.0.1
|
||||
port=5555
|
||||
|
||||
ms_type=async+posix
|
||||
|
||||
# R1: one 1M region per op == one EC shard == one contiguous ptr.
|
||||
payload_frags=0
|
||||
|
||||
[client]
|
||||
receiver=0
|
||||
rw=write
|
||||
|
||||
[server]
|
||||
receiver=1
|
||||
rw=read
|
||||
51
src/test/fio/ceph-messenger-ec-r2.fio
Normal file
51
src/test/fio/ceph-messenger-ec-r2.fio
Normal file
@ -0,0 +1,51 @@
|
||||
# Messenger send-shape model — EC read, R2 "horizontal gather"
|
||||
# (companion to ceph-messenger-ec-r1.fio).
|
||||
#
|
||||
# Models the MOSDOpReply the primary OSD sends back to RGW for an EC
|
||||
# object read when the read op is the full stripe width:
|
||||
#
|
||||
# profile 6+2, stripe_unit 1M => stripe_width = k*stripe_unit = 6M
|
||||
# rgw_obj_stripe_size = rgw_max_chunk_size = 6M
|
||||
# rgw_get_obj_max_req_size = 6M (= stripe_width)
|
||||
#
|
||||
# With rgw_get_obj_max_req_size = stripe_width, every read spans all
|
||||
# k=6 data shards: the primary gathers the shards and the reply to RGW
|
||||
# is the reassembled logical 6M buffer — a concatenation of 6 x 1M
|
||||
# per-shard contributions, i.e. a fragmented, multi-segment send. This
|
||||
# is the read < write asymmetry's reply-fragmentation component.
|
||||
#
|
||||
# What this measures: the l_msgr_send_iov_segments histogram on the
|
||||
# sender Worker should shift UP versus R1 (single-shard), tracking
|
||||
# the k-way fan-out. payload_frags=6 models the per-shard segmentation
|
||||
# of one stripe; payload_frag_unalign models shard boundaries that are
|
||||
# not page-aligned. Run R1 and R2 back to back and diff the histogram —
|
||||
# that delta is the EC reply-fragmentation cost.
|
||||
#
|
||||
# Not a verify job and not an EC cluster — a send-path shape proxy.
|
||||
|
||||
[global]
|
||||
bs=6m
|
||||
size=24g
|
||||
iodepth=128
|
||||
|
||||
ioengine=libfio_ceph_messenger.so
|
||||
#ceph_conf_file=ceph-messenger.conf
|
||||
|
||||
hostname=127.0.0.1
|
||||
port=5555
|
||||
|
||||
ms_type=async+posix
|
||||
|
||||
# R2: one 6M stripe per op, segmented into k=6 x 1M per-shard pieces.
|
||||
payload_frags=6
|
||||
# Shard boundaries in a real reassembled EC buffer are not page
|
||||
# aligned; uncomment to model that skew.
|
||||
#payload_frag_unalign=1
|
||||
|
||||
[client]
|
||||
receiver=0
|
||||
rw=write
|
||||
|
||||
[server]
|
||||
receiver=1
|
||||
rw=read
|
||||
68
src/test/fio/ceph-messenger-verify.fio
Normal file
68
src/test/fio/ceph-messenger-verify.fio
Normal file
@ -0,0 +1,68 @@
|
||||
# Messenger data-integrity gate for the ceph-msgr engine.
|
||||
#
|
||||
# This job round-trips the payload and the ENGINE self-checks it: with
|
||||
# verify_echo=1 the receiver echoes the bytes it actually received, and
|
||||
# the sender compares crc32c + length of the echo against a snapshot
|
||||
# taken just before send. A mismatch fails the io_u (EILSEQ) so fio
|
||||
# aborts the run.
|
||||
#
|
||||
# Why the engine self-checks instead of fio's verify=crc32c: fio's
|
||||
# verify is medium/offset-keyed (write a pattern to offset X, later
|
||||
# read offset X back from the medium), but this engine is a storageless
|
||||
# request/reply pipe keyed by an io_u pointer. fio-native verify
|
||||
# therefore fails even on a correct build (proven: "bad magic header
|
||||
# 0"). The engine-side crc compare needs no medium and directly
|
||||
# catches: corrupted bytes, truncation/partial send, and the zero-copy
|
||||
# defect class — a send buffer mutated/retired before the kernel
|
||||
# finished sending, or recycled by another in-flight request.
|
||||
#
|
||||
# verify_echo adds a full reply payload and a crc scan — this job is
|
||||
# NOT a throughput/bandwidth measurement. Use ceph-messenger.fio and
|
||||
# the EC R1/R2 variants for that.
|
||||
#
|
||||
# IMPORTANT: this job needs ms_crc_data=false (supplied by the
|
||||
# ceph_conf_file below). With the wire CRC on, a payload mutated after
|
||||
# the frame CRC was computed is rejected by the RECEIVER as a frame
|
||||
# error: no reply is generated, so the echo check never runs and the
|
||||
# job stalls rather than reporting a mismatch.
|
||||
#
|
||||
# Broken-build gate (env-gated, no rebuild): a correct build runs to
|
||||
# completion with err=0; running with CEPH_MSGR_VERIFY_ECHO_CORRUPT=1
|
||||
# must report a verify_echo MISMATCH and abort. See
|
||||
# src/test/fio/README.md (incl. the CRC-cache finding).
|
||||
|
||||
[global]
|
||||
bs=4m
|
||||
# Must be well above iodepth*bs, or every io_u is used exactly once
|
||||
# and no buffer is ever reused. fio hands each io_u a disjoint slice
|
||||
# of one allocation, so reuse only happens across completions.
|
||||
size=8g
|
||||
iodepth=128
|
||||
iodepth_batch=32
|
||||
iomem=malloc
|
||||
|
||||
ioengine=libfio_ceph_messenger.so
|
||||
# Required for the gate to report mismatches rather than stall; sets
|
||||
# ms_crc_data=false. Path is relative to the working directory.
|
||||
ceph_conf_file=ceph-messenger.conf
|
||||
|
||||
hostname=127.0.0.1
|
||||
port=5555
|
||||
|
||||
ms_type=async+posix
|
||||
|
||||
# Round-trip + engine self-check.
|
||||
verify_echo=1
|
||||
|
||||
# Fragment the payload (RGW-multipart / EC-scatter shape) so the gate
|
||||
# also exercises multi-segment iovec assembly and fragment boundaries.
|
||||
payload_frags=64
|
||||
#payload_frag_unalign=1
|
||||
|
||||
[client]
|
||||
receiver=0
|
||||
rw=randwrite
|
||||
|
||||
[server]
|
||||
receiver=1
|
||||
rw=read
|
||||
@ -13,6 +13,13 @@ port=5555
|
||||
|
||||
ms_type=async+posix # or async+dpdk or async+rdma
|
||||
|
||||
# Split each send payload into N non-owning bufferptrs to mimic
|
||||
# RGW-multipart / EC scatter (0/1 = single contiguous ptr). Watch the
|
||||
# msgr_send_iov_segments histogram shift in the teardown dump.
|
||||
#payload_frags=64
|
||||
# Skew fragment boundaries so ptr starts are not page-aligned.
|
||||
#payload_frag_unalign=1
|
||||
|
||||
[client]
|
||||
receiver=0
|
||||
rw=write
|
||||
|
||||
@ -17,6 +17,11 @@
|
||||
#include "auth/DummyAuth.h"
|
||||
#include "ring_buffer.h"
|
||||
|
||||
#include <algorithm>
|
||||
#include <cerrno>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include <fio.h>
|
||||
#include <flist.h>
|
||||
#include <optgroup.h>
|
||||
@ -44,6 +49,15 @@ struct ceph_msgr_options {
|
||||
const char *hostname;
|
||||
const char *conffile;
|
||||
enum ceph_msgr_type ms_type;
|
||||
/* Split the send payload into N non-owning bufferptrs to mimic
|
||||
* RGW-multipart / EC scatter. */
|
||||
unsigned int payload_frags;
|
||||
unsigned int payload_frag_unalign;
|
||||
/* verify_echo: round-trip the payload (receiver echoes it in the
|
||||
* reply; the sender compares crc32c+length against a pre-send
|
||||
* snapshot) to detect corrupted/stale sends. Default off so
|
||||
* throughput runs stay lean and unidirectional. */
|
||||
unsigned int verify_echo;
|
||||
};
|
||||
|
||||
class FioDispatcher;
|
||||
@ -73,6 +87,12 @@ struct ceph_msgr_io {
|
||||
struct ceph_msgr_data *data;
|
||||
struct io_u *io_u;
|
||||
MOSDOp *req_msg; /** Cached request, valid only for sender */
|
||||
/* verify_echo: crc32c + length of the payload snapshotted just
|
||||
* before send, compared against the echoed reply.
|
||||
* fio's native verify is medium/offset-keyed and cannot gate this
|
||||
* storageless io_u-pointer-keyed pipe, so the engine self-checks. */
|
||||
uint32_t echo_crc;
|
||||
uint64_t echo_len;
|
||||
};
|
||||
|
||||
struct ceph_msgr_reply_io {
|
||||
@ -123,6 +143,9 @@ static void create_or_get_ceph_context(struct ceph_msgr_options *o)
|
||||
|
||||
common_init_finish(g_ceph_context);
|
||||
g_ceph_context->_conf.apply_changes(NULL);
|
||||
/* Account bufferlist CRC cache hits/misses so the copy baseline can
|
||||
* quantify CRC-scan recomputation. */
|
||||
ceph::buffer::track_cached_crc(true);
|
||||
g_dummy_auth = new DummyAuthClientServer(g_ceph_context);
|
||||
g_dummy_auth->auth_registry.refresh_config();
|
||||
}
|
||||
@ -140,6 +163,25 @@ static void put_ceph_context(void)
|
||||
f->flush(ostr);
|
||||
ostr << ">>>>>>>>>>>>> PERFCOUNTERS END <<<<<<<<<<<<" << std::endl;
|
||||
|
||||
/* Histograms (e.g. msgr_send_iov_segments) are excluded from the
|
||||
* regular dump_formatted path and need an explicit call. */
|
||||
g_ceph_context->get_perfcounters_collection()->dump_formatted_histograms(
|
||||
f, false);
|
||||
ostr << ">>>>>>>>>>>>> PERF HISTOGRAMS BEGIN <<<<<<<<<<<<" << std::endl;
|
||||
f->flush(ostr);
|
||||
ostr << ">>>>>>>>>>>>> PERF HISTOGRAMS END <<<<<<<<<<<<" << std::endl;
|
||||
|
||||
/* Bufferlist CRC cache accounting. These are process-global
|
||||
* atomics, not perfcounters, so emit them alongside. */
|
||||
ostr << ">>>>>>>>>>>>> BUFFER CRC CACHE BEGIN <<<<<<<<<<<<" << std::endl;
|
||||
ostr << "buffer_cached_crc: "
|
||||
<< ceph::buffer::get_cached_crc() << std::endl;
|
||||
ostr << "buffer_cached_crc_adjusted: "
|
||||
<< ceph::buffer::get_cached_crc_adjusted() << std::endl;
|
||||
ostr << "buffer_missed_crc: "
|
||||
<< ceph::buffer::get_missed_crc() << std::endl;
|
||||
ostr << ">>>>>>>>>>>>> BUFFER CRC CACHE END <<<<<<<<<<<<" << std::endl;
|
||||
|
||||
delete f;
|
||||
delete g_dummy_auth;
|
||||
dout(0) << ostr.str() << dendl;
|
||||
@ -148,7 +190,7 @@ static void put_ceph_context(void)
|
||||
g_ceph_context->put();
|
||||
}
|
||||
|
||||
static void ceph_msgr_sender_on_reply(const object_t &oid)
|
||||
static void ceph_msgr_sender_on_reply(MOSDOpReply *rep)
|
||||
{
|
||||
struct ceph_msgr_data *data;
|
||||
struct ceph_msgr_io *io;
|
||||
@ -160,8 +202,39 @@ static void ceph_msgr_sender_on_reply(const object_t &oid)
|
||||
* to search for reply, just send a pointer and get it back.
|
||||
*/
|
||||
|
||||
io = (decltype(io))str_to_ptr(oid.name);
|
||||
io = (decltype(io))str_to_ptr(rep->get_oid().name);
|
||||
data = io->data;
|
||||
|
||||
if (data->o->verify_echo) {
|
||||
/* verify_echo gate: the receiver echoed the bytes it actually
|
||||
* received; compare crc32c + length against what we snapshotted
|
||||
* just before send. A mismatch means the wire bytes differed from
|
||||
* the payload at queue time — corruption, truncation/partial send,
|
||||
* or (with zero-copy sends) a buffer mutated/retired before the
|
||||
* kernel finished sending, or recycled by another in-flight
|
||||
* request. Fail the
|
||||
* io_u so fio aborts the run (continue_on_error defaults off).
|
||||
* Not for bandwidth runs: the reply carries a full echo. */
|
||||
ceph::buffer::list &bl = rep->get_data();
|
||||
uint64_t got_len = bl.length();
|
||||
uint32_t got_crc = bl.crc32c(0);
|
||||
if (getenv("CEPH_MSGR_VERIFY_ECHO_DEBUG"))
|
||||
fprintf(stderr,
|
||||
"fio: verify_echo DEBUG io=%p: sent len=%llu crc=%08x, "
|
||||
"echoed len=%llu crc=%08x\n",
|
||||
(void *)io,
|
||||
(unsigned long long)io->echo_len, io->echo_crc,
|
||||
(unsigned long long)got_len, got_crc);
|
||||
if (got_len != io->echo_len || got_crc != io->echo_crc) {
|
||||
fprintf(stderr,
|
||||
"fio: verify_echo MISMATCH io=%p: sent len=%llu crc=%08x, "
|
||||
"echoed len=%llu crc=%08x\n",
|
||||
(void *)io,
|
||||
(unsigned long long)io->echo_len, io->echo_crc,
|
||||
(unsigned long long)got_len, got_crc);
|
||||
io->io_u->error = EILSEQ;
|
||||
}
|
||||
}
|
||||
ring_buffer_enqueue(&data->io_completed_q, (void *)io);
|
||||
}
|
||||
|
||||
@ -189,6 +262,21 @@ static void ceph_msgr_receiver_on_request(struct ceph_msgr_data *data,
|
||||
rep = new MOSDOpReply(req, 0, 0, 0, false);
|
||||
rep->set_connection(req->get_connection());
|
||||
|
||||
if (data->o->verify_echo && !req->ops.empty()) {
|
||||
/* verify_echo: echo the received write payload back as the reply's op
|
||||
* out-data so the sender can self-check the round-tripped bytes.
|
||||
* MOSDOp::write() carries the payload in the message data
|
||||
* bufferlist (per-op indata stays empty, payload_len 0), so source
|
||||
* the echo from req->get_data(), not ops[].indata. The MOSDOpReply
|
||||
* ctor copied req->ops so sizes match for claim_op_out_data; put
|
||||
* the whole payload on op 0 (the engine only ever issues one write
|
||||
* op). Refcounted bufferlist copy — no payload memcpy on the
|
||||
* receiver. */
|
||||
std::vector<OSDOp> echo_ops = req->ops;
|
||||
echo_ops[0].outdata = req->get_data();
|
||||
rep->claim_op_out_data(echo_ops);
|
||||
}
|
||||
|
||||
pthread_spin_lock(&data->spin);
|
||||
if (data->io_inflight_nr) {
|
||||
struct ceph_msgr_io *io;
|
||||
@ -262,7 +350,7 @@ public:
|
||||
*/
|
||||
|
||||
rep = static_cast<MOSDOpReply*>(m);
|
||||
ceph_msgr_sender_on_reply(rep->get_oid());
|
||||
ceph_msgr_sender_on_reply(rep);
|
||||
}
|
||||
m->put();
|
||||
}
|
||||
@ -482,20 +570,84 @@ static void fio_ceph_msgr_io_u_free(struct thread_data *td, struct io_u *io_u)
|
||||
static enum fio_q_status ceph_msgr_sender_queue(struct thread_data *td,
|
||||
struct io_u *io_u)
|
||||
{
|
||||
struct ceph_msgr_options *o = (decltype(o))td->eo;
|
||||
struct ceph_msgr_data *data;
|
||||
struct ceph_msgr_io *io;
|
||||
|
||||
bufferlist buflist = bufferlist::static_from_mem(
|
||||
(char *)io_u->buf, io_u->buflen);
|
||||
char *buf = (char *)io_u->buf;
|
||||
const size_t total = io_u->buflen;
|
||||
unsigned nfrags = o->payload_frags;
|
||||
|
||||
bufferlist buflist;
|
||||
if (nfrags <= 1 || total < nfrags) {
|
||||
/* Single contiguous non-owning ptr (original behavior). */
|
||||
buflist = bufferlist::static_from_mem(buf, total);
|
||||
} else {
|
||||
/* N non-owning ptrs carved from io_u->buf, summing to total. No
|
||||
* allocation or memcpy of payload — only small raw_static wrappers,
|
||||
* exactly as static_from_mem does for the single-ptr case. */
|
||||
const size_t base = total / nfrags;
|
||||
/* Skew the first boundary so subsequent ptr starts are not
|
||||
* page-aligned, mimicking odd RGW part sizes. Only applied when
|
||||
* the fragments are big enough to give the skew back: otherwise
|
||||
* the first fragment borrows more than the remaining ones have,
|
||||
* and the last length underflows. */
|
||||
const size_t skew =
|
||||
(o->payload_frag_unalign && base + (total % nfrags) >= 17) ? 17 : 0;
|
||||
size_t off = 0;
|
||||
for (unsigned i = 0; i < nfrags; i++) {
|
||||
size_t len;
|
||||
if (i == nfrags - 1)
|
||||
len = total - off; /* last absorbs remainder */
|
||||
else
|
||||
len = base + (i == 0 ? skew : 0);
|
||||
ceph_assert(off + len <= total);
|
||||
buflist.push_back(ceph::buffer::ptr_node::create(
|
||||
ceph::buffer::create_static(len, buf + off)));
|
||||
off += len;
|
||||
}
|
||||
ceph_assert(off == total);
|
||||
}
|
||||
|
||||
io = (decltype(io))io_u->engine_data;
|
||||
data = (decltype(data))td->io_ops_data;
|
||||
|
||||
if (o->verify_echo) {
|
||||
/* Snapshot the payload crc32c + length BEFORE handing it to the
|
||||
* messenger; the reply path compares the echo against this. The
|
||||
* non-owning buflist still aliases io_u->buf here, so a Phase-1
|
||||
* bug that mutates the buffer after this point but before the
|
||||
* kernel finishes sending will change the wire bytes and be
|
||||
* caught. (CRC scan — verify_echo is not a bandwidth run.) */
|
||||
io->echo_len = buflist.length();
|
||||
io->echo_crc = buflist.crc32c(0);
|
||||
}
|
||||
|
||||
/* No handy method to clear ops before reusage? Ok */
|
||||
io->req_msg->ops.clear();
|
||||
|
||||
/* Here we do not care about direction, always send as write */
|
||||
io->req_msg->write(0, io_u->buflen, buflist);
|
||||
|
||||
/* Built-in gate self-test: with CEPH_MSGR_VERIFY_ECHO_CORRUPT set,
|
||||
* mutate one payload byte AFTER the integrity snapshot but before
|
||||
* the send, modelling the Phase-1 defect class — a buffer changed
|
||||
* out from under an in-flight send. invalidate_crc() is required:
|
||||
* the snapshot's buflist.crc32c() cached crc(original) on the
|
||||
* non-owning raw, and an external pointer write does NOT invalidate
|
||||
* it, so without this the messenger ships a STALE data-CRC and the
|
||||
* receiver rejects the frame (no round trip — the corruption is
|
||||
* caught, but as a wire-CRC error, not a content MISMATCH; see
|
||||
* plan §2.4 / Phase-1 design problem 5). Invalidating forces the
|
||||
* messenger to CRC the corrupted bytes, so the frame round-trips
|
||||
* and the echo self-check catches the content divergence cleanly.
|
||||
* Clean run (env unset) must pass; this run must MISMATCH + abort.
|
||||
* No rebuild needed to flip between the two. */
|
||||
if (o->verify_echo && getenv("CEPH_MSGR_VERIFY_ECHO_CORRUPT")) {
|
||||
((char *)io_u->buf)[0] ^= 0xff;
|
||||
io->req_msg->get_data().invalidate_crc();
|
||||
}
|
||||
|
||||
/* Keep message alive */
|
||||
io->req_msg->get();
|
||||
io->req_msg->get_connection()->send_message(io->req_msg);
|
||||
@ -671,6 +823,38 @@ static std::vector<fio_option> options {
|
||||
o.off1 = offsetof(struct ceph_msgr_options, conffile);
|
||||
o.help = "Path to CEPH configuration file";
|
||||
}),
|
||||
make_option([] (fio_option& o) {
|
||||
o.name = "payload_frags";
|
||||
o.lname = "Payload fragment count";
|
||||
o.type = FIO_OPT_INT;
|
||||
o.off1 = offsetof(struct ceph_msgr_options, payload_frags);
|
||||
o.help = "Split each send payload into N non-owning bufferptrs "
|
||||
"(0/1 = single contiguous ptr) to mimic RGW-multipart / "
|
||||
"EC scatter";
|
||||
o.def = "0";
|
||||
o.minval = 0;
|
||||
}),
|
||||
make_option([] (fio_option& o) {
|
||||
o.name = "payload_frag_unalign";
|
||||
o.lname = "Skew fragment boundaries off page alignment";
|
||||
o.type = FIO_OPT_BOOL;
|
||||
o.off1 = offsetof(struct ceph_msgr_options, payload_frag_unalign);
|
||||
o.help = "When set, offset fragment boundaries so ptrs are not "
|
||||
"page-aligned";
|
||||
o.def = "0";
|
||||
}),
|
||||
make_option([] (fio_option& o) {
|
||||
o.name = "verify_echo";
|
||||
o.lname = "Round-trip the payload and self-check it";
|
||||
o.type = FIO_OPT_BOOL;
|
||||
o.off1 = offsetof(struct ceph_msgr_options, verify_echo);
|
||||
o.help = "Receiver echoes the payload in the reply and the sender "
|
||||
"compares crc32c+length against a pre-send snapshot, "
|
||||
"failing the io_u on mismatch. Must be set on BOTH jobs. "
|
||||
"Needs ms_crc_data=false to report mismatches rather than "
|
||||
"stall. Not for bandwidth runs (full reply payload + scan)";
|
||||
o.def = "0";
|
||||
}),
|
||||
{} /* Last NULL */
|
||||
};
|
||||
|
||||
|
||||
@ -31,6 +31,17 @@ add_executable(unittest_frames_v2 test_frames_v2.cc)
|
||||
add_ceph_unittest(unittest_frames_v2)
|
||||
target_link_libraries(unittest_frames_v2 os global ${UNITTEST_LIBS})
|
||||
|
||||
# unittest_psp_netlink_mock. The mock backend is built here rather
|
||||
# than into libceph-common: it is a test double with failure-injection
|
||||
# knobs and has no business in a shipped library.
|
||||
if(HAVE_PSP)
|
||||
add_executable(unittest_psp_netlink_mock
|
||||
test_psp_netlink_mock.cc
|
||||
${PROJECT_SOURCE_DIR}/src/msg/async/PSPNetlinkMock.cc)
|
||||
add_ceph_unittest(unittest_psp_netlink_mock)
|
||||
target_link_libraries(unittest_psp_netlink_mock global ${UNITTEST_LIBS})
|
||||
endif()
|
||||
|
||||
add_executable(unittest_comp_registry
|
||||
test_comp_registry.cc
|
||||
$<TARGET_OBJECTS:unit-main>
|
||||
|
||||
180
src/test/msgr/test_psp_netlink_mock.cc
Normal file
180
src/test/msgr/test_psp_netlink_mock.cc
Normal file
@ -0,0 +1,180 @@
|
||||
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
|
||||
// vim: ts=8 sw=2 smarttab
|
||||
//
|
||||
// Unit tests for the PSP netlink mock backend.
|
||||
//
|
||||
// These tests exercise the NetlinkBackend interface contract
|
||||
// using only the in-memory mock - no kernel, no hardware, no
|
||||
// real netlink. The two-peer rendezvous test models the shape
|
||||
// of the planned in-band PSP handshake without requiring the
|
||||
// protocol code to exist yet.
|
||||
|
||||
#include "msg/async/PSPNetlink.h"
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
using namespace ceph::msgr::psp;
|
||||
|
||||
TEST(PSPNetlinkMock, GetDevCapsReportsSupported) {
|
||||
auto backend = make_mock_backend(MockConfig{});
|
||||
auto caps = backend->get_dev_caps(/*sock_fd*/ 42);
|
||||
ASSERT_TRUE(caps.has_value());
|
||||
EXPECT_TRUE(caps->psp_supported);
|
||||
EXPECT_EQ(caps->ifname, "mock0");
|
||||
EXPECT_EQ(caps->free_tx_assoc_slots, 64u);
|
||||
EXPECT_EQ(caps->free_rx_assoc_slots, 64u);
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, GetDevCapsHonorsUnsupported) {
|
||||
MockConfig cfg;
|
||||
cfg.psp_supported = false;
|
||||
auto backend = make_mock_backend(cfg);
|
||||
auto caps = backend->get_dev_caps(42);
|
||||
ASSERT_TRUE(caps.has_value());
|
||||
EXPECT_FALSE(caps->psp_supported);
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, GetDevCapsHardError) {
|
||||
MockConfig cfg;
|
||||
cfg.fail_get_dev_caps = true;
|
||||
auto backend = make_mock_backend(cfg);
|
||||
EXPECT_FALSE(backend->get_dev_caps(42).has_value());
|
||||
}
|
||||
|
||||
// Models the in-band PSP handshake shape: each peer allocs its
|
||||
// own rx-key locally, the wrapped blob round-trips over the
|
||||
// (mocked) wire, and each peer installs the other's blob as its
|
||||
// tx-key. After this, both peers consider the connection PSP-up.
|
||||
TEST(PSPNetlinkMock, TwoPeerRendezvous) {
|
||||
auto peer_a = make_mock_backend(MockConfig{});
|
||||
auto peer_b = make_mock_backend(MockConfig{});
|
||||
|
||||
auto blob_a = peer_a->alloc_rx_assoc(/*fd*/ 100);
|
||||
ASSERT_TRUE(blob_a.has_value());
|
||||
auto blob_b = peer_b->alloc_rx_assoc(/*fd*/ 200);
|
||||
ASSERT_TRUE(blob_b.has_value());
|
||||
|
||||
EXPECT_EQ(0, peer_a->install_tx_assoc(100, *blob_b));
|
||||
EXPECT_EQ(0, peer_b->install_tx_assoc(200, *blob_a));
|
||||
|
||||
auto caps_a = peer_a->get_dev_caps(0);
|
||||
ASSERT_TRUE(caps_a.has_value());
|
||||
EXPECT_EQ(caps_a->free_tx_assoc_slots, 64u - 1);
|
||||
EXPECT_EQ(caps_a->free_rx_assoc_slots, 64u - 1);
|
||||
}
|
||||
|
||||
// A tx key must come from the peer. Installing a blob this instance
|
||||
// issued is a crossed handshake, not a working association - without
|
||||
// this check the rendezvous test would pass even if the harness never
|
||||
// exchanged anything.
|
||||
TEST(PSPNetlinkMock, RejectsOwnBlobAsTxKey) {
|
||||
auto peer = make_mock_backend(MockConfig{});
|
||||
auto own = peer->alloc_rx_assoc(/*fd*/ 10);
|
||||
ASSERT_TRUE(own.has_value());
|
||||
EXPECT_EQ(-EINVAL, peer->install_tx_assoc(10, *own));
|
||||
}
|
||||
|
||||
// Exercises the tear-down-on-key-error path: alloc succeeds and the
|
||||
// peer's blob arrives, but the kernel-side install rejects it. The
|
||||
// local rx association must then be released, not leaked.
|
||||
TEST(PSPNetlinkMock, InjectedInstallFailure) {
|
||||
MockConfig cfg;
|
||||
cfg.fail_next_install_tx_assoc = 1;
|
||||
auto local = make_mock_backend(cfg);
|
||||
auto remote = make_mock_backend(MockConfig{});
|
||||
|
||||
auto local_rx = local->alloc_rx_assoc(/*fd*/ 50);
|
||||
ASSERT_TRUE(local_rx.has_value());
|
||||
auto remote_rx = remote->alloc_rx_assoc(/*fd*/ 250);
|
||||
ASSERT_TRUE(remote_rx.has_value());
|
||||
|
||||
// -EIO, not -ENOSPC: an injected rejection is not capacity pressure.
|
||||
EXPECT_EQ(-EIO, local->install_tx_assoc(50, *remote_rx));
|
||||
|
||||
// The connection is torn down after a failed key install; the rx
|
||||
// slot taken by alloc_rx_assoc must come back.
|
||||
EXPECT_EQ(0, local->teardown(50));
|
||||
auto caps = local->get_dev_caps(0);
|
||||
ASSERT_TRUE(caps.has_value());
|
||||
EXPECT_EQ(caps->free_rx_assoc_slots, 64u);
|
||||
EXPECT_EQ(caps->free_tx_assoc_slots, 64u);
|
||||
|
||||
// Budget consumed; a subsequent handshake succeeds.
|
||||
auto remote_rx2 = remote->alloc_rx_assoc(/*fd*/ 251);
|
||||
ASSERT_TRUE(remote_rx2.has_value());
|
||||
ASSERT_TRUE(local->alloc_rx_assoc(51).has_value());
|
||||
EXPECT_EQ(0, local->install_tx_assoc(51, *remote_rx2));
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, InjectedAllocFailure) {
|
||||
MockConfig cfg;
|
||||
cfg.fail_next_alloc_rx_assoc = 2;
|
||||
auto backend = make_mock_backend(cfg);
|
||||
|
||||
EXPECT_FALSE(backend->alloc_rx_assoc(60).has_value());
|
||||
EXPECT_FALSE(backend->alloc_rx_assoc(61).has_value());
|
||||
// Budget consumed.
|
||||
EXPECT_TRUE(backend->alloc_rx_assoc(62).has_value());
|
||||
|
||||
// Failed allocs must not have consumed rx slots.
|
||||
auto caps = backend->get_dev_caps(0);
|
||||
ASSERT_TRUE(caps.has_value());
|
||||
EXPECT_EQ(caps->free_rx_assoc_slots, 64u - 1);
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, TxCapacityExhaustion) {
|
||||
MockConfig cfg;
|
||||
cfg.tx_capacity = 2;
|
||||
auto local = make_mock_backend(cfg);
|
||||
auto remote = make_mock_backend(MockConfig{});
|
||||
|
||||
for (int i = 0; i < 2; ++i) {
|
||||
auto peer_blob = remote->alloc_rx_assoc(200 + i);
|
||||
ASSERT_TRUE(peer_blob.has_value());
|
||||
EXPECT_EQ(0, local->install_tx_assoc(100 + i, *peer_blob));
|
||||
}
|
||||
auto peer_blob3 = remote->alloc_rx_assoc(202);
|
||||
ASSERT_TRUE(peer_blob3.has_value());
|
||||
EXPECT_EQ(-ENOSPC, local->install_tx_assoc(102, *peer_blob3));
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, RxCapacityExhaustion) {
|
||||
MockConfig cfg;
|
||||
cfg.rx_capacity = 2;
|
||||
auto backend = make_mock_backend(cfg);
|
||||
|
||||
EXPECT_TRUE(backend->alloc_rx_assoc(1).has_value());
|
||||
EXPECT_TRUE(backend->alloc_rx_assoc(2).has_value());
|
||||
EXPECT_FALSE(backend->alloc_rx_assoc(3).has_value());
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, RejectsMalformedBlob) {
|
||||
auto backend = make_mock_backend(MockConfig{});
|
||||
std::vector<uint8_t> garbage(16, 0xff);
|
||||
EXPECT_EQ(-EINVAL, backend->install_tx_assoc(50, garbage));
|
||||
|
||||
std::vector<uint8_t> wrong_size{'P', 'S', 'P', 'M'};
|
||||
EXPECT_EQ(-EINVAL, backend->install_tx_assoc(50, wrong_size));
|
||||
|
||||
// Right magic and size, unsupported version.
|
||||
std::vector<uint8_t> bad_ver(16, 0);
|
||||
bad_ver[0] = 'P'; bad_ver[1] = 'S'; bad_ver[2] = 'P'; bad_ver[3] = 'M';
|
||||
bad_ver[4] = 0xff; bad_ver[5] = 0xff;
|
||||
EXPECT_EQ(-EINVAL, backend->install_tx_assoc(50, bad_ver));
|
||||
}
|
||||
|
||||
TEST(PSPNetlinkMock, TeardownReleasesSlots) {
|
||||
auto local = make_mock_backend(MockConfig{});
|
||||
auto remote = make_mock_backend(MockConfig{});
|
||||
auto peer_blob = remote->alloc_rx_assoc(/*fd*/ 275);
|
||||
ASSERT_TRUE(peer_blob.has_value());
|
||||
ASSERT_TRUE(local->alloc_rx_assoc(/*fd*/ 75).has_value());
|
||||
ASSERT_EQ(0, local->install_tx_assoc(75, *peer_blob));
|
||||
|
||||
EXPECT_EQ(0, local->teardown(75));
|
||||
|
||||
auto caps = local->get_dev_caps(0);
|
||||
ASSERT_TRUE(caps.has_value());
|
||||
EXPECT_EQ(caps->free_tx_assoc_slots, 64u);
|
||||
EXPECT_EQ(caps->free_rx_assoc_slots, 64u);
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user
