ck_bag: Various Bug Fixes and Cleanup.

Migrate available block list to CK_LIST.
New blocks are only allocated when the available list is exhausted.
Remove bag->avail_tail.
Print out number of writer iterations for unit test.
Lengthen duration of unit test.

include/ck_bag.h

@@ -32,6 +32,7 @@
 #include <ck_pr.h>
 #include <ck_malloc.h>
 #include <ck_stdint.h>
+#include <ck_queue.h>
 #include <stdbool.h>
 #include <stddef.h>
 
@@ -84,15 +85,13 @@ struct ck_bag_block_md {
 
 struct ck_bag_block {
 	struct ck_bag_block_md next;
-	struct ck_bag_block *avail_next;
+	CK_LIST_ENTRY(ck_bag_block) avail_entry;
-	struct ck_bag_block *avail_prev;
 	void *array[];
 } CK_CC_CACHELINE;
 
 struct ck_bag {
 	struct ck_bag_block *head;
-	struct ck_bag_block *avail_head;
+	CK_LIST_HEAD(avail_list, ck_bag_block) avail_blocks;
-	struct ck_bag_block *avail_tail;
 	unsigned int n_entries;
 	unsigned int n_blocks;
 	enum ck_bag_allocation_strategy alloc_strat;

regressions/ck_bag/validate/order.c

@@ -185,7 +185,6 @@ writer_thread(void *unused)
 		if (ck_pr_load_int(&leave) == 1)
 			break;
 
-		//fprintf(stderr, "set: %d", iteration);
 		for (i = 1; i < writer_max; i++) {
 			void *replace = (void *)(uintptr_t)i;
 			if (ck_bag_set_spmc(&bag, (void *)(uintptr_t)i, replace) == false) {
@@ -210,6 +209,7 @@ writer_thread(void *unused)
 		ck_epoch_write_end(&epoch_wr);
 	}
 
+	fprintf(stderr, "Writer %u iterations, %u writes per iteration.\n", iteration, writer_max);
 	while (ck_pr_load_uint(&barrier) != NUM_READER_THREADS)
 		ck_pr_stall();
 
@@ -288,7 +288,7 @@ main(int argc, char **argv)
 		pthread_create(&readers[i], NULL, reader, NULL);
 	}
 
-	sleep(10);
+	sleep(120);
 
 	ck_pr_store_int(&leave, 1);
 	for (i = 0; i < NUM_READER_THREADS; i++)

src/ck_bag.c

@@ -50,7 +50,7 @@ ck_bag_init(struct ck_bag *bag,
 {
 	size_t block_overhead;
 
-	bag->avail_head = bag->avail_tail = NULL;
+	CK_LIST_INIT(&bag->avail_blocks);
 	bag->head = NULL;
 	bag->n_entries = 0;
 	bag->n_blocks = 0;
@@ -109,95 +109,50 @@ ck_bag_allocator_set(struct ck_malloc *m, size_t alloc_overhead)
 bool
 ck_bag_put_spmc(struct ck_bag *bag, void *entry)
 {
-	struct ck_bag_block *cursor, *new_block, *new_block_prev, *new_tail;
+	struct ck_bag_block *cursor, *new_block, *new_bag_head, *prev_block;
 	uint16_t n_entries_block;
 	size_t blocks_alloc, i;
 	uintptr_t next = 0;
 
-	new_block = new_block_prev = new_tail = NULL;
+	new_block = new_bag_head = prev_block = NULL;
 
-	/*
+	cursor = CK_LIST_FIRST(&bag->avail_blocks);
-	 * Blocks with available entries are stored in
-	 * the bag's available list.
-	 */
-	cursor = bag->avail_head;
 	if (cursor != NULL) {
 		n_entries_block = ck_bag_block_count(cursor);
 	} else {
 		/* The bag is full, allocate a new set of blocks */
+		prev_block = CK_LIST_FIRST(&bag->avail_blocks);
 		if (bag->alloc_strat == CK_BAG_ALLOCATE_GEOMETRIC)
 			blocks_alloc = (bag->n_blocks + 1) << 1;
 		else
 			blocks_alloc = 1;
 
-		for (i = 0; i < blocks_alloc-1; i++) {
+		for (i = 0; i < blocks_alloc; i++) {
 			new_block = allocator.malloc(bag->info.bytes);
 
 			if (new_block == NULL)
 				return false;
 
-			/*
+#ifdef __x86_64__
-			 * First node is the tail of the Bag.
+			new_block->next.ptr = NULL;
-			 * Second node is the new tail of the Available list.
+#else
-			 */
-			if (i == 0)
-				new_tail = new_block;
-
-#ifndef __x86_64__
 			new_block->next.n_entries = 0;
 #endif
-
+			if (i == 0) {
-			new_block->next.ptr = new_block_prev;
+				new_bag_head = new_block;
-			new_block->avail_next = new_block_prev;
+				CK_LIST_INSERT_HEAD(&bag->avail_blocks, new_block, avail_entry);
-			if (new_block_prev != NULL)
+			} else {
-				new_block_prev->avail_prev = new_block;
+				CK_LIST_INSERT_AFTER(prev_block, new_block, avail_entry);
-
+			}
-			new_block_prev = new_block;
+
+			prev_block = new_block;
 		}
 
-		/*
+		cursor = new_bag_head;
-		 * Insert entry into last allocated block.
-		 * cursor is new head of available list.
-		 */
-		cursor = allocator.malloc(bag->info.bytes);
-		cursor->avail_next = new_block;
-		cursor->avail_prev = NULL;
-		new_block->avail_prev = cursor;
 		n_entries_block = 0;
-		bag->n_blocks += blocks_alloc; /* n_blocks and n_avail_blocks? */
+		bag->n_blocks += blocks_alloc;
-	}
-
-	/* Update the available list */
-	if (new_block != NULL) {
-		if (bag->avail_tail != NULL) {
-			cursor->avail_prev = bag->avail_tail;
-			bag->avail_tail->avail_next = cursor;
-		} else {
-			/* Available list was previously empty */
-			bag->avail_head = cursor;
-		}
-
-		bag->avail_tail = new_tail;
-	} else if (n_entries_block == bag->info.max-1) {
-	    /* New entry will fill up block, remove from avail list */
-		    if (cursor->avail_prev != NULL)
-			    cursor->avail_prev->avail_next = cursor->avail_next;
-
-		    if (cursor->avail_next != NULL)
-			    cursor->avail_next->avail_prev = cursor->avail_prev;
-
-		    if (bag->avail_head == cursor)
-			    bag->avail_head = cursor->avail_next;
-
-		    if (bag->avail_tail == cursor)
-			    bag->avail_tail = cursor->avail_prev;
-
-		    /* For debugging purposes */
-		    cursor->avail_next = NULL;
-		    cursor->avail_prev = NULL;
 	}
 
-	/* Update array and block->n_entries */
 	cursor->array[n_entries_block++] = entry;
 	ck_pr_fence_store();
 
@@ -205,9 +160,9 @@ ck_bag_put_spmc(struct ck_bag *bag, void *entry)
 	next = ((uintptr_t)n_entries_block << 48);
 #endif
 
-	/* Update bag's list */
-	if (n_entries_block == 1) {
 
+	if (n_entries_block == 1) {
+		/* Place newly filled block at the head of bag list */
 		if (bag->head != NULL) {
 #ifdef __x86_64__
 			next += ((uintptr_t)(void *)ck_bag_block_next(bag->head));
@@ -223,13 +178,17 @@ ck_bag_put_spmc(struct ck_bag *bag, void *entry)
 		ck_pr_store_ptr(&cursor->next.ptr, (void *)next);
 		ck_pr_store_ptr(&bag->head, cursor);
 	} else {
-
+		/* Block is already on bag list, update n_entries */
 #ifdef __x86_64__
 		next += ((uintptr_t)(void *)ck_bag_block_next(cursor->next.ptr));
 		ck_pr_store_ptr(&cursor->next, (void *)next);
 #else
 		ck_pr_store_ptr(&cursor->next.n_entries, (void *)(uintptr_t)n_entries_block);
 #endif
+		/* the block is full, remove from available_list */
+		if (n_entries_block == bag->info.max) {
+			CK_LIST_REMOVE(cursor, avail_entry);
+		}
 
 	}
 
@@ -290,7 +249,6 @@ ck_bag_remove_spmc(struct ck_bag *bag, void *entry)
 found:
 	/* Cursor points to containing block, block_index is index of deletion */
 	if (n_entries == 1) {
-		/* If a block's single entry is being removed, remove the block. */
 		if (prev == NULL) {
 			struct ck_bag_block *new_head = ck_bag_block_next(cursor->next.ptr);
 			ck_pr_store_ptr(&bag->head, new_head);
@@ -305,15 +263,8 @@ found:
 			ck_pr_store_ptr(&prev->next.ptr, (struct ck_bag_block *)next);
 		}
 
-		/* Remove block from available list */
+		CK_LIST_REMOVE(cursor, avail_entry);
-		if (cursor->avail_prev != NULL)
-			cursor->avail_prev->avail_next = cursor->avail_next;
-
-		if (cursor->avail_next != NULL)
-			cursor->avail_next->avail_prev = cursor->avail_prev;
-
 		bag->n_blocks--;
-		copy = cursor->avail_next;
 	} else {
 		uintptr_t next_ptr;
 
@@ -346,22 +297,13 @@ found:
 #endif
 		}
 
-		if (n_entries == bag->info.max - 1) {
+		if (n_entries != bag->info.max-1) {
-			/* Block was previously fully, add to head of avail. list */
+			/* Only remove cursor if it was previously on the avail_list */
-			copy->avail_next = bag->avail_head;
+			CK_LIST_REMOVE(cursor, avail_entry);
-			copy->avail_prev = NULL;
-			bag->avail_head = copy;
 		}
-
+		CK_LIST_INSERT_HEAD(&bag->avail_blocks, copy, avail_entry);
 	}
 
-	/* Update available list. */
-	if (bag->avail_head == cursor)
-		bag->avail_head = copy;
-
-	if (bag->avail_tail == cursor)
-		bag->avail_tail = copy;
-
 	allocator.free(cursor, sizeof(struct ck_bag_block), true);
 	ck_pr_store_uint(&bag->n_entries, bag->n_entries - 1);
 	return true;