/*
*
* Thread-safe Skiplist Using Integer Identifiers
* Copyright 1998-2000 Scott Shambarger (scott@shambarger.net)
*
* This software is open source. Permission to use, copy, modify, and
* distribute this software for any purpose and without fee is hereby granted,
* provided that the above copyright notice appear in all copies. No
* warranty of any kind is expressed or implied. Use at your own risk.
*
* 1/14/2001 blong
* Made it use neo errs... probably need to check locking functions
* for error returns...
*
*/
#include "cs_config.h"
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include "neo_misc.h"
#include "neo_err.h"
#include "skiplist.h"
#include "ulocks.h"
typedef struct skipItem *skipItem;
/* structure is sized on allocation based on its level */
struct skipItem {
UINT32 locks; /* count of locks on value */
UINT32 key; /* item's key */
void *value; /* item's value */
INT32 level; /* item level */
skipItem next[1]; /* array of next items */
};
#define SIZEOFITEM(max) (sizeof(struct skipItem) + \
((max+1) * sizeof(skipItem)))
struct skipList_struct {
INT32 topLevel; /* current max level in any item */
INT32 levelHint; /* hint at level to start search */
skipItem header; /* header item (has all levels) */
skipItem tail; /* tail item (has all levels) */
/* elements to handle cached deleted items */
skipItem deleted; /* cached deleted items (linked by level+1 next entries) */
UINT32 cached; /* number of cached deleted items */
int flushing; /* TRUE if thread waiting to flush cached items */
UINT32 readers; /* number of current readers */
int block; /* TRUE if readers should wait */
pthread_mutex_t read; /* readers count/cond wait mutex */
pthread_mutex_t write; /* writer mutex */
pthread_cond_t resume; /* condition to wait on to resume reads */
pthread_cond_t flush; /* condition to wait on for flush */
/* list constants */
int threaded; /* TRUE if list needs to be thread safe */
UINT32 flushLimit; /* max number of cached deleted items before flush */
INT32 maxLevel; /* max level list can reach */
double randLimit; /* min random value to jump levels */
skipFreeValue freeValue; /* free value callback */
void *freeValueCtx; /* context to pass to <freeValue> callback */
};
static void readLock(skipList list) {
mLock(&list->read);
if(list->block)
cWait(&list->resume, &list->read);
list->readers++;
mUnlock(&list->read);
return;
}
static void readUnlock(skipList list, skipItem x, void **plock) {
int startFlush = FALSE;
if(list->threaded)
mLock(&list->read);
if(plock) {
x->locks++;
*plock = x;
}
if(! list->threaded)
return;
list->readers--;
if((list->readers == 0) && list->block)
startFlush = TRUE;
mUnlock(&list->read);
if(startFlush)
cSignal(&list->flush);
return;
}
static void readBlock(skipList list) {
mLock(&list->read);
list->block = TRUE;
if(list->readers)
cWait(&list->flush, &list->read); /* wait until reader locks released */
return;
}
static void readUnblock(skipList list) {
list->block = FALSE;
mUnlock(&list->read);
cBroadcast(&list->resume);
return;
}
static void writeLock(skipList list) {
mLock(&list->write);
return;
}
static void writeUnlock(skipList list) {
mUnlock(&list->write);
return;
}
static NEOERR *skipAllocItem(skipItem *item, UINT32 level, UINT32 key,
void *value)
{
if(! (*item = malloc(SIZEOFITEM(level))))
return nerr_raise(NERR_NOMEM, "Unable to allocate space for skipItem");
/* init new item */
(*item)->locks = 0;
(*item)->key = key;
(*item)->value = value;
(*item)->level = level;
return STATUS_OK;
}
static void skipFreeItem(skipList list, skipItem item) {
if(list->freeValue)
list->freeValue(item->value, list->freeValueCtx); /* free value */
free(item); /* release item */
return;
}
static void skipFlushDeleted(skipList list, int force) {
skipItem x, y, next;
x = list->deleted;
y = x->next[x->level + 1];
while(y != list->tail) {
next = y->next[y->level + 1];
if(force || (! y->locks)) { /* check if value currently locked */
x->next[x->level + 1] = next; /* set previous item's next link */
skipFreeItem(list, y); /* free item */
list->cached--; /* update cached count */
}
else {
x = y; /* make this item the previous item */
}
y = next; /* advance to next item */
}
return;
}
static void skipWriteUnlock(skipList list) {
int flush;
if(! list->threaded)
return;
if((list->cached > list->flushLimit) && (! list->flushing)) {
list->flushing = TRUE;
flush = TRUE;
}
else {
flush = FALSE;
}
writeUnlock(list); /* let any pending writes complete */
readUnlock(list, NULL, NULL); /* no longer reading */
if(flush) {
/* we are now flushing deleted items */
readBlock(list); /* acquire all read locks */
/* at this point no readers/writers are active */
skipFlushDeleted(list, FALSE); /* flush deleted items */
list->flushing = FALSE; /* done flushing */
readUnblock(list); /* let everyone continue */
}
return;
}
static skipItem skipFind(skipList list, UINT32 key) {
skipItem x, y = NULL;
INT32 i;
if(list->threaded)
readLock(list);
x = list->header; /* header contains all levels */
for(i = list->levelHint; /* loop from levelHint level down to level 0 */
i >= 0;
i--) {
y = x->next[i]; /* get next item at new level */
while(y->key < key) { /* if y has a smaller key, try the next item */
x = y; /* save x in case we overshoot */
y = x->next[i]; /* get next item */
}
}
return y;
}
void *skipSearch(skipList list, UINT32 key, void **plock) {
skipItem y;
void *value;
y = skipFind(list, key); /* find item */
if(y->key == key) { /* y has our key, or it isn't here */
value = y->value;
}
else { /* didn't find item, don't allow locks */
value = NULL;
plock = NULL;
}
readUnlock(list, y, plock);
return value;
}
void *skipNext(skipList list, UINT32 *pkey, void **plock) {
skipItem y;
void *value;
y = skipFind(list, *pkey); /* find item */
if((y->key == *pkey) && (y != list->tail)) /* skip to next if found y */
y = y->next[0];
if(y != list->tail) { /* reset key to next, return value */
*pkey = y->key;
value = y->value;
}
else { /* no next item, don't allow locks */
value = NULL;
plock = NULL;
}
readUnlock(list, y, plock);
return value;
}
void skipRelease(skipList list, void *lock) {
skipItem x;
mLock(&list->read);
x = lock;
x->locks--;
mUnlock(&list->read);
return;
}
/* list is write locked */
static NEOERR *skipNewItem(skipList list, skipItem *item, UINT32 key,
void *value)
{
INT32 level = 0;
while((drand48() < list->randLimit) && (level < list->maxLevel))
level++;
if(level > list->topLevel) {
if(list->topLevel < list->maxLevel)
list->topLevel++;
level = list->topLevel;
}
return skipAllocItem(item, level, key, value);
}
/* list is write locked */
static void skipDeleteItem(skipList list, skipItem item) {
if(list->threaded) {
item->next[item->level + 1] = list->deleted->next[1];
list->cached++;
list->deleted->next[1] = item;
}
else {
skipFreeItem(list, item);
}
return;
}
NEOERR *skipNewList(skipList *skip, int threaded, int root, int maxLevel,
int flushLimit, skipFreeValue freeValue, void *ctx)
{
NEOERR *err;
skipList list;
UINT32 i;
*skip = NULL;
if(! (list = calloc(1, sizeof(struct skipList_struct))))
return nerr_raise(NERR_NOMEM, "Unable to allocate memore for skiplist");
if (maxLevel == 0)
return nerr_raise(NERR_ASSERT, "maxLevel must be greater than 0");
if(maxLevel >= SKIP_MAXLEVEL) /* check limits */
maxLevel = SKIP_MAXLEVEL-1;
if(root > 4)
root = 4;
else if(root < 2)
root = 2;
list->maxLevel = maxLevel; /* init list constants */
list->randLimit = 1.0 / (double)root;
list->threaded = threaded;
list->freeValue = freeValue;
list->freeValueCtx = ctx;
do {
if(threaded) {
list->flushLimit = flushLimit;
err = mCreate(&list->read);
if (err != STATUS_OK) break;
err = mCreate(&list->write);
if (err != STATUS_OK) break;
err = cCreate(&list->resume);
if (err != STATUS_OK) break;
err = cCreate(&list->flush);
if (err != STATUS_OK) break;
}
err = skipAllocItem(&(list->header), list->maxLevel, 0, NULL);
if (err != STATUS_OK) break;
err = skipAllocItem(&(list->tail), list->maxLevel, (UINT32)-1, NULL);
if (err != STATUS_OK) break;
err = skipAllocItem(&(list->deleted), 0, 0, NULL);
if (err != STATUS_OK) break;
for(i = 0; /* init header and tail */
i <= list->maxLevel;
i++) {
list->tail->next[i] = NULL;
list->header->next[i] = list->tail;
}
list->deleted->next[1] = list->tail;
*skip = list;
return STATUS_OK; /* return new list */
} while(FALSE);
if(list->header) /* failed to make list, bail */
free(list->header);
free(list);
return nerr_pass(err);
}
/* list considered locked */
static void skipFreeAllItems(skipList list) {
UINT32 i;
skipItem x, y;
x = list->header->next[0];
while(x != list->tail) {
y = x->next[0]; /* get next item from level 0 pointer */
skipFreeItem(list, x); /* release item */
x = y;
}
/* clear header pointers */
for(i = 0;
i <= list->maxLevel;
i++)
list->header->next[i] = list->tail;
return;
}
void skipFreeList(skipList list) {
skipFlushDeleted(list, TRUE); /* flush deleted items */
skipFreeAllItems(list); /* free list items */
if(list->threaded) {
cDestroy(&list->flush);
cDestroy(&list->resume);
mDestroy(&list->write);
mDestroy(&list->read);
}
free(list->tail); /* free list */
free(list->header);
free(list->deleted);
free(list);
return;
}
/* <list> is locked, <x> is at least level <level>, and <x>->key < <key> */
static skipItem skipClosest(skipItem x, UINT32 key, UINT32 level) {
skipItem y;
y = x->next[level]; /* get next item at this level */
while(y->key < key) { /* ensure that we have the item before the key */
x = y;
y = x->next[level];
}
return x;
}
static skipItem skipLock(skipList list, UINT32 key, skipItem *save, INT32 top) {
INT32 i;
skipItem x, y;
if(list->threaded)
readLock(list);
x = list->header; /* header contains all levels */
for(i = top; /* loop from top level down to level 0 */
i >= 0;
i--) {
y = x->next[i]; /* get next item at this level */
while(y->key < key) { /* if y has a smaller key, try the next item */
x = y; /* save x in case we overshoot */
y = x->next[i]; /* get next item */
}
save[i] = x; /* preserve item with next pointer in save */
}
if(list->threaded)
writeLock(list); /* lock list for update */
/* validate we have the closest previous item */
return skipClosest(x, key, 0);
}
NEOERR *skipInsert(skipList list, UINT32 key, void *value, int allowUpdate)
{
NEOERR *err;
INT32 i, level;
skipItem save[SKIP_MAXLEVEL];
skipItem x, y;
if (value == 0)
return nerr_raise(NERR_ASSERT, "value must be non-zero");
if (key == 0 || key == (UINT32)-1)
return nerr_raise(NERR_ASSERT, "key must not be 0 or -1");
level = list->levelHint;
x = skipLock(list, key, save, level); /* quick search for key */
y = x->next[0];
if(y->key == key) {
if(!allowUpdate)
{
skipWriteUnlock(list);
return nerr_raise(NERR_DUPLICATE, "key %u exists in skiplist", key);
}
y->value = value; /* found the key, update value */
skipWriteUnlock(list);
return STATUS_OK;
}
err = skipNewItem(list, &y, key, value);
if (err != STATUS_OK)
{
skipWriteUnlock(list);
return nerr_pass(err);
}
for(i = level + 1; /* is new item has more levels than <level> */
i <= y->level; /* if so fill in save */
i++)
save[i] = list->header;
for(i = 0; /* populate pointers for all levels */
i <= y->level;
i++) {
if(i) /* check that save is correct for each level */
x = skipClosest(save[i], key, i);
y->next[i] = x->next[i]; /* now insert the item at this level */
x->next[i] = y; /* (order here important for thread-safeness) */
}
while((list->levelHint < list->topLevel) /* update levelHint */
&& (list->header->next[list->levelHint+1] != list->tail))
list->levelHint++;
skipWriteUnlock(list);
return STATUS_OK;
}
void skipDelete(skipList list, UINT32 key) {
INT32 i, level;
skipItem save[SKIP_MAXLEVEL];
skipItem x, y;
assert(key && (key != (UINT32)-1));
level = list->levelHint;
x = skipLock(list, key, save, level); /* quick search for key */
y = x->next[0];
/* check that we found the key, and it isn't deleted */
if((y->key != key) || (y->next[0]->key < key)) {
skipWriteUnlock(list);
return;
}
for(i = level + 1; /* check if item has more levels than <level> */
i <= y->level; /* if so fill in save */
i++)
save[i] = list->header;
for(i = y->level;
i >= 0;
i--) {
/* check that save is correct for each level */
x = skipClosest(save[i], key, i);
x->next[i] = y->next[i]; /* now remove item at this level */
y->next[i] = x; /* (order here is imported for thread-safeness) */
}
skipDeleteItem(list, y); /* put on deleted list */
while((list->levelHint > 0) /* update levelHint */
&& (list->header->next[list->levelHint] == list->tail))
list->levelHint--;
skipWriteUnlock(list);
return;
}