Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature/read transaction for non rmw commands #440

Merged
merged 10 commits into from
Dec 29, 2024
Merged

Feature/read transaction for non rmw commands #440

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
9069ca3
Always require a transaction when operating on the hashtable for the …
danielealbano Oct 24, 2023
f76dcf7
If a chunk is already write locked, no need to lock it again in read
danielealbano Oct 24, 2023
9de6442
Update hashtable tests
danielealbano Oct 24, 2023
9de298f
Update the storage layer to use always transaction, even to read data
danielealbano Oct 24, 2023
96098ba
Update the redis commands of the redis module to use transactions whe…
danielealbano Oct 24, 2023
27df42b
Update tests
danielealbano Oct 24, 2023
8097de6
Keep track of the transaction in the RMW state
danielealbano Aug 18, 2024
484d6d6
Request and use the transaction in the storage db layer
danielealbano Aug 18, 2024
a919e02
Fix required headers in transaction.h
danielealbano Aug 18, 2024
141bba2
Fix benchmark for hashtable mpmc op get and set
danielealbano Aug 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions benches/bench-hashtable-mpmc-op-get.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,11 +61,18 @@ static void hashtable_op_get_not_found_key(benchmark::State& state) {
test_support_set_thread_affinity(state.thread_index());

for (auto _ : state) {
transaction_t transaction = { 0 };
transaction_acquire(&transaction);

benchmark::DoNotOptimize(hashtable_mcmp_op_get(
hashtable,
0,
&transaction,
test_key_1,
test_key_1_len,
&value));

transaction_release(&transaction);
}

if (state.thread_index() == 0) {
Expand Down Expand Up @@ -108,12 +115,19 @@ static void hashtable_op_get_single_key_external(benchmark::State& state) {
test_support_set_thread_affinity(state.thread_index());

for (auto _ : state) {
transaction_t transaction = { 0 };
transaction_acquire(&transaction);

benchmark::DoNotOptimize((result = hashtable_mcmp_op_get(
hashtable,
0,
&transaction,
test_key_1,
test_key_1_len,
&value)));

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand Down
30 changes: 30 additions & 0 deletions benches/bench-hashtable-mpmc-op-set.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,18 +213,24 @@ BENCHMARK_DEFINE_F(HashtableOpSetInsertFixture, hashtable_op_set_insert)(benchma
uint64_t key_index = state.thread_index();
key_index < requested_keyset_size;
key_index += state.threads()) {
transaction_t transaction = { 0 };
transaction_acquire(&transaction);

bool should_free_key = false;
hashtable_bucket_index_t out_bucket_index;
benchmark::DoNotOptimize((result = hashtable_mcmp_op_set(
hashtable,
0,
&transaction,
keyset_slots[key_index].key,
keyset_slots[key_index].key_length,
key_index,
nullptr,
&out_bucket_index,
&should_free_key)));

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand All @@ -245,13 +251,19 @@ BENCHMARK_DEFINE_F(HashtableOpSetInsertFixture, hashtable_op_set_insert)(benchma
key_index += state.threads()) {
hashtable_value_data_t data = 0;

transaction_t transaction = { 0 };
transaction_acquire(&transaction);

result = hashtable_mcmp_op_get(
hashtable,
0,
&transaction,
keyset_slots[key_index].key,
keyset_slots[key_index].key_length,
&data);

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand Down Expand Up @@ -374,18 +386,24 @@ class HashtableOpSetUpdateFixture : public benchmark::Fixture {
uint64_t key_index = state.thread_index();
key_index < this->_requested_keyset_size;
key_index += state.threads()) {
transaction_t transaction = { 0 };
transaction_acquire(&transaction);

bool should_free_key = false;
hashtable_bucket_index_t out_bucket_index;
bool result = hashtable_mcmp_op_set(
(hashtable_t*)static_hashtable,
0,
&transaction,
static_keyset_slots[key_index].key,
static_keyset_slots[key_index].key_length,
key_index,
nullptr,
&out_bucket_index,
&should_free_key);

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand Down Expand Up @@ -473,18 +491,24 @@ BENCHMARK_DEFINE_F(HashtableOpSetUpdateFixture, hashtable_op_set_update)(benchma
uint64_t key_index = state.thread_index();
key_index < requested_keyset_size;
key_index += state.threads()) {
transaction_t transaction = { 0 };
transaction_acquire(&transaction);

bool should_free_key = false;
hashtable_bucket_index_t out_bucket_index;
benchmark::DoNotOptimize((result = hashtable_mcmp_op_set(
hashtable,
0,
&transaction,
keyset_slots[key_index].key,
keyset_slots[key_index].key_length,
key_index,
nullptr,
&out_bucket_index,
&should_free_key)));

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand All @@ -506,13 +530,19 @@ BENCHMARK_DEFINE_F(HashtableOpSetUpdateFixture, hashtable_op_set_update)(benchma
key_index += state.threads()) {
hashtable_value_data_t data = 0;

transaction_t transaction = { 0 };
transaction_acquire(&transaction);

result = hashtable_mcmp_op_get(
hashtable,
0,
&transaction,
keyset_slots[key_index].key,
keyset_slots[key_index].key_length,
&data);

transaction_release(&transaction);

if (!result) {
sprintf(
error_message,
Expand Down
3 changes: 2 additions & 1 deletion src/data_structures/hashtable/mcmp/hashtable.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ extern "C" {
#define HASHTABLE_USE_UINT64 1
#endif

#define HASHTABLE_OP_ITER_END UINT64_MAX

#define HASHTABLE_MCMP_HALF_HASHES_CHUNK_SLOTS_COUNT 14
#define HASHTABLE_HALF_HASHES_CHUNK_SEARCH_MAX 32

Expand Down Expand Up @@ -173,7 +175,6 @@ typedef struct hashtable_mcmp_op_rmw_transaction hashtable_mcmp_op_rmw_status_t;
struct hashtable_mcmp_op_rmw_transaction {
hashtable_hash_t hash;
hashtable_t *hashtable;
transaction_t *transaction;
hashtable_half_hashes_chunk_volatile_t *half_hashes_chunk;
hashtable_key_value_volatile_t *key_value;
hashtable_database_number_t database_number;
Expand Down
50 changes: 28 additions & 22 deletions src/data_structures/hashtable/mcmp/hashtable_op_delete.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
bool hashtable_mcmp_op_delete(
hashtable_t* hashtable,
hashtable_database_number_t database_number,
transaction_t *transaction,
hashtable_key_data_t* key,
hashtable_key_length_t key_length,
hashtable_value_data_t *current_value) {
Expand All @@ -37,7 +38,6 @@ bool hashtable_mcmp_op_delete(
hashtable_chunk_slot_index_t chunk_slot_index = 0;
hashtable_half_hashes_chunk_volatile_t* half_hashes_chunk;
hashtable_key_value_volatile_t* key_value;
transaction_t transaction = { 0 };
bool deleted = false;

// TODO: the deletion algorithm needs to be updated to compact the keys stored in further away slots relying on the
Expand All @@ -52,7 +52,6 @@ bool hashtable_mcmp_op_delete(

LOG_DI("key (%d) = %s", key_length, key);
LOG_DI("hash = 0x%016x", hash);
transaction_acquire(&transaction);

volatile hashtable_data_t* hashtable_data_list[] = {
hashtable->ht_current,
Expand Down Expand Up @@ -80,7 +79,7 @@ bool hashtable_mcmp_op_delete(
key,
key_length,
hash,
&transaction,
transaction,
&chunk_index,
&chunk_slot_index,
&key_value) == false) {
Expand All @@ -93,13 +92,13 @@ bool hashtable_mcmp_op_delete(
half_hashes_chunk = &hashtable_data->half_hashes_chunk[chunk_index];

if (half_hashes_chunk->half_hashes[chunk_slot_index].filled == 0) {
transaction_release(&transaction);
return false;
}

if (unlikely(!transaction_upgrade_lock_for_write(&transaction, &half_hashes_chunk->lock))) {
transaction_release(&transaction);
return false;
if (unlikely(!transaction_rwspinlock_is_owned_by_transaction(&half_hashes_chunk->lock, transaction))) {
if (unlikely(!transaction_upgrade_lock_for_write(transaction, &half_hashes_chunk->lock))) {
return false;
}
}

// The hashtable_mcmp_support_op_search_key operation is lockless, it's necessary to set the lock and validate
Expand Down Expand Up @@ -130,8 +129,6 @@ bool hashtable_mcmp_op_delete(
deleted = true;
}

transaction_release(&transaction);

if (likely(deleted)) {
break;
}
Expand All @@ -147,13 +144,14 @@ bool hashtable_mcmp_op_delete(
bool hashtable_mcmp_op_delete_by_index(
hashtable_t* hashtable,
hashtable_database_number_t database_number,
transaction_t *transaction,
bool already_locked_for_read,
hashtable_bucket_index_t bucket_index,
hashtable_value_data_t *current_value) {
hashtable_chunk_index_t chunk_index = 0;
hashtable_chunk_slot_index_t chunk_slot_index = 0;
hashtable_half_hashes_chunk_volatile_t* half_hashes_chunk;
hashtable_key_value_volatile_t* key_value;
transaction_t transaction = { 0 };
bool deleted = false;

volatile hashtable_data_t* hashtable_data_list[] = {
Expand Down Expand Up @@ -185,15 +183,21 @@ bool hashtable_mcmp_op_delete_by_index(
return false;
}

transaction_acquire(&transaction);
if (unlikely(!transaction_lock_for_write(&transaction, &half_hashes_chunk->lock))) {
return false;
if (already_locked_for_read) {
if (unlikely(!transaction_upgrade_lock_for_write(transaction, &half_hashes_chunk->lock))) {
return false;
}
} else {
if (unlikely(!transaction_rwspinlock_is_owned_by_transaction(&half_hashes_chunk->lock, transaction))) {
if (unlikely(!transaction_lock_for_write(transaction, &half_hashes_chunk->lock))) {
return false;
}
}
}

key_value = &hashtable_data->keys_values[bucket_index];

if (unlikely(key_value->database_number != database_number)) {
transaction_release(&transaction);
return false;
}

Expand All @@ -219,8 +223,6 @@ bool hashtable_mcmp_op_delete_by_index(

deleted = true;

transaction_release(&transaction);

break;
}

Expand All @@ -233,13 +235,14 @@ bool hashtable_mcmp_op_delete_by_index(

bool hashtable_mcmp_op_delete_by_index_all_databases(
hashtable_t* hashtable,
transaction_t *transaction,
bool already_locked_for_read,
hashtable_bucket_index_t bucket_index,
hashtable_value_data_t *current_value) {
hashtable_chunk_index_t chunk_index = 0;
hashtable_chunk_slot_index_t chunk_slot_index = 0;
hashtable_half_hashes_chunk_volatile_t* half_hashes_chunk;
hashtable_key_value_volatile_t* key_value;
transaction_t transaction = { 0 };
bool deleted = false;

volatile hashtable_data_t* hashtable_data_list[] = {
Expand Down Expand Up @@ -271,9 +274,14 @@ bool hashtable_mcmp_op_delete_by_index_all_databases(
return false;
}

transaction_acquire(&transaction);
if (unlikely(!transaction_lock_for_write(&transaction, &half_hashes_chunk->lock))) {
return false;
if (already_locked_for_read) {
if (unlikely(!transaction_upgrade_lock_for_write(transaction, &half_hashes_chunk->lock))) {
return false;
}
} else {
if (unlikely(!transaction_lock_for_write(transaction, &half_hashes_chunk->lock))) {
return false;
}
}

key_value = &hashtable_data->keys_values[bucket_index];
Expand All @@ -300,8 +308,6 @@ bool hashtable_mcmp_op_delete_by_index_all_databases(

deleted = true;

transaction_release(&transaction);

break;
}

Expand Down
5 changes: 5 additions & 0 deletions src/data_structures/hashtable/mcmp/hashtable_op_delete.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,18 +8,23 @@ extern "C" {
bool hashtable_mcmp_op_delete(
hashtable_t* hashtable,
hashtable_database_number_t database_number,
transaction_t *transaction,
hashtable_key_data_t* key,
hashtable_key_length_t key_length,
hashtable_value_data_t *current_value);

bool hashtable_mcmp_op_delete_by_index(
hashtable_t* hashtable,
hashtable_database_number_t database_number,
transaction_t *transaction,
bool already_locked_for_read,
hashtable_bucket_index_t bucket_index,
hashtable_value_data_t *current_value);

bool hashtable_mcmp_op_delete_by_index_all_databases(
hashtable_t* hashtable,
transaction_t *transaction,
bool already_locked_for_read,
hashtable_bucket_index_t bucket_index,
hashtable_value_data_t *current_value);

Expand Down
Loading
Loading