[PATCH 3/4] idr: Rewrite ida

[PATCH 3/4] idr: Rewrite ida

[Kent Overstreet]

This is a new, from scratch implementation of ida that should be
simpler, faster and more space efficient.

Two primary reasons for the rewrite:
 * A future patch will reimplement idr on top of this ida implementation +
   radix trees. Once that's done, the end result will be ~1k fewer lines
   of code, much simpler and easier to understand and it should be quite
   a bit faster.

 * The performance improvements and addition of ganged allocation should
   make ida more suitable for use by a percpu id/tag allocator, which
   would then act as a frontend to this allocator.

The old ida implementation was done with the idr data structures - this
was IMO backwards. I'll soon be reimplementing idr on top of this new
ida implementation and radix trees - using a separate dedicated data
structure for the free ID bitmap should actually make idr faster, and
the end result is _significantly_ less code.

This implementation conceptually isn't that different from the old one -
it's a tree of bitmaps, where one bit in a given node indicates whether
or not there are free bits in a child node.

The main difference (and advantage) over the old version is that the
tree isn't implemented with pointers - it's implemented in an array,
like how heaps are implemented, which both better space efficiency and
it'll be faster since there's no pointer chasing.

This does mean that the entire bitmap is stored in one contiguous memory
allocation - and as currently implemented we won't be able to allocate
_quite_ as many ids as with the previous implementation.

I don't expect this to be an issue in practice since anywhere this is
used, an id corresponds to a struct allocation somewher else - we can't
allocate an unbounded number of ids, we'll run out of memory somewhere
else eventually, and I expect that to be the limiting factor in
practice.

If a user/use case does come up where this matters I can add some
sharding (or perhaps add a separate big_ida implementation) - but the
extra complexity would adversely affect performance for the users that
don't need > millions of ids, so I intend to leave the implementation as
is until if and when this becomes an issue.

Signed-off-by: Kent Overstreet <koverstreet@google.com>
Cc: Andrew Morton <akpm@linux-foundation.org>
Cc: Tejun Heo <tj@kernel.org>
Cc: Stephen Rothwell <sfr@canb.auug.org.au>
Cc: Fengguang Wu <fengguang.wu@intel.com>
---
 include/linux/idr.h | 118 +++++---
 lib/idr.c           | 776 +++++++++++++++++++++++++++++++++-------------------
 2 files changed, 573 insertions(+), 321 deletions(-)

diff --git a/include/linux/idr.h b/include/linux/idr.h
index c0e0c54..9169e18 100644
--- a/include/linux/idr.h
+++ b/include/linux/idr.h
@@ -17,6 +17,88 @@
 #include <linux/init.h>
 #include <linux/rcupdate.h>
 
+/* IDA */
+
+#define IDA_INLINE_NODES	4
+
+struct ida {
+	spinlock_t		lock;
+
+	/*
+	 * cur_id and allocated_ids are for ida_alloc_cyclic. For cyclic
+	 * allocations we search for new ids to allocate starting from the last
+	 * id allocated - cur_id is the next id to try allocating.
+	 *
+	 * But we also don't want the allocated ids to be arbitrarily sparse -
+	 * the memory usage for the bitmap could be arbitrarily bad, and if
+	 * they're used as keys in a radix tree the memory overhead of the radix
+	 * tree could be quite bad as well. So we use allocated_ids to decide
+	 * when to restart cur_id from 0, and bound how sparse the bitmap can
+	 * be.
+	 */
+	unsigned		cur_id;
+	unsigned		allocated_ids;
+
+	/* size of ida->tree */
+	unsigned		nodes;
+
+	/*
+	 * Index of first leaf node in ida->tree; equal to the number of non
+	 * leaf nodes, ida->nodes - ida->first_leaf == number of leaf nodes
+	 */
+	unsigned		first_leaf;
+
+	unsigned long		*tree;
+	unsigned long		inline_nodes[IDA_INLINE_NODES];
+};
+
+#define IDA_INIT(name)						\
+{								\
+	.lock		= __SPIN_LOCK_UNLOCKED(name.lock),	\
+	.nodes		= IDA_INLINE_NODES,			\
+	.first_leaf	= 1,					\
+	.tree		= name.inline_nodes,			\
+}
+#define DEFINE_IDA(name)	struct ida name = IDA_INIT(name)
+
+void ida_remove(struct ida *ida, unsigned id);
+int ida_alloc_range(struct ida *ida, unsigned int start,
+		  unsigned int end, gfp_t gfp);
+int ida_alloc_cyclic(struct ida *ida, unsigned start, unsigned end, gfp_t gfp);
+void ida_destroy(struct ida *ida);
+int ida_init_prealloc(struct ida *ida, unsigned prealloc);
+
+/**
+ * ida_alloc_range - allocate a new id.
+ * @ida: the (initialized) ida.
+ * @gfp_mask: memory allocation flags
+ *
+ * Allocates an id in the range [0, INT_MAX]. Returns -ENOSPC if no ids are
+ * available, or -ENOMEM on memory allocation failure.
+ *
+ * Returns the smallest available id
+ *
+ * Use ida_remove() to get rid of an id.
+ */
+static inline int ida_alloc(struct ida *ida, gfp_t gfp_mask)
+{
+	return ida_alloc_range(ida, 0, 0, gfp_mask);
+}
+
+/**
+ * ida_init - initialize ida handle
+ * @ida:	ida handle
+ *
+ * This function is use to set up the handle (@ida) that you will pass
+ * to the rest of the functions.
+ */
+static inline void ida_init(struct ida *ida)
+{
+	ida_init_prealloc(ida, 0);
+}
+
+/* IDR */
+
 /*
  * We want shallower trees and thus more bits covered at each layer.  8
  * bits gives us large enough first layer for most use cases and maximum
@@ -195,42 +277,6 @@ static inline void __deprecated idr_remove_all(struct idr *idp)
 	__idr_remove_all(idp);
 }
 
-/*
- * IDA - IDR based id allocator, use when translation from id to
- * pointer isn't necessary.
- *
- * IDA_BITMAP_LONGS is calculated to be one less to accommodate
- * ida_bitmap->nr_busy so that the whole struct fits in 128 bytes.
- */
-#define IDA_CHUNK_SIZE		128	/* 128 bytes per chunk */
-#define IDA_BITMAP_LONGS	(IDA_CHUNK_SIZE / sizeof(long) - 1)
-#define IDA_BITMAP_BITS 	(IDA_BITMAP_LONGS * sizeof(long) * 8)
-
-struct ida_bitmap {
-	long			nr_busy;
-	unsigned long		bitmap[IDA_BITMAP_LONGS];
-};
-
-struct ida {
-	struct idr		idr;
-	struct ida_bitmap	*free_bitmap;
-};
-
-#define IDA_INIT(name)		{ .idr = IDR_INIT((name).idr), .free_bitmap = NULL, }
-#define DEFINE_IDA(name)	struct ida name = IDA_INIT(name)
-
-void ida_destroy(struct ida *ida);
-void ida_init(struct ida *ida);
-
-int ida_alloc_range(struct ida *ida, unsigned int start, unsigned int end,
-		   gfp_t gfp_mask);
-void ida_remove(struct ida *ida, unsigned int id);
-
-static inline int ida_alloc(struct ida *ida, gfp_t gfp_mask)
-{
-	return ida_alloc_range(ida, 0, 0, gfp_mask);
-}
-
 void __init idr_init_cache(void);
 
 #endif /* __IDR_H__ */
diff --git a/lib/idr.c b/lib/idr.c
index df2d32e..7d89587 100644
--- a/lib/idr.c
+++ b/lib/idr.c
@@ -8,6 +8,8 @@
  *
  * Modified by Nadia Derbey to make it RCU safe.
  *
+ * IDA completely rewritten by Kent Overstreet <koverstreet@google.com>
+ *
  * Small id to pointer translation service.
  *
  * It uses a radix tree like structure as a sparse array indexed
@@ -38,6 +40,495 @@
 #include <linux/percpu.h>
 #include <linux/hardirq.h>
 
+#define IDA_TREE_ARY		BITS_PER_LONG
+
+/**
+ * DOC: IDA description
+ * IDA - ID (small integer) allocator
+ *
+ * This works much like using a simple bitmap to allocate indices - ida_alloc()
+ * is equivalent to find_first_zero_bit() then __set_bit(), and ida_remove() is
+ * equivalent to __clear_bit(). But it's much more efficient than a large
+ * bitmap, and resizes itself as needed.
+ *
+ * It's implemented as a tree of bitmaps: a node in the tree is a single
+ * unsigned long. The leaf nodes of the tree are segments of the entire bitmap -
+ * a cleared bit indicates a free id, and a set bit indicates an allocated one.
+ * Bits in the parent nodes indicate whether or not there are free bits in the
+ * corresponding child node - when all the bits in a parent node are set, none
+ * of its children have bits free.
+ *
+ * The splay factor of the tree (IDA_TREE_ARY) == BITS_PER_LONG - parent nodes
+ * have 32 or 64 children.
+ *
+ * The tree itself is implemented with an array instead of pointers - exactly
+ * like the textbook implementation of D-ary heaps. The root of the bitmap tree
+ * is at ida->tree[0]. The children of node i are at i * IDA_TREE_ARY + 1 + j,
+ * where j is in the range [0, 63], and the parent of node i is at (i - 1) /
+ * IDA_TREE_ARY.
+ *
+ * This conveniently means that our leaf nodes are all contiguous in memory -
+ * the bit for id i is bit id % BITS_PER_LONG in ida->tree[ida->first_leaf + i /
+ * BITS_PER_LONG].
+ *
+ * Note that the number of ids we can allocate is limited by the amount of
+ * memory we can contiguously allocate. The amount of memory used for the bitmap
+ * tree is only slightly more than a flat bitmap would use - about 1 / TREE_ARY
+ * * (sizeof flat bitmap).
+ *
+ * So for 1 mb of memory (and allocating more than that should be fine with
+ * CONFIG_COMPACTION) you get slightly under 8 million IDs.
+ */
+
+/*
+ * For a given number of nodes, calculate how many are going to be parent nodes
+ * (equal to ida->first_leaf) and by extension how my will be leaves.
+ */
+static unsigned first_leaf_from_nodes(unsigned nodes)
+{
+	unsigned ret = 0;
+
+	while (ret * IDA_TREE_ARY + 1 < nodes)
+		ret = ret * IDA_TREE_ARY + 1;
+
+	return ret;
+}
+
+/*
+ * Given some number of leaf nodes, calculate the number of parent nodes the
+ * bitmap tree will require - i.e. the new ida->first_leaf
+ */
+static unsigned first_leaf_from_leaves(unsigned leaves)
+{
+	unsigned ret = 0;
+
+	while (ret * IDA_TREE_ARY + 1 < ret + leaves)
+		ret = ret * IDA_TREE_ARY + 1;
+
+	return ret;
+}
+
+static void __ida_remove(struct ida *ida, unsigned int id)
+{
+	unsigned i = ida->first_leaf + id / BITS_PER_LONG;
+	unsigned bit = id % BITS_PER_LONG;
+
+	BUG_ON(i >= ida->nodes);
+
+	--ida->allocated_ids;
+
+	while (1) {
+		unsigned long *node = ida->tree + i, old = *node;
+
+		WARN_ON(!test_bit(bit, node));
+		__clear_bit(bit, node);
+
+		if (~old || !i)
+			break;
+
+		/*
+		 * If this node's bits were all 1s before we cleared this bit,
+		 * we need to clear this node's bit in the parent node - and so
+		 * on up to the root.
+		 */
+
+		bit = (i - 1) % IDA_TREE_ARY;
+		i = (i - 1) / IDA_TREE_ARY;
+	}
+}
+
+/**
+ * ida_remove - remove an allocated id.
+ * @ida: the (initialized) ida.
+ * @id: the id returned by ida_alloc_range.
+ */
+void ida_remove(struct ida *ida, unsigned int id)
+{
+	unsigned long flags;
+	spin_lock_irqsave(&ida->lock, flags);
+	__ida_remove(ida, id);
+	spin_unlock_irqrestore(&ida->lock, flags);
+}
+EXPORT_SYMBOL(ida_remove);
+
+static void ida_free_tree(unsigned long *tree, unsigned nodes)
+{
+	size_t bytes = nodes * sizeof(unsigned long);
+
+	if (bytes < PAGE_SIZE)
+		kfree(tree);
+	else
+		free_pages((unsigned long) tree,
+			   get_order(bytes));
+
+}
+
+/*
+ * Attempt to double the size of the tree. We have to drop ida->lock to allocate
+ * memory, so we might race with another allocation that also tries to resize.
+ * So if the tree's not the size it originally was when we retake ida->lock,
+ * just return 0 - but the caller needs to recheck for the tree being full in
+ * case we _really_ raced badly.
+ */
+static inline int __ida_resize(struct ida *ida, unsigned max_id,
+			       gfp_t gfp, unsigned long *flags)
+	__releases(&ida->lock)
+	__acquires(&ida->lock)
+{
+	unsigned long *tree;
+	unsigned old_nodes = ida->nodes;
+	unsigned cur_leaves = ida->nodes - ida->first_leaf;
+	unsigned new_nodes = roundup_pow_of_two(ida->nodes + 1);
+	unsigned first_leaf = first_leaf_from_nodes(new_nodes);
+	size_t bytes;
+
+	if (cur_leaves >= BITS_TO_LONGS(max_id))
+		return -ENOSPC;
+
+	spin_unlock_irqrestore(&ida->lock, *flags);
+
+	bytes = new_nodes * sizeof(unsigned long);
+
+	tree = bytes < PAGE_SIZE
+		? kmalloc(bytes, gfp)
+		: (void *) __get_free_pages(gfp, get_order(bytes));
+
+	spin_lock_irqsave(&ida->lock, *flags);
+
+	if (!tree)
+		return -ENOMEM;
+
+	if (old_nodes != ida->nodes) {
+		ida_free_tree(tree, new_nodes);
+		return 0;
+	}
+
+	if (first_leaf == ida->first_leaf) {
+		/* Depth doesn't change, just appending leaf nodes */
+		memcpy(tree, ida->tree, ida->nodes * sizeof(unsigned long));
+	} else {
+		unsigned i, j, bit;
+
+		/* Zero out new parent nodes, reconstructing them below */
+		memset(tree, 0, first_leaf * sizeof(unsigned long));
+
+		memcpy(tree + first_leaf,
+		       ida->tree + ida->first_leaf,
+		       cur_leaves * sizeof(unsigned long));
+
+		for (i = first_leaf; i < first_leaf + cur_leaves; i++) {
+			j = i;
+
+			while (!~tree[j] && j) {
+				bit = (j - 1) % IDA_TREE_ARY;
+				j = (j - 1) / IDA_TREE_ARY;
+
+				__set_bit(bit, tree + j);
+			}
+		}
+	}
+
+	/* Zero out new leaf nodes */
+	memset(tree + first_leaf + cur_leaves, 0,
+	       (new_nodes - first_leaf - cur_leaves) * sizeof(unsigned long));
+
+	if (ida->tree != ida->inline_nodes)
+		ida_free_tree(ida->tree, ida->nodes);
+
+	ida->nodes	= new_nodes;
+	ida->first_leaf = first_leaf;
+	ida->tree	= tree;
+
+	return 0;
+}
+
+/*
+ * Ganged allocation - amortize locking and tree traversal for when we've got
+ * another allocator (i.e. a percpu version) acting as a frontend to this code
+ */
+static int __ida_alloc_range_multiple(struct ida *ida, unsigned *ids,
+				      unsigned nr_ids, unsigned min_id,
+				      unsigned max_id, gfp_t gfp,
+				      unsigned long *flags)
+	__releases(&ida->lock)
+	__acquires(&ida->lock)
+{
+	unsigned i = 0, bit, bit_offset, id, ids_found = 0;
+	unsigned long *node = ida->tree;
+	int err = 0;
+
+	if (!max_id)
+		max_id = (unsigned) INT_MAX + 1;
+
+	if (min_id >= max_id)
+		return -ENOSPC;
+
+	while (ids_found < nr_ids) {
+		/*
+		 * If all bits are set in the root, no bits free and we need to
+		 * resize.
+		 */
+		while (!~*node) {
+resize:
+			err = __ida_resize(ida, max_id, gfp, flags);
+			if (err)
+				break;
+
+			i = 0;
+			node = ida->tree;
+		}
+
+		if (min_id) {
+			/*
+			 * If we're starting from a specific index, skip to that
+			 * leaf node and start looking there:
+			 */
+			bit_offset = min_id % BITS_PER_LONG;
+			i = ida->first_leaf + min_id / BITS_PER_LONG;
+
+			if (i >= ida->nodes)
+				goto resize;
+
+			while (1) {
+				node = ida->tree + i;
+				bit = ffz(*node >> bit_offset) + bit_offset;
+
+				/*
+				 * We might have had to go back up the tree
+				 * before we found a free bit - so skip down to
+				 * where we recurse down the tree.
+				 */
+				if (~*node && bit < BITS_PER_LONG)
+					goto found;
+
+				if (!i)
+					goto resize;
+
+				/*
+				 * Ok, no bits available in this node - go up a
+				 * level. But we have to update bit_offset so we
+				 * start searching in the parent _after_ the
+				 * node we're currently at
+				 */
+				bit_offset = (i - 1) % IDA_TREE_ARY + 1;
+				i = (i - 1) / IDA_TREE_ARY;
+			}
+		}
+
+		/*
+		 * Recurse down the tree looking for a free bit. We already
+		 * checked to make sure there _were_ free bits, but we might end
+		 * up at a leaf node we haven't allocated yet.
+		 */
+		while (1) {
+			bit = ffz(*node);
+found:
+			/*
+			 * Found a bit - if we're at a leaf node, great! We're
+			 * done:
+			 */
+			if (i >= ida->first_leaf)
+				break;
+
+			i = i * IDA_TREE_ARY + 1 + bit;
+			node = ida->tree + i;
+
+			/*
+			 * Recurse. But if we'd recurse to a node that hasn't
+			 * been allocated yet, resize:
+			 */
+
+			if (i >= ida->nodes)
+				goto resize;
+
+			BUG_ON(!~*node);
+		}
+
+		/*
+		 * Our leaves are contiguous, so we can calculate the id we
+		 * allocated from the node we're at and the bit we found within
+		 * that node:
+		 */
+		id = (i - ida->first_leaf) * BITS_PER_LONG + bit;
+		BUG_ON(id < min_id);
+
+		if (id >= max_id) {
+			err = -ENOSPC;
+			break;
+		}
+
+		ids[ids_found++] = id;
+		ida->allocated_ids++;
+
+		/*
+		 * Mark the id as allocated. If all the bits are now set in this
+		 * node, set this node's bit in the parent node - and so on up
+		 * to the root:
+		 */
+		while (1) {
+			__set_bit(bit, node);
+
+			if (~*node || !i)
+				break;
+
+			bit = (i - 1) % IDA_TREE_ARY;
+			i = (i - 1) / IDA_TREE_ARY;
+
+			node = ida->tree + i;
+		}
+	}
+
+	return ids_found ? ids_found : err;
+}
+
+/**
+ * ida_alloc_range - allocate a new id.
+ * @ida: the (initialized) ida.
+ * @start: the minimum id (inclusive, <= INT_MAX)
+ * @end: the maximum id (exclusive, <= INT_MAX + 1 or 0 for unlimited)
+ * @gfp_mask: memory allocation flags
+ *
+ * Allocates an id in the range [start, end). Returns -ENOSPC if no ids are
+ * available, or -ENOMEM on memory allocation failure.
+ *
+ * Returns the smallest free id >= start.
+ *
+ * Use ida_remove() to get rid of an id.
+ */
+int ida_alloc_range(struct ida *ida, unsigned int start,
+		    unsigned int end, gfp_t gfp)
+{
+	int ret;
+	unsigned id;
+	unsigned long flags;
+
+	spin_lock_irqsave(&ida->lock, flags);
+	ret = __ida_alloc_range_multiple(ida, &id, 1, start, end, gfp, &flags);
+	spin_unlock_irqrestore(&ida->lock, flags);
+
+	return ret == 1 ? id : ret;
+}
+EXPORT_SYMBOL(ida_alloc_range);
+
+static int __ida_alloc_cyclic(struct ida *ida, unsigned start, unsigned end,
+			      gfp_t gfp, unsigned long *flags)
+	__releases(&ida->lock)
+	__acquires(&ida->lock)
+{
+	int ret;
+	unsigned id;
+
+	ret = __ida_alloc_range_multiple(ida, &id, 1,
+					 max(start, ida->cur_id),
+					 end, gfp, flags);
+
+	if (ret == -ENOSPC)
+		ret = __ida_alloc_range_multiple(ida, &id, 1, start,
+						 end, gfp, flags);
+	if (ret == 1) {
+		ida->cur_id = id + 1;
+		if ((ida->cur_id - start) / 2 > max(1024U, ida->allocated_ids))
+			ida->cur_id = 0;
+
+		return id;
+	}
+
+	return ret;
+}
+
+/**
+ * ida_alloc_cyclic - allocate new ids cyclically
+ * @ida: the (initialized) ida.
+ * @start: the minimum id (inclusive, <= INT_MAX)
+ * @end: the maximum id (exclusive, <= INT_MAX + 1 or 0 for unlimited)
+ * @gfp_mask: memory allocation flags
+ *
+ * Allocates an id in the range start <= id < end, or returns -ENOSPC.
+ * On memory allocation failure, returns -ENOMEM.
+ *
+ * Instead of returning the smallest free id, start searching from the position
+ * where the last id was allocated - i.e. it won't reuse freed ids right away.
+ *
+ * To avoid the allocated id space (and internal bitmap) becoming arbitrarily
+ * sparse, it can wrap before reaching the maximum id - if less than half of our
+ * current id space is allocated, it resets cur_id to 0
+ *
+ * But we don't want to wrap when the id space is small, so we use the maximum
+ * of (1024, allocated_ids) - see __ida_alloc_cyclic().
+ *
+ * Use ida_remove() to get rid of an id.
+ */
+int ida_alloc_cyclic(struct ida *ida, unsigned start, unsigned end, gfp_t gfp)
+{
+	int ret;
+	unsigned long flags;
+
+	spin_lock_irqsave(&ida->lock, flags);
+	ret = __ida_alloc_cyclic(ida, start, end, gfp, &flags);
+	spin_unlock_irqrestore(&ida->lock, flags);
+
+	return ret;
+}
+EXPORT_SYMBOL(ida_alloc_cyclic);
+
+/**
+ * ida_destroy - release all cached layers within an ida tree
+ * @ida:		ida handle
+ */
+void ida_destroy(struct ida *ida)
+{
+	if (ida->tree != ida->inline_nodes)
+		ida_free_tree(ida->tree, ida->nodes);
+}
+EXPORT_SYMBOL(ida_destroy);
+
+/**
+ * ida_init_prealloc - initialize ida handle
+ * @ida:	ida handle
+ * @prealloc:	number of ids to preallocate memory for
+ *
+ * Initialize an ida, and preallocate enough memory that ida_alloc() will never
+ * return -ENOMEM if passed max_id <= prealloc.
+ */
+int ida_init_prealloc(struct ida *ida, unsigned prealloc)
+{
+	memset(ida, 0, sizeof(*ida));
+
+	spin_lock_init(&ida->lock);
+
+	ida->nodes	= IDA_INLINE_NODES;
+	ida->first_leaf = 0;
+	ida->tree	= ida->inline_nodes;
+
+	if (prealloc) {
+		unsigned leaves = BITS_TO_LONGS(prealloc);
+		unsigned first_leaf = first_leaf_from_leaves(leaves);
+		unsigned nodes = first_leaf + leaves;
+		size_t bytes;
+
+		if (nodes > ida->nodes) {
+			nodes = roundup_pow_of_two(nodes);
+			bytes = nodes * sizeof(unsigned long);
+
+			ida->tree = bytes < PAGE_SIZE
+				? kzalloc(bytes, GFP_KERNEL)
+				: (void *) __get_free_pages(GFP_KERNEL|__GFP_ZERO,
+							    get_order(bytes));
+			if (!ida->tree)
+				return -ENOMEM;
+
+			ida->nodes = nodes;
+			ida->first_leaf = first_leaf;
+		}
+	}
+
+	return 0;
+
+}
+EXPORT_SYMBOL(ida_init_prealloc);
+
+/* IDR */
+
 #define MAX_IDR_SHIFT		(sizeof(int) * 8 - 1)
 #define MAX_IDR_BIT		(1U << MAX_IDR_SHIFT)
 
@@ -50,7 +541,6 @@
 static struct kmem_cache *idr_layer_cache;
 static DEFINE_PER_CPU(struct idr_layer *, idr_preload_head);
 static DEFINE_PER_CPU(int, idr_preload_cnt);
-static DEFINE_SPINLOCK(simple_ida_lock);
 
 /* the maximum ID which can be allocated given idr->layers */
 static int idr_max(int layers)
@@ -870,287 +1360,3 @@ void idr_init(struct idr *idp)
 	spin_lock_init(&idp->lock);
 }
 EXPORT_SYMBOL(idr_init);
-
-
-/**
- * DOC: IDA description
- * IDA - IDR based ID allocator
- *
- * This is id allocator without id -> pointer translation.  Memory
- * usage is much lower than full blown idr because each id only
- * occupies a bit.  ida uses a custom leaf node which contains
- * IDA_BITMAP_BITS slots.
- *
- * 2007-04-25  written by Tejun Heo <htejun@gmail.com>
- */
-
-static void free_bitmap(struct ida *ida, struct ida_bitmap *bitmap)
-{
-	unsigned long flags;
-
-	if (!ida->free_bitmap) {
-		spin_lock_irqsave(&ida->idr.lock, flags);
-		if (!ida->free_bitmap) {
-			ida->free_bitmap = bitmap;
-			bitmap = NULL;
-		}
-		spin_unlock_irqrestore(&ida->idr.lock, flags);
-	}
-
-	kfree(bitmap);
-}
-
-/**
- * ida_pre_get - reserve resources for ida allocation
- * @ida:	ida handle
- * @gfp_mask:	memory allocation flag
- *
- * This function should be called prior to locking and calling the
- * following function.  It preallocates enough memory to satisfy the
- * worst possible allocation.
- *
- * If the system is REALLY out of memory this function returns %0,
- * otherwise %1.
- */
-static int ida_pre_get(struct ida *ida, gfp_t gfp_mask)
-{
-	/* allocate idr_layers */
-	if (!__idr_pre_get(&ida->idr, gfp_mask))
-		return 0;
-
-	/* allocate free_bitmap */
-	if (!ida->free_bitmap) {
-		struct ida_bitmap *bitmap;
-
-		bitmap = kmalloc(sizeof(struct ida_bitmap), gfp_mask);
-		if (!bitmap)
-			return 0;
-
-		free_bitmap(ida, bitmap);
-	}
-
-	return 1;
-}
-
-/**
- * ida_get_new_above - allocate new ID above or equal to a start id
- * @ida:	ida handle
- * @starting_id: id to start search at
- * @p_id:	pointer to the allocated handle
- *
- * Allocate new ID above or equal to @starting_id.  It should be called
- * with any required locks.
- *
- * If memory is required, it will return %-EAGAIN, you should unlock
- * and go back to the ida_pre_get() call.  If the ida is full, it will
- * return %-ENOSPC.
- *
- * @p_id returns a value in the range @starting_id ... %0x7fffffff.
- */
-static int ida_get_new_above(struct ida *ida, int starting_id, int *p_id)
-{
-	struct idr_layer *pa[MAX_IDR_LEVEL + 1];
-	struct ida_bitmap *bitmap;
-	unsigned long flags;
-	int idr_id = starting_id / IDA_BITMAP_BITS;
-	int offset = starting_id % IDA_BITMAP_BITS;
-	int t, id;
-
- restart:
-	/* get vacant slot */
-	t = idr_get_empty_slot(&ida->idr, idr_id, pa, 0, &ida->idr);
-	if (t < 0)
-		return t == -ENOMEM ? -EAGAIN : t;
-
-	if (t * IDA_BITMAP_BITS >= MAX_IDR_BIT)
-		return -ENOSPC;
-
-	if (t != idr_id)
-		offset = 0;
-	idr_id = t;
-
-	/* if bitmap isn't there, create a new one */
-	bitmap = (void *)pa[0]->ary[idr_id & IDR_MASK];
-	if (!bitmap) {
-		spin_lock_irqsave(&ida->idr.lock, flags);
-		bitmap = ida->free_bitmap;
-		ida->free_bitmap = NULL;
-		spin_unlock_irqrestore(&ida->idr.lock, flags);
-
-		if (!bitmap)
-			return -EAGAIN;
-
-		memset(bitmap, 0, sizeof(struct ida_bitmap));
-		rcu_assign_pointer(pa[0]->ary[idr_id & IDR_MASK],
-				(void *)bitmap);
-		pa[0]->count++;
-	}
-
-	/* lookup for empty slot */
-	t = find_next_zero_bit(bitmap->bitmap, IDA_BITMAP_BITS, offset);
-	if (t == IDA_BITMAP_BITS) {
-		/* no empty slot after offset, continue to the next chunk */
-		idr_id++;
-		offset = 0;
-		goto restart;
-	}
-
-	id = idr_id * IDA_BITMAP_BITS + t;
-	if (id >= MAX_IDR_BIT)
-		return -ENOSPC;
-
-	__set_bit(t, bitmap->bitmap);
-	if (++bitmap->nr_busy == IDA_BITMAP_BITS)
-		idr_mark_full(pa, idr_id);
-
-	*p_id = id;
-
-	/* Each leaf node can handle nearly a thousand slots and the
-	 * whole idea of ida is to have small memory foot print.
-	 * Throw away extra resources one by one after each successful
-	 * allocation.
-	 */
-	if (ida->idr.id_free_cnt || ida->free_bitmap) {
-		struct idr_layer *p = get_from_free_list(&ida->idr);
-		if (p)
-			kmem_cache_free(idr_layer_cache, p);
-	}
-
-	return 0;
-}
-
-static void __ida_remove(struct ida *ida, int id)
-{
-	struct idr_layer *p = ida->idr.top;
-	int shift = (ida->idr.layers - 1) * IDR_BITS;
-	int idr_id = id / IDA_BITMAP_BITS;
-	int offset = id % IDA_BITMAP_BITS;
-	int n;
-	struct ida_bitmap *bitmap;
-
-	/* clear full bits while looking up the leaf idr_layer */
-	while ((shift > 0) && p) {
-		n = (idr_id >> shift) & IDR_MASK;
-		__clear_bit(n, p->bitmap);
-		p = p->ary[n];
-		shift -= IDR_BITS;
-	}
-
-	if (p == NULL)
-		goto err;
-
-	n = idr_id & IDR_MASK;
-	__clear_bit(n, p->bitmap);
-
-	bitmap = (void *)p->ary[n];
-	if (!test_bit(offset, bitmap->bitmap))
-		goto err;
-
-	/* update bitmap and remove it if empty */
-	__clear_bit(offset, bitmap->bitmap);
-	if (--bitmap->nr_busy == 0) {
-		__set_bit(n, p->bitmap);	/* to please idr_remove() */
-		idr_remove(&ida->idr, idr_id);
-		free_bitmap(ida, bitmap);
-	}
-
-	return;
-
- err:
-	printk(KERN_WARNING
-	       "ida_remove called for id=%d which is not allocated.\n", id);
-}
-
-/**
- * ida_destroy - release all cached layers within an ida tree
- * @ida:		ida handle
- */
-void ida_destroy(struct ida *ida)
-{
-	idr_destroy(&ida->idr);
-	kfree(ida->free_bitmap);
-}
-EXPORT_SYMBOL(ida_destroy);
-
-/**
- * ida_alloc_range - get a new id.
- * @ida: the (initialized) ida.
- * @start: the minimum id (inclusive, < 0x8000000)
- * @end: the maximum id (exclusive, < 0x8000000 or 0)
- * @gfp_mask: memory allocation flags
- *
- * Allocates an id in the range start <= id < end, or returns -ENOSPC.
- * On memory allocation failure, returns -ENOMEM.
- *
- * Use ida_remove() to get rid of an id.
- */
-int ida_alloc_range(struct ida *ida, unsigned int start, unsigned int end,
-		   gfp_t gfp_mask)
-{
-	int ret, id;
-	unsigned int max;
-	unsigned long flags;
-
-	BUG_ON((int)start < 0);
-	BUG_ON((int)end < 0);
-
-	if (end == 0)
-		max = 0x80000000;
-	else {
-		BUG_ON(end < start);
-		max = end - 1;
-	}
-
-again:
-	if (!ida_pre_get(ida, gfp_mask))
-		return -ENOMEM;
-
-	spin_lock_irqsave(&simple_ida_lock, flags);
-	ret = ida_get_new_above(ida, start, &id);
-	if (!ret) {
-		if (id > max) {
-			__ida_remove(ida, id);
-			ret = -ENOSPC;
-		} else {
-			ret = id;
-		}
-	}
-	spin_unlock_irqrestore(&simple_ida_lock, flags);
-
-	if (unlikely(ret == -EAGAIN))
-		goto again;
-
-	return ret;
-}
-EXPORT_SYMBOL(ida_alloc_range);
-
-/**
- * ida_remove - remove an allocated id.
- * @ida: the (initialized) ida.
- * @id: the id returned by ida_alloc_range.
- */
-void ida_remove(struct ida *ida, unsigned int id)
-{
-	unsigned long flags;
-
-	BUG_ON((int)id < 0);
-	spin_lock_irqsave(&simple_ida_lock, flags);
-	__ida_remove(ida, id);
-	spin_unlock_irqrestore(&simple_ida_lock, flags);
-}
-EXPORT_SYMBOL(ida_remove);
-
-/**
- * ida_init - initialize ida handle
- * @ida:	ida handle
- *
- * This function is use to set up the handle (@ida) that you will pass
- * to the rest of the functions.
- */
-void ida_init(struct ida *ida)
-{
-	memset(ida, 0, sizeof(struct ida));
-	idr_init(&ida->idr);
-
-}
-EXPORT_SYMBOL(ida_init);
-- 
1.8.3.1

[PATCH 4/4] idr: Percpu ida

[Kent Overstreet]

Percpu frontend for allocating ids. With percpu allocation (that works),
it's impossible to guarantee it will always be possible to allocate all
nr_tags - typically, some will be stuck on a remote percpu freelist
where the current job can't get to them.

We do guarantee that it will always be possible to allocate at least
(nr_tags / 2) tags - this is done by keeping track of which and how many
cpus have tags on their percpu freelists. On allocation failure if
enough cpus have tags that there could potentially be (nr_tags / 2) tags
stuck on remote percpu freelists, we then pick a remote cpu at random to
steal from.

Note that the synchronization is _definitely_ tricky - we're using
xchg()/cmpxchg() on the percpu lists, to synchronize between
steal_tags().

The alternative would've been adding a spinlock to protect the percpu
freelists, but that would've required some different tricky code to
avoid deadlock because of the lock ordering.

Note that there's no cpu hotplug notifier - we don't care, because
steal_tags() will eventually get the down cpu's tags. We _could_ satisfy
more allocations if we had a notifier - but we'll still meet our
guarantees and it's absolutely not a correctness issue, so I don't think
it's worth the extra code.

Signed-off-by: Kent Overstreet <koverstreet@google.com>
Cc: Tejun Heo <tj@kernel.org>
Cc: Oleg Nesterov <oleg@redhat.com>
Cc: Christoph Lameter <cl@linux-foundation.org>
Cc: Ingo Molnar <mingo@redhat.com>
Cc: Andi Kleen <andi@firstfloor.org>
Cc: Jens Axboe <axboe@kernel.dk>
Cc: "Nicholas A. Bellinger" <nab@linux-iscsi.org>
---
 include/linux/idr.h |  48 ++++++++
 lib/idr.c           | 312 ++++++++++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 352 insertions(+), 8 deletions(-)

diff --git a/include/linux/idr.h b/include/linux/idr.h
index 9169e18..afaa071 100644
--- a/include/linux/idr.h
+++ b/include/linux/idr.h
@@ -16,6 +16,8 @@
 #include <linux/bitops.h>
 #include <linux/init.h>
 #include <linux/rcupdate.h>
+#include <linux/spinlock_types.h>
+#include <linux/wait.h>
 
 /* IDA */
 
@@ -97,6 +99,52 @@ static inline void ida_init(struct ida *ida)
 	ida_init_prealloc(ida, 0);
 }
 
+/* Percpu IDA/tag allocator */
+
+struct percpu_ida_cpu;
+
+struct percpu_ida {
+	/*
+	 * number of tags available to be allocated, as passed to
+	 * percpu_ida_init()
+	 */
+	unsigned			nr_tags;
+
+	struct percpu_ida_cpu __percpu	*tag_cpu;
+
+	/*
+	 * Bitmap of cpus that (may) have tags on their percpu freelists:
+	 * steal_tags() uses this to decide when to steal tags, and which cpus
+	 * to try stealing from.
+	 *
+	 * It's ok for a freelist to be empty when its bit is set - steal_tags()
+	 * will just keep looking - but the bitmap _must_ be set whenever a
+	 * percpu freelist does have tags.
+	 */
+	unsigned long			*cpus_have_tags;
+
+	struct {
+		/*
+		 * When we go to steal tags from another cpu (see steal_tags()),
+		 * we want to pick a cpu at random. Cycling through them every
+		 * time we steal is a bit easier and more or less equivalent:
+		 */
+		unsigned		cpu_last_stolen;
+
+		/* For sleeping on allocation failure */
+		wait_queue_head_t	wait;
+
+		/* Global freelist */
+		struct ida		ida;
+	} ____cacheline_aligned_in_smp;
+};
+
+int percpu_ida_alloc(struct percpu_ida *pool, gfp_t gfp);
+void percpu_ida_free(struct percpu_ida *pool, unsigned tag);
+
+void percpu_ida_destroy(struct percpu_ida *pool);
+int percpu_ida_init(struct percpu_ida *pool, unsigned long nr_tags);
+
 /* IDR */
 
 /*
diff --git a/lib/idr.c b/lib/idr.c
index 7d89587..401933f 100644
--- a/lib/idr.c
+++ b/lib/idr.c
@@ -28,17 +28,20 @@
  * with the slab allocator.
  */
 
-#ifndef TEST                        // to test in user space...
-#include <linux/slab.h>
-#include <linux/init.h>
-#include <linux/export.h>
-#endif
+#include <linux/bitmap.h>
+#include <linux/bitops.h>
+#include <linux/bug.h>
 #include <linux/err.h>
-#include <linux/string.h>
+#include <linux/export.h>
+#include <linux/hardirq.h>
 #include <linux/idr.h>
-#include <linux/spinlock.h>
+#include <linux/init.h>
+#include <linux/kernel.h>
 #include <linux/percpu.h>
-#include <linux/hardirq.h>
+#include <linux/sched.h>
+#include <linux/slab.h>
+#include <linux/string.h>
+#include <linux/spinlock.h>
 
 #define IDA_TREE_ARY		BITS_PER_LONG
 
@@ -527,6 +530,299 @@ int ida_init_prealloc(struct ida *ida, unsigned prealloc)
 }
 EXPORT_SYMBOL(ida_init_prealloc);
 
+/* Percpu IDA */
+
+/*
+ * Number of tags we move between the percpu freelist and the global freelist at
+ * a time
+ */
+#define TAG_CPU_BATCH_MOVE	32U
+
+/* Max size of percpu freelist, */
+#define TAG_CPU_SIZE		(TAG_CPU_BATCH_MOVE + TAG_CPU_BATCH_MOVE / 2)
+
+struct percpu_ida_cpu {
+	spinlock_t			lock;
+	unsigned			nr_free;
+	unsigned			freelist[];
+};
+
+/*
+ * Try to steal tags from a remote cpu's percpu freelist.
+ *
+ * We first check how many percpu freelists have tags - we don't steal tags
+ * unless enough percpu freelists have tags on them that it's possible more than
+ * half the total tags could be stuck on remote percpu freelists.
+ *
+ * Then we iterate through the cpus until we find some tags - we don't attempt
+ * to find the "best" cpu to steal from, to keep cacheline bouncing to a
+ * minimum.
+ *
+ * Returns true on success (our percpu freelist is no longer empty), false on
+ * failure.
+ */
+static inline bool steal_tags(struct percpu_ida *pool,
+			      struct percpu_ida_cpu *tags)
+{
+	unsigned cpus_have_tags, cpu = pool->cpu_last_stolen;
+	struct percpu_ida_cpu *remote;
+
+	for (cpus_have_tags = bitmap_weight(pool->cpus_have_tags, nr_cpu_ids);
+	     cpus_have_tags * TAG_CPU_SIZE > pool->nr_tags / 2;
+	     cpus_have_tags--) {
+		cpu = find_next_bit(pool->cpus_have_tags, nr_cpu_ids, cpu);
+
+		if (cpu == nr_cpu_ids)
+			cpu = find_first_bit(pool->cpus_have_tags, nr_cpu_ids);
+
+		if (cpu == nr_cpu_ids)
+			BUG();
+
+		pool->cpu_last_stolen = cpu;
+		remote = per_cpu_ptr(pool->tag_cpu, cpu);
+
+		clear_bit(cpu, pool->cpus_have_tags);
+
+		if (remote == tags)
+			continue;
+
+		spin_lock(&remote->lock);
+
+		if (remote->nr_free) {
+			memcpy(tags->freelist,
+			       remote->freelist,
+			       sizeof(unsigned) * remote->nr_free);
+
+			tags->nr_free = remote->nr_free;
+			remote->nr_free = 0;
+		}
+
+		spin_unlock(&remote->lock);
+
+		if (tags->nr_free)
+			return true;
+	}
+
+	return false;
+}
+
+static inline bool alloc_global_tags(struct percpu_ida *pool,
+				     struct percpu_ida_cpu *tags)
+{
+	int nr_free = __ida_alloc_range_multiple(&pool->ida, tags->freelist,
+						 TAG_CPU_BATCH_MOVE, 0,
+						 pool->nr_tags, GFP_NOWAIT,
+						 NULL);
+
+	if (nr_free <= 0)
+		return false;
+
+	tags->nr_free = nr_free;
+	return true;
+}
+
+static inline unsigned alloc_local_tag(struct percpu_ida *pool,
+				       struct percpu_ida_cpu *tags)
+{
+	int tag = -ENOSPC;
+
+	spin_lock(&tags->lock);
+	if (tags->nr_free)
+			tag = tags->freelist[--tags->nr_free];
+	spin_unlock(&tags->lock);
+
+	return tag;
+}
+
+/**
+ * percpu_ida_alloc - allocate a tag
+ * @pool: pool to allocate from
+ * @gfp: gfp flags
+ *
+ * Returns a tag - an integer in the range [0..nr_tags) (passed to
+ * tag_pool_init()), or otherwise -ENOSPC on allocation failure.
+ *
+ * Safe to be called from interrupt context (assuming it isn't passed
+ * __GFP_WAIT, of course).
+ *
+ * Will not fail if passed __GFP_WAIT.
+ */
+int percpu_ida_alloc(struct percpu_ida *pool, gfp_t gfp)
+{
+	DEFINE_WAIT(wait);
+	struct percpu_ida_cpu *tags;
+	unsigned long flags;
+	unsigned this_cpu;
+	int tag;
+
+	local_irq_save(flags);
+	this_cpu = smp_processor_id();
+	tags = per_cpu_ptr(pool->tag_cpu, this_cpu);
+
+	/* Fastpath */
+	tag = alloc_local_tag(pool, tags);
+	if (likely(tag >= 0)) {
+		local_irq_restore(flags);
+		return tag;
+	}
+
+	while (1) {
+		spin_lock(&pool->ida.lock);
+
+		/*
+		 * prepare_to_wait() must come before steal_tags(), in case
+		 * percpu_ida_free() on another cpu flips a bit in
+		 * cpus_have_tags
+		 */
+		prepare_to_wait(&pool->wait, &wait, TASK_UNINTERRUPTIBLE);
+
+		/*
+		 * alloc_global_tags(), steal_tags() return true iff we now have
+		 * tags on our percpu freelist
+		 */
+		if (tags->nr_free ||
+		    alloc_global_tags(pool, tags) ||
+		    steal_tags(pool, tags)) {
+			/*
+			 * Global lock held and irqs disabled, don't need percpu
+			 * lock
+			 */
+			tag = tags->freelist[--tags->nr_free];
+			if (tags->nr_free)
+				set_bit(this_cpu, pool->cpus_have_tags);
+		}
+
+		spin_unlock(&pool->ida.lock);
+		local_irq_restore(flags);
+
+		if (tag >= 0 || !(gfp & __GFP_WAIT))
+			break;
+
+		schedule();
+
+		local_irq_save(flags);
+		this_cpu = smp_processor_id();
+		tags = per_cpu_ptr(pool->tag_cpu, this_cpu);
+	}
+
+	finish_wait(&pool->wait, &wait);
+	return tag;
+}
+EXPORT_SYMBOL_GPL(percpu_ida_alloc);
+
+/**
+ * percpu_ida_free - free a tag
+ * @pool: pool @tag was allocated from
+ * @tag: a tag previously allocated with percpu_ida_alloc()
+ *
+ * Safe to be called from interrupt context.
+ */
+void percpu_ida_free(struct percpu_ida *pool, unsigned tag)
+{
+	struct percpu_ida_cpu *tags;
+	unsigned long flags;
+	unsigned nr_free, this_cpu;
+
+	BUG_ON(tag >= pool->nr_tags);
+
+	local_irq_save(flags);
+	this_cpu = smp_processor_id();
+	tags = per_cpu_ptr(pool->tag_cpu, this_cpu);
+
+	spin_lock(&tags->lock);
+	nr_free = tags->nr_free;
+	tags->freelist[tags->nr_free++] = tag;
+	spin_unlock(&tags->lock);
+
+	if (!nr_free) {
+		set_bit(this_cpu, pool->cpus_have_tags);
+		wake_up(&pool->wait);
+	}
+
+	if (nr_free + 1 == TAG_CPU_SIZE) {
+		spin_lock(&pool->ida.lock);
+
+		/*
+		 * Global lock held and irqs disabled, don't need percpu
+		 * lock
+		 */
+		while (tags->nr_free > TAG_CPU_SIZE - TAG_CPU_BATCH_MOVE)
+			__ida_remove(&pool->ida,
+				     tags->freelist[--tags->nr_free]);
+
+		wake_up(&pool->wait);
+		spin_unlock(&pool->ida.lock);
+	}
+
+	local_irq_restore(flags);
+}
+EXPORT_SYMBOL_GPL(percpu_ida_free);
+
+/**
+ * percpu_ida_destroy - release a tag pool's resources
+ * @pool: pool to free
+ *
+ * Frees the resources allocated by percpu_ida_init().
+ */
+void percpu_ida_destroy(struct percpu_ida *pool)
+{
+	free_percpu(pool->tag_cpu);
+	kfree(pool->cpus_have_tags);
+	ida_destroy(&pool->ida);
+}
+EXPORT_SYMBOL_GPL(percpu_ida_destroy);
+
+/**
+ * percpu_ida_init - initialize a percpu tag pool
+ * @pool: pool to initialize
+ * @nr_tags: number of tags that will be available for allocation
+ *
+ * Initializes @pool so that it can be used to allocate tags - integers in the
+ * range [0, nr_tags). Typically, they'll be used by driver code to refer to a
+ * preallocated array of tag structures.
+ *
+ * Allocation is percpu, but sharding is limited by nr_tags - for best
+ * performance, the workload should not span more cpus than nr_tags / 128.
+ */
+int percpu_ida_init(struct percpu_ida *pool, unsigned long nr_tags)
+{
+	unsigned cpu;
+
+	memset(pool, 0, sizeof(*pool));
+
+	init_waitqueue_head(&pool->wait);
+	pool->nr_tags = nr_tags;
+
+	/* Guard against overflow */
+	if (nr_tags > (unsigned) INT_MAX + 1) {
+		pr_err("tags.c: nr_tags too large\n");
+		return -EINVAL;
+	}
+
+	if (ida_init_prealloc(&pool->ida, nr_tags))
+		return -ENOMEM;
+
+	pool->cpus_have_tags = kzalloc(BITS_TO_LONGS(nr_cpu_ids) *
+				       sizeof(unsigned long), GFP_KERNEL);
+	if (!pool->cpus_have_tags)
+		goto err;
+
+	pool->tag_cpu = __alloc_percpu(sizeof(struct percpu_ida_cpu) +
+				       TAG_CPU_SIZE * sizeof(unsigned),
+				       sizeof(unsigned));
+	if (!pool->tag_cpu)
+		goto err;
+
+	for_each_possible_cpu(cpu)
+		spin_lock_init(&per_cpu_ptr(pool->tag_cpu, cpu)->lock);
+
+	return 0;
+err:
+	percpu_ida_destroy(pool);
+	return -ENOMEM;
+}
+EXPORT_SYMBOL_GPL(percpu_ida_init);
+
 /* IDR */
 
 #define MAX_IDR_SHIFT		(sizeof(int) * 8 - 1)
-- 
1.8.3.1

Re: [PATCH 4/4] idr: Percpu ida

[Christoph Lameter]

On Mon, 17 Jun 2013, Kent Overstreet wrote:

> +static inline unsigned alloc_local_tag(struct percpu_ida *pool,
> +				       struct percpu_ida_cpu *tags)
> +{
> +	int tag = -ENOSPC;
> +
> +	spin_lock(&tags->lock);
> +	if (tags->nr_free)
> +			tag = tags->freelist[--tags->nr_free];
> +	spin_unlock(&tags->lock);
> +
> +	return tag;
> +}

This could be made much faster by avoiding real atomics (coming with
spinlocks) and using per cpu atomics instead. Slub f.e. uses a single
linked per cpu list managed via this_cpu_cmpxchg.

In order to avoid locking completely the state of the percpu tag list must
fit into one word that can be handled via the local cmpxchg (or two with
the cmpxchg_double). May require some changes to the data structure.

Re: [PATCH 4/4] idr: Percpu ida

[Kent Overstreet]

On Tue, Jun 18, 2013 at 02:14:53PM +0000, Christoph Lameter wrote:
> On Mon, 17 Jun 2013, Kent Overstreet wrote:
> 
> > +static inline unsigned alloc_local_tag(struct percpu_ida *pool,
> > +				       struct percpu_ida_cpu *tags)
> > +{
> > +	int tag = -ENOSPC;
> > +
> > +	spin_lock(&tags->lock);
> > +	if (tags->nr_free)
> > +			tag = tags->freelist[--tags->nr_free];
> > +	spin_unlock(&tags->lock);
> > +
> > +	return tag;
> > +}
> 
> This could be made much faster by avoiding real atomics (coming with
> spinlocks) and using per cpu atomics instead. Slub f.e. uses a single
> linked per cpu list managed via this_cpu_cmpxchg.

Actually, we do need the atomic ops - they're protecting against a
different cpu grabbing our freelist with steal_tags().

The alternative (which Andrew Morton suggested and Jens implemented in
his fork of an earlier version of this code) would be to use IPIs, but
there are reasonable use cases (high ratio of cpus:nr_tags) where tag
stealing is going to be reasonably common so we don't want it to suck
too much. Plus, steal_tags() needs to try different cpu's freelists
until it finds one with tags, IPIs would make that loop... problematic.

I actually originally used cmpxchg(), but then later in review I
realized I had an ABA and couldn't figure out how to solve it, so that's
when I switched to spinlocks (which Andrew Morton preferred anyways).

But I just realized a few minutes ago I've seen your solution to ABA
with a stack, in the slub code - double word cmpxchg() with a
transaction id :) I may try that again just to see how the code looks,
but I'm not sure the difference in performance will be big enough to
matter, give that this lock should basically never be contended.