forked from mooncake-track/Mooncake
[Store] Enlarge the Default KV TTL to 5 Seconds (#660)
* change default kv lease ttl to 5 sec
This commit is contained in:
ykwd
GitHub
parent
ac3d7934f6
commit
1f71517d72
|
|
@ -94,7 +94,9 @@ jobs:
|
|||
- name: Start Mooncake Master
|
||||
run: |
|
||||
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib
|
||||
mooncake_master &
|
||||
# Set a small kv lease ttl to make the test faster.
|
||||
# Must be consistent with the client test parameters.
|
||||
mooncake_master --default_kv_lease_ttl=500 &
|
||||
shell: bash
|
||||
|
||||
- name: Test (in build env)
|
||||
|
|
@ -102,7 +104,7 @@ jobs:
|
|||
cd build
|
||||
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib
|
||||
ldconfig -v || echo "always continue"
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata make test -j ARGS="-V"
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata DEFAULT_KV_LEASE_TTL=500 make test -j ARGS="-V"
|
||||
shell: bash
|
||||
|
||||
- name: Stop Mooncake Master Service
|
||||
|
|
|
|||
|
|
@ -384,7 +384,7 @@ it initiates evict operations. The eviction target is to clean an additional `-e
|
|||
|
||||
To avoid data conflicts, a per-object lease will be granted whenever an `ExistKey` request or a `GetReplicaListRequest` request succeeds. An object is guaranteed to be protected from `Remove` request, `RemoveAll` request and `Eviction` task until its lease expires. A `Remove` request on a leased object will fail. A `RemoveAll` request will only remove objects without a lease.
|
||||
|
||||
The default lease TTL is 200 ms and is configurable via a startup parameter of `master_service`.
|
||||
The default lease TTL is 5 seconds and is configurable via a startup parameter of `master_service`.
|
||||
|
||||
### Soft Pin
|
||||
|
||||
|
|
|
|||
|
|
@ -397,7 +397,7 @@ virtual std::shared_ptr<BufHandle> Allocate(
|
|||
|
||||
为避免数据冲突，每当 `ExistKey` 请求或 `GetReplicaListRequest` 请求成功时，系统会为对应对象授予一个租约。在租约过期前，该对象将受到保护，不会被 `Remove`、`RemoveAll` 或替换任务删除。对有租约的对象执行 `Remove` 请求会失败；`RemoveAll` 请求则只会删除没有租约的对象。
|
||||
|
||||
默认的租约时间为 200 毫秒，并可通过 `master_service` 的启动参数进行配置。
|
||||
默认的租约时间为 5 秒，并可通过 `master_service` 的启动参数进行配置。
|
||||
|
||||
### 软固定机制
|
||||
|
||||
|
|
|
|||
|
|
@ -393,7 +393,7 @@ it initiates evict operations. The eviction target is to clean an additional `-e
|
|||
|
||||
To avoid data conflicts, a per-object lease will be granted whenever an `ExistKey` request or a `GetReplicaListRequest` request succeeds. An object is guaranteed to be protected from `Remove` request, `RemoveAll` request and `Eviction` task until its lease expires. A `Remove` request on a leased object will fail. A `RemoveAll` request will only remove objects without a lease.
|
||||
|
||||
The default lease TTL is 200 ms and is configurable via a startup parameter of `master_service`.
|
||||
The default lease TTL is 5 seconds and is configurable via a startup parameter of `master_service`.
|
||||
|
||||
### Soft Pin
|
||||
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@ static constexpr uint64_t WRONG_VERSION = 0;
|
|||
static constexpr uint64_t DEFAULT_VALUE = UINT64_MAX;
|
||||
static constexpr uint64_t ERRNO_BASE = DEFAULT_VALUE - 1000;
|
||||
static constexpr uint64_t DEFAULT_DEFAULT_KV_LEASE_TTL =
|
||||
200; // in milliseconds
|
||||
5000; // in milliseconds
|
||||
static constexpr uint64_t DEFAULT_KV_SOFT_PIN_TTL_MS =
|
||||
30 * 60 * 1000; // 30 minutes
|
||||
static constexpr bool DEFAULT_ALLOW_EVICT_SOFT_PINNED_OBJECTS = true;
|
||||
|
|
|
|||
|
|
@ -61,6 +61,13 @@ class ClientIntegrationTest : public ::testing::Test {
|
|||
<< ", Device name: " << FLAGS_device_name
|
||||
<< ", Metadata URL: " << FLAGS_transfer_engine_metadata_url;
|
||||
|
||||
if (getenv("DEFAULT_KV_LEASE_TTL")) {
|
||||
default_kv_lease_ttl_ = std::stoul(getenv("DEFAULT_KV_LEASE_TTL"));
|
||||
} else {
|
||||
default_kv_lease_ttl_ = FLAGS_default_kv_lease_ttl;
|
||||
}
|
||||
LOG(INFO) << "Default KV lease TTL: " << default_kv_lease_ttl_;
|
||||
|
||||
InitializeClients();
|
||||
InitializeSegment();
|
||||
}
|
||||
|
|
@ -154,6 +161,7 @@ class ClientIntegrationTest : public ::testing::Test {
|
|||
static size_t ram_buffer_size_;
|
||||
static void* test_client_segment_ptr_;
|
||||
static size_t test_client_ram_buffer_size_;
|
||||
static uint64_t default_kv_lease_ttl_;
|
||||
};
|
||||
|
||||
// Static members initialization
|
||||
|
|
@ -166,6 +174,7 @@ std::unique_ptr<SimpleAllocator>
|
|||
ClientIntegrationTest::client_buffer_allocator_ = nullptr;
|
||||
size_t ClientIntegrationTest::ram_buffer_size_ = 0;
|
||||
size_t ClientIntegrationTest::test_client_ram_buffer_size_ = 0;
|
||||
uint64_t ClientIntegrationTest::default_kv_lease_ttl_ = 0;
|
||||
|
||||
// Test basic Put/Get operations through the client
|
||||
TEST_F(ClientIntegrationTest, BasicPutGetOperations) {
|
||||
|
|
@ -208,7 +217,7 @@ TEST_F(ClientIntegrationTest, BasicPutGetOperations) {
|
|||
ASSERT_TRUE(put_result2.has_value())
|
||||
<< "Second Put operation failed: " << toString(put_result2.error());
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(FLAGS_default_kv_lease_ttl));
|
||||
std::chrono::milliseconds(default_kv_lease_ttl_));
|
||||
auto remove_result = test_client_->Remove(key);
|
||||
ASSERT_TRUE(remove_result.has_value())
|
||||
<< "Remove operation failed: " << toString(remove_result.error());
|
||||
|
|
@ -297,7 +306,7 @@ TEST_F(ClientIntegrationTest, LocalPreferredAllocationTest) {
|
|||
|
||||
// Clean up
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(FLAGS_default_kv_lease_ttl));
|
||||
std::chrono::milliseconds(default_kv_lease_ttl_));
|
||||
auto remove_result2 = test_client_->Remove(key);
|
||||
ASSERT_TRUE(remove_result2.has_value())
|
||||
<< "Remove operation failed: " << toString(remove_result2.error());
|
||||
|
|
@ -424,7 +433,7 @@ TEST_F(ClientIntegrationTest, LargeAllocateTest) {
|
|||
|
||||
// Remove the key
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(FLAGS_default_kv_lease_ttl));
|
||||
std::chrono::milliseconds(default_kv_lease_ttl_));
|
||||
auto remove_result = test_client_->Remove(key);
|
||||
ASSERT_TRUE(remove_result.has_value())
|
||||
<< "Remove operation failed: " << toString(remove_result.error());
|
||||
|
|
@ -589,7 +598,7 @@ TEST_F(ClientIntegrationTest, BatchIsExistOperations) {
|
|||
test_data_list[i].size());
|
||||
}
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(FLAGS_default_kv_lease_ttl));
|
||||
std::chrono::milliseconds(default_kv_lease_ttl_));
|
||||
for (int i = 0; i < batch_size / 2; i++) {
|
||||
auto remove_result = test_client_->Remove(keys[i]);
|
||||
ASSERT_TRUE(remove_result.has_value())
|
||||
|
|
@ -642,7 +651,7 @@ TEST_F(ClientIntegrationTest, BatchPutDuplicateKeys) {
|
|||
|
||||
// Clean up the key that was successfully put
|
||||
std::this_thread::sleep_for(
|
||||
std::chrono::milliseconds(FLAGS_default_kv_lease_ttl));
|
||||
std::chrono::milliseconds(default_kv_lease_ttl_));
|
||||
auto remove_result = test_client_->Remove(key);
|
||||
// Remove might fail if the key wasn't actually put, which is fine
|
||||
ASSERT_TRUE(remove_result);
|
||||
|
|
|
|||
|
|
@ -7,7 +7,9 @@ from mooncake.store import MooncakeDistributedStore
|
|||
|
||||
# The lease time of the kv object, should be set equal to
|
||||
# the master's value.
|
||||
DEFAULT_KV_LEASE_TTL = 200 # 200 milliseconds
|
||||
DEFAULT_DEFAULT_KV_LEASE_TTL = 5000 # 5000 milliseconds
|
||||
# Use environment variable if set, otherwise use default
|
||||
default_kv_lease_ttl = int(os.getenv("DEFAULT_KV_LEASE_TTL", DEFAULT_DEFAULT_KV_LEASE_TTL))
|
||||
|
||||
def get_client(store):
|
||||
"""Initialize and setup the distributed store client."""
|
||||
|
|
@ -89,7 +91,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(self.store.put(key, test_data), 0)
|
||||
|
||||
# Remove the key
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
def test_batch_is_exist_operations(self):
|
||||
|
|
@ -134,7 +136,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(non_existent_result[0], 0)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in existing_keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
|
|
@ -194,7 +196,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertLess(bytes_read, 0, "get_into should fail with small buffer")
|
||||
|
||||
# Cleanup
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(buffer_ptr), 0, "Buffer unregistration should succeed")
|
||||
self.assertEqual(self.store.unregister_buffer(small_buffer_ptr), 0)
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
|
@ -270,7 +272,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(len(empty_results), 0, "Should return empty results for empty input")
|
||||
|
||||
# Cleanup
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(large_buffer_ptr), 0, "Buffer unregistration should succeed")
|
||||
for key in keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
|
@ -343,7 +345,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(len(empty_results), 0, "Should return empty results for empty input")
|
||||
|
||||
# Cleanup
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(large_buffer_ptr), 0, "Buffer unregistration should succeed")
|
||||
for key in keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
|
@ -398,7 +400,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
get_barrier.wait()
|
||||
|
||||
# Remove all keys
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in thread_keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
|
|
@ -517,7 +519,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
print(record)
|
||||
raise e
|
||||
# Cleanup: ensure all remaining keys are removed
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in list(reference.keys()):
|
||||
self.store.remove(key)
|
||||
|
||||
|
|
@ -561,7 +563,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, test_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
# Test with custom config
|
||||
|
|
@ -577,7 +579,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, test_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.remove(key2), 0)
|
||||
|
||||
with self.assertRaises(TypeError):
|
||||
|
|
@ -607,7 +609,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, expected_value)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
|
|
@ -625,7 +627,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, expected_value)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in keys2:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
|
|
@ -657,7 +659,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, test_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
# Test with custom config
|
||||
|
|
@ -674,7 +676,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, test_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(buffer_ptr), 0)
|
||||
self.assertEqual(self.store.remove(key2), 0)
|
||||
|
||||
|
|
@ -727,7 +729,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, expected_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
for key in keys:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
||||
|
|
@ -748,7 +750,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertEqual(retrieved_data, expected_data)
|
||||
|
||||
# Clean up
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(large_buffer_ptr), 0)
|
||||
for key in keys2:
|
||||
self.assertEqual(self.store.remove(key), 0)
|
||||
|
|
|
|||
|
|
@ -7,6 +7,11 @@ import threading
|
|||
import random
|
||||
from mooncake.store import MooncakeDistributedStore
|
||||
|
||||
# The lease time of the kv object, should be set equal to
|
||||
# the master's value.
|
||||
DEFAULT_DEFAULT_KV_LEASE_TTL = 5000 # 5000 milliseconds
|
||||
# Use environment variable if set, otherwise use default
|
||||
default_kv_lease_ttl = int(os.getenv("DEFAULT_KV_LEASE_TTL", DEFAULT_DEFAULT_KV_LEASE_TTL))
|
||||
|
||||
# Define a test class for serialization
|
||||
class TestClass:
|
||||
|
|
@ -108,6 +113,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
self.assertTrue(torch.equal(tensor_rand, retrieved_rand))
|
||||
|
||||
# Clean up
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.store.remove(key)
|
||||
self.store.remove(key_int)
|
||||
self.store.remove(key_bool)
|
||||
|
|
|
|||
|
|
@ -10,7 +10,9 @@ from collections import defaultdict
|
|||
|
||||
# The lease time of the kv object, should be set equal to
|
||||
# the master's value.
|
||||
DEFAULT_KV_LEASE_TTL = 200 # 200 milliseconds
|
||||
DEFAULT_DEFAULT_KV_LEASE_TTL = 5000 # 5000 milliseconds
|
||||
# Use environment variable if set, otherwise use default
|
||||
default_kv_lease_ttl = int(os.getenv("DEFAULT_KV_LEASE_TTL", DEFAULT_DEFAULT_KV_LEASE_TTL))
|
||||
ADD_OPERATION_COUNT_ONE = 1
|
||||
ADD_OPERATION_COUNT_ZERO = 0
|
||||
NO_ADD_LATENCY = 0
|
||||
|
|
@ -286,7 +288,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
# --------------------------
|
||||
# Phase 3: Cleanup
|
||||
# --------------------------
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
index = 0
|
||||
while index < MAX_REQUESTS:
|
||||
key = "k_" + str(index)
|
||||
|
|
@ -417,7 +419,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
throughput_stats.print_throughput(thread_put_stats, thread_get_stats)
|
||||
|
||||
# Cleanup
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
index = 0
|
||||
while index < OPERATIONS_PER_THREAD:
|
||||
key = "k_" + str(index)
|
||||
|
|
@ -563,7 +565,7 @@ class TestDistributedObjectStore(unittest.TestCase):
|
|||
# --------------------------
|
||||
# Phase 3: Cleanup (still using single remove for simplicity)
|
||||
# --------------------------
|
||||
time.sleep(DEFAULT_KV_LEASE_TTL / 1000)
|
||||
time.sleep(default_kv_lease_ttl / 1000)
|
||||
self.assertEqual(self.store.unregister_buffer(large_buffer_ptr), 0, "Buffer unregistration should succeed")
|
||||
index = 0
|
||||
while index < MAX_REQUESTS:
|
||||
|
|
|
|||
|
|
@ -22,14 +22,16 @@ which mooncake_master 2>/dev/null | grep -q '/usr/local/bin/mooncake_master' &&
|
|||
echo "mooncake_master not found in /usr/local/bin, installed by python"
|
||||
|
||||
echo "mooncake_master found, running tests..."
|
||||
mooncake_master &
|
||||
# Set a small kv lease ttl to make the test faster.
|
||||
# Must be consistent with the client test parameters.
|
||||
mooncake_master --default_kv_lease_ttl=500 &
|
||||
MASTER_PID=$!
|
||||
sleep 1
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata python test_distributed_object_store.py
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata DEFAULT_KV_LEASE_TTL=500 python test_distributed_object_store.py
|
||||
sleep 1
|
||||
|
||||
pip install torch numpy
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata python test_put_get_tensor.py
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata DEFAULT_KV_LEASE_TTL=500 python test_put_get_tensor.py
|
||||
kill $MASTER_PID || true
|
||||
|
||||
|
||||
|
|
@ -39,10 +41,12 @@ if [ -n "$TEST_SSD_OFFLOAD_IN_EVICT" ]; then
|
|||
mkdir -p $TEST_ROOT_DIR
|
||||
echo "MOONCAKE_STORAGE_ROOT_DIR is set to: $TEST_ROOT_DIR"
|
||||
echo "Running with ssd offload in evict tests..."
|
||||
mooncake_master &
|
||||
# Set a small kv lease ttl to make the test faster.
|
||||
# Must be consistent with the client test parameters.
|
||||
mooncake_master --default_kv_lease_ttl=500 &
|
||||
MASTER_PID=$!
|
||||
sleep 1
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata MOONCAKE_STORAGE_ROOT_DIR=$TEST_ROOT_DIR python test_ssd_offload_in_evict.py
|
||||
MC_METADATA_SERVER=http://127.0.0.1:8080/metadata MOONCAKE_STORAGE_ROOT_DIR=$TEST_ROOT_DIR DEFAULT_KV_LEASE_TTL=500 python test_ssd_offload_in_evict.py
|
||||
kill $MASTER_PID || true
|
||||
rm -rf $TEST_ROOT_DIR
|
||||
else
|
||||
|
|
|
|||
Loading…
Reference in New Issue