mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 04:16:38 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2c3e68ad89 | |||
| 2f0918f60b | |||
| 5b951de2b7 |
@@ -400,7 +400,7 @@ jobs:
|
|||||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||||
needs: [ quick-checks ]
|
needs: [ quick-checks ]
|
||||||
runs-on: sm-standard-4
|
runs-on: sm-standard-4
|
||||||
timeout-minutes: 45
|
timeout-minutes: 90
|
||||||
env:
|
env:
|
||||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||||
steps:
|
steps:
|
||||||
@@ -440,7 +440,7 @@ jobs:
|
|||||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||||
needs: [ quick-checks ]
|
needs: [ quick-checks ]
|
||||||
runs-on: sm-standard-4
|
runs-on: sm-standard-4
|
||||||
timeout-minutes: 60
|
timeout-minutes: 90
|
||||||
env:
|
env:
|
||||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||||
steps:
|
steps:
|
||||||
@@ -470,7 +470,7 @@ jobs:
|
|||||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||||
needs: [ quick-checks ]
|
needs: [ quick-checks ]
|
||||||
runs-on: sm-standard-4
|
runs-on: sm-standard-4
|
||||||
timeout-minutes: 60
|
timeout-minutes: 90
|
||||||
strategy:
|
strategy:
|
||||||
# On a PR, one failing protocol leg is enough to know the PR is not ready,
|
# On a PR, one failing protocol leg is enough to know the PR is not ready,
|
||||||
# so stop the sibling leg instead of paying another ~40 minutes for it.
|
# so stop the sibling leg instead of paying another ~40 minutes for it.
|
||||||
|
|||||||
@@ -57,6 +57,13 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
|
|||||||
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
|
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
|
||||||
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
|
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
|
||||||
|
|
||||||
|
/// Dedicated blocking thread pool for fsync/fdatasync operations.
|
||||||
|
/// When > 1, fsync operations are isolated from the main blocking pool to
|
||||||
|
/// prevent device-bound fsync from starving read operations (pread/stat/open).
|
||||||
|
/// Default 0 means auto (no isolation, use main runtime).
|
||||||
|
pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
|
||||||
|
pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
|
||||||
|
|
||||||
// Dial9 Tokio Telemetry Default values
|
// Dial9 Tokio Telemetry Default values
|
||||||
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
|
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
|
||||||
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
|
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
|
||||||
|
|||||||
@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
|
|||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
{
|
{
|
||||||
let dir = dir.as_ref().to_path_buf();
|
let dir = dir.as_ref().to_path_buf();
|
||||||
tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
|
fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(not(unix))]
|
#[cfg(not(unix))]
|
||||||
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
let dir = group.dir.clone();
|
let dir = group.dir.clone();
|
||||||
let dir_file = group.dir_file.clone();
|
let dir_file = group.dir_file.clone();
|
||||||
tokio::task::spawn_blocking(move || {
|
fsync_spawn_blocking(move || {
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
{
|
{
|
||||||
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
|
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
|
||||||
@@ -1080,6 +1080,44 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
|
|||||||
|
|
||||||
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
|
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
|
||||||
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
|
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||||
|
|
||||||
|
/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
|
||||||
|
/// configured with >1 threads, isolates device-bound fsync from the main
|
||||||
|
/// blocking pool so reads (pread/stat/open) are not starved. `None` means
|
||||||
|
/// fall back to the main runtime (zero behavior change).
|
||||||
|
static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
|
||||||
|
let threads =
|
||||||
|
rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
|
||||||
|
if threads <= 1 {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let mut builder = tokio::runtime::Builder::new_multi_thread();
|
||||||
|
builder
|
||||||
|
.worker_threads(num_cpus::get().min(8))
|
||||||
|
.max_blocking_threads(threads)
|
||||||
|
.thread_name("rustfs-fsync")
|
||||||
|
.thread_stack_size(512 * 1024)
|
||||||
|
.enable_all();
|
||||||
|
match builder.build() {
|
||||||
|
Ok(rt) => {
|
||||||
|
tracing::info!(threads, "fsync dedicated blocking pool enabled");
|
||||||
|
Some(rt)
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
/// Spawn a blocking task on the fsync-dedicated runtime if configured,
|
||||||
|
/// otherwise fall back to the main tokio blocking pool.
|
||||||
|
fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
|
||||||
|
match FSYNC_RUNTIME.as_ref() {
|
||||||
|
Some(rt) => rt.spawn_blocking(f),
|
||||||
|
None => tokio::task::spawn_blocking(f),
|
||||||
|
}
|
||||||
|
}
|
||||||
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
|
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
|
||||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||||
type NamespaceMutationLock = AsyncMutex<()>;
|
type NamespaceMutationLock = AsyncMutex<()>;
|
||||||
@@ -1217,7 +1255,7 @@ where
|
|||||||
F: FnOnce() -> io::Result<T> + Send + 'static,
|
F: FnOnce() -> io::Result<T> + Send + 'static,
|
||||||
{
|
{
|
||||||
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
|
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
|
||||||
let result = tokio::task::spawn_blocking(move || {
|
let result = fsync_spawn_blocking(move || {
|
||||||
let _disk_permit = disk_permit;
|
let _disk_permit = disk_permit;
|
||||||
work()
|
work()
|
||||||
})
|
})
|
||||||
@@ -2146,7 +2184,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
|
|||||||
wait_started,
|
wait_started,
|
||||||
);
|
);
|
||||||
let disk_permit = admission.disk_permit.clone();
|
let disk_permit = admission.disk_permit.clone();
|
||||||
let result = tokio::task::spawn_blocking(move || {
|
let result = fsync_spawn_blocking(move || {
|
||||||
let _lease = lease;
|
let _lease = lease;
|
||||||
let _disk_permit = disk_permit;
|
let _disk_permit = disk_permit;
|
||||||
operation()
|
operation()
|
||||||
|
|||||||
@@ -206,6 +206,13 @@ def check_runner_selection(root: Path) -> list[str]:
|
|||||||
return errors
|
return errors
|
||||||
|
|
||||||
|
|
||||||
|
def check_s3_tests_runner(root: Path) -> list[str]:
|
||||||
|
runner = (root / "scripts/s3-tests/run.sh").read_text()
|
||||||
|
if "--showlocals" in runner:
|
||||||
|
return ["scripts/s3-tests/run.sh: pytest failure diagnostics must not dump local values"]
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
def profile_selection(root: Path, profile: str) -> str:
|
def profile_selection(root: Path, profile: str) -> str:
|
||||||
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
|
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
|
||||||
raise ValueError(f"invalid e2e profile name: {profile}")
|
raise ValueError(f"invalid e2e profile name: {profile}")
|
||||||
@@ -272,6 +279,7 @@ def validate(root: Path) -> list[str]:
|
|||||||
errors.extend(check_e2e_modules(root))
|
errors.extend(check_e2e_modules(root))
|
||||||
errors.extend(check_fuzz_targets(root))
|
errors.extend(check_fuzz_targets(root))
|
||||||
errors.extend(check_runner_selection(root))
|
errors.extend(check_runner_selection(root))
|
||||||
|
errors.extend(check_s3_tests_runner(root))
|
||||||
errors.extend(check_profile_definitions(root))
|
errors.extend(check_profile_definitions(root))
|
||||||
return errors
|
return errors
|
||||||
|
|
||||||
@@ -341,6 +349,23 @@ class SelfTests(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
self.assertEqual(len(check_fuzz_targets(root)), 1)
|
self.assertEqual(len(check_fuzz_targets(root)), 1)
|
||||||
|
|
||||||
|
def test_s3_runner_rejects_unbounded_failure_locals(self) -> None:
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
root = Path(tmp)
|
||||||
|
runner = root / "scripts/s3-tests/run.sh"
|
||||||
|
runner.parent.mkdir(parents=True)
|
||||||
|
runner.write_text("tox -- -vv -ra --tb=long\n")
|
||||||
|
self.assertEqual(check_s3_tests_runner(root), [])
|
||||||
|
runner.write_text("tox -- -vv -ra --showlocals --tb=long\n")
|
||||||
|
self.assertEqual(len(check_s3_tests_runner(root)), 1)
|
||||||
|
with (
|
||||||
|
mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
|
||||||
|
mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
|
||||||
|
mock.patch(__name__ + ".check_runner_selection", return_value=[]),
|
||||||
|
mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
|
||||||
|
):
|
||||||
|
self.assertEqual(len(validate(root)), 1)
|
||||||
|
|
||||||
def test_profile_listing_enforces_selection(self) -> None:
|
def test_profile_listing_enforces_selection(self) -> None:
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
root = Path(tmp)
|
root = Path(tmp)
|
||||||
@@ -411,7 +436,7 @@ def main() -> int:
|
|||||||
for error in errors:
|
for error in errors:
|
||||||
print(f"ERROR: {error}", file=sys.stderr)
|
print(f"ERROR: {error}", file=sys.stderr)
|
||||||
return 1
|
return 1
|
||||||
print("OK: e2e modules, runner selection, fuzz matrices, and profile guards are wired")
|
print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1028,10 +1028,11 @@ else
|
|||||||
fi
|
fi
|
||||||
|
|
||||||
# Run tests from s3tests/functional
|
# Run tests from s3tests/functional
|
||||||
|
# Failure locals can contain multi-MiB request bodies; keep tracebacks without expanding local values.
|
||||||
set +e
|
set +e
|
||||||
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
|
S3TEST_CONF="${CONF_OUTPUT_PATH}" \
|
||||||
tox -- \
|
tox -- \
|
||||||
-vv -ra --showlocals --tb=long \
|
-vv -ra --tb=long \
|
||||||
--maxfail="${MAXFAIL}" \
|
--maxfail="${MAXFAIL}" \
|
||||||
--timeout="${TEST_TIMEOUT}" \
|
--timeout="${TEST_TIMEOUT}" \
|
||||||
--junitxml="${ARTIFACTS_DIR}/junit.xml" \
|
--junitxml="${ARTIFACTS_DIR}/junit.xml" \
|
||||||
|
|||||||
Reference in New Issue
Block a user