Skip to content

Robj/shmalloc #655

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

Merged
rtjohnso merged 9 commits into from
Jan 22, 2026
Merged

Robj/shmalloc #655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
8ce48d3
incorporated new shm code
rtjohnso Jan 20, 2026
336c15a
change splinter_perf range lookup counts
rtjohnso Jan 20, 2026
e5c2f47
add shmalloc.[hc]
rtjohnso Jan 20, 2026
1464ee4
dial down heap capacities
rtjohnso Jan 20, 2026
0b69db9
add physical next to shmalloc nodes
rtjohnso Jan 21, 2026
771be22
move some allocations out of global heap
rtjohnso Jan 21, 2026
da2f42b
move more allocations to private heap
rtjohnso Jan 21, 2026
d6e68e1
remove space usage printf
rtjohnso Jan 21, 2026
1a01d1f
delete racey assert
rtjohnso Jan 22, 2026
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
2 changes: 1 addition & 1 deletion Makefile
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,7 @@ PLATFORM_SYS = $(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/platform_assert.o \
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/platform_log.o \
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/platform_mutex.o \
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/platform_threads.o \
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/shmem.o
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/shmalloc.o
PLATFORM_IO_SYS = $(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/platform_io.o \
$(OBJDIR)/$(SRCDIR)/$(PLATFORM_DIR)/laio.o

Expand Down
2 changes: 0 additions & 2 deletions src/clockcache.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -810,8 +810,6 @@ clockcache_try_set_writeback(clockcache *cc,
entry_number,
cc->cfg->page_capacity);

platform_assert(cc->entry[entry_number].waiters.head == NULL);

volatile uint32 *status = &cc->entry[entry_number].status;
if (__sync_bool_compare_and_swap(
status, CC_CLEANABLE1_STATUS, CC_WRITEBACK1_STATUS))
Expand Down
58 changes: 31 additions & 27 deletions src/core.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,7 @@ core_set_super_block(core_handle *spl,
if (root_addr != 0) {
super->root_addr = root_addr;
rc = trunk_inc_ref(spl->cfg.trunk_node_cfg,
spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
spl->cc,
spl->al,
spl->ts,
Expand Down Expand Up @@ -205,7 +205,7 @@ core_set_super_block(core_handle *spl,

if (old_root_addr != 0 && !is_create) {
rc = trunk_dec_ref(spl->cfg.trunk_node_cfg,
spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
spl->cc,
spl->al,
spl->ts,
Expand Down Expand Up @@ -384,8 +384,12 @@ core_memtable_insert(core_handle *spl, key tuple_key, message msg)
// this call is safe because we hold the insert lock
memtable *mt = core_get_memtable(spl, generation);
uint64 leaf_generation; // used for ordering the log
rc = memtable_insert(
&spl->mt_ctxt, mt, spl->heap_id, tuple_key, msg, &leaf_generation);
rc = memtable_insert(&spl->mt_ctxt,
mt,
PROCESS_PRIVATE_HEAP_ID,
tuple_key,
msg,
&leaf_generation);
if (!SUCCESS(rc)) {
goto unlock_insert_lock;
}
Expand Down Expand Up @@ -822,10 +826,10 @@ core_range_iterator_init(core_handle *spl,
range_itor->can_prev = TRUE;
range_itor->can_next = TRUE;

key_buffer_init(&range_itor->min_key, spl->heap_id);
key_buffer_init(&range_itor->max_key, spl->heap_id);
key_buffer_init(&range_itor->local_min_key, spl->heap_id);
key_buffer_init(&range_itor->local_max_key, spl->heap_id);
key_buffer_init(&range_itor->min_key, PROCESS_PRIVATE_HEAP_ID);
key_buffer_init(&range_itor->max_key, PROCESS_PRIVATE_HEAP_ID);
key_buffer_init(&range_itor->local_min_key, PROCESS_PRIVATE_HEAP_ID);
key_buffer_init(&range_itor->local_max_key, PROCESS_PRIVATE_HEAP_ID);

if (core_key_compare(spl, min_key, start_key) > 0) {
// in bounds, start at min
Expand Down Expand Up @@ -971,7 +975,7 @@ core_range_iterator_init(core_handle *spl,
range_itor->itor[i] = &btree_itor->super;
}

rc = merge_iterator_create(spl->heap_id,
rc = merge_iterator_create(PROCESS_PRIVATE_HEAP_ID,
spl->cfg.data_cfg,
range_itor->num_branches,
range_itor->itor,
Expand All @@ -994,7 +998,7 @@ core_range_iterator_init(core_handle *spl,
if (core_key_compare(spl, local_max, max_key) < 0) {
key_buffer local_max_buffer;
rc = key_buffer_init_from_key(
&local_max_buffer, spl->heap_id, local_max);
&local_max_buffer, PROCESS_PRIVATE_HEAP_ID, local_max);
core_range_iterator_deinit(range_itor);
if (!SUCCESS(rc)) {
return rc;
Expand Down Expand Up @@ -1023,7 +1027,7 @@ core_range_iterator_init(core_handle *spl,
if (core_key_compare(spl, local_min, min_key) > 0) {
key_buffer local_min_buffer;
rc = key_buffer_init_from_key(
&local_min_buffer, spl->heap_id, local_min);
&local_min_buffer, PROCESS_PRIVATE_HEAP_ID, local_min);
core_range_iterator_deinit(range_itor);
if (!SUCCESS(rc)) {
return rc;
Expand Down Expand Up @@ -1075,21 +1079,21 @@ core_range_iterator_next(iterator *itor)
if (!range_itor->can_next) {
KEY_CREATE_LOCAL_COPY(rc,
min_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->min_key));
if (!SUCCESS(rc)) {
return rc;
}
KEY_CREATE_LOCAL_COPY(rc,
max_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->max_key));
if (!SUCCESS(rc)) {
return rc;
}
KEY_CREATE_LOCAL_COPY(rc,
local_max_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->local_max_key));
if (!SUCCESS(rc)) {
return rc;
Expand Down Expand Up @@ -1134,21 +1138,21 @@ core_range_iterator_prev(iterator *itor)
if (!range_itor->can_prev) {
KEY_CREATE_LOCAL_COPY(rc,
min_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->min_key));
if (!SUCCESS(rc)) {
return rc;
}
KEY_CREATE_LOCAL_COPY(rc,
max_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->max_key));
if (!SUCCESS(rc)) {
return rc;
}
KEY_CREATE_LOCAL_COPY(rc,
local_min_key,
range_itor->spl->heap_id,
PROCESS_PRIVATE_HEAP_ID,
key_buffer_key(&range_itor->local_min_key));
if (!SUCCESS(rc)) {
return rc;
Expand Down Expand Up @@ -1198,7 +1202,7 @@ core_range_iterator_deinit(core_range_iterator *range_itor)
{
core_handle *spl = range_itor->spl;
if (range_itor->merge_itor != NULL) {
merge_iterator_destroy(range_itor->spl->heap_id, &range_itor->merge_itor);
merge_iterator_destroy(PROCESS_PRIVATE_HEAP_ID, &range_itor->merge_itor);
for (uint64 i = 0; i < range_itor->num_branches; i++) {
btree_iterator *btree_itor = &range_itor->btree_itor[i];
if (range_itor->compacted[i]) {
Expand Down Expand Up @@ -1766,22 +1770,22 @@ core_print_insertion_stats(platform_log_handle *log_handle, const core_handle *s

core_stats *global;

global = TYPED_ZALLOC(spl->heap_id, global);
global = TYPED_ZALLOC(PROCESS_PRIVATE_HEAP_ID, global);
if (global == NULL) {
platform_error_log("Out of memory for statistics");
return;
}

histogram_handle insert_lat_accum, update_lat_accum, delete_lat_accum;
histogram_create(spl->heap_id,
histogram_create(PROCESS_PRIVATE_HEAP_ID,
LATENCYHISTO_SIZE + 1,
latency_histo_buckets,
&insert_lat_accum);
histogram_create(spl->heap_id,
histogram_create(PROCESS_PRIVATE_HEAP_ID,
LATENCYHISTO_SIZE + 1,
latency_histo_buckets,
&update_lat_accum);
histogram_create(spl->heap_id,
histogram_create(PROCESS_PRIVATE_HEAP_ID,
LATENCYHISTO_SIZE + 1,
latency_histo_buckets,
&delete_lat_accum);
Expand Down Expand Up @@ -1840,9 +1844,9 @@ core_print_insertion_stats(platform_log_handle *log_handle, const core_handle *s
histogram_print(insert_lat_accum, "Insert Latency Histogram (ns):", log_handle);
histogram_print(update_lat_accum, "Update Latency Histogram (ns):", log_handle);
histogram_print(delete_lat_accum, "Delete Latency Histogram (ns):", log_handle);
histogram_destroy(spl->heap_id, &insert_lat_accum);
histogram_destroy(spl->heap_id, &update_lat_accum);
histogram_destroy(spl->heap_id, &delete_lat_accum);
histogram_destroy(PROCESS_PRIVATE_HEAP_ID, &insert_lat_accum);
histogram_destroy(PROCESS_PRIVATE_HEAP_ID, &update_lat_accum);
histogram_destroy(PROCESS_PRIVATE_HEAP_ID, &delete_lat_accum);


platform_log(log_handle, "Flush Statistics\n");
Expand Down Expand Up @@ -1892,7 +1896,7 @@ core_print_insertion_stats(platform_log_handle *log_handle, const core_handle *s
platform_log(log_handle, "------------------------------------------------------------------------------------\n");
cache_print_stats(log_handle, spl->cc);
platform_log(log_handle, "\n");
platform_free(spl->heap_id, global);
platform_free(PROCESS_PRIVATE_HEAP_ID, global);
}

void
Expand Down Expand Up @@ -1929,7 +1933,7 @@ void
core_print_lookup(core_handle *spl, key target, platform_log_handle *log_handle)
{
merge_accumulator data;
merge_accumulator_init(&data, spl->heap_id);
merge_accumulator_init(&data, PROCESS_PRIVATE_HEAP_ID);

platform_stream_handle stream;
platform_open_log_stream(&stream);
Expand Down
23 changes: 14 additions & 9 deletions src/platform_linux/platform_heap.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

#include "platform_heap.h"
#include "platform_status.h"
#include <sys/mman.h>

/*
* Declare globals to track heap handle/ID that may have been created when
Expand All @@ -25,24 +26,28 @@ platform_heap_create(platform_module_id UNUSED_PARAM(module_id),
bool use_shmem,
platform_heap_id *heap_id)
{
*heap_id = PROCESS_PRIVATE_HEAP_ID;

if (use_shmem) {
platform_status rc = platform_shmcreate(max, (shmem_heap **)heap_id);
if (SUCCESS(rc)) {
Heap_id = *heap_id;
shmallocator *shm = mmap(
NULL, max, PROT_READ | PROT_WRITE, MAP_ANONYMOUS | MAP_SHARED, -1, 0);
if (shm == MAP_FAILED) {
return STATUS_NO_MEMORY;
}
return rc;
shmallocator_init(shm, max / 4096, max);
*heap_id = (platform_heap_id)shm;

} else {
*heap_id = PROCESS_PRIVATE_HEAP_ID;
}
*heap_id = NULL;
return STATUS_OK;
}

void
platform_heap_destroy(platform_heap_id *heap_id)
{
// If shared segment was allocated, it's being tracked thru heap ID.
if (*heap_id) {
return platform_shmdestroy((shmem_heap **)heap_id);
size_t size = shmallocator_size((shmallocator *)*heap_id);
shmallocator_deinit((shmallocator *)*heap_id);
munmap((void *)*heap_id, size);
*heap_id = NULL;
}
}
46 changes: 9 additions & 37 deletions src/platform_linux/platform_heap.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
#include "platform_util.h"
#include "platform_machine.h"
#include "platform_log.h"
#include "shmem.h"
#include "shmalloc.h"
#include <stddef.h>
#include <stdlib.h>

Expand Down Expand Up @@ -73,24 +73,13 @@ platform_aligned_malloc(const platform_heap_id heap_id,
{
// Requirement for aligned_alloc
platform_assert(IS_POWER_OF_2(alignment));
size_t aligned_size = (size + alignment - 1) & ~((uintptr_t)alignment - 1);

/*
* aligned_alloc requires size to be a multiple of alignment
* round up to nearest multiple of alignment
*
* Note that since this is inlined, the compiler will turn the constant
* (power of 2) alignment mod operations into bitwise &
*/
// RESOLVE: Delete this padding from caller. Push this down to
// platform_shm_alloc().
const size_t padding = platform_align_bytes_reqd(alignment, size);
const size_t required = (size + padding);

void *retptr =
(heap_id
? platform_shm_alloc(heap_id, required, objname, func, file, lineno)
: aligned_alloc(alignment, required));
return retptr;
if (heap_id) {
return shmalloc(heap_id, alignment, size);
} else {
return aligned_alloc(alignment, aligned_size);
}
}

/*
Expand All @@ -115,16 +104,7 @@ platform_realloc(const platform_heap_id heap_id,

// Farm control off to shared-memory based realloc, if it's configured
if (heap_id) {
// The shmem-based allocator is expecting all memory requests to be of
// aligned sizes, as that's what platform_aligned_malloc() does. So, to
// keep that allocator happy, align this memory request if needed.
// As this is the case of realloc, we assume that it would suffice to
// align at platform's natural cacheline boundary.
const size_t padding =
platform_align_bytes_reqd(PLATFORM_CACHELINE_SIZE, newsize);
const size_t required = (newsize + padding);
return platform_shm_realloc(
heap_id, ptr, oldsize, required, __func__, __FILE__, __LINE__);
return shrealloc(heap_id, ptr, newsize);
} else {
return realloc(ptr, newsize);
}
Expand All @@ -139,14 +119,12 @@ platform_free_from_heap(platform_heap_id heap_id,
int lineno)
{
if (heap_id) {
platform_shm_free(heap_id, ptr, objname, func, file, lineno);
shfree(heap_id, ptr);
} else {
free(ptr);
}
}

typedef struct shmem_heap shmem_heap;

platform_status
platform_heap_create(platform_module_id module_id,
size_t max,
Expand All @@ -156,12 +134,6 @@ platform_heap_create(platform_module_id module_id,
void
platform_heap_destroy(platform_heap_id *heap_id);

void
platform_shm_set_splinterdb_handle(platform_heap_id heap_id, void *addr);

shmem_heap *
platform_heap_id_to_shmaddr(platform_heap_id hid);

/*
* Similar to the TYPED_MALLOC functions, for all the free functions we need to
* call platform_get_heap_id() from a macro instead of an inline function
Expand Down
Loading