summaryrefslogtreecommitdiff
path: root/src/lib/ssm
diff options
context:
space:
mode:
Diffstat (limited to 'src/lib/ssm')
-rw-r--r--src/lib/ssm/flow_set.c42
-rw-r--r--src/lib/ssm/pool.c122
-rw-r--r--src/lib/ssm/rbuff.c60
-rw-r--r--src/lib/ssm/ssm.h.in21
-rw-r--r--src/lib/ssm/tests/pool_sharding_test.c69
-rw-r--r--src/lib/ssm/tests/pool_test.c12
6 files changed, 177 insertions, 149 deletions
diff --git a/src/lib/ssm/flow_set.c b/src/lib/ssm/flow_set.c
index 73d0db55..2e33b408 100644
--- a/src/lib/ssm/flow_set.c
+++ b/src/lib/ssm/flow_set.c
@@ -58,9 +58,9 @@
#define QUEUESIZE ((SSM_RBUFF_SIZE) * sizeof(struct flowevent))
#define SSM_FSET_FILE_SIZE (SYS_MAX_FLOWS * sizeof(ssize_t) \
- + PROG_MAX_FQUEUES * sizeof(size_t) \
- + PROG_MAX_FQUEUES * sizeof(pthread_cond_t) \
- + PROG_MAX_FQUEUES * QUEUESIZE \
+ + PROC_MAX_FQUEUES * sizeof(size_t) \
+ + PROC_MAX_FQUEUES * sizeof(pthread_cond_t) \
+ + PROC_MAX_FQUEUES * QUEUESIZE \
+ sizeof(pthread_mutex_t))
#define fqueue_ptr(fs, idx) (fs->fqueues + (SSM_RBUFF_SIZE) * idx)
@@ -104,10 +104,10 @@ static struct ssm_flow_set * flow_set_create(pid_t pid,
set->mtable = shm_base;
set->heads = (size_t *) (set->mtable + SYS_MAX_FLOWS);
- set->conds = (pthread_cond_t *)(set->heads + PROG_MAX_FQUEUES);
- set->fqueues = (struct flowevent *) (set->conds + PROG_MAX_FQUEUES);
+ set->conds = (pthread_cond_t *)(set->heads + PROC_MAX_FQUEUES);
+ set->fqueues = (struct flowevent *) (set->conds + PROC_MAX_FQUEUES);
set->lock = (pthread_mutex_t *)
- (set->fqueues + PROG_MAX_FQUEUES * (SSM_RBUFF_SIZE));
+ (set->fqueues + PROC_MAX_FQUEUES * (SSM_RBUFF_SIZE));
return set;
@@ -164,7 +164,7 @@ struct ssm_flow_set * ssm_flow_set_create(pid_t pid)
if (pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK))
goto fail_condattr_set;
#endif
- for (i = 0; i < PROG_MAX_FQUEUES; ++i) {
+ for (i = 0; i < PROC_MAX_FQUEUES; ++i) {
set->heads[i] = 0;
if (pthread_cond_init(&set->conds[i], &cattr))
goto fail_init;
@@ -222,7 +222,7 @@ void ssm_flow_set_zero(struct ssm_flow_set * set,
ssize_t i = 0;
assert(set);
- assert(idx < PROG_MAX_FQUEUES);
+ assert(idx < PROC_MAX_FQUEUES);
pthread_mutex_lock(set->lock);
@@ -242,7 +242,7 @@ int ssm_flow_set_add(struct ssm_flow_set * set,
{
assert(set);
assert(!(flow_id < 0) && flow_id < SYS_MAX_FLOWS);
- assert(idx < PROG_MAX_FQUEUES);
+ assert(idx < PROC_MAX_FQUEUES);
pthread_mutex_lock(set->lock);
@@ -264,7 +264,7 @@ void ssm_flow_set_del(struct ssm_flow_set * set,
{
assert(set);
assert(!(flow_id < 0) && flow_id < SYS_MAX_FLOWS);
- assert(idx < PROG_MAX_FQUEUES);
+ assert(idx < PROC_MAX_FQUEUES);
pthread_mutex_lock(set->lock);
@@ -282,7 +282,7 @@ int ssm_flow_set_has(struct ssm_flow_set * set,
assert(set);
assert(!(flow_id < 0) && flow_id < SYS_MAX_FLOWS);
- assert(idx < PROG_MAX_FQUEUES);
+ assert(idx < PROC_MAX_FQUEUES);
pthread_mutex_lock(set->lock);
@@ -299,26 +299,34 @@ void ssm_flow_set_notify(struct ssm_flow_set * set,
int event)
{
struct flowevent * e;
+ ssize_t idx;
assert(set);
assert(!(flow_id < 0) && flow_id < SYS_MAX_FLOWS);
pthread_mutex_lock(set->lock);
- if (set->mtable[flow_id] == -1) {
+ idx = set->mtable[flow_id];
+ if (idx == -1) {
pthread_mutex_unlock(set->lock);
return;
}
- e = fqueue_ptr(set, set->mtable[flow_id]) +
- set->heads[set->mtable[flow_id]];
+ /* Ring full: drop redundant FLOW_PKT, reserve a slot for ctrl. */
+ if (set->heads[idx] >= SSM_RBUFF_SIZE
+ || (event == FLOW_PKT && set->heads[idx] >= SSM_RBUFF_SIZE - 1)) {
+ pthread_mutex_unlock(set->lock);
+ return;
+ }
+
+ e = fqueue_ptr(set, idx) + set->heads[idx];
e->flow_id = flow_id;
e->event = event;
- ++set->heads[set->mtable[flow_id]];
+ ++set->heads[idx];
- pthread_cond_signal(&set->conds[set->mtable[flow_id]]);
+ pthread_cond_signal(&set->conds[idx]);
pthread_mutex_unlock(set->lock);
}
@@ -332,7 +340,7 @@ ssize_t ssm_flow_set_wait(const struct ssm_flow_set * set,
ssize_t ret = 0;
assert(set);
- assert(idx < PROG_MAX_FQUEUES);
+ assert(idx < PROC_MAX_FQUEUES);
assert(fqueue);
#ifndef HAVE_ROBUST_MUTEX
diff --git a/src/lib/ssm/pool.c b/src/lib/ssm/pool.c
index 5c98b515..705de147 100644
--- a/src/lib/ssm/pool.c
+++ b/src/lib/ssm/pool.c
@@ -24,6 +24,7 @@
#include "config.h"
+#include <ouroboros/atomics.h>
#include <ouroboros/errno.h>
#include <ouroboros/pthread.h>
#include <ouroboros/ssm_pool.h>
@@ -37,10 +38,20 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
+#include <time.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/stat.h>
+static __inline__ uint64_t pool_now_ns(void)
+{
+ struct timespec ts;
+
+ clock_gettime(CLOCK_MONOTONIC, &ts);
+
+ return (uint64_t) ts.tv_sec * 1000000000ULL + (uint64_t) ts.tv_nsec;
+}
+
/* Global Shared Packet Pool (GSPP) configuration */
static const struct ssm_size_class_cfg ssm_gspp_cfg[SSM_POOL_MAX_CLASSES] = {
{ (1 << 8), SSM_GSPP_256_BLOCKS },
@@ -75,26 +86,6 @@ static const struct ssm_size_class_cfg ssm_pup_cfg[SSM_POOL_MAX_CLASSES] = {
#define GET_SHARD_FOR_PID(pid) ((int)((pid) % SSM_POOL_SHARDS))
-#define LOAD_RELAXED(ptr) \
- (__atomic_load_n(ptr, __ATOMIC_RELAXED))
-
-#define LOAD_ACQUIRE(ptr) \
- (__atomic_load_n(ptr, __ATOMIC_ACQUIRE))
-
-#define STORE_RELEASE(ptr, val) \
- (__atomic_store_n(ptr, val, __ATOMIC_RELEASE))
-
-#define LOAD(ptr) \
- (__atomic_load_n(ptr, __ATOMIC_SEQ_CST))
-
-#define STORE(ptr, val) \
- (__atomic_store_n(ptr, val, __ATOMIC_SEQ_CST))
-
-#define FETCH_ADD(ptr, val) \
- (__atomic_fetch_add(ptr, val, __ATOMIC_SEQ_CST))
-
-#define FETCH_SUB(ptr, val) \
- (__atomic_fetch_sub(ptr, val, __ATOMIC_SEQ_CST))
#define SSM_FILE_SIZE (SSM_POOL_TOTAL_SIZE + sizeof(struct _ssm_pool_hdr))
#define SSM_GSPP_FILE_SIZE (SSM_GSPP_TOTAL_SIZE + sizeof(struct _ssm_pool_hdr))
@@ -107,6 +98,8 @@ static const struct ssm_size_class_cfg ssm_pup_cfg[SSM_POOL_MAX_CLASSES] = {
: SSM_PUP_FILE_SIZE)
#define GET_POOL_CFG(uid) (IS_GSPP(uid) ? ssm_gspp_cfg : ssm_pup_cfg)
+#define NEEDS_CHOWN(uid, gid) ((uid) != geteuid() || (gid) != getegid())
+
struct ssm_pool {
uint8_t * shm_base; /* start of blocks */
struct _ssm_pool_hdr * hdr; /* shared memory header */
@@ -163,29 +156,6 @@ static __inline__ void list_add_head(struct _ssm_list_head * head,
STORE(&head->count, LOAD(&head->count) + 1);
}
-static __inline__ int select_size_class(struct ssm_pool * pool,
- size_t len)
-{
- size_t sz;
- int i;
-
- assert(pool != NULL);
-
- /* Total space needed: header + headspace + data + tailspace */
- sz = sizeof(struct ssm_pk_buff) + SSM_PK_BUFF_HEADSPACE + len
- + SSM_PK_BUFF_TAILSPACE;
-
- for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
- struct _ssm_size_class * sc;
-
- sc = &pool->hdr->size_classes[i];
- if (sc->object_size > 0 && sz <= sc->object_size)
- return i;
- }
-
- return -1;
-}
-
static __inline__ int find_size_class_for_offset(struct ssm_pool * pool,
size_t offset)
{
@@ -276,6 +246,7 @@ static void init_size_classes(struct ssm_pool * pool)
STORE(&blk->refcount, 0);
blk->allocator_pid = 0;
+ blk->alloc_ts = 0;
STORE(&blk->next_offset, 0);
list_add_head(&sc->shards[0].free_list, blk,
@@ -306,19 +277,31 @@ static size_t reclaim_pid_from_sc(struct _ssm_size_class * sc,
size_t i;
size_t recovered = 0;
struct ssm_pk_buff * blk;
+ uint64_t now;
+ uint64_t min_age_ns;
- region = (uint8_t *) pool_base + sc->pool_start;
+ region = (uint8_t *) pool_base + sc->pool_start;
+ now = pool_now_ns();
+ min_age_ns = (uint64_t) SSM_POOL_RECLAIM_AGE_S * 1000000000ULL;
for (i = 0; i < sc->object_count; ++i) {
blk = (struct ssm_pk_buff *)(region + i * sc->object_size);
- if (blk->allocator_pid == pid && LOAD(&blk->refcount) > 0) {
- STORE(&blk->refcount, 0);
- blk->allocator_pid = 0;
- list_add_head(&shard->free_list, blk, pool_base);
- FETCH_ADD(&shard->free_count, 1);
- recovered++;
- }
+ if (blk->allocator_pid != pid)
+ continue;
+
+ if (LOAD(&blk->refcount) == 0)
+ continue;
+
+ /* Recent: a live consumer may still hold the handoff. */
+ if (now - blk->alloc_ts < min_age_ns)
+ continue;
+
+ STORE(&blk->refcount, 0);
+ blk->allocator_pid = 0;
+ list_add_head(&shard->free_list, blk, pool_base);
+ FETCH_ADD(&shard->free_count, 1);
+ recovered++;
}
return recovered;
@@ -379,6 +362,7 @@ static __inline__ ssize_t init_block(struct ssm_pool * pool,
{
STORE(&blk->refcount, 1);
blk->allocator_pid = getpid();
+ blk->alloc_ts = pool_now_ns();
blk->size = (uint32_t) (sc->object_size -
sizeof(struct ssm_pk_buff));
blk->pk_head = SSM_PK_BUFF_HEADSPACE;
@@ -548,7 +532,7 @@ static struct ssm_pool * __pool_create(const char * name,
if (flags & O_CREAT) {
if (ftruncate(fd, (off_t) file_size) < 0)
goto fail_truncate;
- if (uid != geteuid() && fchown(fd, uid, gid) < 0)
+ if (NEEDS_CHOWN(uid, gid) && fchown(fd, uid, gid) < 0)
goto fail_truncate;
}
@@ -700,7 +684,7 @@ ssize_t ssm_pool_alloc(struct ssm_pool * pool,
assert(pool != NULL);
assert(spb != NULL);
- idx = select_size_class(pool, count);
+ idx = select_size_class(pool->hdr, count);
if (idx >= 0)
return alloc_from_sc(pool, idx, count, ptr, spb);
@@ -718,7 +702,7 @@ ssize_t ssm_pool_alloc_b(struct ssm_pool * pool,
assert(pool != NULL);
assert(spb != NULL);
- idx = select_size_class(pool, count);
+ idx = select_size_class(pool->hdr, count);
if (idx >= 0)
return alloc_from_sc_b(pool, idx, count, ptr, spb, abstime);
@@ -744,7 +728,7 @@ ssize_t ssm_pool_read(uint8_t ** dst,
}
struct ssm_pk_buff * ssm_pool_get(struct ssm_pool * pool,
- size_t off)
+ size_t off)
{
struct ssm_pk_buff * blk;
@@ -823,36 +807,36 @@ int ssm_pool_remove(struct ssm_pool * pool,
return 0;
}
-size_t ssm_pk_buff_get_idx(struct ssm_pk_buff * spb)
+size_t ssm_pk_buff_get_off(const struct ssm_pk_buff * spb)
{
assert(spb != NULL);
return spb->off;
}
-uint8_t * ssm_pk_buff_head(struct ssm_pk_buff * spb)
+uint8_t * ssm_pk_buff_head(const struct ssm_pk_buff * spb)
{
assert(spb != NULL);
- return spb->data + spb->pk_head;
+ return (uint8_t *) spb->data + spb->pk_head;
}
-uint8_t * ssm_pk_buff_tail(struct ssm_pk_buff * spb)
+uint8_t * ssm_pk_buff_tail(const struct ssm_pk_buff * spb)
{
assert(spb != NULL);
- return spb->data + spb->pk_tail;
+ return (uint8_t *) spb->data + spb->pk_tail;
}
-size_t ssm_pk_buff_len(struct ssm_pk_buff * spb)
+size_t ssm_pk_buff_len(const struct ssm_pk_buff * spb)
{
assert(spb != NULL);
return spb->pk_tail - spb->pk_head;
}
-uint8_t * ssm_pk_buff_head_alloc(struct ssm_pk_buff * spb,
- size_t size)
+uint8_t * ssm_pk_buff_push(struct ssm_pk_buff * spb,
+ size_t size)
{
assert(spb != NULL);
@@ -864,8 +848,8 @@ uint8_t * ssm_pk_buff_head_alloc(struct ssm_pk_buff * spb,
return spb->data + spb->pk_head;
}
-uint8_t * ssm_pk_buff_tail_alloc(struct ssm_pk_buff * spb,
- size_t size)
+uint8_t * ssm_pk_buff_push_tail(struct ssm_pk_buff * spb,
+ size_t size)
{
uint8_t * buf;
@@ -881,8 +865,8 @@ uint8_t * ssm_pk_buff_tail_alloc(struct ssm_pk_buff * spb,
return buf;
}
-uint8_t * ssm_pk_buff_head_release(struct ssm_pk_buff * spb,
- size_t size)
+uint8_t * ssm_pk_buff_pop(struct ssm_pk_buff * spb,
+ size_t size)
{
uint8_t * buf;
@@ -896,8 +880,8 @@ uint8_t * ssm_pk_buff_head_release(struct ssm_pk_buff * spb,
return buf;
}
-uint8_t * ssm_pk_buff_tail_release(struct ssm_pk_buff * spb,
- size_t size)
+uint8_t * ssm_pk_buff_pop_tail(struct ssm_pk_buff * spb,
+ size_t size)
{
assert(spb != NULL);
assert(!(size > spb->pk_tail - spb->pk_head));
diff --git a/src/lib/ssm/rbuff.c b/src/lib/ssm/rbuff.c
index e4558c31..c149c306 100644
--- a/src/lib/ssm/rbuff.c
+++ b/src/lib/ssm/rbuff.c
@@ -80,6 +80,7 @@ struct ssm_rbuff {
pthread_cond_t * del; /* signal when data removed */
pid_t pid; /* pid of the owner */
int flow_id; /* flow_id of the flow */
+ size_t n_users; /* in-flight users */
};
#define MM_FLAGS (PROT_READ | PROT_WRITE)
@@ -119,6 +120,7 @@ static struct ssm_rbuff * rbuff_create(pid_t pid,
rb->del = rb->add + 1;
rb->pid = pid;
rb->flow_id = flow_id;
+ rb->n_users = 0;
return rb;
@@ -228,11 +230,20 @@ void ssm_rbuff_close(struct ssm_rbuff * rb)
{
assert(rb);
+ /*
+ * Caller must set ACL_FLOWDOWN first; if a user becomes
+ * cancellable, push a cleanup that decrements n_users.
+ */
+ while (__atomic_load_n(&rb->n_users, __ATOMIC_SEQ_CST) > 0) {
+ struct timespec tic = { 0, 100000 };
+ nanosleep(&tic, NULL);
+ }
+
rbuff_destroy(rb);
}
int ssm_rbuff_write(struct ssm_rbuff * rb,
- size_t idx)
+ size_t off)
{
size_t acl;
bool was_empty;
@@ -240,6 +251,8 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
assert(rb != NULL);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
if (acl != ACL_RDWR) {
if (acl & ACL_FLOWDOWN) {
@@ -261,7 +274,7 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
was_empty = IS_EMPTY(rb);
- HEAD(rb) = (ssize_t) idx;
+ HEAD(rb) = (ssize_t) off;
ADVANCE_HEAD(rb);
if (was_empty)
@@ -269,16 +282,18 @@ int ssm_rbuff_write(struct ssm_rbuff * rb,
pthread_mutex_unlock(rb->mtx);
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return 0;
fail_mutex:
pthread_mutex_unlock(rb->mtx);
fail_acl:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
int ssm_rbuff_write_b(struct ssm_rbuff * rb,
- size_t idx,
+ size_t off,
const struct timespec * abstime)
{
size_t acl;
@@ -287,6 +302,8 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
assert(rb != NULL);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
if (acl != ACL_RDWR) {
if (acl & ACL_FLOWDOWN) {
@@ -316,7 +333,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
if (ret != -ETIMEDOUT && ret != -EFLOWDOWN) {
was_empty = IS_EMPTY(rb);
- HEAD(rb) = (ssize_t) idx;
+ HEAD(rb) = (ssize_t) off;
ADVANCE_HEAD(rb);
if (was_empty)
pthread_cond_broadcast(rb->add);
@@ -325,6 +342,7 @@ int ssm_rbuff_write_b(struct ssm_rbuff * rb,
pthread_mutex_unlock(rb->mtx);
fail_acl:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
@@ -351,11 +369,21 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb)
assert(rb != NULL);
- if (IS_EMPTY(rb))
- return check_rb_acl(rb);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
+ if (IS_EMPTY(rb)) {
+ ret = check_rb_acl(rb);
+ goto out;
+ }
robust_mutex_lock(rb->mtx);
+ if (IS_EMPTY(rb)) {
+ pthread_mutex_unlock(rb->mtx);
+ ret = check_rb_acl(rb);
+ goto out;
+ }
+
ret = TAIL(rb);
ADVANCE_TAIL(rb);
@@ -363,6 +391,8 @@ ssize_t ssm_rbuff_read(struct ssm_rbuff * rb)
pthread_mutex_unlock(rb->mtx);
+ out:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return ret;
}
@@ -374,9 +404,13 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb,
assert(rb != NULL);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
acl = __atomic_load_n(rb->acl, __ATOMIC_SEQ_CST);
- if (IS_EMPTY(rb) && (acl & ACL_FLOWDOWN))
- return -EFLOWDOWN;
+ if (IS_EMPTY(rb) && (acl & ACL_FLOWDOWN)) {
+ idx = -EFLOWDOWN;
+ goto out;
+ }
robust_mutex_lock(rb->mtx);
@@ -402,6 +436,8 @@ ssize_t ssm_rbuff_read_b(struct ssm_rbuff * rb,
assert(idx != -EAGAIN);
+ out:
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
return idx;
}
@@ -410,7 +446,11 @@ void ssm_rbuff_set_acl(struct ssm_rbuff * rb,
{
assert(rb != NULL);
+ robust_mutex_lock(rb->mtx);
__atomic_store_n(rb->acl, (size_t) flags, __ATOMIC_SEQ_CST);
+ pthread_cond_broadcast(rb->add);
+ pthread_cond_broadcast(rb->del);
+ pthread_mutex_unlock(rb->mtx);
}
uint32_t ssm_rbuff_get_acl(struct ssm_rbuff * rb)
@@ -424,6 +464,8 @@ void ssm_rbuff_fini(struct ssm_rbuff * rb)
{
assert(rb != NULL);
+ __atomic_fetch_add(&rb->n_users, 1, __ATOMIC_SEQ_CST);
+
robust_mutex_lock(rb->mtx);
pthread_cleanup_push(__cleanup_mutex_unlock, rb->mtx);
@@ -432,6 +474,8 @@ void ssm_rbuff_fini(struct ssm_rbuff * rb)
robust_wait(rb->del, rb->mtx, NULL);
pthread_cleanup_pop(true);
+
+ __atomic_fetch_sub(&rb->n_users, 1, __ATOMIC_SEQ_CST);
}
size_t ssm_rbuff_queued(struct ssm_rbuff * rb)
diff --git a/src/lib/ssm/ssm.h.in b/src/lib/ssm/ssm.h.in
index b9246c8b..57febae4 100644
--- a/src/lib/ssm/ssm.h.in
+++ b/src/lib/ssm/ssm.h.in
@@ -38,7 +38,6 @@
#define SSM_RBUFF_PREFIX "@SSM_RBUFF_PREFIX@"
#define SSM_FLOW_SET_PREFIX "@SSM_FLOW_SET_PREFIX@"
#define SSM_POOL_NAME "@SSM_POOL_NAME@"
-#define SSM_POOL_BLOCKS @SSM_POOL_BLOCKS@
#define SSM_RBUFF_SIZE @SSM_RBUFF_SIZE@
/* Packet buffer space reservation */
@@ -84,6 +83,7 @@
/* Size class configuration */
#define SSM_POOL_MAX_CLASSES 9
#define SSM_POOL_SHARDS @SSM_POOL_SHARDS@
+#define SSM_POOL_RECLAIM_AGE_S @SSM_POOL_RECLAIM_AGE_S@
/* Internal structures - exposed for testing */
#ifdef __cplusplus
@@ -126,6 +126,7 @@ struct ssm_pk_buff {
uint32_t pk_head; /* Head offset into data */
uint32_t pk_tail; /* Tail offset into data */
uint32_t off; /* Block offset in pool */
+ uint64_t alloc_ts; /* CLOCK_MONOTONIC ns at alloc */
uint8_t data[]; /* Packet data */
};
@@ -164,6 +165,24 @@ struct _ssm_pool_hdr {
struct _ssm_size_class size_classes[SSM_POOL_MAX_CLASSES];
};
+#define SSM_PK_BUFF_TOTALSPACE (SSM_PK_BUFF_HEADSPACE + SSM_PK_BUFF_TAILSPACE)
+static __inline__ int select_size_class(struct _ssm_pool_hdr * hdr,
+ size_t len)
+{
+ size_t sz;
+ int i;
+
+ sz = sizeof(struct ssm_pk_buff) + SSM_PK_BUFF_TOTALSPACE + len;
+
+ for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
+ struct _ssm_size_class * sc = &hdr->size_classes[i];
+ if (sc->object_size > 0 && sz <= sc->object_size)
+ return i;
+ }
+
+ return -1;
+}
+
#ifdef __cplusplus
}
#endif
diff --git a/src/lib/ssm/tests/pool_sharding_test.c b/src/lib/ssm/tests/pool_sharding_test.c
index c53105e3..ec464a92 100644
--- a/src/lib/ssm/tests/pool_sharding_test.c
+++ b/src/lib/ssm/tests/pool_sharding_test.c
@@ -80,19 +80,13 @@ static int test_lazy_distribution(void)
goto fail_pool;
}
- /* Find the first size class with blocks */
- sc_idx = -1;
- for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
- if (hdr->size_classes[i].object_count > 0) {
- sc_idx = i;
- break;
- }
- }
-
+ /* Inspect the class that TEST_SIZE allocations will use */
+ sc_idx = select_size_class(hdr, TEST_SIZE);
if (sc_idx < 0) {
- printf("No size classes configured.\n");
+ printf("No size class fits TEST_SIZE=%d.\n", TEST_SIZE);
for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
- printf(" Class %d: count=%zu\n", i,
+ printf(" Class %d: object_size=%zu count=%zu\n", i,
+ hdr->size_classes[i].object_size,
hdr->size_classes[i].object_count);
}
goto fail_pool;
@@ -137,7 +131,6 @@ static int test_shard_migration(void)
ssize_t off;
int shard_idx;
int sc_idx;
- int i;
TEST_START();
@@ -149,18 +142,11 @@ static int test_shard_migration(void)
hdr = get_pool_hdr(pool);
- /* Find the first size class with blocks */
- sc_idx = -1;
- for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
- if (hdr->size_classes[i].object_count > 0) {
- sc_idx = i;
- break;
- }
- }
-
+ /* Inspect the class that TEST_SIZE allocations will use */
+ sc_idx = select_size_class(hdr, TEST_SIZE);
if (sc_idx < 0) {
- printf("No size classes configured.\n");
- goto fail;
+ printf("No size class fits TEST_SIZE=%d.\n", TEST_SIZE);
+ goto fail_pool;
}
sc = &hdr->size_classes[sc_idx];
@@ -209,7 +195,6 @@ static int test_fallback_stealing(void)
size_t total_free;
size_t i;
int sc_idx;
- int c;
TEST_START();
@@ -221,18 +206,11 @@ static int test_fallback_stealing(void)
hdr = get_pool_hdr(pool);
- /* Find the first size class with blocks */
- sc_idx = -1;
- for (c = 0; c < SSM_POOL_MAX_CLASSES; c++) {
- if (hdr->size_classes[c].object_count > 0) {
- sc_idx = c;
- break;
- }
- }
-
+ /* Inspect the class that TEST_SIZE allocations will use */
+ sc_idx = select_size_class(hdr, TEST_SIZE);
if (sc_idx < 0) {
- printf("No size classes configured.\n");
- goto fail;
+ printf("No size class fits TEST_SIZE=%d.\n", TEST_SIZE);
+ goto fail_pool;
}
sc = &hdr->size_classes[sc_idx];
@@ -261,7 +239,7 @@ static int test_fallback_stealing(void)
/* Free them all - they go to local_shard */
for (i = 0; i < total_blocks / 2; i++) {
- size_t off = ssm_pk_buff_get_idx(spbs[i]);
+ size_t off = ssm_pk_buff_get_off(spbs[i]);
if (ssm_pool_remove(pool, off) != 0) {
printf("Remove %zu failed.\n", i);
free(spbs);
@@ -299,7 +277,7 @@ static int test_fallback_stealing(void)
/* Now all allocated blocks are in use again */
/* Cleanup - free all allocated blocks */
for (i = 0; i < total_blocks / 2; i++) {
- size_t off = ssm_pk_buff_get_idx(spbs[i]);
+ size_t off = ssm_pk_buff_get_off(spbs[i]);
ssm_pool_remove(pool, off);
}
@@ -396,20 +374,15 @@ static int test_multiprocess_sharding(void)
/* Verify blocks distributed across shards */
hdr = get_pool_hdr(pool);
- /* Find the first size class with blocks */
- sc = NULL;
- for (i = 0; i < SSM_POOL_MAX_CLASSES; i++) {
- if (hdr->size_classes[i].object_count > 0) {
- sc = &hdr->size_classes[i];
- break;
- }
- }
-
- if (sc == NULL) {
- printf("No size classes configured.\n");
+ /* Inspect the class that TEST_SIZE allocations used */
+ i = select_size_class(hdr, TEST_SIZE);
+ if (i < 0) {
+ printf("No size class fits TEST_SIZE=%d.\n", TEST_SIZE);
goto fail_pool;
}
+ sc = &hdr->size_classes[i];
+
/* After children allocate and free, blocks should be in shards
* (though exact distribution depends on PID values)
*/
diff --git a/src/lib/ssm/tests/pool_test.c b/src/lib/ssm/tests/pool_test.c
index 3fc19cd5..0f9db24d 100644
--- a/src/lib/ssm/tests/pool_test.c
+++ b/src/lib/ssm/tests/pool_test.c
@@ -741,14 +741,14 @@ static int test_ssm_pk_buff_operations(void)
memcpy(head, data, dlen);
- tail = ssm_pk_buff_tail_alloc(spb, 32);
+ tail = ssm_pk_buff_push_tail(spb, 32);
if (tail == NULL) {
- printf("Tail_alloc failed.\n");
+ printf("push_tail failed.\n");
goto fail_ops;
}
if (ssm_pk_buff_len(spb) != POOL_256 + 32) {
- printf("Length after tail_alloc: %zu.\n",
+ printf("Length after push_tail: %zu.\n",
ssm_pk_buff_len(spb));
goto fail_ops;
}
@@ -758,14 +758,14 @@ static int test_ssm_pk_buff_operations(void)
goto fail_ops;
}
- tail = ssm_pk_buff_tail_release(spb, 32);
+ tail = ssm_pk_buff_pop_tail(spb, 32);
if (tail == NULL) {
- printf("Tail_release failed.\n");
+ printf("pop_tail failed.\n");
goto fail_ops;
}
if (ssm_pk_buff_len(spb) != POOL_256) {
- printf("Length after tail_release: %zu.\n",
+ printf("Length after pop_tail: %zu.\n",
ssm_pk_buff_len(spb));
goto fail_ops;
}
generated by cgit v1.2.3 (git 2.39.1) at 2026-06-26 06:44:15 +0000