From 380ae5cbb53d2ad7e0893f99c15d8426f1486672 Mon Sep 17 00:00:00 2001 From: Shintaro Iwasaki Date: Sun, 15 Jul 2018 15:44:51 -0500 Subject: [PATCH 1/3] mem/pool: add a scalable memory allocator. --- configure.ac | 1 + src/include/Makefile.mk | 1 + src/include/mem/zm_pool.h | 19 +++ src/mem/Makefile.mk | 4 +- src/mem/zm_pool.c | 288 ++++++++++++++++++++++++++++++++++++ test/regres/Makefile.am | 2 +- test/regres/mem/Makefile.am | 20 +++ test/regres/mem/mem_pool.c | 127 ++++++++++++++++ 8 files changed, 459 insertions(+), 3 deletions(-) create mode 100644 src/include/mem/zm_pool.h create mode 100644 src/mem/zm_pool.c create mode 100644 test/regres/mem/Makefile.am create mode 100644 test/regres/mem/mem_pool.c diff --git a/configure.ac b/configure.ac index 6979174..c2339e3 100644 --- a/configure.ac +++ b/configure.ac @@ -344,6 +344,7 @@ AC_CONFIG_FILES([Makefile test/regres/Makefile test/regres/lock/Makefile test/regres/cond/Makefile + test/regres/mem/Makefile test/regres/list/Makefile test/regres/queue/Makefile test/perf/Makefile diff --git a/src/include/Makefile.mk b/src/include/Makefile.mk index f43c111..2d7c443 100644 --- a/src/include/Makefile.mk +++ b/src/include/Makefile.mk @@ -19,6 +19,7 @@ zm_headers = \ include/cond/zm_cond_types.h \ include/cond/zm_ccond.h \ include/cond/zm_scount.h \ + include/mem/zm_pool.h \ include/queue/zm_queue_types.h \ include/queue/zm_glqueue.h \ include/queue/zm_swpqueue.h \ diff --git a/src/include/mem/zm_pool.h b/src/include/mem/zm_pool.h new file mode 100644 index 0000000..271c50d --- /dev/null +++ b/src/include/mem/zm_pool.h @@ -0,0 +1,19 @@ +/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */ +/* + * See COPYRIGHT in top-level directory. + */ + +#ifndef _ZM_POOL_H_ +#define _ZM_POOL_H_ + +#include "common/zm_common.h" +#include + +typedef zm_ptr_t *zm_pool_t; + +int zm_pool_create(size_t element_size, zm_pool_t *handle); +int zm_pool_destroy(zm_pool_t *handle); +int zm_pool_alloc(zm_pool_t handle, int tid, void **ptr); +int zm_pool_free(zm_pool_t handle, int tid, void *ptr); + +#endif /* _ZM_POOL_H_ */ \ No newline at end of file diff --git a/src/mem/Makefile.mk b/src/mem/Makefile.mk index 3272a02..4af11c4 100644 --- a/src/mem/Makefile.mk +++ b/src/mem/Makefile.mk @@ -4,5 +4,5 @@ # zm_sources += \ - mem/zm_hzdptr.c - + mem/zm_hzdptr.c \ + mem/zm_pool.c diff --git a/src/mem/zm_pool.c b/src/mem/zm_pool.c new file mode 100644 index 0000000..e590b40 --- /dev/null +++ b/src/mem/zm_pool.c @@ -0,0 +1,288 @@ +/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */ +/* + * See COPYRIGHT in top-level directory. + */ + +#include "mem/zm_pool.h" +#include +#include + +// Bulk size (bytes) +#define PAGESIZE (4 * 1024 * 1024) +// Block size +#define BLOCKSIZE 256 +// Local pool capacity +#define LOCALPOOL_NUM_BLOCKS 2 +// Number of blocks that are taken from a global pool. +#define GLOBAL_TO_LOCAL_NUM_BLOCKS 1 +// Number of blocks that are returned to a global pool. +#define LOCAL_TO_GLOBAL_NUM_BLOCKS 1 + +struct element { + struct element *next; + // element is the first sizeof(element) bytes of an element. +}; + +struct memory_bulk { + struct memory_bulk *next; + // memory_bulk is the first sizeof(memory_bulk) bytes of a page. +} __attribute__((aligned(ZM_CACHELINE_SIZE))); + +struct block { + int num_elements; + struct element *head; + struct element *tail; +}; + +struct local_pool { + int num_elements; + struct block *blocks; +} __attribute__((aligned(ZM_CACHELINE_SIZE))); + +struct global_pool { + // Assumes tids range from [0.. max_threads) + size_t element_size; + int max_threads; + struct local_pool **local_pools; + __attribute__((aligned(ZM_CACHELINE_SIZE))) + int lock; + __attribute__((aligned(ZM_CACHELINE_SIZE))) + int num_elements; + int len_blocks; + struct block *blocks; + struct memory_bulk *bulk; +} __attribute__((aligned(ZM_CACHELINE_SIZE))); + +static inline void *element_to_ptr(struct element *element) { + return (void *)(((char *)element) + sizeof(struct element)); +} + +static inline struct element *ptr_to_element(void *ptr) { + return (struct element *)(((char *)ptr) - sizeof(struct element)); +} + +static inline void *allocate(size_t size) { + void *ptr; + if (posix_memalign(&ptr, ZM_CACHELINE_SIZE, size)) + return NULL; + return ptr; +} + +static inline int get_max_threads() { + hwloc_topology_t topo; + hwloc_topology_init(&topo); + hwloc_topology_load(topo); + int max_threads = hwloc_get_nbobjs_by_type(topo, HWLOC_OBJ_PU); + hwloc_topology_destroy(topo); + return max_threads; +} + +static inline void lock_create(struct global_pool *global_pool) { + char *p_lock = (char *)&global_pool->lock; + *p_lock = 0; +} + +static inline void lock_acquire(struct global_pool *global_pool) { + char *p_lock = (char *)&global_pool->lock; + while (zm_atomic_flag_test_and_set(p_lock, zm_memord_acquire) == 1) { + while (zm_atomic_load(p_lock, zm_memord_acquire) != 0) { + ; // pause? + } + } +} + +static inline void lock_release(struct global_pool *global_pool) { + char *p_lock = (char *)&global_pool->lock; + zm_atomic_flag_clear(p_lock, zm_memord_release); +} + +static inline void lock_destroy(struct global_pool *global_pool) { + // Do nothing. +} + +static struct local_pool *local_pool_create(size_t element_size) { + struct local_pool *local_pool = (struct local_pool *)allocate(sizeof(struct local_pool)); + local_pool->num_elements = 0; + local_pool->blocks = (struct block *)allocate(sizeof(struct block) * LOCALPOOL_NUM_BLOCKS); + memset(local_pool->blocks, 0, sizeof(struct block) * LOCALPOOL_NUM_BLOCKS); + return local_pool; +} + +static void local_pool_destroy(struct local_pool *local_pool) { + free(local_pool->blocks); + free(local_pool); +} + +int zm_pool_create(size_t element_size, zm_pool_t *handle) { + struct global_pool *global_pool = (struct global_pool *)allocate(sizeof(struct global_pool)); + global_pool->element_size = element_size; + global_pool->max_threads = get_max_threads(); + global_pool->local_pools = (struct local_pool **)allocate(sizeof(struct local_pool *) * global_pool->max_threads); + for (int tid = 0; tid < global_pool->max_threads; tid++) { + global_pool->local_pools[tid] = local_pool_create(element_size); + } + lock_create(global_pool); + global_pool->num_elements = 0; + global_pool->len_blocks = 0; + global_pool->blocks = NULL; + global_pool->bulk = NULL; + *handle = (zm_pool_t)global_pool; + return 0; +} + +int zm_pool_destroy(zm_pool_t *handle) { + struct global_pool *global_pool = *(struct global_pool **)handle; + // Free local pools. + for (int tid = 0; tid < global_pool->max_threads; tid++) { + local_pool_destroy(global_pool->local_pools[tid]); + } + free(global_pool->local_pools); + // Free lock. + lock_destroy(global_pool); + // Free blocks. + free(global_pool->blocks); + // Free bulks. + struct memory_bulk *bulk = global_pool->bulk; + while (bulk) { + struct memory_bulk *next = bulk->next; + free(bulk); + bulk = next; + } + + free(global_pool); + return 0; +} + +int zm_pool_alloc(zm_pool_t handle, int tid, void **ptr) { + struct global_pool *global_pool = (struct global_pool *)handle; + struct local_pool *local_pool = global_pool->local_pools[tid]; + if (local_pool->num_elements == 0) { + lock_acquire(global_pool); + while (global_pool->num_elements < GLOBAL_TO_LOCAL_NUM_BLOCKS * BLOCKSIZE) { + // Allocate a new bulk. + const size_t element_size = ((sizeof(struct element) + global_pool->element_size + ZM_CACHELINE_SIZE - 1) / ZM_CACHELINE_SIZE) * ZM_CACHELINE_SIZE; + size_t bulk_size = PAGESIZE; + char *mem = (char *)allocate(bulk_size); + struct memory_bulk *bulk = (struct memory_bulk *)mem; + mem += sizeof(struct memory_bulk); + bulk_size -= sizeof(struct memory_bulk); + // Add a new bulk. + bulk->next = global_pool->bulk; + global_pool->bulk = bulk; + // Create elements. + int block_i = global_pool->num_elements / BLOCKSIZE; + struct block *blocks = global_pool->blocks; + int len_blocks = global_pool->len_blocks; + int num_elements = global_pool->num_elements; + while (bulk_size >= element_size) { + struct element *new_element = (struct element *)mem; + new_element->next = NULL; + mem += element_size; + bulk_size -= element_size; + if (block_i >= len_blocks) { + // Extend blocks. + int new_len_blocks = (len_blocks == 0) ? 1 : (len_blocks * 2); + struct block *new_blocks = (struct block *)allocate(sizeof(struct block) * new_len_blocks); + memcpy(new_blocks, blocks, sizeof(struct block) * len_blocks); + memset(&new_blocks[len_blocks], 0, sizeof(struct block) * (new_len_blocks - len_blocks)); + len_blocks = new_len_blocks; + free(blocks); + blocks = new_blocks; + } + // Put it to a tail block. + struct block *block = &blocks[block_i]; + if (block->num_elements == 0) { + block->head = new_element; + block->tail = new_element; + } else { + block->tail->next = new_element; + block->tail = new_element; + } + block->num_elements++; + if (block->num_elements == BLOCKSIZE) + block_i++; + num_elements++; + } + global_pool->blocks = blocks; + global_pool->len_blocks = len_blocks; + global_pool->num_elements = num_elements; + } + // Take blocks from a global pool. + int global_block_i = global_pool->num_elements / BLOCKSIZE; + memcpy(local_pool->blocks, &global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], sizeof(struct block) * GLOBAL_TO_LOCAL_NUM_BLOCKS); + local_pool->num_elements = BLOCKSIZE * GLOBAL_TO_LOCAL_NUM_BLOCKS; + global_pool->num_elements -= BLOCKSIZE * GLOBAL_TO_LOCAL_NUM_BLOCKS; + if (global_pool->num_elements % BLOCKSIZE != 0) { + // Copy the last block of global_pool and clear the remaining blocks. + memcpy(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], &global_pool->blocks[global_block_i], sizeof(struct block)); + memset(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS + 1], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS + 1))); + } else { + // Clear the remaining blocks. + memset(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS))); + } + lock_release(global_pool); + } + { + // Take an element from a local pool. + local_pool->num_elements -= 1; + int block_i = local_pool->num_elements / BLOCKSIZE; + struct block *block = &local_pool->blocks[block_i]; + struct element *element = block->head; + struct element *next = element->next; + block->head = next; + block->num_elements -= 1; + *ptr = element_to_ptr(element); + } + return 0; +} + +int zm_pool_free(zm_pool_t handle, int tid, void *ptr) { + struct global_pool *global_pool = (struct global_pool *)handle; + struct local_pool *local_pool = global_pool->local_pools[tid]; + if (local_pool->num_elements == LOCALPOOL_NUM_BLOCKS * BLOCKSIZE) { + lock_acquire(global_pool); + int len_required_blocks = (global_pool->num_elements + BLOCKSIZE - 1 + LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE) / BLOCKSIZE; + struct block* blocks = global_pool->blocks; + if (global_pool->len_blocks < len_required_blocks) { + // Extend blocks. + const int len_blocks = global_pool->len_blocks; + int new_len_blocks = (len_blocks * 2 < len_required_blocks) ? len_required_blocks : (len_blocks * 2); + struct block *new_blocks = (struct block *)allocate(sizeof(struct block) * new_len_blocks); + memcpy(new_blocks, blocks, sizeof(struct block) * len_blocks); + memset(&new_blocks[len_blocks], 0, sizeof(struct block) * (new_len_blocks - len_blocks)); + global_pool->len_blocks = new_len_blocks; + free(blocks); + blocks = new_blocks; + global_pool->blocks = new_blocks; + } + // Return blocks to a global pool. + int block_i = global_pool->num_elements / BLOCKSIZE; + if (global_pool->num_elements % BLOCKSIZE != 0) { + // Copy the last block of a global pool. + memcpy(&blocks[block_i + LOCAL_TO_GLOBAL_NUM_BLOCKS], &blocks[block_i], sizeof(struct block)); + } + memcpy(&blocks[block_i], &local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); + memset(&local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], 0, sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); + global_pool->num_elements += LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE; + local_pool->num_elements -= LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE; + lock_release(global_pool); + } + { + // Return an element to a local pool. + int block_i = local_pool->num_elements / BLOCKSIZE; + local_pool->num_elements += 1; + struct block *block = &local_pool->blocks[block_i]; + struct element *element = ptr_to_element(ptr); + if (block->num_elements == 0) { + block->head = element; + block->tail = element; + element->next = NULL; + } else { + element->next = block->head; + block->head = element; + } + block->num_elements += 1; + } + return 0; +} + diff --git a/test/regres/Makefile.am b/test/regres/Makefile.am index 7faea86..c269f88 100644 --- a/test/regres/Makefile.am +++ b/test/regres/Makefile.am @@ -3,5 +3,5 @@ # See COPYRIGHT in top-level directory. # -SUBDIRS = lock cond list queue +SUBDIRS = lock cond mem list queue DIST_SUBDIRS = $(SUBDIRS) diff --git a/test/regres/mem/Makefile.am b/test/regres/mem/Makefile.am new file mode 100644 index 0000000..15c63f6 --- /dev/null +++ b/test/regres/mem/Makefile.am @@ -0,0 +1,20 @@ +# -*- Mode: Makefile; -*- +# +# See COPYRIGHT in top-level directory. +# + +TESTS = \ + mem_pool + +XFAIL_TESTS = + +check_PROGRAMS = $(TESTS) +noinst_PROGRAMS = $(TESTS) + +include $(top_srcdir)/test/Makefile.mk + +mem_pool_SOURCES = mem_pool.c + +mem_pool_CFLAGS = -D_GNU_SOURCE + +mem_pool_LDFLAGS = -pthread \ No newline at end of file diff --git a/test/regres/mem/mem_pool.c b/test/regres/mem/mem_pool.c new file mode 100644 index 0000000..e51053e --- /dev/null +++ b/test/regres/mem/mem_pool.c @@ -0,0 +1,127 @@ +/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */ +/* + * See COPYRIGHT in top-level directory. + */ + +#include +#include +#include +#include +#include + +#define TEST_NTHREADS 3 +#define TEST_NELEMTS 20000 +#define NUM_OPERATIONS 30000 + +typedef struct thread_data thread_data_t; +struct thread_data { + int tid; + size_t element_size; + zm_pool_t pool; +}; + +static void* func(void *arg) { + thread_data_t *data = (thread_data_t*) arg; + const int tid = data->tid; + const int id = tid % 128; + const size_t element_size = data->element_size; + zm_pool_t pool = data->pool; + + unsigned short xsubi[3] = {tid, id, element_size}; + void **ptrs = (void *)malloc(sizeof(void *) * TEST_NELEMTS); + + int r, num_elements = 0; + do { + double rand = erand48(xsubi); + if (rand < 0.5) { + // Allocate elements + int num_allocs = (rand * 2) * 1000; + if (num_elements + num_allocs >= TEST_NELEMTS) { + num_allocs = TEST_NELEMTS - num_elements; + } + if (num_allocs == 0) + continue; + for (int i = 0; i < num_allocs; i++) { + void *ptr = NULL; + zm_pool_alloc(pool, tid, &ptr); + memset(ptr, id, element_size); + ptrs[num_elements++] = ptr; + } + } else { + // Free elements + int num_frees = ((rand - 0.5)* 2) * 1000; + if (num_frees > num_elements) { + num_frees = num_elements; + } + if (num_frees == 0) + continue; + for (int i = 0; i < num_frees; i++) { + void *ptr = ptrs[--num_elements]; + for (int j = 0; j < (int)element_size; j++) { + char val = ((char *)ptr)[j]; + if (val != id) { + printf("Failed: got ptrs[%d][%d] %d instead of %d\n", num_elements, j, (int) val, id); + abort(); + } + } + zm_pool_free(pool, tid, ptr); + } + } + + } while(r++ <= NUM_OPERATIONS); + + // Free remaining elements. + for (int i = 0; i < num_elements; i++) { + void *ptr = ptrs[i]; + for (int j = 0; j < (int)element_size; j++) { + char val = ((char *)ptr)[j]; + if (val != id) { + printf("Failed: got ptrs[%d][%d] %d instead of %d\n", i, j, (int) val, id); + abort(); + } + } + zm_pool_free(pool, tid, ptr); + } + + free(ptrs); + return 0; +} + +/*------------------------------------------------------------------------- + * Function: run + * + * Purpose: Test the pool operations by repeating alloc() and free() by + * multiple threads. + * + * Return: Success: 0 + * Failure: 1 + *------------------------------------------------------------------------- + */ +static void run() { + pthread_t threads[TEST_NTHREADS]; + thread_data_t data[TEST_NTHREADS]; + + const size_t element_sizes[] = {64, 128, 256, 1024}; + + for (int i = 0; i < sizeof(element_sizes) / sizeof(const size_t); i++) { + zm_pool_t pool; + zm_pool_create(element_sizes[i], &pool); + + for (int tid = 0; tid < TEST_NTHREADS; tid++) { + data[tid].tid = tid; + data[tid].element_size = element_sizes[i]; + data[tid].pool = pool; + } + for (int tid = 1; tid < TEST_NTHREADS; tid++) + pthread_create(&threads[tid], NULL, func, (void*) &data[tid]); + func(&data[0]); + for (int tid = 1; tid < TEST_NTHREADS; tid++) + pthread_join(threads[tid], NULL); + + zm_pool_destroy(&pool); + } +} + +int main(int argc, char **argv) { + run(); +} From c483f7fc10d756d2ef65d024e41057f4ab179c7a Mon Sep 17 00:00:00 2001 From: Halim Amer Date: Thu, 23 Aug 2018 17:40:49 +0000 Subject: [PATCH 2/3] swpq: playing around with the new memory pool --- src/include/queue/zm_queue_types.h | 2 + src/include/queue/zm_swpqueue.h | 5 ++ src/queue/zm_swpqueue.c | 41 +++++++++++++++ test/perf/queue/Makefile.am | 8 ++- test/perf/queue/thread_scale.c | 4 +- test/perf/queue/thscale_swp.c | 81 ++++++++++++++++++++++++++++++ test/regres/mem/mem_pool.c | 2 +- 7 files changed, 138 insertions(+), 5 deletions(-) create mode 100644 test/perf/queue/thscale_swp.c diff --git a/src/include/queue/zm_queue_types.h b/src/include/queue/zm_queue_types.h index e48da4a..18da40a 100644 --- a/src/include/queue/zm_queue_types.h +++ b/src/include/queue/zm_queue_types.h @@ -6,6 +6,7 @@ #ifndef _ZM_QUEUE_TYPES_H #define _ZM_QUEUE_TYPES_H #include "common/zm_common.h" +#include "mem/zm_pool.h" #include #include @@ -40,6 +41,7 @@ struct zm_msqnode { struct zm_msqueue { zm_atomic_ptr_t head ZM_ALLIGN_TO_CACHELINE; zm_atomic_ptr_t tail ZM_ALLIGN_TO_CACHELINE; + zm_pool_t pool; }; /* faqueue */ diff --git a/src/include/queue/zm_swpqueue.h b/src/include/queue/zm_swpqueue.h index 2a4c621..95420a6 100644 --- a/src/include/queue/zm_swpqueue.h +++ b/src/include/queue/zm_swpqueue.h @@ -15,4 +15,9 @@ int zm_swpqueue_dequeue(zm_swpqueue_t* q, void **data); int zm_swpqueue_isempty_weak(zm_swpqueue_t* q); int zm_swpqueue_isempty_strong(zm_swpqueue_t* q); +int zm_swpqueue_init_explicit(zm_swpqueue_t *, int); +int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data, int); +int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data, int); + + #endif /* _ZM_SWPQUEUE_H */ diff --git a/src/queue/zm_swpqueue.c b/src/queue/zm_swpqueue.c index a33b338..07174cb 100644 --- a/src/queue/zm_swpqueue.c +++ b/src/queue/zm_swpqueue.c @@ -53,3 +53,44 @@ int zm_swpqueue_isempty_strong(zm_swpqueue_t* q) { tail = (zm_swpqnode_t*) zm_atomic_load(&q->tail, zm_memord_acquire); return ((head->next == ZM_NULL) && (head == tail)); } + +int zm_swpqueue_init_explicit(zm_swpqueue_t *q, int tid) { + + zm_pool_create(sizeof(zm_swpqnode_t), &q->pool); + zm_swpqnode_t* node; + zm_pool_alloc(q->pool, tid, (void**)&node); + node->data = NULL; + node->next = ZM_NULL; + zm_atomic_store(&q->head, (zm_ptr_t)node, zm_memord_release); + zm_atomic_store(&q->tail, (zm_ptr_t)node, zm_memord_release); + return 0; +} + +int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data, int tid) { + zm_swpqnode_t* pred; + zm_swpqnode_t* node; + zm_pool_alloc(q->pool, tid, (void**)&node); + node->data = data; + zm_atomic_store(&node->next, ZM_NULL, zm_memord_release); + pred = (zm_swpqnode_t*)zm_atomic_exchange(&q->tail, (zm_ptr_t)node, zm_memord_acq_rel); + zm_atomic_store(&pred->next, (zm_ptr_t)node, zm_memord_release); + return 0; +} + +int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data, int tid) { + zm_swpqnode_t* head; + zm_ptr_t next; + *data = NULL; + head = (zm_swpqnode_t*) zm_atomic_load(&q->head, zm_memord_acquire); + /* At least one element in the queue: + ==> head != tail + ==> no consistency issues between enqueuers and dequeuers */ + if (head->next != ZM_NULL) { + next = (zm_ptr_t) zm_atomic_load(&head->next, zm_memord_acquire); + zm_atomic_store(&q->head, next, zm_memord_release); + *data = ((zm_swpqnode_t*)next)->data; + zm_pool_free(q->pool, tid, (void*)head); + } + return 1; +} + diff --git a/test/perf/queue/Makefile.am b/test/perf/queue/Makefile.am index f40ed32..eb7d906 100644 --- a/test/perf/queue/Makefile.am +++ b/test/perf/queue/Makefile.am @@ -16,7 +16,8 @@ TESTS = \ thread_scale_spmc_ms \ enq_deq_pairs_fa \ thread_scale_mpsc_fa \ - thscale_mpb + thscale_mpb \ + thscale_swp #XFAIL_TESTS = enq_deq_pairs_fa \ # thread_scale_mpsc_fa @@ -39,6 +40,7 @@ thread_scale_mpsc_ms_SOURCES = thread_scale.c thread_scale_spmc_gl_SOURCES = thread_scale.c thread_scale_spmc_ms_SOURCES = thread_scale.c thscale_mpb_SOURCES = thscale_mpb.c +thscale_swp_SOURCES = thscale_swp.c enq_deq_pairs_gl_CFLAGS = -DZMTEST_USE_GLQUEUE -DZMTEST_ALLOC_QELEM -fopenmp enq_deq_pairs_ms_CFLAGS = -DZMTEST_USE_MSQUEUE -DZMTEST_ALLOC_QELEM -fopenmp @@ -47,12 +49,13 @@ enq_deq_pairs_fa_CFLAGS = -DZMTEST_USE_FAQUEUE -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_mpmc_gl_CFLAGS = -DZMTEST_USE_GLQUEUE -DZMTEST_MPMC -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_mpmc_ms_CFLAGS = -DZMTEST_USE_MSQUEUE -DZMTEST_MPMC -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_mpsc_gl_CFLAGS = -DZMTEST_USE_GLQUEUE -DZMTEST_MPSC -DZMTEST_ALLOC_QELEM -fopenmp -thread_scale_mpsc_swp_CFLAGS = -DZMTEST_USE_SWPQUEUE -DZMTEST_MPSC -fopenmp +thread_scale_mpsc_swp_CFLAGS = -DZMTEST_USE_SWPQUEUE -DZMTEST_MPSC -fopenmp -DZMTEST_ALLOC_QELEM thread_scale_mpsc_fa_CFLAGS = -DZMTEST_USE_FAQUEUE -DZMTEST_MPSC -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_mpsc_ms_CFLAGS = -DZMTEST_USE_MSQUEUE -DZMTEST_MPSC -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_spmc_gl_CFLAGS = -DZMTEST_USE_GLQUEUE -DZMTEST_SPMC -DZMTEST_ALLOC_QELEM -fopenmp thread_scale_spmc_ms_CFLAGS = -DZMTEST_USE_MSQUEUE -DZMTEST_SPMC -DZMTEST_ALLOC_QELEM -fopenmp thscale_mpb_CFLAGS = -fopenmp +thscale_swp_CFLAGS = -fopenmp enq_deq_pairs_gl_LDFLAGS = -fopenmp enq_deq_pairs_ms_LDFLAGS = -fopenmp @@ -67,3 +70,4 @@ thread_scale_mpsc_ms_LDFLAGS = -fopenmp thread_scale_spmc_gl_LDFLAGS = -fopenmp thread_scale_spmc_ms_LDFLAGS = -fopenmp thscale_mpb_LDFLAGS = -fopenmp +thscale_swp_LDFLAGS = -fopenmp diff --git a/test/perf/queue/thread_scale.c b/test/perf/queue/thread_scale.c index 2f0f79a..94940d4 100644 --- a/test/perf/queue/thread_scale.c +++ b/test/perf/queue/thread_scale.c @@ -6,7 +6,7 @@ #include #include #include "zmtest_absqueue.h" -#define TEST_NELEMTS 64 +#define TEST_NELEMTS 440 #define NITER (1024*32) /*------------------------------------------------------------------------- @@ -27,7 +27,7 @@ static inline void run() { printf("#threads \t throughput ops/s\n"); int nthreads; - for (nthreads = 2; nthreads <= omp_get_max_threads(); nthreads ++) { + for (nthreads = 2; nthreads <= omp_get_max_threads(); nthreads *= 2 ) { zm_absqueue_init(&queue); int nelem_enq, nelem_deq; diff --git a/test/perf/queue/thscale_swp.c b/test/perf/queue/thscale_swp.c new file mode 100644 index 0000000..cdf857e --- /dev/null +++ b/test/perf/queue/thscale_swp.c @@ -0,0 +1,81 @@ +/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */ +/* + * See COPYRIGHT in top-level directory. + */ +#include +#include +#include +#include "queue/zm_swpqueue.h" +#include "mem/zm_pool.h" +#define TEST_NELEMTS 440 +#define NITER (1024*32) + +zm_pool_t pool; + +/*------------------------------------------------------------------------- + * Function: run + * + * Purpose: Test the correctness of queue operations by counting the number + * of dequeued elements to the expected number + * + * Return: Success: 0 + * Failure: 1 + *------------------------------------------------------------------------- + */ +static inline void run() { + unsigned test_counter = 0; + zm_swpqueue_t queue; + double t1, t2; + + printf("#threads \t throughput ops/s\n"); + + int nthreads; + for (nthreads = 2; nthreads <= omp_get_max_threads(); nthreads *= 2) { + zm_swpqueue_init_explicit(&queue, omp_get_thread_num()); + int nelem_enq, nelem_deq; + + nelem_enq = TEST_NELEMTS/(nthreads-1); + nelem_deq = (nthreads-1)*nelem_enq; + + t1 = omp_get_wtime(); + + #pragma omp parallel num_threads(nthreads) + { + int tid, producer_b; + int *input; + tid = omp_get_thread_num(); + producer_b = (tid != 0); + int elem; + + for(int i = 0; i Date: Tue, 4 Sep 2018 23:36:48 +0000 Subject: [PATCH 3/3] mem/pool: improve the scalable memory allocator. - thread ID is no longer needed. - Add an embedded option "USE_PAGE". With USE_PAGE, allocation becomes page-level. Without USE_PAGE, allocation becomes object-level. --- src/include/mem/zm_pool.h | 4 +- src/include/queue/zm_queue_types.h | 2 +- src/include/queue/zm_swpqueue.h | 6 +- src/mem/zm_pool.c | 852 +++++++++++++++++++++++------ src/queue/zm_swpqueue.c | 12 +- test/perf/queue/thscale_swp.c | 10 +- test/regres/mem/mem_pool.c | 27 +- 7 files changed, 717 insertions(+), 196 deletions(-) diff --git a/src/include/mem/zm_pool.h b/src/include/mem/zm_pool.h index 271c50d..f4636e8 100644 --- a/src/include/mem/zm_pool.h +++ b/src/include/mem/zm_pool.h @@ -13,7 +13,7 @@ typedef zm_ptr_t *zm_pool_t; int zm_pool_create(size_t element_size, zm_pool_t *handle); int zm_pool_destroy(zm_pool_t *handle); -int zm_pool_alloc(zm_pool_t handle, int tid, void **ptr); -int zm_pool_free(zm_pool_t handle, int tid, void *ptr); +int zm_pool_alloc(zm_pool_t handle, void **ptr); +int zm_pool_free(zm_pool_t handle, void *ptr); #endif /* _ZM_POOL_H_ */ \ No newline at end of file diff --git a/src/include/queue/zm_queue_types.h b/src/include/queue/zm_queue_types.h index 18da40a..b1d685b 100644 --- a/src/include/queue/zm_queue_types.h +++ b/src/include/queue/zm_queue_types.h @@ -41,7 +41,7 @@ struct zm_msqnode { struct zm_msqueue { zm_atomic_ptr_t head ZM_ALLIGN_TO_CACHELINE; zm_atomic_ptr_t tail ZM_ALLIGN_TO_CACHELINE; - zm_pool_t pool; + zm_pool_t pool ZM_ALLIGN_TO_CACHELINE; }; /* faqueue */ diff --git a/src/include/queue/zm_swpqueue.h b/src/include/queue/zm_swpqueue.h index 95420a6..9b19128 100644 --- a/src/include/queue/zm_swpqueue.h +++ b/src/include/queue/zm_swpqueue.h @@ -15,9 +15,9 @@ int zm_swpqueue_dequeue(zm_swpqueue_t* q, void **data); int zm_swpqueue_isempty_weak(zm_swpqueue_t* q); int zm_swpqueue_isempty_strong(zm_swpqueue_t* q); -int zm_swpqueue_init_explicit(zm_swpqueue_t *, int); -int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data, int); -int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data, int); +int zm_swpqueue_init_explicit(zm_swpqueue_t *); +int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data); +int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data); #endif /* _ZM_SWPQUEUE_H */ diff --git a/src/mem/zm_pool.c b/src/mem/zm_pool.c index e590b40..2065aee 100644 --- a/src/mem/zm_pool.c +++ b/src/mem/zm_pool.c @@ -5,275 +5,787 @@ #include "mem/zm_pool.h" #include -#include +#include +#include +#include + +//////////////////////////////////////////////////////////////////////////////// +// Configurable parameters +//////////////////////////////////////////////////////////////////////////////// -// Bulk size (bytes) -#define PAGESIZE (4 * 1024 * 1024) // Block size -#define BLOCKSIZE 256 +#define BLOCKSIZE_LOG 12 + +#define BLOCKSIZE (1 << BLOCKSIZE_LOG) // Local pool capacity -#define LOCALPOOL_NUM_BLOCKS 2 -// Number of blocks that are taken from a global pool. -#define GLOBAL_TO_LOCAL_NUM_BLOCKS 1 -// Number of blocks that are returned to a global pool. -#define LOCAL_TO_GLOBAL_NUM_BLOCKS 1 +#define LOCALPOOL_NUM_BLOCKS 4 +// Number of blocks that are taken from a global pool. It should be smaller than LOCALPOOL_NUM_BLOCKS +#define GLOBAL_TO_LOCAL_NUM_BLOCKS 2 + +#if LOCALPOOL_NUM_BLOCKS <= GLOBAL_TO_LOCAL_NUM_BLOCKS +#error "LOCALPOOL_NUM_BLOCKS must be smaller than GLOBAL_TO_LOCAL_NUM_BLOCKS." +#endif + +// Number of blocks that are returned to a global pool. It should be smaller than LOCALPOOL_NUM_BLOCKS +#define LOCAL_TO_GLOBAL_NUM_BLOCKS 2 + +#if LOCALPOOL_NUM_BLOCKS <= LOCAL_TO_GLOBAL_NUM_BLOCKS +#error "LOCALPOOL_NUM_BLOCKS must be smaller than LOCAL_TO_GLOBAL_NUM_BLOCKS." +#endif + +#define MALLOC_ALIGNMENT 4 + +// Define if page-level allocation is used. +// Page-level allocation +// Pros: First-time allocation becomes fast. +// Cons: Memory is never freed once allocated until pool itself gets freed. +// Object-level allocation +// Pros: Memory can be freed if it gets beyond the threshold. +// Cons: First-time allocation is slow. + +#define USE_PAGE + +#ifdef USE_PAGE + +// Bulk size (bytes). +#define PAGESIZE (8 * 1024 * 1024) + +#else + +// The capacity of the global pool. It should be larger than GLOBAL_TO_LOCAL_NUM_BLOCKS and LOCAL_TO_GLOBAL_NUM_BLOCKS +#define GLOBALPOOL_NUM_BLOCKS 8 + +#if GLOBALPOOL_NUM_BLOCKS <= GLOBAL_TO_LOCAL_NUM_BLOCKS +#error "GLOBALPOOL_NUM_BLOCKS must be larger than GLOBAL_TO_LOCAL_NUM_BLOCKS." +#endif + +#if GLOBALPOOL_NUM_BLOCKS <= LOCAL_TO_GLOBAL_NUM_BLOCKS +#error "GLOBALPOOL_NUM_BLOCKS must be larger than LOCAL_TO_GLOBAL_NUM_BLOCKS." +#endif + +#endif // USE_PAGE + +// Define if array-based block management is used. +// Array-based block management +// Pros: No additional data segment is needed. +// Cons: Block size is large, so copying blocks is costly. +// Linked list-based block management +// Pros: Block size is small, so copying blocks is lightweight. +// Cons: Each element has an additional data segment for linked list. + +#define USE_ARRAY_BASED_BLOCK + +//////////////////////////////////////////////////////////////////////////////// +// Utility +//////////////////////////////////////////////////////////////////////////////// + +static inline void lock_acquire(int *p_int_lock) { + // The caller assumes a spinlock. Do not yield inside when ULT is used. + char *p_lock = (char *)p_int_lock; + while (zm_atomic_flag_test_and_set(p_lock, zm_memord_acquire) == 1) { + while (zm_atomic_load(p_lock, zm_memord_acquire) != 0) { +#if defined(__GNUC__) && defined(__x86_64__) + __asm__ __volatile__ ("pause"); +#endif + } + } +} + +static inline void lock_release(int *p_int_lock) { + // The caller expects immediate return after unlock. Do not yield inside when ULT is used. + char *p_lock = (char *)p_int_lock; + zm_atomic_flag_clear(p_lock, zm_memord_release); +} + +static inline void fast_memset(void *ptr, int value, size_t num) { +#ifdef __GNUC__ + // Since num is usually very small, the function calling overhead is not negligible. + __builtin_memset(ptr, value, num); +#else + memset(ptr, value, num); +#endif +} + +static inline void fast_memcpy(void *dst, const void *src, size_t num) { +#ifdef __GNUC__ + // Since num is usually very small, the function calling overhead is not negligible. + __builtin_memcpy(dst, src, num); +#else + memcpy(dst, src, num); +#endif +} + +static inline void *aligned_malloc(size_t size) { + size_t alloc_size = (size + ZM_CACHELINE_SIZE - 1) & (~(ZM_CACHELINE_SIZE - 1)); + void *mem; + int ret = posix_memalign(&mem, ZM_CACHELINE_SIZE, alloc_size); + if (zm_likely(ret == 0)) + return mem; + return NULL; +} + +static inline void *aligned_calloc(size_t size) { + void *mem = aligned_malloc(size); + if (zm_likely(mem)) { + fast_memset(mem, 0, size); + return mem; + } + return NULL; +} + +static inline void aligned_free(void *ptr) { + free(ptr); +} + +static inline void *aligned_realloc(void *ptr, size_t old_size, size_t new_size) { + void *mem = aligned_malloc(new_size); + fast_memcpy(mem, ptr, old_size); + aligned_free(ptr); + return mem; +} + +//////////////////////////////////////////////////////////////////////////////// +// Extensible thread-local entry +//////////////////////////////////////////////////////////////////////////////// + +struct local_pool; +static struct local_pool *local_pool_create(); +static void local_pool_free(struct local_pool *p_local_pool); + +struct thread_local_entry { // thread local + size_t local_pools_len; + // read local_pools and local_pools[x] : no lock + // update local_pools and local_pools[x]: thread_local_entry_manager.lock + struct local_pool **local_pools; +} __attribute__((aligned(ZM_CACHELINE_SIZE))); + +__attribute__((aligned(ZM_CACHELINE_SIZE))) zm_thread_local struct thread_local_entry *lp_thread_local_entry; + +struct thread_local_entry_manager { // global + // The lock owner may not yield when ULT is used. + int lock; + __attribute__((aligned(ZM_CACHELINE_SIZE))) + // All these variables require thread_local_entry_manager.lock to read or update. + size_t entry_list_len; + char *entry_list; + size_t thread_local_entries_len; + struct thread_local_entry **thread_local_entries; + int is_initialized_thread_local_entry_key; + pthread_key_t thread_local_entry_key; + int num_alive_threads; +} __attribute__((aligned(ZM_CACHELINE_SIZE))); + +__attribute__((aligned(ZM_CACHELINE_SIZE))) struct thread_local_entry_manager g_thread_local_entry_manager; + +void thread_local_entry_destructor(void *value) { + struct thread_local_entry *p_entry = (struct thread_local_entry *)value; + struct thread_local_entry_manager *p_entry_manager = &g_thread_local_entry_manager; + lock_acquire(&p_entry_manager->lock); + // Remove this entry from the p_entry_manager. + for (int i = 0, iend = p_entry_manager->thread_local_entries_len; i < iend; i++) { + if (p_entry_manager->thread_local_entries[i] == p_entry) { + p_entry_manager->thread_local_entries[i] = NULL; + break; + } + } + p_entry_manager->num_alive_threads--; + if (p_entry_manager->num_alive_threads == 0) { + // Free dynamically allocated objects. + // When all entries are freed, no other threads access thread local objects, so we can safely free all the objects. + aligned_free(p_entry_manager->entry_list); + p_entry_manager->entry_list = NULL; + p_entry_manager->entry_list_len = 0; + aligned_free(p_entry_manager->thread_local_entries); + p_entry_manager->thread_local_entries = NULL; + p_entry_manager->thread_local_entries_len = 0; + } + lock_release(&p_entry_manager->lock); + // Free local pools. + for (int entry_index = 0; entry_index < p_entry->local_pools_len; entry_index++) { + if (p_entry->local_pools[entry_index]) + local_pool_free(p_entry->local_pools[entry_index]); + } + if (p_entry->local_pools) + aligned_free(p_entry->local_pools); + aligned_free(p_entry); +} + +static inline struct local_pool *get_thread_local_pool(int entry_index) { + // First, check its thread_local_entry + struct thread_local_entry *p_entry = lp_thread_local_entry; + if (zm_unlikely(!p_entry)) { + // Allocate a new local entry. + struct thread_local_entry_manager *p_entry_manager = &g_thread_local_entry_manager; + p_entry = (struct thread_local_entry *)aligned_calloc(sizeof(struct thread_local_entry)); + // Save a key to free p_entry when the caller dies. + lock_acquire(&p_entry_manager->lock); + if (!p_entry_manager->is_initialized_thread_local_entry_key) { + pthread_key_create(&p_entry_manager->thread_local_entry_key, thread_local_entry_destructor); + p_entry_manager->is_initialized_thread_local_entry_key = 1; + } + p_entry_manager->num_alive_threads++; + pthread_setspecific(p_entry_manager->thread_local_entry_key, p_entry); + lock_release(&p_entry_manager->lock); + lp_thread_local_entry = p_entry; + } + if (zm_unlikely(p_entry->local_pools_len <= entry_index)) { + // Allocate or extend local_pools. This allocation is serialized. + struct thread_local_entry_manager *p_entry_manager = &g_thread_local_entry_manager; + lock_acquire(&p_entry_manager->lock); + if (!p_entry->local_pools) { + // Allocate new local_pools. + size_t new_len = entry_index + 1; + p_entry->local_pools = (struct local_pool **)aligned_calloc(sizeof(struct local_pool *) * new_len); + p_entry->local_pools_len = new_len; + int is_inserted = 0; + for (int i = 0; i < p_entry_manager->thread_local_entries_len; i++) { + if (!p_entry_manager->thread_local_entries[i]) { + p_entry_manager->thread_local_entries[i] = p_entry; + is_inserted = 1; + break; + } + } + if (!is_inserted) { + // Extend thread_local_entries_len. + size_t old_thread_local_entries_len = p_entry_manager->thread_local_entries_len; + size_t new_thread_local_entries_len = (old_thread_local_entries_len == 0) ? 2 : (old_thread_local_entries_len * 2); + p_entry_manager->thread_local_entries = (struct thread_local_entry **)aligned_realloc(p_entry_manager->thread_local_entries, sizeof(struct thread_local_entry *) * old_thread_local_entries_len, sizeof(struct thread_local_entry *) * new_thread_local_entries_len); + fast_memset(p_entry_manager->thread_local_entries + old_thread_local_entries_len, 0, sizeof(struct thread_local_entry *) * (new_thread_local_entries_len - old_thread_local_entries_len)); + p_entry_manager->thread_local_entries_len = new_thread_local_entries_len; + p_entry_manager->thread_local_entries[old_thread_local_entries_len] = p_entry; + } + } else { + const size_t old_len = p_entry->local_pools_len; + if (old_len <= entry_index) { + // Extend new local_pools. + size_t new_len = entry_index + 1; + struct local_pool **new_local_pools = (struct local_pool **)aligned_realloc(p_entry->local_pools, sizeof(struct local_pool *) * old_len, sizeof(struct local_pool *) * new_len); + fast_memset(new_local_pools + old_len, 0, sizeof(struct local_pool *) * (new_len - old_len)); + p_entry->local_pools_len = new_len; + p_entry->local_pools = new_local_pools; + } + } + lock_release(&p_entry_manager->lock); + } + // Next, check if local_pools[entry_index] has an entity. + struct local_pool **local_pools = p_entry->local_pools; + struct local_pool *local_pool = local_pools[entry_index]; + if (zm_unlikely(!local_pool)) { + // Allocate and initialize local_pools[entry_index]. + local_pool = local_pool_create(); + local_pools[entry_index] = local_pool; + } + return local_pool; +} + +// Return newly created entry_index. +static int create_new_entry() { + struct thread_local_entry_manager *p_entry_manager = &g_thread_local_entry_manager; + lock_acquire(&p_entry_manager->lock); + size_t entry_list_len = p_entry_manager->entry_list_len; + if (entry_list_len == 0) { + // Allocate entry_list. + entry_list_len = 8; + p_entry_manager->entry_list = (char *)aligned_calloc(sizeof(char) * entry_list_len); + p_entry_manager->entry_list_len = entry_list_len; + } + while (1) { + // Allocate entry_list. + char *entry_list = p_entry_manager->entry_list; + for (int entry_index = 0; entry_index < entry_list_len; entry_index++) { + if (entry_list[entry_index] == 0) { + entry_list[entry_index] = 1; + lock_release(&p_entry_manager->lock); + return entry_index; + } + } + // Extend entry list. + size_t new_entry_list_len = entry_list_len * 2; + entry_list = (char *)aligned_realloc(entry_list, entry_list_len, new_entry_list_len); + fast_memset(entry_list + entry_list_len, 0, sizeof(char) * (new_entry_list_len - entry_list_len)); + p_entry_manager->entry_list = entry_list; + p_entry_manager->entry_list_len = new_entry_list_len; + entry_list_len = new_entry_list_len; + } + // Unreachable. +} + +// Destroy an entry. +static void free_entry(int entry_index) { + struct thread_local_entry_manager *p_entry_manager = &g_thread_local_entry_manager; + lock_acquire(&p_entry_manager->lock); + // Free entry_index. + p_entry_manager->entry_list[entry_index] = 0; + // Free all local objects. + for (int i = 0, iend = p_entry_manager->thread_local_entries_len; i < iend; i++) { + struct thread_local_entry *p_entry = p_entry_manager->thread_local_entries[i]; + if (p_entry && entry_index < p_entry->local_pools_len && p_entry->local_pools[entry_index]) { + local_pool_free(p_entry->local_pools[entry_index]); + p_entry->local_pools[entry_index] = NULL; + } + } + lock_release(&p_entry_manager->lock); +} + +//////////////////////////////////////////////////////////////////////////////// +// Synchronization-free-read extensible array. +//////////////////////////////////////////////////////////////////////////////// + +struct extensible_array { + // This array will be freed if all values are set to NULL. + void **values; // This array is extensible. + int lock; + int length; + void *next_unused; + // values has the following structure. + // | len (8 bytes) | values (sizeof(void *) * len bytes)| next (8 bytes) + // ^ pointed by values. + // next points to the head of buffer, not head of values. +}; + +static inline void *allocate_extensible_array_values(size_t length) { + char *values_buffer = aligned_malloc(length * sizeof(void *) + 16); + *((uint64_t *)values_buffer) = (uint64_t)length; + *((void **)(values_buffer + sizeof(void *) * length + 8)) = NULL; + return (void *)(values_buffer + 8); +} + +static void free_extensible_array(struct extensible_array *p_array, int index) { + // Take a lock and check if no one uses it. + lock_acquire(&p_array->lock); + p_array->values[index] = NULL; + int is_empty = 1; + for (int i = 0; i < p_array->length; i++) { + if (p_array->values[i] != NULL) { + is_empty = 0; + break; + } + } + if (is_empty) { + aligned_free(((char *)p_array->values) - 8); + void *next_unused = p_array->next_unused; + while (next_unused) { + void *prev_unused = next_unused; + size_t length = *((uint64_t *)prev_unused); + next_unused = *(void **)(((char *)prev_unused) + sizeof(void *) * length + 8); + aligned_free(prev_unused); + } + p_array->values = NULL; + p_array->length = 0; + p_array->next_unused = NULL; + } + lock_release(&p_array->lock); +} + +static void write_extensible_array(struct extensible_array *p_array, int index, void *value) { + assert(value != NULL); + // Take a lock and update it. + lock_acquire(&p_array->lock); + if (p_array->length <= index) { + int new_length = (p_array->length * 2 <= (index + 1)) ? (index + 1) : (p_array->length * 2); + // Create new values and replace the old one. + void **new_values = allocate_extensible_array_values(new_length); + // Initialize new_values. + fast_memcpy(new_values, p_array->values, sizeof(void *) * p_array->length); + fast_memset(new_values + p_array->length, 0, sizeof(void *) * (new_length - p_array->length)); + // NEVER free the old values (since other threads might be accessing them) + // Instead, add the old values to next_unused. + if (p_array->values) { + char *old_values = (((char *)p_array->values) - 8); + *(void **)(old_values + sizeof(void *) * (*((uint64_t*)old_values)) + 8) = p_array->next_unused; + p_array->next_unused = old_values; + } + // Update values first. + p_array->values = new_values; + // Update length at last. + zm_atomic_store(&p_array->length, new_length, zm_memord_release); + } + // Write values. + p_array->values[index] = value; + lock_release(&p_array->lock); +} + +//////////////////////////////////////////////////////////////////////////////// +// Pools +//////////////////////////////////////////////////////////////////////////////// struct element { +#ifndef USE_ARRAY_BASED_BLOCK struct element *next; // element is the first sizeof(element) bytes of an element. +#endif }; +#ifdef USE_PAGE struct memory_bulk { struct memory_bulk *next; // memory_bulk is the first sizeof(memory_bulk) bytes of a page. } __attribute__((aligned(ZM_CACHELINE_SIZE))); +#endif struct block { - int num_elements; +#ifdef USE_ARRAY_BASED_BLOCK + struct element *elements[BLOCKSIZE]; // This is part of block, so never freed. +#else struct element *head; struct element *tail; +#endif }; struct local_pool { int num_elements; - struct block *blocks; +#ifdef USE_PAGE + size_t extra_mem_size; + void *p_extra_mem_ptr; +#endif + struct block blocks[LOCALPOOL_NUM_BLOCKS]; } __attribute__((aligned(ZM_CACHELINE_SIZE))); struct global_pool { - // Assumes tids range from [0.. max_threads) + int entry_index; size_t element_size; - int max_threads; - struct local_pool **local_pools; __attribute__((aligned(ZM_CACHELINE_SIZE))) int lock; __attribute__((aligned(ZM_CACHELINE_SIZE))) int num_elements; int len_blocks; struct block *blocks; +#ifdef USE_PAGE + // This is not protected by lock; accessed by swapping. + __attribute__((aligned(ZM_CACHELINE_SIZE))) struct memory_bulk *bulk; +#endif } __attribute__((aligned(ZM_CACHELINE_SIZE))); -static inline void *element_to_ptr(struct element *element) { - return (void *)(((char *)element) + sizeof(struct element)); -} - -static inline struct element *ptr_to_element(void *ptr) { - return (struct element *)(((char *)ptr) - sizeof(struct element)); -} - -static inline void *allocate(size_t size) { - void *ptr; - if (posix_memalign(&ptr, ZM_CACHELINE_SIZE, size)) - return NULL; - return ptr; -} - -static inline int get_max_threads() { - hwloc_topology_t topo; - hwloc_topology_init(&topo); - hwloc_topology_load(topo); - int max_threads = hwloc_get_nbobjs_by_type(topo, HWLOC_OBJ_PU); - hwloc_topology_destroy(topo); - return max_threads; -} +struct global_pools { + struct extensible_array pools; +} __attribute__((aligned(ZM_CACHELINE_SIZE))); -static inline void lock_create(struct global_pool *global_pool) { - char *p_lock = (char *)&global_pool->lock; - *p_lock = 0; -} +__attribute__((aligned(ZM_CACHELINE_SIZE))) struct global_pools g_global_pools; -static inline void lock_acquire(struct global_pool *global_pool) { - char *p_lock = (char *)&global_pool->lock; - while (zm_atomic_flag_test_and_set(p_lock, zm_memord_acquire) == 1) { - while (zm_atomic_load(p_lock, zm_memord_acquire) != 0) { - ; // pause? - } +#ifdef USE_PAGE +static inline void atomic_insert_bulk(struct global_pool *p_global_pool, struct memory_bulk *bulk) { + while (1) { + struct memory_bulk *cur_bulk = p_global_pool->bulk; + bulk->next = cur_bulk; + if (zm_atomic_compare_exchange_weak(&p_global_pool->bulk, &cur_bulk, bulk, zm_memord_acq_rel, zm_memord_acquire)) + return; } } +#endif -static inline void lock_release(struct global_pool *global_pool) { - char *p_lock = (char *)&global_pool->lock; - zm_atomic_flag_clear(p_lock, zm_memord_release); +static inline void *element_to_ptr(struct element *element) { +#ifdef USE_ARRAY_BASED_BLOCK + return (void *)element; +#else + return (void *)(((char *)element) + sizeof(struct element)); +#endif } -static inline void lock_destroy(struct global_pool *global_pool) { - // Do nothing. +static inline struct element *ptr_to_element(void *ptr) { +#ifdef USE_ARRAY_BASED_BLOCK + return (struct element *)ptr; +#else + return (struct element *)(((char *)ptr) - sizeof(struct element)); +#endif } -static struct local_pool *local_pool_create(size_t element_size) { - struct local_pool *local_pool = (struct local_pool *)allocate(sizeof(struct local_pool)); - local_pool->num_elements = 0; - local_pool->blocks = (struct block *)allocate(sizeof(struct block) * LOCALPOOL_NUM_BLOCKS); - memset(local_pool->blocks, 0, sizeof(struct block) * LOCALPOOL_NUM_BLOCKS); +static struct local_pool *local_pool_create() { + struct local_pool *local_pool = (struct local_pool *)aligned_calloc(sizeof(struct local_pool)); return local_pool; } -static void local_pool_destroy(struct local_pool *local_pool) { - free(local_pool->blocks); - free(local_pool); +static void local_pool_free(struct local_pool *p_local_pool) { +#ifndef USE_PAGE + // Free all objects. + const size_t blocki_from = 0; + const size_t blocki_to = (p_local_pool->num_elements + BLOCKSIZE - 1) >> BLOCKSIZE_LOG; + for (int block_i = blocki_from; block_i < blocki_to; block_i++) { + struct element *element = p_local_pool->blocks[block_i].head; + while(element) { + struct element *next = element->next; + aligned_free(element); + element = next; + } + } +#endif + aligned_free(p_local_pool); } int zm_pool_create(size_t element_size, zm_pool_t *handle) { - struct global_pool *global_pool = (struct global_pool *)allocate(sizeof(struct global_pool)); - global_pool->element_size = element_size; - global_pool->max_threads = get_max_threads(); - global_pool->local_pools = (struct local_pool **)allocate(sizeof(struct local_pool *) * global_pool->max_threads); - for (int tid = 0; tid < global_pool->max_threads; tid++) { - global_pool->local_pools[tid] = local_pool_create(element_size); - } - lock_create(global_pool); - global_pool->num_elements = 0; - global_pool->len_blocks = 0; - global_pool->blocks = NULL; - global_pool->bulk = NULL; - *handle = (zm_pool_t)global_pool; + struct global_pool *p_global_pool = (struct global_pool *)aligned_calloc(sizeof(struct global_pool)); + const int entry_index = create_new_entry(); + p_global_pool->entry_index = entry_index; + p_global_pool->element_size = element_size; + write_extensible_array(&g_global_pools.pools, entry_index, (void *)p_global_pool); + *handle = (zm_pool_t)((intptr_t)entry_index); return 0; } int zm_pool_destroy(zm_pool_t *handle) { - struct global_pool *global_pool = *(struct global_pool **)handle; - // Free local pools. - for (int tid = 0; tid < global_pool->max_threads; tid++) { - local_pool_destroy(global_pool->local_pools[tid]); - } - free(global_pool->local_pools); + const int entry_index = (int)((intptr_t)*handle); + struct global_pool *p_global_pool = (void *)g_global_pools.pools.values[entry_index]; // Free lock. - lock_destroy(global_pool); - // Free blocks. - free(global_pool->blocks); + p_global_pool->lock = 0; +#ifdef USE_PAGE // Free bulks. - struct memory_bulk *bulk = global_pool->bulk; + struct memory_bulk *bulk = p_global_pool->bulk; while (bulk) { struct memory_bulk *next = bulk->next; free(bulk); bulk = next; } - - free(global_pool); +#else + // Free all objects. + const size_t blocki_from = 0; + const size_t blocki_to = (p_global_pool->num_elements + BLOCKSIZE - 1) >> BLOCKSIZE_LOG; + for (int block_i = blocki_from; block_i < blocki_to; block_i++) { + struct element *element = p_global_pool->blocks[block_i].head; + while(element) { + struct element *next = element->next; + aligned_free(element); + element = next; + } + } +#endif + // Free blocks. + free(p_global_pool->blocks); + // Free local pools. + free_entry(entry_index); + free_extensible_array(&g_global_pools.pools, entry_index); + free(p_global_pool); return 0; } -int zm_pool_alloc(zm_pool_t handle, int tid, void **ptr) { - struct global_pool *global_pool = (struct global_pool *)handle; - struct local_pool *local_pool = global_pool->local_pools[tid]; - if (local_pool->num_elements == 0) { - lock_acquire(global_pool); - while (global_pool->num_elements < GLOBAL_TO_LOCAL_NUM_BLOCKS * BLOCKSIZE) { - // Allocate a new bulk. - const size_t element_size = ((sizeof(struct element) + global_pool->element_size + ZM_CACHELINE_SIZE - 1) / ZM_CACHELINE_SIZE) * ZM_CACHELINE_SIZE; - size_t bulk_size = PAGESIZE; - char *mem = (char *)allocate(bulk_size); - struct memory_bulk *bulk = (struct memory_bulk *)mem; - mem += sizeof(struct memory_bulk); - bulk_size -= sizeof(struct memory_bulk); - // Add a new bulk. - bulk->next = global_pool->bulk; - global_pool->bulk = bulk; - // Create elements. - int block_i = global_pool->num_elements / BLOCKSIZE; - struct block *blocks = global_pool->blocks; - int len_blocks = global_pool->len_blocks; - int num_elements = global_pool->num_elements; - while (bulk_size >= element_size) { - struct element *new_element = (struct element *)mem; - new_element->next = NULL; - mem += element_size; - bulk_size -= element_size; - if (block_i >= len_blocks) { - // Extend blocks. - int new_len_blocks = (len_blocks == 0) ? 1 : (len_blocks * 2); - struct block *new_blocks = (struct block *)allocate(sizeof(struct block) * new_len_blocks); - memcpy(new_blocks, blocks, sizeof(struct block) * len_blocks); - memset(&new_blocks[len_blocks], 0, sizeof(struct block) * (new_len_blocks - len_blocks)); - len_blocks = new_len_blocks; - free(blocks); - blocks = new_blocks; +int zm_pool_alloc(zm_pool_t handle, void **ptr) { + const int entry_index = (int)((intptr_t)handle); + struct local_pool *local_pool = get_thread_local_pool(entry_index); + if (zm_unlikely(local_pool->num_elements == 0)) { + struct global_pool *global_pool = (struct global_pool *)g_global_pools.pools.values[entry_index]; + int num_taken_blocks = 0; + size_t element_size = 0; + // Critical section. Take blocks from a global pool. + lock_acquire(&global_pool->lock); + { + int num_global_pool_full_blocks = global_pool->num_elements >> BLOCKSIZE_LOG; + num_taken_blocks = (num_global_pool_full_blocks < GLOBAL_TO_LOCAL_NUM_BLOCKS) ? num_global_pool_full_blocks : GLOBAL_TO_LOCAL_NUM_BLOCKS; + if (num_taken_blocks != 0) { + int global_block_i = global_pool->num_elements >> BLOCKSIZE_LOG; + fast_memcpy(local_pool->blocks, &global_pool->blocks[global_block_i - num_taken_blocks], sizeof(struct block) * num_taken_blocks); + // local_pool->num_elements = BLOCKSIZE * num_taken_blocks; // Update later. + global_pool->num_elements -= BLOCKSIZE * num_taken_blocks; + if ((global_pool->num_elements & (BLOCKSIZE - 1)) != 0) { // global_pool->num_elements % BLOCKSIZE != 0 + // Copy the last block of global_pool and clear the remaining blocks. + fast_memcpy(&global_pool->blocks[global_block_i - num_taken_blocks], &global_pool->blocks[global_block_i], sizeof(struct block)); + // fast_memset(&global_pool->blocks[global_block_i - num_taken_blocks + 1], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - num_taken_blocks + 1))); + } else { + // Clear the remaining blocks. + // fast_memset(&global_pool->blocks[global_block_i - num_taken_blocks], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - num_taken_blocks))); + } + } + if (GLOBAL_TO_LOCAL_NUM_BLOCKS != num_taken_blocks) { +#ifdef USE_ARRAY_BASED_BLOCK + element_size = (global_pool->element_size + MALLOC_ALIGNMENT - 1) & (~(MALLOC_ALIGNMENT - 1)); +#else + element_size = (sizeof(struct element) + global_pool->element_size + MALLOC_ALIGNMENT - 1) & (~(MALLOC_ALIGNMENT - 1)); +#endif + } + } + lock_release(&global_pool->lock); + int num_remaining_blocks = GLOBAL_TO_LOCAL_NUM_BLOCKS - num_taken_blocks; + if (num_remaining_blocks) { + int num_remaining_elements = num_remaining_blocks << BLOCKSIZE_LOG; + assert(element_size); + // Allocate remaining blocks. +#ifdef USE_PAGE + // Use extra_mem_ptr. + int block_i = num_taken_blocks; + struct block *blocks = local_pool->blocks; + size_t extra_mem_size = local_pool->extra_mem_size; + char *p_extra_mem_ptr = (char *)local_pool->p_extra_mem_ptr; + + struct block *block = &blocks[block_i]; + int block_num_elements = 0; +#ifdef USE_ARRAY_BASED_BLOCK + struct element **block_elements = block->elements; +#else + struct element *block_head = NULL; + struct element *block_tail = NULL; +#endif + while (num_remaining_elements) { + if (!p_extra_mem_ptr) { + // Allocate new page and extract elements from it. + extra_mem_size = PAGESIZE; + p_extra_mem_ptr = (char *)aligned_malloc(extra_mem_size); + struct memory_bulk *bulk = (struct memory_bulk *)p_extra_mem_ptr; + p_extra_mem_ptr += sizeof(struct memory_bulk); + extra_mem_size -= sizeof(struct memory_bulk); + // Add a new bulk. + atomic_insert_bulk(global_pool, bulk); + } + while (num_remaining_elements && extra_mem_size >= element_size) { + struct element *new_element = (struct element *)p_extra_mem_ptr; + p_extra_mem_ptr += element_size; + extra_mem_size -= element_size; +#ifdef USE_ARRAY_BASED_BLOCK + block_elements[block_num_elements] = new_element; +#else + new_element->next = NULL; + if (block_num_elements == 0) { + block_head = new_element; + block_tail = new_element; + } else { + block_tail->next = new_element; + block_tail = new_element; + } +#endif + block_num_elements++; + if (block_num_elements == BLOCKSIZE) { +#ifndef USE_ARRAY_BASED_BLOCK + block->head = block_head; + block->tail = block_tail; +#endif + block_i++; + block = &blocks[block_i]; + block_num_elements = 0; +#ifdef USE_ARRAY_BASED_BLOCK + block_elements = block->elements; +#endif + } + num_remaining_elements--; + } + if (extra_mem_size < element_size) { + extra_mem_size = 0; + p_extra_mem_ptr = NULL; } + } + local_pool->extra_mem_size = extra_mem_size; + local_pool->p_extra_mem_ptr = p_extra_mem_ptr; + assert(block_num_elements == 0); +#else // !USE_PAGE + int block_i = num_taken_blocks; + struct block *blocks = local_pool->blocks; + struct block *block = &blocks[block_i]; + int block_num_elements = 0; +#ifdef USE_ARRAY_BASED_BLOCK + struct element **block_elements = block->elements; +#else + struct element *block_head = NULL; + struct element *block_tail = NULL; +#endif + for (int i = 0; i < num_remaining_elements; i++) { + // Allocate all single elements. + struct element *new_element = (struct element *)aligned_malloc(element_size); // Put it to a tail block. - struct block *block = &blocks[block_i]; - if (block->num_elements == 0) { - block->head = new_element; - block->tail = new_element; +#ifdef USE_ARRAY_BASED_BLOCK + block_elements[block_num_elements] = new_element; +#else + new_element->next = NULL; + if (block_num_elements == 0) { + block_head = new_element; + block_tail = new_element; } else { - block->tail->next = new_element; - block->tail = new_element; + block_tail->next = new_element; + block_tail = new_element; } - block->num_elements++; - if (block->num_elements == BLOCKSIZE) +#endif + block_num_elements++; + if (block_num_elements == BLOCKSIZE) { +#ifndef USE_ARRAY_BASED_BLOCK + block->head = block_head; + block->tail = block_tail; +#endif block_i++; - num_elements++; + block = &blocks[block_i]; + block_num_elements = 0; +#ifdef USE_ARRAY_BASED_BLOCK + block_elements = block->elements; +#endif + } } - global_pool->blocks = blocks; - global_pool->len_blocks = len_blocks; - global_pool->num_elements = num_elements; - } - // Take blocks from a global pool. - int global_block_i = global_pool->num_elements / BLOCKSIZE; - memcpy(local_pool->blocks, &global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], sizeof(struct block) * GLOBAL_TO_LOCAL_NUM_BLOCKS); - local_pool->num_elements = BLOCKSIZE * GLOBAL_TO_LOCAL_NUM_BLOCKS; - global_pool->num_elements -= BLOCKSIZE * GLOBAL_TO_LOCAL_NUM_BLOCKS; - if (global_pool->num_elements % BLOCKSIZE != 0) { - // Copy the last block of global_pool and clear the remaining blocks. - memcpy(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], &global_pool->blocks[global_block_i], sizeof(struct block)); - memset(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS + 1], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS + 1))); - } else { - // Clear the remaining blocks. - memset(&global_pool->blocks[global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS], 0, sizeof(struct block) * (global_pool->len_blocks - (global_block_i - GLOBAL_TO_LOCAL_NUM_BLOCKS))); + assert(block_num_elements == 0); +#endif // USE_PAGE } - lock_release(global_pool); + local_pool->num_elements = GLOBAL_TO_LOCAL_NUM_BLOCKS * BLOCKSIZE; // Update local_pool variable here. } { // Take an element from a local pool. - local_pool->num_elements -= 1; - int block_i = local_pool->num_elements / BLOCKSIZE; + int local_pool_num_elements = --local_pool->num_elements; + int block_i = local_pool_num_elements >> BLOCKSIZE_LOG; + int block_num_elements = local_pool_num_elements & (BLOCKSIZE - 1); struct block *block = &local_pool->blocks[block_i]; +#ifdef USE_ARRAY_BASED_BLOCK + struct element *element = block->elements[block_num_elements]; +#else struct element *element = block->head; struct element *next = element->next; block->head = next; - block->num_elements -= 1; +#endif *ptr = element_to_ptr(element); } return 0; } -int zm_pool_free(zm_pool_t handle, int tid, void *ptr) { - struct global_pool *global_pool = (struct global_pool *)handle; - struct local_pool *local_pool = global_pool->local_pools[tid]; - if (local_pool->num_elements == LOCALPOOL_NUM_BLOCKS * BLOCKSIZE) { - lock_acquire(global_pool); - int len_required_blocks = (global_pool->num_elements + BLOCKSIZE - 1 + LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE) / BLOCKSIZE; +int zm_pool_free(zm_pool_t handle, void *ptr) { + const int entry_index = (int)((intptr_t)handle); + struct local_pool *local_pool = get_thread_local_pool(entry_index); + if (zm_unlikely(local_pool->num_elements == LOCALPOOL_NUM_BLOCKS * BLOCKSIZE)) { + struct global_pool *global_pool = (struct global_pool *)g_global_pools.pools.values[entry_index]; + lock_acquire(&global_pool->lock); + int len_required_blocks = (global_pool->num_elements + BLOCKSIZE - 1 + LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE) >> BLOCKSIZE_LOG; struct block* blocks = global_pool->blocks; if (global_pool->len_blocks < len_required_blocks) { // Extend blocks. const int len_blocks = global_pool->len_blocks; int new_len_blocks = (len_blocks * 2 < len_required_blocks) ? len_required_blocks : (len_blocks * 2); - struct block *new_blocks = (struct block *)allocate(sizeof(struct block) * new_len_blocks); - memcpy(new_blocks, blocks, sizeof(struct block) * len_blocks); - memset(&new_blocks[len_blocks], 0, sizeof(struct block) * (new_len_blocks - len_blocks)); + struct block *new_blocks = (struct block *)aligned_realloc(blocks, sizeof(struct block) * len_blocks, sizeof(struct block) * new_len_blocks); + // fast_memset(&new_blocks[len_blocks], 0, sizeof(struct block) * (new_len_blocks - len_blocks)); global_pool->len_blocks = new_len_blocks; - free(blocks); blocks = new_blocks; global_pool->blocks = new_blocks; } // Return blocks to a global pool. - int block_i = global_pool->num_elements / BLOCKSIZE; - if (global_pool->num_elements % BLOCKSIZE != 0) { - // Copy the last block of a global pool. - memcpy(&blocks[block_i + LOCAL_TO_GLOBAL_NUM_BLOCKS], &blocks[block_i], sizeof(struct block)); + { + int block_i = global_pool->num_elements >> BLOCKSIZE_LOG; + if ((global_pool->num_elements & (BLOCKSIZE - 1)) != 0) { // global_pool->num_elements % BLOCKSIZE != 0 + // Copy the last block of a global pool. + fast_memcpy(&blocks[block_i + LOCAL_TO_GLOBAL_NUM_BLOCKS], &blocks[block_i], sizeof(struct block)); + } + fast_memcpy(&blocks[block_i], &local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); } - memcpy(&blocks[block_i], &local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); - memset(&local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], 0, sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); global_pool->num_elements += LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE; +#ifndef USE_PAGE + // Free objects if # of blocks is more than GLOBALPOOL_NUM_BLOCKS. + if (global_pool->num_elements > GLOBALPOOL_NUM_BLOCKS * BLOCKSIZE) { + const size_t blocki_from = GLOBALPOOL_NUM_BLOCKS; + const size_t blocki_to = (global_pool->num_elements + BLOCKSIZE - 1) >> BLOCKSIZE_LOG; + for (int block_i = blocki_from; block_i < blocki_to; block_i++) { + struct element *element = blocks[block_i].head; + while(element) { + struct element *next = element->next; + aligned_free(element); + element = next; + } + } + global_pool->num_elements = GLOBALPOOL_NUM_BLOCKS * BLOCKSIZE; + } +#endif + lock_release(&global_pool->lock); + // Local work which can be done outside the critical section. + // fast_memset(&local_pool->blocks[LOCALPOOL_NUM_BLOCKS - LOCAL_TO_GLOBAL_NUM_BLOCKS], 0, sizeof(struct block) * LOCAL_TO_GLOBAL_NUM_BLOCKS); local_pool->num_elements -= LOCAL_TO_GLOBAL_NUM_BLOCKS * BLOCKSIZE; - lock_release(global_pool); } { // Return an element to a local pool. - int block_i = local_pool->num_elements / BLOCKSIZE; - local_pool->num_elements += 1; + int block_i = local_pool->num_elements >> BLOCKSIZE_LOG; + int local_pool_num_elements = local_pool->num_elements++; + int block_num_elements = local_pool_num_elements & (BLOCKSIZE - 1); struct block *block = &local_pool->blocks[block_i]; struct element *element = ptr_to_element(ptr); - if (block->num_elements == 0) { +#ifdef USE_ARRAY_BASED_BLOCK + block->elements[block_num_elements] = element; +#else + if (zm_unlikely(block_num_elements == 0)) { block->head = element; block->tail = element; element->next = NULL; @@ -281,7 +793,7 @@ int zm_pool_free(zm_pool_t handle, int tid, void *ptr) { element->next = block->head; block->head = element; } - block->num_elements += 1; +#endif } return 0; } diff --git a/src/queue/zm_swpqueue.c b/src/queue/zm_swpqueue.c index 07174cb..5e0389c 100644 --- a/src/queue/zm_swpqueue.c +++ b/src/queue/zm_swpqueue.c @@ -54,11 +54,11 @@ int zm_swpqueue_isempty_strong(zm_swpqueue_t* q) { return ((head->next == ZM_NULL) && (head == tail)); } -int zm_swpqueue_init_explicit(zm_swpqueue_t *q, int tid) { +int zm_swpqueue_init_explicit(zm_swpqueue_t *q) { zm_pool_create(sizeof(zm_swpqnode_t), &q->pool); zm_swpqnode_t* node; - zm_pool_alloc(q->pool, tid, (void**)&node); + zm_pool_alloc(q->pool, (void**)&node); node->data = NULL; node->next = ZM_NULL; zm_atomic_store(&q->head, (zm_ptr_t)node, zm_memord_release); @@ -66,10 +66,10 @@ int zm_swpqueue_init_explicit(zm_swpqueue_t *q, int tid) { return 0; } -int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data, int tid) { +int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data) { zm_swpqnode_t* pred; zm_swpqnode_t* node; - zm_pool_alloc(q->pool, tid, (void**)&node); + zm_pool_alloc(q->pool, (void**)&node); node->data = data; zm_atomic_store(&node->next, ZM_NULL, zm_memord_release); pred = (zm_swpqnode_t*)zm_atomic_exchange(&q->tail, (zm_ptr_t)node, zm_memord_acq_rel); @@ -77,7 +77,7 @@ int zm_swpqueue_enqueue_explicit(zm_swpqueue_t* q, void *data, int tid) { return 0; } -int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data, int tid) { +int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data) { zm_swpqnode_t* head; zm_ptr_t next; *data = NULL; @@ -89,7 +89,7 @@ int zm_swpqueue_dequeue_explicit(zm_swpqueue_t* q, void **data, int tid) { next = (zm_ptr_t) zm_atomic_load(&head->next, zm_memord_acquire); zm_atomic_store(&q->head, next, zm_memord_release); *data = ((zm_swpqnode_t*)next)->data; - zm_pool_free(q->pool, tid, (void*)head); + zm_pool_free(q->pool, (void*)head); } return 1; } diff --git a/test/perf/queue/thscale_swp.c b/test/perf/queue/thscale_swp.c index cdf857e..73a999a 100644 --- a/test/perf/queue/thscale_swp.c +++ b/test/perf/queue/thscale_swp.c @@ -31,7 +31,7 @@ static inline void run() { int nthreads; for (nthreads = 2; nthreads <= omp_get_max_threads(); nthreads *= 2) { - zm_swpqueue_init_explicit(&queue, omp_get_thread_num()); + zm_swpqueue_init_explicit(&queue); int nelem_enq, nelem_deq; nelem_enq = TEST_NELEMTS/(nthreads-1); @@ -50,18 +50,18 @@ static inline void run() { for(int i = 0; i #include -#define TEST_NTHREADS 3 +#define TEST_NTHREADS 8 +#define TEST_NTHREADS_PARENT 8 #define TEST_NELEMTS 20000 #define NUM_OPERATIONS 30000 @@ -43,8 +44,8 @@ static void* func(void *arg) { continue; for (int i = 0; i < num_allocs; i++) { void *ptr = NULL; - zm_pool_alloc(pool, tid, &ptr); - memset(ptr, id, element_size); + zm_pool_alloc(pool, &ptr); + __builtin_memset(ptr, id, element_size); ptrs[num_elements++] = ptr; } } else { @@ -64,7 +65,7 @@ static void* func(void *arg) { abort(); } } - zm_pool_free(pool, tid, ptr); + zm_pool_free(pool, ptr); } } @@ -80,7 +81,7 @@ static void* func(void *arg) { abort(); } } - zm_pool_free(pool, tid, ptr); + zm_pool_free(pool, ptr); } free(ptrs); @@ -97,16 +98,18 @@ static void* func(void *arg) { * Failure: 1 *------------------------------------------------------------------------- */ -static void run() { +static void *run(void *arg) { pthread_t threads[TEST_NTHREADS]; thread_data_t data[TEST_NTHREADS]; - const size_t element_sizes[] = {64, 128, 256, 1024}; + // const size_t element_sizes[] = {1, 64, 256, 1024}; + const size_t element_sizes[] = {1, 2, 4, 8, 16 ,7 ,5, 3, 2, 1, 64, 128, 256, 1024, 1, 3, 5, 6, 8}; + // const size_t element_sizes[] = {1, 2}; for (int i = 0; i < sizeof(element_sizes) / sizeof(const size_t); i++) { zm_pool_t pool; zm_pool_create(element_sizes[i], &pool); - + printf("trying %d\n", (int) element_sizes[i]); for (int tid = 0; tid < TEST_NTHREADS; tid++) { data[tid].tid = tid; data[tid].element_size = element_sizes[i]; @@ -120,8 +123,14 @@ static void run() { zm_pool_destroy(&pool); } + return NULL; } int main(int argc, char **argv) { - run(); + pthread_t threads[TEST_NTHREADS_PARENT]; + for (int tid = 1; tid < TEST_NTHREADS_PARENT; tid++) + pthread_create(&threads[tid], NULL, run, NULL); + run(NULL); + for (int tid = 1; tid < TEST_NTHREADS_PARENT; tid++) + pthread_join(threads[tid], NULL); }