[BUG]:read after write failed

[shuoranliu]

Describe the bug

python3 test_write_read_consistency.py --path /mnt/curvine/lsr-test/fs-test/

#!/usr/bin/env python3
"""
Test script to verify write-read consistency issue in curvine-fuse.
This reproduces the issue seen in YOLO training where results.csv read fails
after being written due to S3 InvalidRange error.
"""

import os
import sys
import time
import tempfile
import argparse

def test_write_then_read(base_path: str, file_name: str = "test_consistency.csv"):
    """
    Test write-then-read consistency.
    Simulates what YOLO does: write to a file, then immediately read it.
    """
    file_path = os.path.join(base_path, file_name)

    print(f"[TEST] Testing write-read consistency on: {file_path}")

    # Clean up if exists
    if os.path.exists(file_path):
        print(f"[CLEANUP] Removing existing file: {file_path}")
        os.remove(file_path)
        time.sleep(0.5)

    # Test 1: Write new file, then read
    print("\n=== Test 1: Create new file and read immediately ===")
    test_data_1 = "epoch,loss,accuracy\n1,0.5,0.8\n"

    print(f"[WRITE] Writing {len(test_data_1)} bytes...")
    with open(file_path, 'w') as f:
        f.write(test_data_1)
    print("[WRITE] Write completed")

    print("[READ] Reading file immediately after write...")
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        print(f"[READ] Read {len(content)} bytes successfully")
        if content == test_data_1:
            print("[PASS] Content matches!")
        else:
            print(f"[FAIL] Content mismatch!")
            print(f"  Expected: {repr(test_data_1)}")
            print(f"  Got: {repr(content)}")
            return False
    except Exception as e:
        print(f"[FAIL] Read failed: {e}")
        return False

    # Test 2: Append to file, then read (simulates results.csv append mode)
    print("\n=== Test 2: Append to file and read immediately ===")
    test_data_2 = "2,0.4,0.85\n"

    print(f"[APPEND] Appending {len(test_data_2)} bytes...")
    with open(file_path, 'a') as f:
        f.write(test_data_2)
    print("[APPEND] Append completed")

    print("[READ] Reading file immediately after append...")
    try:
        with open(file_path, 'r') as f:
            content = f.read()
        expected = test_data_1 + test_data_2
        print(f"[READ] Read {len(content)} bytes successfully")
        if content == expected:
            print("[PASS] Content matches!")
        else:
            print(f"[FAIL] Content mismatch!")
            print(f"  Expected: {repr(expected)}")
            print(f"  Got: {repr(content)}")
            return False
    except Exception as e:
        print(f"[FAIL] Read failed: {e}")
        return False

    # Test 3: Multiple rapid append-read cycles
    print("\n=== Test 3: Multiple rapid append-read cycles ===")
    for i in range(3, 6):
        append_data = f"{i},{0.5-i*0.05:.2f},{0.8+i*0.02:.2f}\n"
        print(f"[CYCLE {i}] Appending: {append_data.strip()}")

        with open(file_path, 'a') as f:
            f.write(append_data)

        try:
            with open(file_path, 'r') as f:
                content = f.read()
            lines = content.strip().split('\n')
            print(f"[CYCLE {i}] Read OK, {len(lines)} lines total")
        except Exception as e:
            print(f"[CYCLE {i}] Read FAILED: {e}")
            return False

    # Test 4: Check file stat after operations
    print("\n=== Test 4: File stat consistency ===")
    try:
        stat = os.stat(file_path)
        print(f"[STAT] File size: {stat.st_size} bytes")

        with open(file_path, 'r') as f:
            content = f.read()
        actual_len = len(content.encode('utf-8'))

        if stat.st_size == actual_len:
            print(f"[PASS] Stat size matches content length: {actual_len}")
        else:
            print(f"[FAIL] Stat size ({stat.st_size}) != content length ({actual_len})")
            return False
    except Exception as e:
        print(f"[FAIL] Stat check failed: {e}")
        return False

    # Cleanup
    print(f"\n[CLEANUP] Removing test file: {file_path}")
    os.remove(file_path)

    return True


def test_concurrent_write_read(base_path: str):
    """
    Test concurrent write and read operations.
    """
    import threading
    import random

    file_path = os.path.join(base_path, "test_concurrent.txt")
    errors = []

    print(f"\n=== Test 5: Concurrent write-read ===")
    print(f"[TEST] File: {file_path}")

    # Initialize file
    with open(file_path, 'w') as f:
        f.write("initial\n")

    def writer(thread_id, iterations):
        for i in range(iterations):
            try:
                with open(file_path, 'a') as f:
                    f.write(f"writer-{thread_id}-{i}\n")
            except Exception as e:
                errors.append(f"Writer {thread_id} failed at {i}: {e}")
            time.sleep(random.uniform(0.01, 0.05))

    def reader(thread_id, iterations):
        for i in range(iterations):
            try:
                with open(file_path, 'r') as f:
                    content = f.read()
                    _ = len(content)
            except Exception as e:
                errors.append(f"Reader {thread_id} failed at {i}: {e}")
            time.sleep(random.uniform(0.01, 0.05))

    threads = []
    for i in range(2):
        t = threading.Thread(target=writer, args=(i, 10))
        threads.append(t)
    for i in range(3):
        t = threading.Thread(target=reader, args=(i, 15))
        threads.append(t)

    print(f"[START] Starting {len(threads)} threads...")
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    if errors:
        print(f"[FAIL] {len(errors)} errors occurred:")
        for err in errors[:5]:
            print(f"  - {err}")
        return False
    else:
        print("[PASS] Concurrent test completed without errors")

    # Cleanup
    os.remove(file_path)
    return True


def test_large_file(base_path: str, size_mb: float = 1.0):
    """
    Test write-read with larger file.
    """
    file_path = os.path.join(base_path, "test_large.bin")
    size_bytes = int(size_mb * 1024 * 1024)

    print(f"\n=== Test 6: Large file ({size_mb} MB) ===")
    print(f"[TEST] File: {file_path}")

    # Generate test data
    test_data = os.urandom(size_bytes)

    print(f"[WRITE] Writing {size_bytes} bytes...")
    start = time.time()
    with open(file_path, 'wb') as f:
        f.write(test_data)
    write_time = time.time() - start
    print(f"[WRITE] Completed in {write_time:.2f}s ({size_mb/write_time:.2f} MB/s)")

    print("[READ] Reading file immediately...")
    start = time.time()
    try:
        with open(file_path, 'rb') as f:
            content = f.read()
        read_time = time.time() - start
        print(f"[READ] Read {len(content)} bytes in {read_time:.2f}s ({size_mb/read_time:.2f} MB/s)")

        if content == test_data:
            print("[PASS] Content matches!")
        else:
            print(f"[FAIL] Content mismatch! Got {len(content)} bytes, expected {len(test_data)}")
            return False
    except Exception as e:
        print(f"[FAIL] Read failed: {e}")
        return False

    # Cleanup
    os.remove(file_path)
    return True


def main():
    parser = argparse.ArgumentParser(description='Test write-read consistency on curvine-fuse')
    parser.add_argument('--path', '-p', default='/mnt/curvine/lsr-test/torch-test',
                        help='Base path to test (default: /mnt/curvine/lsr-test/torch-test)')
    parser.add_argument('--local', '-l', action='store_true',
                        help='Also run on local filesystem for comparison')
    parser.add_argument('--size', '-s', type=float, default=1.0,
                        help='Size in MB for large file test (default: 1.0)')
    args = parser.parse_args()

    results = {}

    # Test on curvine
    print("=" * 60)
    print(f"Testing on CURVINE: {args.path}")
    print("=" * 60)

    if not os.path.exists(args.path):
        print(f"[ERROR] Path does not exist: {args.path}")
        print("Make sure curvine-fuse is mounted at /mnt/curvine")
        sys.exit(1)

    results['curvine_basic'] = test_write_then_read(args.path)
    results['curvine_concurrent'] = test_concurrent_write_read(args.path)
    results['curvine_large'] = test_large_file(args.path, args.size)

    # Test on local filesystem for comparison
    if args.local:
        print("\n" + "=" * 60)
        print("Testing on LOCAL filesystem for comparison")
        print("=" * 60)

        with tempfile.TemporaryDirectory() as tmpdir:
            results['local_basic'] = test_write_then_read(tmpdir)
            results['local_concurrent'] = test_concurrent_write_read(tmpdir)
            results['local_large'] = test_large_file(tmpdir, args.size)

    # Summary
    print("\n" + "=" * 60)
    print("SUMMARY")
    print("=" * 60)
    all_passed = True
    for name, passed in results.items():
        status = "PASS" if passed else "FAIL"
        print(f"  {name}: {status}")
        if not passed:
            all_passed = False

    print("=" * 60)
    if all_passed:
        print("All tests PASSED!")
        sys.exit(0)
    else:
        print("Some tests FAILED!")
        sys.exit(1)


if __name__ == '__main__':
    main()

root@curvine-train-test-pod:/mnt/curvine/lsr-test# python3 test_write_read_consistency.py --path /mnt/curvine/lsr-test/fs-test/
============================================================
Testing on CURVINE: /mnt/curvine/lsr-test/fs-test/
============================================================
[TEST] Testing write-read consistency on: /mnt/curvine/lsr-test/fs-test/test_consistency.csv

=== Test 1: Create new file and read immediately ===
[WRITE] Writing 30 bytes...
[WRITE] Write completed
[READ] Reading file immediately after write...
[READ] Read 30 bytes successfully
[PASS] Content matches!

=== Test 2: Append to file and read immediately ===
[APPEND] Appending 11 bytes...
[APPEND] Append completed
[READ] Reading file immediately after append...
[READ] Read 41 bytes successfully
[PASS] Content matches!

=== Test 3: Multiple rapid append-read cycles ===
[CYCLE 3] Appending: 3,0.35,0.86
[CYCLE 3] Read OK, 4 lines total
[CYCLE 4] Appending: 4,0.30,0.88
[CYCLE 4] Read OK, 5 lines total
[CYCLE 5] Appending: 5,0.25,0.90
[CYCLE 5] Read OK, 6 lines total

=== Test 4: File stat consistency ===
[STAT] File size: 77 bytes
[PASS] Stat size matches content length: 77

[CLEANUP] Removing test file: /mnt/curvine/lsr-test/fs-test/test_consistency.csv

=== Test 5: Concurrent write-read ===
[TEST] File: /mnt/curvine/lsr-test/fs-test/test_concurrent.txt
[START] Starting 5 threads...
[FAIL] 6 errors occurred:
  - Writer 1 failed at 6: [Errno 5] Input/output error
  - Writer 0 failed at 6: [Errno 5] Input/output error
  - Writer 0 failed at 8: [Errno 5] Input/output error
  - Writer 1 failed at 8: [Errno 5] Input/output error
  - Reader 0 failed at 6: [Errno 5] Input/output error

=== Test 6: Large file (1.0 MB) ===
[TEST] File: /mnt/curvine/lsr-test/fs-test/test_large.bin
[WRITE] Writing 1048576 bytes...
[WRITE] Completed in 0.18s (5.57 MB/s)
[READ] Reading file immediately...
[READ] Read 1048576 bytes in 0.00s (370.03 MB/s)
[PASS] Content matches!

============================================================
SUMMARY
============================================================
  curvine_basic: PASS
  curvine_concurrent: FAIL
  curvine_large: PASS
============================================================
Some tests FAILED!

To Reproduce
Steps to reproduce the behavior:

Expected behavior
A clear and concise description of what you expected to happen.

Screenshots
If applicable, add screenshots to help explain your problem.

OS Version (please complete the following information):

Additional context
Add any other context about the problem here.

[shuoranliu]

python3 test_concurrent_io.py --path /mnt/curvine/lsr-test/fs-test/

#!/usr/bin/env python3
"""
Test script to diagnose concurrent I/O errors in curvine-fuse.
Reproduces the [Errno 5] Input/output error seen during concurrent access.
"""

import os
import sys
import time
import threading
import argparse
from datetime import datetime
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed


def ts():
    """Return current timestamp string for log correlation."""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]


def test_concurrent_append(base_path: str, num_writers: int = 2, iterations: int = 20):
    """
    Test concurrent append to same file.
    """
    file_path = os.path.join(base_path, "concurrent_append.txt")
    errors = []
    success_count = [0]
    lock = threading.Lock()

    print(f"\n[{ts()}] === Test: Concurrent append ({num_writers} writers, {iterations} iterations) ===")

    # Initialize file
    with open(file_path, 'w') as f:
        f.write("init\n")

    def writer(thread_id):
        for i in range(iterations):
            try:
                with open(file_path, 'a') as f:
                    f.write(f"T{thread_id}-I{i}\n")
                with lock:
                    success_count[0] += 1
            except Exception as e:
                errors.append((ts(), thread_id, i, type(e).__name__, str(e)))
            time.sleep(0.01)

    threads = [threading.Thread(target=writer, args=(i,)) for i in range(num_writers)]
    start = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    elapsed = time.time() - start

    total_ops = num_writers * iterations
    print(f"[{ts()}]   Success: {success_count[0]}/{total_ops} ({100*success_count[0]/total_ops:.1f}%)")
    print(f"[{ts()}]   Errors: {len(errors)}")
    print(f"[{ts()}]   Time: {elapsed:.2f}s")

    if errors:
        print(f"[{ts()}]   Error details:")
        for err_ts, tid, i, etype, msg in errors[:5]:
            print(f"    [{err_ts}] Thread {tid}, iter {i}: {etype}: {msg}")

    # Cleanup
    try:
        os.remove(file_path)
    except:
        pass

    return len(errors) == 0


def test_concurrent_read_same_file(base_path: str, num_readers: int = 5, iterations: int = 20):
    """
    Test concurrent read of same file.
    """
    file_path = os.path.join(base_path, "concurrent_read.txt")
    errors = []
    success_count = [0]
    lock = threading.Lock()

    print(f"\n[{ts()}] === Test: Concurrent read ({num_readers} readers, {iterations} iterations) ===")

    # Create test file
    test_content = "test line\n" * 100
    with open(file_path, 'w') as f:
        f.write(test_content)

    time.sleep(0.5)  # Let it sync

    def reader(thread_id):
        for i in range(iterations):
            try:
                with open(file_path, 'r') as f:
                    content = f.read()
                if len(content) > 0:
                    with lock:
                        success_count[0] += 1
            except Exception as e:
                errors.append((ts(), thread_id, i, type(e).__name__, str(e)))
            time.sleep(0.01)

    threads = [threading.Thread(target=reader, args=(i,)) for i in range(num_readers)]
    start = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    elapsed = time.time() - start

    total_ops = num_readers * iterations
    print(f"[{ts()}]   Success: {success_count[0]}/{total_ops} ({100*success_count[0]/total_ops:.1f}%)")
    print(f"[{ts()}]   Errors: {len(errors)}")
    print(f"[{ts()}]   Time: {elapsed:.2f}s")

    if errors:
        print(f"[{ts()}]   Error details:")
        for err_ts, tid, i, etype, msg in errors[:5]:
            print(f"    [{err_ts}] Thread {tid}, iter {i}: {etype}: {msg}")

    # Cleanup
    try:
        os.remove(file_path)
    except:
        pass

    return len(errors) == 0


def test_concurrent_read_different_files(base_path: str, num_files: int = 10, num_readers: int = 5):
    """
    Test concurrent read of different files (simulates dataloader).
    """
    errors = []
    success_count = [0]
    lock = threading.Lock()

    print(f"\n[{ts()}] === Test: Concurrent read different files ({num_files} files, {num_readers} readers) ===")

    # Create test files
    test_dir = os.path.join(base_path, "multi_files")
    os.makedirs(test_dir, exist_ok=True)

    files = []
    for i in range(num_files):
        fp = os.path.join(test_dir, f"file_{i:03d}.txt")
        with open(fp, 'w') as f:
            f.write(f"content of file {i}\n" * 50)
        files.append(fp)

    time.sleep(0.5)

    def reader(file_list):
        for fp in file_list:
            try:
                with open(fp, 'r') as f:
                    content = f.read()
                if len(content) > 0:
                    with lock:
                        success_count[0] += 1
            except Exception as e:
                errors.append((ts(), fp, type(e).__name__, str(e)))

    # Each reader reads all files
    threads = [threading.Thread(target=reader, args=(files,)) for _ in range(num_readers)]
    start = time.time()
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    elapsed = time.time() - start

    total_ops = num_files * num_readers
    print(f"[{ts()}]   Success: {success_count[0]}/{total_ops} ({100*success_count[0]/total_ops:.1f}%)")
    print(f"[{ts()}]   Errors: {len(errors)}")
    print(f"[{ts()}]   Time: {elapsed:.2f}s")

    if errors:
        print(f"[{ts()}]   Error details:")
        for err_ts, fp, etype, msg in errors[:5]:
            print(f"    [{err_ts}] {os.path.basename(fp)}: {etype}: {msg}")

    # Cleanup
    for fp in files:
        try:
            os.remove(fp)
        except:
            pass
    try:
        os.rmdir(test_dir)
    except:
        pass

    return len(errors) == 0


def test_read_while_write(base_path: str, duration_sec: float = 3.0):
    """
    Test reading while another thread is writing.
    """
    file_path = os.path.join(base_path, "read_while_write.txt")
    errors = defaultdict(list)
    counts = {'read': 0, 'write': 0}
    lock = threading.Lock()
    stop_flag = [False]

    print(f"\n[{ts()}] === Test: Read while write ({duration_sec}s) ===")

    # Initialize
    with open(file_path, 'w') as f:
        f.write("init\n")

    def writer():
        i = 0
        while not stop_flag[0]:
            try:
                with open(file_path, 'a') as f:
                    f.write(f"line-{i}\n")
                with lock:
                    counts['write'] += 1
                i += 1
            except Exception as e:
                errors['write'].append((ts(), i, type(e).__name__, str(e)))
            time.sleep(0.02)

    def reader():
        i = 0
        while not stop_flag[0]:
            try:
                with open(file_path, 'r') as f:
                    _ = f.read()
                with lock:
                    counts['read'] += 1
                i += 1
            except Exception as e:
                errors['read'].append((ts(), i, type(e).__name__, str(e)))
            time.sleep(0.01)

    writer_thread = threading.Thread(target=writer)
    reader_threads = [threading.Thread(target=reader) for _ in range(3)]

    writer_thread.start()
    for t in reader_threads:
        t.start()

    time.sleep(duration_sec)
    stop_flag[0] = True

    writer_thread.join()
    for t in reader_threads:
        t.join()

    print(f"[{ts()}]   Write ops: {counts['write']}, errors: {len(errors['write'])}")
    print(f"[{ts()}]   Read ops: {counts['read']}, errors: {len(errors['read'])}")

    all_errors = errors['write'] + errors['read']
    if all_errors:
        print(f"[{ts()}]   Error details:")
        for item in all_errors[:5]:
            print(f"    [{item[0]}] iter {item[1]}: {item[2]}: {item[3]}")

    # Cleanup
    try:
        os.remove(file_path)
    except:
        pass

    return len(all_errors) == 0


def test_open_close_rapid(base_path: str, iterations: int = 100):
    """
    Test rapid open/close cycles.
    """
    file_path = os.path.join(base_path, "rapid_open_close.txt")
    errors = []

    print(f"\n[{ts()}] === Test: Rapid open/close ({iterations} iterations) ===")

    # Create file
    with open(file_path, 'w') as f:
        f.write("test content\n" * 10)

    time.sleep(0.5)

    start = time.time()
    for i in range(iterations):
        try:
            with open(file_path, 'r') as f:
                _ = f.read()
        except Exception as e:
            errors.append((ts(), i, type(e).__name__, str(e)))

    elapsed = time.time() - start

    print(f"[{ts()}]   Success: {iterations - len(errors)}/{iterations}")
    print(f"[{ts()}]   Errors: {len(errors)}")
    print(f"[{ts()}]   Time: {elapsed:.2f}s ({iterations/elapsed:.1f} ops/s)")

    if errors:
        print(f"[{ts()}]   Error details:")
        for err_ts, i, etype, msg in errors[:5]:
            print(f"    [{err_ts}] iter {i}: {etype}: {msg}")

    # Cleanup
    try:
        os.remove(file_path)
    except:
        pass

    return len(errors) == 0


def test_threadpool_read(base_path: str, num_files: int = 20, max_workers: int = 8):
    """
    Test with ThreadPoolExecutor (similar to PyTorch DataLoader).
    """
    errors = []
    success_count = [0]

    print(f"\n[{ts()}] === Test: ThreadPoolExecutor read ({num_files} files, {max_workers} workers) ===")

    # Create test files
    test_dir = os.path.join(base_path, "threadpool_test")
    os.makedirs(test_dir, exist_ok=True)

    files = []
    for i in range(num_files):
        fp = os.path.join(test_dir, f"data_{i:03d}.bin")
        with open(fp, 'wb') as f:
            f.write(os.urandom(10000))  # 10KB each
        files.append(fp)

    time.sleep(0.5)

    def read_file(fp):
        with open(fp, 'rb') as f:
            return f.read()

    start = time.time()
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(read_file, fp): fp for fp in files}
        for future in as_completed(futures):
            fp = futures[future]
            try:
                data = future.result()
                if len(data) > 0:
                    success_count[0] += 1
            except Exception as e:
                errors.append((ts(), os.path.basename(fp), type(e).__name__, str(e)))

    elapsed = time.time() - start

    print(f"[{ts()}]   Success: {success_count[0]}/{num_files}")
    print(f"[{ts()}]   Errors: {len(errors)}")
    print(f"[{ts()}]   Time: {elapsed:.2f}s")

    if errors:
        print(f"[{ts()}]   Error details:")
        for err_ts, fp, etype, msg in errors[:5]:
            print(f"    [{err_ts}] {fp}: {etype}: {msg}")

    # Cleanup
    for fp in files:
        try:
            os.remove(fp)
        except:
            pass
    try:
        os.rmdir(test_dir)
    except:
        pass

    return len(errors) == 0


def main():
    parser = argparse.ArgumentParser(description='Test concurrent I/O on curvine-fuse')
    parser.add_argument('--path', '-p', default='/mnt/curvine/lsr-test/fs-test',
                        help='Base path to test')
    parser.add_argument('--workers', '-w', type=int, default=8,
                        help='Number of workers for threadpool test')
    args = parser.parse_args()

    if not os.path.exists(args.path):
        print(f"[{ts()}] [ERROR] Path does not exist: {args.path}")
        sys.exit(1)

    print("=" * 60)
    print(f"[{ts()}] Concurrent I/O Test on: {args.path}")
    print("=" * 60)

    results = {}
    results['concurrent_append'] = test_concurrent_append(args.path)
    results['concurrent_read_same'] = test_concurrent_read_same_file(args.path)
    results['concurrent_read_diff'] = test_concurrent_read_different_files(args.path)
    results['read_while_write'] = test_read_while_write(args.path)
    results['rapid_open_close'] = test_open_close_rapid(args.path)
    results['threadpool_read'] = test_threadpool_read(args.path, max_workers=args.workers)

    # Summary
    print("\n" + "=" * 60)
    print(f"[{ts()}] SUMMARY")
    print("=" * 60)
    passed = 0
    for name, ok in results.items():
        status = "PASS" if ok else "FAIL"
        print(f"  {name}: {status}")
        if ok:
            passed += 1

    print(f"\n[{ts()}]   Total: {passed}/{len(results)} passed")
    print("=" * 60)

    sys.exit(0 if passed == len(results) else 1)


if __name__ == '__main__':
    main()

root@curvine-train-test-pod:/workspace# python3 test_concurrent_io.py --path /mnt/curvine/lsr-test/fs-test/
============================================================
[2026-02-11 07:06:42.341] Concurrent I/O Test on: /mnt/curvine/lsr-test/fs-test/
============================================================

[2026-02-11 07:06:42.341] === Test: Concurrent append (2 writers, 20 iterations) ===
[2026-02-11 07:06:42.852]   Success: 40/40 (100.0%)
[2026-02-11 07:06:42.852]   Errors: 0
[2026-02-11 07:06:42.852]   Time: 0.25s

[2026-02-11 07:06:42.951] === Test: Concurrent read (5 readers, 20 iterations) ===
[2026-02-11 07:06:43.830]   Success: 100/100 (100.0%)
[2026-02-11 07:06:43.830]   Errors: 0
[2026-02-11 07:06:43.830]   Time: 0.22s

[2026-02-11 07:06:43.922] === Test: Concurrent read different files (10 files, 5 readers) ===
[2026-02-11 07:06:46.219]   Success: 50/50 (100.0%)
[2026-02-11 07:06:46.219]   Errors: 0
[2026-02-11 07:06:46.219]   Time: 0.02s

[2026-02-11 07:06:47.201] === Test: Read while write (3.0s) ===
[2026-02-11 07:06:50.426]   Write ops: 19, errors: 16
[2026-02-11 07:06:50.426]   Read ops: 138, errors: 18
[2026-02-11 07:06:50.426]   Error details:
    [2026-02-11 07:06:47.767] iter 8: OSError: [Errno 5] Input/output error
    [2026-02-11 07:06:47.870] iter 9: OSError: [Errno 5] Input/output error
    [2026-02-11 07:06:48.057] iter 9: OSError: [Errno 5] Input/output error
    [2026-02-11 07:06:48.499] iter 16: OSError: [Errno 5] Input/output error
    [2026-02-11 07:06:48.672] iter 19: OSError: [Errno 5] Input/output error

[2026-02-11 07:06:50.515] === Test: Rapid open/close (100 iterations) ===
Traceback (most recent call last):
  File "/workspace/test_concurrent_io.py", line 421, in <module>
    main()
  File "/workspace/test_concurrent_io.py", line 400, in main
    results['rapid_open_close'] = test_open_close_rapid(args.path)
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/workspace/test_concurrent_io.py", line 283, in test_open_close_rapid
    with open(file_path, 'w') as f:
OSError: [Errno 5] Input/output error

[jlon]

Status update: issue kept OPEN as requested.\n\nWe pushed a follow-up fix in PR #655 (commit c24a580) and re-ran all scripts posted in this issue/comment thread:\n- test_write_read_consistency.py\n- test_concurrent_io.py\n\nCurrent local results: all tests pass, including repeated stability rounds.\n\nPR details: #655

[shuoranliu]

root@curvine-train-test-pod:/workspace# python3 test_concurrent_io.py --path /mnt/curvine/lsr-test/fs-test/
============================================================
[2026-02-11 14:45:04.956] Concurrent I/O Test on: /mnt/curvine/lsr-test/fs-test/
============================================================

[2026-02-11 14:45:04.956] === Test: Concurrent append (2 writers, 20 iterations) ===
[2026-02-11 14:45:06.907]   Success: 40/40 (100.0%)
[2026-02-11 14:45:06.907]   Errors: 0
[2026-02-11 14:45:06.907]   Time: 1.73s

[2026-02-11 14:45:07.014] === Test: Concurrent read (5 readers, 20 iterations) ===
[2026-02-11 14:45:08.228]   Success: 100/100 (100.0%)
[2026-02-11 14:45:08.228]   Errors: 0
[2026-02-11 14:45:08.228]   Time: 0.42s

[2026-02-11 14:45:08.320] === Test: Concurrent read different files (10 files, 5 readers) ===
[2026-02-11 14:45:11.480]   Success: 50/50 (100.0%)
[2026-02-11 14:45:11.480]   Errors: 0
[2026-02-11 14:45:11.480]   Time: 0.14s

[2026-02-11 14:45:13.086] === Test: Read while write (3.0s) ===
[2026-02-11 14:45:16.374]   Write ops: 38, errors: 0
[2026-02-11 14:45:16.374]   Read ops: 129, errors: 9
[2026-02-11 14:45:16.374]   Error details:
    [2026-02-11 14:45:13.754] iter 10: OSError: [Errno 5] Input/output error
    [2026-02-11 14:45:14.290] iter 13: OSError: [Errno 5] Input/output error
    [2026-02-11 14:45:14.738] iter 15: OSError: [Errno 5] Input/output error
    [2026-02-11 14:45:14.848] iter 19: OSError: [Errno 5] Input/output error
    [2026-02-11 14:45:14.848] iter 15: OSError: [Errno 5] Input/output error

[2026-02-11 14:45:16.451] === Test: Rapid open/close (100 iterations) ===
[2026-02-11 14:45:17.314]   Success: 100/100
[2026-02-11 14:45:17.314]   Errors: 0
[2026-02-11 14:45:17.314]   Time: 0.17s (578.1 ops/s)

[2026-02-11 14:45:17.395] === Test: ThreadPoolExecutor read (20 files, 8 workers) ===
[2026-02-11 14:45:22.490]   Success: 20/20
[2026-02-11 14:45:22.490]   Errors: 0
[2026-02-11 14:45:22.490]   Time: 0.13s

============================================================
[2026-02-11 14:45:25.487] SUMMARY
============================================================
  concurrent_append: PASS
  concurrent_read_same: PASS
  concurrent_read_diff: PASS
  read_while_write: FAIL
  rapid_open_close: PASS
  threadpool_read: PASS

[2026-02-11 14:45:25.487]   Total: 5/6 passed
============================================================

问题分析
根本原因
read_while_write 测试失败的核心问题是 读写并发时的元数据不一致，日志显示了两种错误模式：

1. S3 InvalidRange 错误 (14:45:13.754)


S3Error { code: "InvalidRange", message: "The requested range is not satisfiable", 
  range: 0-11 }
Reader 获取到旧的文件长度 (len=54)
Writer 同时在 append，文件大小在变化 (47→54→61...)
当 Reader 尝试从 S3 读取时，文件已被 truncate/rewrite，导致 range 请求失效
2. 路径状态不一致错误

FuseError { errno: 5, error: "path is not curvine path" }
出现在 unified_filesystem.rs:98 - get_mount() 检查失败
原因：在 cache sync 期间，路径的 Curvine 元数据状态不完整
3. Hydrate job 冲突


Reject hydrate load for ... because an active load job is already touching this path
Writer 的 sync job 正在运行时，Reader 尝试触发 hydrate 被拒绝
这导致 fallback 到 UFS 读取，但 UFS 的状态也不一致
事件序列

14:45:13.730 - Writer creates at pos=47
14:45:13.731 - Reader reads from cache (len=54, 旧缓存)
14:45:13.754 - S3 返回 InvalidRange (文件已被修改)
14:45:13.787 - Hydrate job 被拒绝 (active job conflict)
14:45:14.289 - "path is not curvine path" 错误
是否需要 Master/Worker 日志？
不一定需要，当前 FUSE 日志已经清楚显示了问题：

Master 日志只会显示更多 hydrate job rejection 细节
Worker 日志可能有 cache write 的详细信息
但如果需要验证以下情况，可以查看：

Master 日志：确认 sync job 的完整生命周期和冲突处理逻辑
Worker 日志：查看 cache 块的写入状态和 invalidation 时序
修复方向
这是已知的并发写读场景的竞态问题，可能的修复：

Reader 在遇到 InvalidRange 时应该重新获取元数据并重试
在有活跃 write job 时，read 应该等待或直接读 UFS 最新数据
get_mount() 应该对 transitional 状态更宽容

[shuoranliu]

# Curvine read_while_write Test Error Summary
# Date: 2026-02-12
# Test: python3 test_concurrent_io.py --path /mnt/curvine/cube-studio-eu/curvine/lsr-test/fs-test/

================================================================================
TEST RESULT
================================================================================
read_while_write: FAIL
- Write ops: 21, errors: 6
- Read ops: 73, errors: 11

================================================================================
ERROR TYPE 1: Not found block (NEW)
================================================================================
Location: curvine-server/src/master/meta/inode/inode_file.rs:281

[02:12:31.726] Not found block, block_id = 527475671040
[02:12:31.890] Not found block, block_id = 527475671041
[02:10:08.949] Not found block, block_id = 527089795082
[02:10:09.136] Not found block, block_id = 527089795083
[02:10:09.941] Not found block, block_id = 527089795087

Root Cause: Block metadata lost during concurrent write operations.
            Writer creates block but metadata is cleared/reset by another operation.

================================================================================
ERROR TYPE 2: Complete len mismatch (NEW)
================================================================================
Location: curvine-server/src/master/meta/inode/inode_file.rs:313

[02:10:09.472] Complete len is not equal to file len, complete_len = 0, file_len = 8
[02:10:09.709] Complete len is not equal to file len, complete_len = 0, file_len = 16

Root Cause: Write operation not properly committed before close.
            Inode shows file_len but no blocks are marked complete.

================================================================================
ERROR TYPE 3: No available worker (NEW)
================================================================================
Location: curvine-client/src/block/block_writer.rs:149

[02:12:32.090] There is no available worker

Root Cause: Block allocation fails because location info is missing/stale.
            Likely caused by inode reset clearing block locations.

================================================================================
ERROR TYPE 4: Path is not curvine path
================================================================================
Location: curvine-client/src/unified/unified_filesystem.rs:89 (was line 98)

[02:12:31.726] path is not curvine path
[02:12:31.852] path is not curvine path
[02:12:31.853] path is not curvine path

Root Cause: get_mount() check fails during transitional state.
            Path metadata is incomplete when reader accesses it.

================================================================================
ERROR TYPE 5: S3 InvalidRange (HTTP 416)
================================================================================
Location: curvine-client/src/services.rs:192

[02:12:32.351] S3Error { code: "InvalidRange", message: "The requested range is not satisfiable" }
             uri: https://s3.eu-central-1.amazonaws.com/cube-studio-eu/curvine/lsr-test/fs-test/read_while_write.txt
             range: 0-4

Root Cause: Reader uses stale metadata (len=14) but S3 file has different size.
            File was modified by concurrent writer after metadata fetch.

================================================================================
ERROR TYPE 6: Channel closed
================================================================================
[02:12:31.726] IO([]channel closed)
[02:12:32.090] IO([]channel closed)

Root Cause: Writer channel dropped due to upstream error (block not found / no worker).
            Secondary error caused by primary failures above.

================================================================================
TIMELINE (02:12:31 - 02:12:32)
================================================================================
02:12:31.726 [Writer] Create writer pos=0, len=0
02:12:31.726 [Master] ERROR: Not found block, block_id = 527475671040
02:12:31.726 [FUSE]   errno 5: path is not curvine path
02:12:31.890 [Master] ERROR: Not found block, block_id = 527475671041
02:12:32.090 [Writer] ERROR: There is no available worker
02:12:32.090 [FUSE]   errno 5: channel closed
02:12:32.351 [S3]     ERROR: InvalidRange, range: 0-4

================================================================================
COMPARISON: 2026-02-11 vs 2026-02-12 (after code change)
================================================================================
| Error Type                    | 02-11 | 02-12 |
|-------------------------------|-------|-------|
| S3 InvalidRange               |  YES  |  YES  |
| path is not curvine path      |  L98  |  L89  | <- line changed
| Reject hydrate (active job)   |  YES  |  NO   | <- disappeared
| Reject hydrate (incomplete)   |  YES  |  NO   | <- disappeared
| Not found block               |  NO   |  YES  | <- NEW
| Complete len mismatch         |  NO   |  YES  | <- NEW
| No available worker           |  NO   |  YES  | <- NEW

================================================================================
DIAGNOSIS
================================================================================
The code change introduced new block metadata management issues:

1. Block IDs are being allocated but not properly registered in inode metadata
2. When writer tries to commit, master cannot find the block -> EIO
3. This cascades to "no available worker" and "channel closed" errors
4. The hydrate rejection errors disappeared, suggesting that path was modified

Key files to investigate:
- curvine-server/src/master/meta/inode/inode_file.rs (lines 281, 313)
- curvine-client/src/unified/unified_filesystem.rs (line 89)
- curvine-client/src/block/block_writer.rs (line 149)

================================================================================

[shuoranliu]

=== test-log.txt ===

[2026-02-12 02:32:27.190] === Test: Read while write (3.0s) ===
[2026-02-12 02:32:30.458]   Write ops: 39, errors: 6
[2026-02-12 02:32:30.458]   Read ops: 204, errors: 9
[2026-02-12 02:32:30.458]   Error details:
    [2026-02-12 02:32:27.572] iter 1: OSError: [Errno 5] Input/output error
    [2026-02-12 02:32:27.841] iter 2: OSError: [Errno 5] Input/output error
    [2026-02-12 02:32:28.345] iter 5: OSError: [Errno 5] Input/output error
    [2026-02-12 02:32:29.160] iter 29: OSError: [Errno 5] Input/output error
    [2026-02-12 02:32:30.165] iter 37: OSError: [Errno 5] Input/output error
[2026-02-12 02:32:30.548]   read_while_write: FAIL
[2026-02-12 02:32:30.548]   Total: 0/1 passed

=== fuse.log.2026-02-12 (errno: 5 / EIO errors) ===

26/02/12 02:32:27.544 WARN fuse_response.rs:120 send_rep unique 122: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:27.571 WARN fuse_response.rs:120 send_rep unique 162: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:27.571 WARN fuse_response.rs:120 send_rep unique 164: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:27.572 WARN fuse_response.rs:120 send_rep unique 166: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:27.572 WARN fuse_response.rs:120 send_rep unique 168: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:27.841 WARN fuse_response.rs:120 send_rep unique 306: FuseError { errno: 5, error: Common([][curvine-master] ERROR: Not found block, block_id = 527492448257(curvine-server/src/master/meta/inode/inode_file.rs:281)) }
26/02/12 02:32:27.841 WARN fuse_response.rs:120 send_rep unique 308: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:28.067 WARN fuse_response.rs:120 send_rep unique 508: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:28.162 WARN fuse_response.rs:120 send_rep unique 538: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:28.345 WARN fuse_response.rs:120 send_rep unique 624: FuseError { errno: 5, error: Common([][curvine-master] ERROR: Not found block, block_id = 527492448260(curvine-server/src/master/meta/inode/inode_file.rs:281)) }
26/02/12 02:32:28.345 WARN fuse_response.rs:120 send_rep unique 632: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:28.488 WARN fuse_response.rs:120 send_rep unique 692: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:29.134 WARN fuse_response.rs:120 send_rep unique 3006: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:29.160 WARN fuse_response.rs:120 send_rep unique 3028: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:29.160 WARN fuse_response.rs:120 send_rep unique 3030: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:29.160 WARN fuse_response.rs:120 send_rep unique 3032: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:29.160 WARN fuse_response.rs:120 send_rep unique 3034: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:29.424 WARN fuse_response.rs:120 send_rep unique 3158: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:29.682 WARN fuse_response.rs:120 send_rep unique 3514: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:29.909 WARN fuse_response.rs:120 send_rep unique 3622: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:29.993 WARN fuse_response.rs:120 send_rep unique 3656: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: path is not curvine path(curvine-client/src/unified/unified_filesystem.rs:89)) }
26/02/12 02:32:30.165 WARN fuse_response.rs:120 send_rep unique 3752: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.165 WARN fuse_response.rs:120 send_rep unique 3754: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.165 WARN fuse_response.rs:120 send_rep unique 3756: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.165 WARN fuse_response.rs:120 send_rep unique 3758: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.323 WARN fuse_response.rs:120 send_rep unique 3880: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.323 WARN fuse_response.rs:120 send_rep unique 3882: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.323 WARN fuse_response.rs:120 send_rep unique 3884: FuseError { errno: 5, error: IO([]channel closed) }
26/02/12 02:32:30.323 WARN fuse_response.rs:120 send_rep unique 3886: FuseError { errno: 5, error: IO([]channel closed) }

[shuoranliu]

# rm -rf /mnt/curvine/cube-studio-eu/curvine/lsr-test/fs-test/*
rm: cannot remove '/mnt/curvine/cube-studio-eu/curvine/lsr-test/fs-test/concurrent_append.txt': Input/output error

26/02/12 09:23:38.465 INFO web_server.rs:14 FUSE metrics server listening on 0.0.0.0:9002
26/02/12 09:23:38.466 INFO block_client_pool.rs:69 block client pool created: enable=true, max_idle_size=128, idle_time_ms=60000ms    
26/02/12 09:23:38.466 INFO curvine_filesystem.rs:58 Create new filesystem, git version: 0edf025, masters: 10.95.184.55:31000,10.95.173.218:31001,10.95.183.242:31002, threads: 32-32, buffer(rw): 131072-131072, conn timeout(ms): 30000-60000, rpc timeout(ms): 120000-120000, data timeout(ms): 120000    
26/02/12 09:23:38.466 INFO fuse_pure.rs:139 sys-mount options: fd=12,rootmode=40755,user_id=0,group_id=0,allow_other,async; flags: 0x0    
26/02/12 09:23:38.467 INFO fuse_pure.rs:178 Mounted at /mnt/curvine    
26/02/12 09:23:38.467 INFO fuse_mnt.rs:52 fuse mount success, path "/mnt/curvine", fd 12    
26/02/12 09:23:38.470 INFO fuse_session.rs:66 Create fuse session, git version: 0edf025, mnt number: 1, loop task number: 32,io threads: 32, worker threads: 32, fuse channel size: 0    
26/02/12 09:23:38.470 INFO fuse_session.rs:94 fuse session started running    
26/02/12 09:23:49.593 INFO buffer_client.rs:222 Request(-5218258813106666972,[10.95.166.80:35288] -> [10.95.173.218:31001]) callback execute fail: Ok(RpcMessage { protocol: Protocol { code: 14, status: Status(Rpc, Error), req_id: -5218258813106666972, seq_id: -1 }, header: None, data: Buffer(b"\0\0\0\x02\0\0\0@Not a leader master, code=GetMasterInfo, client_ip=10.95.173.218\0\0\0\0") })    
26/02/12 09:24:19.269 WARN fuse_response.rs:120 send_rep unique 204: FuseError { errno: 5, error: Common([][curvine-fuse] ERROR: job job_e5eb8c1ffc27dcd33886ad4366b164ef Failed: task job_e5eb8c1ffc27dcd33886ad4366b164ef_task_0 failed: File /cube-studio-eu/curvine/lsr-test/fs-test/concurrent_append.txt not found: (curvine-client/src/rpc/job_master_client.rs:128)) }    
26/02/12 09:25:49.595 INFO buffer_client.rs:97 close idle connection: [10.95.166.80:35288] -> [10.95.173.218:31001]    
26/02/12 09:25:49.595 INFO buffer_client.rs:97 close idle connection: [10.95.166.80:43766] -> [10.95.184.55:31000]