wehub-resource-sync
1939 lines
64 KiB
Python
1939 lines
64 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""
|
|
Unit tests for L1Manager.
|
|
|
|
These tests verify the behavior of L1Manager as described in the
|
|
interface docstrings. The tests focus on:
|
|
|
|
1. reserve_read() - Reserve read access for given keys
|
|
- Returns KEY_NOT_EXIST if key does not exist
|
|
- Returns KEY_NOT_READABLE if key exists but is write-locked
|
|
- Returns SUCCESS and MemoryObj if key is readable
|
|
|
|
2. unsafe_read() - Unsafe read without acquiring new read locks
|
|
- Returns KEY_NOT_EXIST if key does not exist
|
|
- Returns KEY_NOT_READABLE if key is not read-locked
|
|
- Returns SUCCESS and MemoryObj if key is read-locked
|
|
|
|
3. finish_read() - Finish read access for given keys
|
|
- Returns KEY_NOT_EXIST if key does not exist
|
|
- Returns KEY_IN_WRONG_STATE if key is write-locked or non-read-locked
|
|
- Returns SUCCESS on successful unlock
|
|
- Deletes temporary objects when read count reaches zero
|
|
|
|
4. reserve_write() - Reserve write access for given keys
|
|
- Returns KEY_NOT_WRITABLE if key exists but cannot be written
|
|
- Returns OUT_OF_MEMORY if allocation fails
|
|
- Returns SUCCESS and MemoryObj on success
|
|
|
|
5. finish_write() - Finish write access for given keys
|
|
- Returns KEY_NOT_EXIST if key does not exist
|
|
- Returns KEY_IN_WRONG_STATE if not write-locked or read-locked
|
|
- Returns SUCCESS on successful unlock
|
|
|
|
6. delete() - Delete keys from L1 cache
|
|
- Returns KEY_NOT_EXIST if key does not exist
|
|
- Returns KEY_IS_LOCKED if key is locked
|
|
- Returns SUCCESS on successful deletion
|
|
|
|
7. get_object_state() - Debugging API to get internal state
|
|
|
|
8. close() - Close the L1Manager and free all resources
|
|
"""
|
|
|
|
# Standard
|
|
import threading
|
|
|
|
# Third Party
|
|
import pytest
|
|
import torch
|
|
|
|
# First Party
|
|
from lmcache.v1.distributed.api import MemoryLayoutDesc, ObjectKey
|
|
from lmcache.v1.distributed.config import (
|
|
L1ManagerConfig,
|
|
L1MemoryManagerConfig,
|
|
)
|
|
from lmcache.v1.distributed.error import L1Error
|
|
|
|
try:
|
|
# First Party
|
|
from lmcache.v1.distributed.l1_manager import L1Manager
|
|
except ImportError:
|
|
# Skip tests if L1Manager cannot be imported
|
|
pytest.skip(
|
|
"Skipping because L1 manager cannot be imported", allow_module_level=True
|
|
)
|
|
|
|
# Skip all tests in this module if CUDA is not available
|
|
pytestmark = pytest.mark.skipif(
|
|
not torch.cuda.is_available(), reason="CUDA is not available"
|
|
)
|
|
|
|
|
|
def should_use_lazy_alloc() -> bool:
|
|
"""Determine if lazy allocation should be used based on CUDA availability."""
|
|
return torch.cuda.is_available()
|
|
|
|
|
|
# =============================================================================
|
|
# Fixtures
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def basic_memory_config():
|
|
"""Create a basic L1MemoryManagerConfig for testing."""
|
|
return L1MemoryManagerConfig(
|
|
size_in_bytes=128 * 1024 * 1024, # 128MB
|
|
use_lazy=should_use_lazy_alloc(),
|
|
init_size_in_bytes=64 * 1024 * 1024, # 64MB
|
|
align_bytes=0x1000, # 4KB
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def small_memory_config():
|
|
"""Create a small L1MemoryManagerConfig to test memory exhaustion."""
|
|
return L1MemoryManagerConfig(
|
|
size_in_bytes=64 * 1024 * 1024, # 64MB
|
|
use_lazy=should_use_lazy_alloc(),
|
|
init_size_in_bytes=64 * 1024 * 1024, # 64MB
|
|
align_bytes=0x1000,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def basic_l1_config(basic_memory_config):
|
|
"""Create a basic L1ManagerConfig for testing."""
|
|
return L1ManagerConfig(
|
|
memory_config=basic_memory_config,
|
|
write_ttl_seconds=600,
|
|
read_ttl_seconds=300,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def small_l1_config(small_memory_config):
|
|
"""Create a small L1ManagerConfig to test memory exhaustion."""
|
|
return L1ManagerConfig(
|
|
memory_config=small_memory_config,
|
|
write_ttl_seconds=600,
|
|
read_ttl_seconds=300,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def basic_layout():
|
|
"""Create a basic MemoryLayoutDesc for testing."""
|
|
return MemoryLayoutDesc(
|
|
shapes=[torch.Size([100, 2, 512])],
|
|
dtypes=[torch.bfloat16],
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def large_layout():
|
|
"""Create a large MemoryLayoutDesc that will exhaust small memory.
|
|
|
|
Each allocation is 8MB (2M elements * 4 bytes).
|
|
"""
|
|
return MemoryLayoutDesc(
|
|
shapes=[torch.Size([2048, 1024])], # 2M elements * 4 bytes = 8MB
|
|
dtypes=[torch.float32],
|
|
)
|
|
|
|
|
|
def make_object_key(chunk_hash: int, model_name: str = "test_model", kv_rank: int = 0):
|
|
"""Helper to create ObjectKey instances."""
|
|
hash_bytes = ObjectKey.IntHash2Bytes(chunk_hash)
|
|
return ObjectKey(chunk_hash=hash_bytes, model_name=model_name, kv_rank=kv_rank)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.reserve_read()
|
|
# =============================================================================
|
|
|
|
|
|
class TestReserveRead:
|
|
"""
|
|
Tests for L1Manager.reserve_read() method.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_NOT_READABLE: The key exists but is not readable.
|
|
- Returns (L1Error, Optional[MemoryObj]) for each key.
|
|
"""
|
|
|
|
def test_reserve_read_non_existing_key_returns_key_not_exist(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_read returns KEY_NOT_EXIST for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.reserve_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_NOT_EXIST
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_write_locked_key_returns_key_not_readable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_read returns KEY_NOT_READABLE for write-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Reserve write (but don't finish) - key is now write-locked
|
|
write_result = manager.reserve_write([key], [False], basic_layout)
|
|
assert write_result[key][0] == L1Error.SUCCESS
|
|
|
|
# Try to reserve read on a write-locked key
|
|
read_result = manager.reserve_read([key])
|
|
|
|
assert key in read_result
|
|
error, mem_obj = read_result[key]
|
|
assert error == L1Error.KEY_NOT_READABLE
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_ready_key_returns_success(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_read returns SUCCESS for ready (unlocked) keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create object: reserve write -> finish write
|
|
write_result = manager.reserve_write([key], [False], basic_layout)
|
|
assert write_result[key][0] == L1Error.SUCCESS
|
|
finish_result = manager.finish_write([key])
|
|
assert finish_result[key] == L1Error.SUCCESS
|
|
|
|
# Now reserve read
|
|
read_result = manager.reserve_read([key])
|
|
|
|
assert key in read_result
|
|
error, mem_obj = read_result[key]
|
|
assert error == L1Error.SUCCESS
|
|
assert mem_obj is not None
|
|
assert mem_obj.is_valid()
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_multiple_keys(self, basic_l1_config, basic_layout):
|
|
"""Test reserve_read with multiple keys in a single call."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key1 = make_object_key(1)
|
|
key2 = make_object_key(2)
|
|
key3 = make_object_key(3)
|
|
|
|
# Create key1 as ready object
|
|
manager.reserve_write([key1], [False], basic_layout)
|
|
manager.finish_write([key1])
|
|
|
|
# key2 does not exist
|
|
# key3 is write-locked
|
|
manager.reserve_write([key3], [False], basic_layout)
|
|
|
|
# Reserve read on all three
|
|
result = manager.reserve_read([key1, key2, key3])
|
|
|
|
assert result[key1][0] == L1Error.SUCCESS
|
|
assert result[key1][1] is not None
|
|
assert result[key2][0] == L1Error.KEY_NOT_EXIST
|
|
assert result[key2][1] is None
|
|
assert result[key3][0] == L1Error.KEY_NOT_READABLE
|
|
assert result[key3][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_can_be_called_multiple_times(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that multiple read reservations can be made on the same key."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Multiple read reservations
|
|
result1 = manager.reserve_read([key])
|
|
result2 = manager.reserve_read([key])
|
|
result3 = manager.reserve_read([key])
|
|
|
|
assert result1[key][0] == L1Error.SUCCESS
|
|
assert result2[key][0] == L1Error.SUCCESS
|
|
assert result3[key][0] == L1Error.SUCCESS
|
|
|
|
# Verify using get_object_state that read lock is held
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
# Check via available_for_read (should still be true since
|
|
# read-locked is readable)
|
|
assert state.available_for_read() is True
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_with_extra_count(self, basic_l1_config, basic_layout):
|
|
"""Test reserve_read(extra_count=N) acquires 1+N locks."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve with extra_count=2 -> total 3 locks
|
|
result = manager.reserve_read([key], extra_count=2)
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
assert result[key][1] is not None
|
|
|
|
# Need 3 finish_read() to fully release
|
|
manager.finish_read([key])
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert state.read_lock.is_locked()
|
|
|
|
manager.finish_read([key])
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert state.read_lock.is_locked()
|
|
|
|
manager.finish_read([key])
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert not state.read_lock.is_locked()
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_read_extra_count_default_is_zero(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Default extra_count=0 acquires exactly 1 lock."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
manager.reserve_read([key])
|
|
manager.finish_read([key])
|
|
|
|
# Lock fully released after single finish_read
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert not state.read_lock.is_locked()
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.unsafe_read()
|
|
# =============================================================================
|
|
|
|
|
|
class TestUnsafeRead:
|
|
"""
|
|
Tests for L1Manager.unsafe_read() method.
|
|
|
|
Per the docstring:
|
|
- This method does not acquire read locks.
|
|
- Caller must ensure unsafe_read is called between reserve_read and finish_read.
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_NOT_READABLE: The key is not readable (not read-locked).
|
|
- Returns (L1Error, Optional[MemoryObj]) for each key.
|
|
"""
|
|
|
|
def test_unsafe_read_non_existing_key_returns_key_not_exist(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that unsafe_read returns KEY_NOT_EXIST for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.unsafe_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_NOT_EXIST
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_non_read_locked_returns_key_not_readable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that unsafe_read returns KEY_NOT_READABLE if not read-locked."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object (not read-locked)
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Try unsafe_read without reserve_read
|
|
result = manager.unsafe_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_NOT_READABLE
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_write_locked_returns_key_not_readable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that unsafe_read returns KEY_NOT_READABLE for write-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create write-locked object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
# Try unsafe_read on write-locked key
|
|
result = manager.unsafe_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_NOT_READABLE
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_read_locked_returns_success(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that unsafe_read returns SUCCESS for read-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object and reserve read
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
reserve_result = manager.reserve_read([key])
|
|
assert reserve_result[key][0] == L1Error.SUCCESS
|
|
|
|
# unsafe_read should succeed on read-locked key
|
|
result = manager.unsafe_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.SUCCESS
|
|
assert mem_obj is not None
|
|
assert mem_obj.is_valid()
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_returns_same_memory_obj_as_reserve_read(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that unsafe_read returns the same MemoryObj as reserve_read."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read and get memory object
|
|
reserve_result = manager.reserve_read([key])
|
|
assert reserve_result[key][0] == L1Error.SUCCESS
|
|
reserved_mem_obj = reserve_result[key][1]
|
|
|
|
# unsafe_read should return the same memory object
|
|
unsafe_result = manager.unsafe_read([key])
|
|
assert unsafe_result[key][0] == L1Error.SUCCESS
|
|
unsafe_mem_obj = unsafe_result[key][1]
|
|
|
|
# Should be the same object
|
|
assert reserved_mem_obj is unsafe_mem_obj
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_multiple_times_without_adding_read_count(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that multiple unsafe_reads don't add to read lock count."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read once
|
|
manager.reserve_read([key])
|
|
|
|
# Multiple unsafe_reads
|
|
for _ in range(5):
|
|
result = manager.unsafe_read([key])
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
# Single finish_read should release the lock and delete temp object
|
|
manager.finish_read([key])
|
|
|
|
# Object should be deleted (only 1 read lock was held, not 6)
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_multiple_keys(self, basic_l1_config, basic_layout):
|
|
"""Test unsafe_read with multiple keys in a single call."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key1 = make_object_key(1)
|
|
key2 = make_object_key(2)
|
|
key3 = make_object_key(3)
|
|
|
|
# key1: read-locked
|
|
manager.reserve_write([key1], [False], basic_layout)
|
|
manager.finish_write([key1])
|
|
manager.reserve_read([key1])
|
|
|
|
# key2: does not exist
|
|
|
|
# key3: ready but not read-locked
|
|
manager.reserve_write([key3], [False], basic_layout)
|
|
manager.finish_write([key3])
|
|
|
|
# unsafe_read on all three
|
|
result = manager.unsafe_read([key1, key2, key3])
|
|
|
|
assert result[key1][0] == L1Error.SUCCESS
|
|
assert result[key1][1] is not None
|
|
assert result[key2][0] == L1Error.KEY_NOT_EXIST
|
|
assert result[key2][1] is None
|
|
assert result[key3][0] == L1Error.KEY_NOT_READABLE
|
|
assert result[key3][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_unsafe_read_between_reserve_and_finish(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test proper usage: unsafe_read between reserve_read and finish_read."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Proper workflow: reserve_read -> unsafe_read -> finish_read
|
|
reserve_result = manager.reserve_read([key])
|
|
assert reserve_result[key][0] == L1Error.SUCCESS
|
|
|
|
unsafe_result = manager.unsafe_read([key])
|
|
assert unsafe_result[key][0] == L1Error.SUCCESS
|
|
assert unsafe_result[key][1] is not None
|
|
|
|
finish_result = manager.finish_read([key])
|
|
assert finish_result[key] == L1Error.SUCCESS
|
|
|
|
# After finish_read, unsafe_read should fail (not read-locked)
|
|
result = manager.unsafe_read([key])
|
|
assert result[key][0] == L1Error.KEY_NOT_READABLE
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.finish_read()
|
|
# =============================================================================
|
|
|
|
|
|
class TestFinishRead:
|
|
"""
|
|
Tests for L1Manager.finish_read() method.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_IN_WRONG_STATE: The key is write-locked or non-read-locked.
|
|
- Will delete the object if it is temporary and read count reaches zero.
|
|
"""
|
|
|
|
def test_finish_read_non_existing_key_returns_key_not_exist(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that finish_read returns KEY_NOT_EXIST for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.finish_read([key])
|
|
|
|
assert key in result
|
|
assert result[key] == L1Error.KEY_NOT_EXIST
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_success(self, basic_l1_config, basic_layout):
|
|
"""Test that finish_read returns SUCCESS after proper read reservation."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read, then finish read
|
|
manager.reserve_read([key])
|
|
result = manager.finish_read([key])
|
|
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_non_read_locked_returns_wrong_state(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that finish_read returns KEY_IN_WRONG_STATE if not read-locked."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object (not read-locked)
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Try to finish read without reserving
|
|
result = manager.finish_read([key])
|
|
|
|
assert result[key] == L1Error.KEY_IN_WRONG_STATE
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_write_locked_returns_wrong_state(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that finish_read returns KEY_IN_WRONG_STATE if write-locked."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create write-locked object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
# Try to finish read on write-locked key
|
|
result = manager.finish_read([key])
|
|
|
|
assert result[key] == L1Error.KEY_IN_WRONG_STATE
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_temporary_object_deleted_when_count_zero(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that temporary objects are deleted when read count reaches zero."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout) # is_temporary=True
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read
|
|
manager.reserve_read([key])
|
|
|
|
# Verify object exists
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Finish read - should delete the temporary object
|
|
result = manager.finish_read([key])
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
# Verify object is deleted
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_multiple_reads_count_down(self, basic_l1_config, basic_layout):
|
|
"""Test that multiple finish_reads count down properly."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read three times
|
|
manager.reserve_read([key])
|
|
manager.reserve_read([key])
|
|
manager.reserve_read([key])
|
|
|
|
# Finish read twice - object should still exist
|
|
manager.finish_read([key])
|
|
manager.finish_read([key])
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Finish read third time - temporary object should be deleted
|
|
manager.finish_read([key])
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_with_extra_count_releases_multiple(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""finish_read(extra_count=2) releases 3 locks at once."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Acquire 3 read locks (1 + extra_count=2)
|
|
manager.reserve_read([key], extra_count=2)
|
|
|
|
# Release all 3 at once
|
|
result = manager.finish_read([key], extra_count=2)
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert not state.read_lock.is_locked()
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_extra_count_partial_release(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Partial extra_count release leaves remaining locks."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# 3 locks total (1 + extra_count=2)
|
|
manager.reserve_read([key], extra_count=2)
|
|
|
|
# Release 2 of 3 (1 + extra_count=1)
|
|
manager.finish_read([key], extra_count=1)
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert state.read_lock.is_locked()
|
|
|
|
# Release last one (1 + extra_count=0)
|
|
manager.finish_read([key])
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert not state.read_lock.is_locked()
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_extra_count_deletes_temporary(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Temp objects deleted when extra_count releases all."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Acquire 3 read locks at once (1 + extra_count=2)
|
|
manager.reserve_read([key], extra_count=2)
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Release all 3 at once -> temp deleted
|
|
result = manager.finish_read([key], extra_count=2)
|
|
assert result[key] == L1Error.SUCCESS
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_finish_read_extra_count_temp_survives_partial(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Temp object survives partial extra_count release."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# 4 locks total (1 + extra_count=3)
|
|
manager.reserve_read([key], extra_count=3)
|
|
|
|
# Release 2 of 4 -> still locked
|
|
manager.finish_read([key], extra_count=1)
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Release remaining 2 -> deleted
|
|
manager.finish_read([key], extra_count=1)
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.reserve_write()
|
|
# =============================================================================
|
|
|
|
|
|
class TestReserveWrite:
|
|
"""
|
|
Tests for L1Manager.reserve_write() method.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_WRITABLE: The key exists but is not writable.
|
|
- OUT_OF_MEMORY: Not enough memory to allocate for the object.
|
|
- Returns (L1Error, Optional[MemoryObj]) for each key.
|
|
"""
|
|
|
|
def test_reserve_write_new_key_returns_success(self, basic_l1_config, basic_layout):
|
|
"""Test that reserve_write returns SUCCESS for new keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.reserve_write([key], [False], basic_layout)
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.SUCCESS
|
|
assert mem_obj is not None
|
|
assert mem_obj.is_valid()
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_write_locked_key_returns_not_writable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_write returns KEY_NOT_WRITABLE for write-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# First reserve write
|
|
result1 = manager.reserve_write([key], [False], basic_layout)
|
|
assert result1[key][0] == L1Error.SUCCESS
|
|
|
|
# Try to reserve write again while still write-locked
|
|
result2 = manager.reserve_write([key], [False], basic_layout)
|
|
|
|
assert result2[key][0] == L1Error.KEY_NOT_WRITABLE
|
|
assert result2[key][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_read_locked_key_returns_not_writable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_write returns KEY_NOT_WRITABLE for read-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read
|
|
manager.reserve_read([key])
|
|
|
|
# Try to reserve write while read-locked
|
|
result = manager.reserve_write([key], [False], basic_layout)
|
|
|
|
assert result[key][0] == L1Error.KEY_NOT_WRITABLE
|
|
assert result[key][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_temporary_key_returns_not_writable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that reserve_write returns KEY_NOT_WRITABLE for temporary objects."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout) # is_temporary=True
|
|
manager.finish_write([key])
|
|
|
|
# Try to reserve write on temporary object
|
|
result = manager.reserve_write([key], [False], basic_layout)
|
|
|
|
assert result[key][0] == L1Error.KEY_NOT_WRITABLE
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_multiple_keys(self, basic_l1_config, basic_layout):
|
|
"""Test reserve_write with multiple keys in a single call."""
|
|
manager = L1Manager(basic_l1_config)
|
|
keys = [make_object_key(i) for i in range(5)]
|
|
is_temporary = [False] * 5
|
|
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout)
|
|
|
|
for key in keys:
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.SUCCESS
|
|
assert mem_obj is not None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_out_of_memory(self, small_l1_config, large_layout):
|
|
"""Test that reserve_write returns OUT_OF_MEMORY when allocation fails."""
|
|
manager = L1Manager(small_l1_config)
|
|
keys = [make_object_key(i) for i in range(10)]
|
|
is_temporary = [False] * 10
|
|
|
|
# Request more memory than available (8MB * 10 = 80MB > 64MB)
|
|
result = manager.reserve_write(keys, is_temporary, large_layout)
|
|
|
|
# All keys should return OUT_OF_MEMORY
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.OUT_OF_MEMORY
|
|
assert result[key][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_new_mode(self, basic_l1_config, basic_layout):
|
|
"""Test that reserve_write returns KEY_NOT_WRITABLE for existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
keys = [make_object_key(i) for i in range(5)]
|
|
is_temporary = [False] * 5
|
|
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout, mode="new")
|
|
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
assert result[key][1] is not None
|
|
|
|
# Commit the write
|
|
result = manager.finish_write(keys)
|
|
for key in keys:
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
# Now try to reserve write again with mode="new"
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout, mode="new")
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.KEY_NOT_WRITABLE
|
|
assert result[key][1] is None
|
|
|
|
manager.close()
|
|
|
|
def test_reserve_write_update_mode(self, basic_l1_config, basic_layout):
|
|
"""Test that reserve_write returns KEY_NOT_WRITABLE for new keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
keys = [make_object_key(i) for i in range(5)]
|
|
is_temporary = [False] * 5
|
|
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout, mode="update")
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.KEY_NOT_WRITABLE
|
|
assert result[key][1] is None
|
|
|
|
# Cannot finish write in update mode because keys not exist
|
|
result = manager.finish_write(keys)
|
|
for key in keys:
|
|
assert result[key] == L1Error.KEY_NOT_EXIST
|
|
|
|
# Now try to reserve write again with mode="new"
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout, mode="new")
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
assert result[key][1] is not None
|
|
|
|
# Commit the write
|
|
result = manager.finish_write(keys)
|
|
for key in keys:
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
# Now try to reserve write again with mode="update"
|
|
result = manager.reserve_write(keys, is_temporary, basic_layout, mode="update")
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
assert result[key][1] is not None
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.finish_write()
|
|
# =============================================================================
|
|
|
|
|
|
class TestFinishWrite:
|
|
"""
|
|
Tests for L1Manager.finish_write() method.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_IN_WRONG_STATE: The key is not write-locked, or it's read-locked.
|
|
"""
|
|
|
|
def test_finish_write_non_existing_key_returns_key_not_exist(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that finish_write returns KEY_NOT_EXIST for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.finish_write([key])
|
|
|
|
assert key in result
|
|
assert result[key] == L1Error.KEY_NOT_EXIST
|
|
|
|
manager.close()
|
|
|
|
def test_finish_write_success(self, basic_l1_config, basic_layout):
|
|
"""Test that finish_write returns SUCCESS after proper write reservation."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Reserve write
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
# Finish write
|
|
result = manager.finish_write([key])
|
|
|
|
assert result[key] == L1Error.SUCCESS
|
|
|
|
# Verify object is now ready (not write-locked)
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert state.available_for_read() is True
|
|
assert state.available_for_write() is True
|
|
|
|
manager.close()
|
|
|
|
def test_finish_write_non_write_locked_returns_wrong_state(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that finish_write returns KEY_IN_WRONG_STATE if not write-locked."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object (not write-locked)
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Try to finish write again
|
|
result = manager.finish_write([key])
|
|
|
|
assert result[key] == L1Error.KEY_IN_WRONG_STATE
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.finish_write_and_reserve_read()
|
|
# =============================================================================
|
|
|
|
|
|
class TestFinishWriteAndReserveRead:
|
|
"""
|
|
Tests for L1Manager.finish_write_and_reserve_read() method.
|
|
|
|
This method atomically finishes write and acquires read lock,
|
|
preventing a race window where eviction could interfere.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_IN_WRONG_STATE: Not write-locked, or already read-locked.
|
|
- SUCCESS: Write unlocked and read lock acquired atomically.
|
|
"""
|
|
|
|
def test_normal_transition(self, basic_l1_config, basic_layout):
|
|
"""Test normal write-locked -> read-locked transition."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Reserve write (key is now write-locked)
|
|
write_result = manager.reserve_write([key], [False], basic_layout)
|
|
assert write_result[key][0] == L1Error.SUCCESS
|
|
|
|
# Atomically finish write and reserve read
|
|
result = manager.finish_write_and_reserve_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.SUCCESS
|
|
assert mem_obj is not None
|
|
|
|
# Verify state: write unlocked, read locked
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert not state.write_lock.is_locked()
|
|
assert state.read_lock.is_locked()
|
|
|
|
# Should be readable (not write-locked)
|
|
assert state.available_for_read() is True
|
|
# Should not be writable (read-locked)
|
|
assert state.available_for_write() is False
|
|
|
|
# Clean up read lock
|
|
manager.finish_read([key])
|
|
manager.close()
|
|
|
|
def test_key_not_exist(self, basic_l1_config, basic_layout):
|
|
"""Test that non-existing key returns KEY_NOT_EXIST."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.finish_write_and_reserve_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_NOT_EXIST
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_not_write_locked(self, basic_l1_config, basic_layout):
|
|
"""Test that non-write-locked key returns KEY_IN_WRONG_STATE."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object (not write-locked)
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
result = manager.finish_write_and_reserve_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_IN_WRONG_STATE
|
|
assert mem_obj is None
|
|
|
|
manager.close()
|
|
|
|
def test_already_read_locked(self, basic_l1_config, basic_layout):
|
|
"""Test that key with both write+read locks returns KEY_IN_WRONG_STATE.
|
|
|
|
This is an unexpected state — normally a key shouldn't be both
|
|
write-locked and read-locked simultaneously.
|
|
"""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create write-locked object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
# Force a read lock via internal state (unusual state)
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
state.read_lock.lock()
|
|
|
|
result = manager.finish_write_and_reserve_read([key])
|
|
|
|
assert key in result
|
|
error, mem_obj = result[key]
|
|
assert error == L1Error.KEY_IN_WRONG_STATE
|
|
assert mem_obj is None
|
|
|
|
# Clean up
|
|
state.read_lock.unlock()
|
|
manager.close()
|
|
|
|
def test_multiple_keys_mixed_results(self, basic_l1_config, basic_layout):
|
|
"""Test with multiple keys where some succeed and some fail."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key1 = make_object_key(1)
|
|
key2 = make_object_key(2) # will not exist
|
|
key3 = make_object_key(3)
|
|
|
|
# key1: write-locked (should succeed)
|
|
manager.reserve_write([key1], [False], basic_layout)
|
|
|
|
# key3: ready, not write-locked (should fail)
|
|
manager.reserve_write([key3], [False], basic_layout)
|
|
manager.finish_write([key3])
|
|
|
|
result = manager.finish_write_and_reserve_read([key1, key2, key3])
|
|
|
|
# key1: SUCCESS
|
|
assert result[key1][0] == L1Error.SUCCESS
|
|
assert result[key1][1] is not None
|
|
|
|
# key2: KEY_NOT_EXIST
|
|
assert result[key2][0] == L1Error.KEY_NOT_EXIST
|
|
assert result[key2][1] is None
|
|
|
|
# key3: KEY_IN_WRONG_STATE (not write-locked)
|
|
assert result[key3][0] == L1Error.KEY_IN_WRONG_STATE
|
|
assert result[key3][1] is None
|
|
|
|
# Clean up
|
|
manager.finish_read([key1])
|
|
manager.close()
|
|
|
|
def test_can_unsafe_read_after_transition(self, basic_l1_config, basic_layout):
|
|
"""Test that unsafe_read works on the transitioned key."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Write and transition
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
result = manager.finish_write_and_reserve_read([key])
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
# unsafe_read should work (key is read-locked)
|
|
read_result = manager.unsafe_read([key])
|
|
assert read_result[key][0] == L1Error.SUCCESS
|
|
assert read_result[key][1] is not None
|
|
|
|
manager.finish_read([key])
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.delete()
|
|
# =============================================================================
|
|
|
|
|
|
class TestDelete:
|
|
"""
|
|
Tests for L1Manager.delete() method.
|
|
|
|
Per the docstring:
|
|
- KEY_NOT_EXIST: The key does not exist.
|
|
- KEY_IS_LOCKED: The key is locked (either write-locked or read-locked).
|
|
"""
|
|
|
|
def test_delete_non_existing_key_returns_key_not_exist(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that delete returns KEY_NOT_EXIST for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
result = manager.delete([key])
|
|
|
|
assert key in result
|
|
assert result[key] == L1Error.KEY_NOT_EXIST
|
|
|
|
manager.close()
|
|
|
|
def test_delete_success(self, basic_l1_config, basic_layout):
|
|
"""Test that delete returns SUCCESS for unlocked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Verify object exists
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Delete
|
|
result = manager.delete([key])
|
|
|
|
assert result[key] == L1Error.SUCCESS
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_delete_write_locked_returns_key_is_locked(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that delete returns KEY_IS_LOCKED for write-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create write-locked object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
# Try to delete
|
|
result = manager.delete([key])
|
|
|
|
assert result[key] == L1Error.KEY_IS_LOCKED
|
|
|
|
manager.close()
|
|
|
|
def test_delete_read_locked_returns_key_is_locked(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that delete returns KEY_IS_LOCKED for read-locked keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read
|
|
manager.reserve_read([key])
|
|
|
|
# Try to delete
|
|
result = manager.delete([key])
|
|
|
|
assert result[key] == L1Error.KEY_IS_LOCKED
|
|
|
|
manager.close()
|
|
|
|
def test_delete_multiple_keys(self, basic_l1_config, basic_layout):
|
|
"""Test delete with multiple keys in a single call."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key1 = make_object_key(1)
|
|
key2 = make_object_key(2)
|
|
key3 = make_object_key(3)
|
|
|
|
# key1: ready (unlocked)
|
|
manager.reserve_write([key1], [False], basic_layout)
|
|
manager.finish_write([key1])
|
|
|
|
# key2: does not exist
|
|
|
|
# key3: write-locked
|
|
manager.reserve_write([key3], [False], basic_layout)
|
|
|
|
result = manager.delete([key1, key2, key3])
|
|
|
|
assert result[key1] == L1Error.SUCCESS
|
|
assert result[key2] == L1Error.KEY_NOT_EXIST
|
|
assert result[key3] == L1Error.KEY_IS_LOCKED
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.get_object_state()
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetObjectState:
|
|
"""
|
|
Tests for L1Manager.get_object_state() method.
|
|
|
|
Per the docstring:
|
|
- Returns the L1ObjectState if the object exists, None otherwise.
|
|
"""
|
|
|
|
def test_get_object_state_non_existing_returns_none(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that get_object_state returns None for non-existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
state = manager.get_object_state(key)
|
|
|
|
assert state is None
|
|
|
|
manager.close()
|
|
|
|
def test_get_object_state_existing_returns_state(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test that get_object_state returns L1ObjectState for existing keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
state = manager.get_object_state(key)
|
|
|
|
assert state is not None
|
|
# Verify we can use the state's methods
|
|
assert state.available_for_read() is True
|
|
assert state.available_for_write() is True
|
|
|
|
manager.close()
|
|
|
|
def test_get_object_state_write_locked(self, basic_l1_config, basic_layout):
|
|
"""Test get_object_state for write-locked objects."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create write-locked object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
state = manager.get_object_state(key)
|
|
|
|
assert state is not None
|
|
assert state.available_for_read() is False
|
|
assert state.available_for_write() is False
|
|
|
|
manager.close()
|
|
|
|
def test_get_object_state_read_locked(self, basic_l1_config, basic_layout):
|
|
"""Test get_object_state for read-locked objects."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object, then read lock it
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
manager.reserve_read([key])
|
|
|
|
state = manager.get_object_state(key)
|
|
|
|
assert state is not None
|
|
# Read-locked is still readable
|
|
assert state.available_for_read() is True
|
|
# But not writable
|
|
assert state.available_for_write() is False
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.close()
|
|
# =============================================================================
|
|
|
|
|
|
class TestClose:
|
|
"""
|
|
Tests for L1Manager.close() method.
|
|
|
|
Per the docstring:
|
|
- Close the L1Manager and free all resources.
|
|
"""
|
|
|
|
def test_close_empty_manager(self, basic_l1_config):
|
|
"""Test that close works on an empty manager."""
|
|
manager = L1Manager(basic_l1_config)
|
|
|
|
# Should not raise any exceptions
|
|
manager.close()
|
|
|
|
def test_close_with_objects(self, basic_l1_config, basic_layout):
|
|
"""Test that close frees all objects in the manager."""
|
|
manager = L1Manager(basic_l1_config)
|
|
|
|
# Create multiple objects
|
|
keys = [make_object_key(i) for i in range(5)]
|
|
manager.reserve_write(keys, [False] * 5, basic_layout)
|
|
|
|
# Close should free all objects
|
|
manager.close()
|
|
|
|
def test_close_clears_objects(self, basic_l1_config, basic_layout):
|
|
"""Test that close clears all objects from the manager."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Verify object exists before close
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
# Close
|
|
manager.close()
|
|
|
|
# After close, get_object_state should return None
|
|
# (objects dict should be cleared)
|
|
assert manager.get_object_state(key) is None
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for state machine transitions (integration)
|
|
# =============================================================================
|
|
|
|
|
|
class TestStateMachineTransitions:
|
|
"""
|
|
Integration tests verifying the state machine transitions as described in the
|
|
L1Manager class docstring.
|
|
"""
|
|
|
|
def test_full_write_read_cycle(self, basic_l1_config, basic_layout):
|
|
"""Test full cycle: None -> write_locked -> ready -> read_locked -> ready."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# None state (key doesn't exist)
|
|
assert manager.get_object_state(key) is None
|
|
|
|
# reserve_write: None -> write_locked
|
|
result = manager.reserve_write([key], [False], basic_layout)
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
state = manager.get_object_state(key)
|
|
assert state.available_for_read() is False
|
|
assert state.available_for_write() is False
|
|
|
|
# finish_write: write_locked -> ready
|
|
result = manager.finish_write([key])
|
|
assert result[key] == L1Error.SUCCESS
|
|
state = manager.get_object_state(key)
|
|
assert state.available_for_read() is True
|
|
assert state.available_for_write() is True
|
|
|
|
# reserve_read: ready -> read_locked
|
|
result = manager.reserve_read([key])
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
state = manager.get_object_state(key)
|
|
assert state.available_for_read() is True
|
|
assert state.available_for_write() is False
|
|
|
|
# finish_read: read_locked -> ready
|
|
result = manager.finish_read([key])
|
|
assert result[key] == L1Error.SUCCESS
|
|
state = manager.get_object_state(key)
|
|
assert state.available_for_read() is True
|
|
assert state.available_for_write() is True
|
|
|
|
manager.close()
|
|
|
|
def test_full_write_read_with_unsafe_read(self, basic_l1_config, basic_layout):
|
|
"""Test cycle with unsafe_read between reserve_read and finish_read."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read
|
|
reserve_result = manager.reserve_read([key])
|
|
assert reserve_result[key][0] == L1Error.SUCCESS
|
|
|
|
# Multiple unsafe_reads should all succeed
|
|
for _ in range(3):
|
|
unsafe_result = manager.unsafe_read([key])
|
|
assert unsafe_result[key][0] == L1Error.SUCCESS
|
|
|
|
# Finish read
|
|
finish_result = manager.finish_read([key])
|
|
assert finish_result[key] == L1Error.SUCCESS
|
|
|
|
# Object should still exist (not temporary)
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
manager.close()
|
|
|
|
def test_delete_from_ready_state(self, basic_l1_config, basic_layout):
|
|
"""Test deletion from ready state."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# delete: ready -> None
|
|
result = manager.delete([key])
|
|
assert result[key] == L1Error.SUCCESS
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_temporary_object_lifecycle(self, basic_l1_config, basic_layout):
|
|
"""Test temporary object lifecycle: deleted after last read."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Read and release
|
|
manager.reserve_read([key])
|
|
manager.finish_read([key])
|
|
|
|
# Object should be deleted
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_temporary_object_with_unsafe_read(self, basic_l1_config, basic_layout):
|
|
"""Test temporary object with unsafe_read doesn't affect deletion."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create temporary object
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read
|
|
manager.reserve_read([key])
|
|
|
|
# Multiple unsafe_reads
|
|
for _ in range(5):
|
|
result = manager.unsafe_read([key])
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
# Single finish_read should delete the object
|
|
manager.finish_read([key])
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
def test_multi_reader_lifecycle_with_extra_count(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Full lifecycle using extra_count (MLA TP>1 scenario).
|
|
|
|
Simulates multiple workers sharing the same key:
|
|
reserve_read(extra_count=N-1) acquires N locks,
|
|
finish_read(extra_count=N-1) releases them all.
|
|
"""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
extra = 3 # total locks = 1 + 3 = 4
|
|
|
|
# write -> ready
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# reserve_read with extra_count
|
|
result = manager.reserve_read([key], extra_count=extra)
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
# unsafe_read should work while read-locked
|
|
ur = manager.unsafe_read([key])
|
|
assert ur[key][0] == L1Error.SUCCESS
|
|
|
|
# finish_read with same extra_count
|
|
fr = manager.finish_read([key], extra_count=extra)
|
|
assert fr[key] == L1Error.SUCCESS
|
|
|
|
# All locks released -> writable again
|
|
state = manager.get_object_state(key)
|
|
assert state is not None
|
|
assert state.available_for_write() is True
|
|
|
|
manager.close()
|
|
|
|
def test_temp_object_multi_reader_deletion(self, basic_l1_config, basic_layout):
|
|
"""Temporary object deleted after extra_count release."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
extra = 2 # total locks = 1 + 2 = 3
|
|
|
|
manager.reserve_write([key], [True], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
manager.reserve_read([key], extra_count=extra)
|
|
assert manager.get_object_state(key) is not None
|
|
|
|
manager.finish_read([key], extra_count=extra)
|
|
assert manager.get_object_state(key) is None
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Thread safety tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestThreadSafety:
|
|
"""Tests verifying thread-safety of L1Manager operations."""
|
|
|
|
def test_concurrent_reserve_write_different_keys(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test concurrent reserve_write on different keys."""
|
|
manager = L1Manager(basic_l1_config)
|
|
num_threads = 8
|
|
keys_per_thread = 5
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def worker(thread_id):
|
|
thread_keys = [
|
|
make_object_key(thread_id * 1000 + i) for i in range(keys_per_thread)
|
|
]
|
|
is_temporary = [False] * keys_per_thread
|
|
result = manager.reserve_write(thread_keys, is_temporary, basic_layout)
|
|
with lock:
|
|
results.append((thread_keys, result))
|
|
|
|
threads = [
|
|
threading.Thread(target=worker, args=(i,)) for i in range(num_threads)
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All writes should succeed
|
|
assert len(results) == num_threads
|
|
for keys, result in results:
|
|
for key in keys:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
manager.close()
|
|
|
|
def test_concurrent_read_same_key(self, basic_l1_config, basic_layout):
|
|
"""Test concurrent reads on the same key."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
num_threads = 10
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def worker():
|
|
result = manager.reserve_read([key])
|
|
with lock:
|
|
results.append(result)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(num_threads)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All reads should succeed
|
|
assert len(results) == num_threads
|
|
for result in results:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
manager.close()
|
|
|
|
def test_concurrent_unsafe_read_same_key(self, basic_l1_config, basic_layout):
|
|
"""Test concurrent unsafe_reads on the same read-locked key."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(12345)
|
|
|
|
# Create ready object and reserve read
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
manager.reserve_read([key])
|
|
|
|
num_threads = 10
|
|
results = []
|
|
lock = threading.Lock()
|
|
|
|
def worker():
|
|
result = manager.unsafe_read([key])
|
|
with lock:
|
|
results.append(result)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(num_threads)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# All unsafe_reads should succeed
|
|
assert len(results) == num_threads
|
|
for result in results:
|
|
assert result[key][0] == L1Error.SUCCESS
|
|
|
|
manager.close()
|
|
|
|
def test_concurrent_read_write_mixed_operations(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""Test concurrent mixed operations don't cause crashes."""
|
|
manager = L1Manager(basic_l1_config)
|
|
num_threads = 8
|
|
operations_per_thread = 10
|
|
errors = []
|
|
lock = threading.Lock()
|
|
|
|
def worker(thread_id):
|
|
try:
|
|
for i in range(operations_per_thread):
|
|
key = make_object_key(thread_id * 1000 + i)
|
|
|
|
# Write cycle
|
|
write_result = manager.reserve_write([key], [False], basic_layout)
|
|
if write_result[key][0] == L1Error.SUCCESS:
|
|
manager.finish_write([key])
|
|
|
|
# Read cycle with unsafe_read
|
|
read_result = manager.reserve_read([key])
|
|
if read_result[key][0] == L1Error.SUCCESS:
|
|
# Do some unsafe_reads
|
|
manager.unsafe_read([key])
|
|
manager.unsafe_read([key])
|
|
manager.finish_read([key])
|
|
|
|
# Delete
|
|
manager.delete([key])
|
|
except Exception as e:
|
|
with lock:
|
|
errors.append(e)
|
|
|
|
threads = [
|
|
threading.Thread(target=worker, args=(i,)) for i in range(num_threads)
|
|
]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
# No exceptions should have occurred
|
|
assert len(errors) == 0, f"Thread safety errors: {errors}"
|
|
|
|
manager.close()
|
|
|
|
|
|
# =============================================================================
|
|
# Tests for L1Manager.is_key_evictable()
|
|
# =============================================================================
|
|
|
|
|
|
class TestIsKeyEvictable:
|
|
"""Tests for L1Manager.is_key_evictable() method."""
|
|
|
|
def test_evictable_key_in_ready_state(self, basic_l1_config, basic_layout):
|
|
"""A key that has been written and finished should be evictable."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Write and finish -> key is in ready state
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
assert manager.is_key_evictable(key) is True
|
|
|
|
manager.close()
|
|
|
|
def test_write_locked_key_is_not_evictable(self, basic_l1_config, basic_layout):
|
|
"""A write-locked key should not be evictable."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Reserve write but don't finish -> key is write-locked
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Cleanup
|
|
manager.finish_write([key])
|
|
manager.close()
|
|
|
|
def test_read_locked_key_is_not_evictable(self, basic_l1_config, basic_layout):
|
|
"""A read-locked key should not be evictable."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Write, finish, then reserve read -> key is read-locked
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
manager.reserve_read([key])
|
|
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Cleanup
|
|
manager.finish_read([key])
|
|
manager.close()
|
|
|
|
def test_nonexistent_key_is_not_evictable(self, basic_l1_config):
|
|
"""A key that does not exist should not be evictable."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(999)
|
|
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
manager.close()
|
|
|
|
def test_key_becomes_evictable_after_read_unlock(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""A key should become evictable after all read locks are released."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Write and finish
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read -> not evictable
|
|
manager.reserve_read([key])
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Finish read -> evictable again
|
|
manager.finish_read([key])
|
|
assert manager.is_key_evictable(key) is True
|
|
|
|
manager.close()
|
|
|
|
def test_key_becomes_evictable_after_write_unlock(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""A key should become evictable after write lock is released."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Reserve write -> not evictable
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Finish write -> evictable
|
|
manager.finish_write([key])
|
|
assert manager.is_key_evictable(key) is True
|
|
|
|
manager.close()
|
|
|
|
def test_multiple_keys_mixed_evictability(self, basic_l1_config, basic_layout):
|
|
"""Test evictability with multiple keys in different states."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key_ready = make_object_key(1)
|
|
key_write_locked = make_object_key(2)
|
|
key_read_locked = make_object_key(3)
|
|
|
|
# key_ready: write + finish -> ready state
|
|
manager.reserve_write([key_ready], [False], basic_layout)
|
|
manager.finish_write([key_ready])
|
|
|
|
# key_write_locked: write only -> write-locked
|
|
manager.reserve_write([key_write_locked], [False], basic_layout)
|
|
|
|
# key_read_locked: write + finish + read -> read-locked
|
|
manager.reserve_write([key_read_locked], [False], basic_layout)
|
|
manager.finish_write([key_read_locked])
|
|
manager.reserve_read([key_read_locked])
|
|
|
|
assert manager.is_key_evictable(key_ready) is True
|
|
assert manager.is_key_evictable(key_write_locked) is False
|
|
assert manager.is_key_evictable(key_read_locked) is False
|
|
|
|
# Cleanup
|
|
manager.finish_write([key_write_locked])
|
|
manager.finish_read([key_read_locked])
|
|
manager.close()
|
|
|
|
def test_deleted_key_is_not_evictable(self, basic_l1_config, basic_layout):
|
|
"""A key that has been deleted should not be evictable."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Write, finish, then delete
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
manager.delete([key])
|
|
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
manager.close()
|
|
|
|
def test_key_with_multiple_read_locks_not_evictable(
|
|
self, basic_l1_config, basic_layout
|
|
):
|
|
"""A key with extra_count read locks should not be evictable
|
|
until all locks are released."""
|
|
manager = L1Manager(basic_l1_config)
|
|
key = make_object_key(1)
|
|
|
|
# Write and finish
|
|
manager.reserve_write([key], [False], basic_layout)
|
|
manager.finish_write([key])
|
|
|
|
# Reserve read with extra_count=2 (total 3 locks)
|
|
manager.reserve_read([key], extra_count=2)
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Release only 1 lock (extra_count=0) -> still locked
|
|
manager.finish_read([key], extra_count=0)
|
|
assert manager.is_key_evictable(key) is False
|
|
|
|
# Release remaining 2 locks (extra_count=1)
|
|
manager.finish_read([key], extra_count=1)
|
|
assert manager.is_key_evictable(key) is True
|
|
|
|
manager.close()
|