Re: Proposal: block-based vector allocator

[Dmitry Antipov <dmantipov@yandex.ru>]

[-- Attachment #1: Type: text/plain, Size: 2859 bytes --]

On 12/10/2011 01:04 AM, Stefan Monnier wrote:

> BTW, your code seems to compute the free list index by dividing by
> word_size (= 4 on 32bit systems) rather than by 8, so half of the slots
> will always be unused, IIUC.

You're correct, and this is fixed.

> Let us how it turns out.

It needs to be run again since I introduced some fixes, including
those that suggested by you.

> If it's an arbitrary choice, it should not be called "page_size" but
> "vector_block_size" or some such.  Otherwise, we should fetch its value
> from the OS.

Better name is OK (old one was a trace of mmap()-related experiments :-).

> Where do you make use of this alignment?

Not needed any more.

> This test looks dangerous (we if we allocate a large vector of this
> size?).

It's impossible because VECTOR_BLOCK_BYTES is 'exact-fit' for the largest
possible block-allocated vector. That's why I'm using xmalloc/mem_insert
and not lisp_malloc: VECTOR_BLOCK_SIZE bytes are allocated for the block,
but only block->data..block->data + VECTOR_BLOCK_BYTES range is marked as
vector memory within mem_node tree. So, any MEM_TYPE_VECTORLIKE mem_node
with VECTOR_BLOCK_BYTES size memory range is guaranteed to match the
vector block, and this vector block may contain the only vector of
VECTOR_BLOCK_BYTES bytes size.

> Hmm... I didn't think of it, but this is a disadvantage of
> non-segregated blocks: during conservative scanning, you need to scan
> the vector block to determine if we really have a live_vector_p whereas
> segregated blocks would just divide the offset by the block's
> vector size.

Yes, but this linear scan may be optimized:
  1) scan is required only if examined pointer is less than
     block bump allocation pointer;
  2) scan should not look at addresses beyond examined pointer.

> For small enough vector blocks, it's probably not a big deal, but for
> 10-slot vectors on 32bit systems with 4KB blocks, that's about
> 100 iterations.

This is a worst-case scenario when it's needed to scan up to the end of vector
block, and the block is full of small vectors. Although the most vectors
are small, small vectors usually allocated and died (becomes unreachable)
in groups, and such a group of adjacent died vectors is a subject for coalescing,
which reduces the number of vectors in the block. For the byte-compile
benchmark, this loop does ~12 iterations on the average (for 64-bit code; expect
larger value for 32-bit since average vector size is ~2x smaller).

BTW, what do you think about hooking buffer allocation into common vector
allocation code? I found it controversial (buffers are mid-sized vector-like
objects, so it might be reasonable to generalize buffer allocation with
other vectors; on the other side, buffers needs special treatment for
liveness detection and special processing during sweep), but want to try
anyway.

Dmitry

[-- Attachment #2: vector_alloc.patch --]
[-- Type: text/plain, Size: 13305 bytes --]

=== modified file 'src/alloc.c'
--- src/alloc.c	2011-11-20 03:07:02 +0000
+++ src/alloc.c	2011-12-11 10:15:11 +0000
@@ -21,7 +21,7 @@
 #include <stdio.h>
 #include <limits.h>		/* For CHAR_BIT.  */
 #include <setjmp.h>
-
+#include <sys/param.h>		/* For roundup.  */
 #include <signal.h>
 
 #ifdef HAVE_PTHREAD
@@ -2896,17 +2896,309 @@
 			   Vector Allocation
  ***********************************************************************/
 
-/* Singly-linked list of all vectors.  */
-
-static struct Lisp_Vector *all_vectors;
+#define VECTOR_BLOCK_SIZE (4096)
 
 /* Handy constants for vectorlike objects.  */
 enum
   {
     header_size = offsetof (struct Lisp_Vector, contents),
-    word_size = sizeof (Lisp_Object)
+    word_size = sizeof (Lisp_Object),
+    roundup_size = 8
   };
 
+/* One pointer is malloc overhead, others are BUMP and NEXT fields of struct
+   vector_block.  Rounding helps to maintain alignment constraints.  */
+
+#define VECTOR_BLOCK_BYTES \
+  (VECTOR_BLOCK_SIZE - roundup (3 * sizeof (void *), roundup_size))
+
+/* Maximum amount of vectors allocated from the vector block.  */
+
+#define VECTORS_PER_BLOCK_MAX \
+  (VECTOR_BLOCK_BYTES / sizeof (struct vectorlike_header))
+
+/* Free lists for all possible block-allocated vectors.  */
+
+#define VECTOR_FREE_LISTS ((VECTOR_BLOCK_BYTES / roundup_size) + 1)
+
+/* When the vector is on a free list, vectorlike_header.SIZE is set to
+   this special value ORed with vector's memory footprint size.  */
+
+#define VECTOR_FREE_LIST_SIZE \
+  (((EMACS_INT) ~0) & ~(ARRAY_MARK_FLAG | PSEUDOVECTOR_FLAG | \
+			(VECTOR_BLOCK_SIZE - 1)))
+
+/* Common shortcut to advance vector pointer over a block data.  */
+
+#define ADVANCE(v,nbytes) \
+  (struct Lisp_Vector *) ((unsigned char *) (v) + (nbytes))
+
+/* Common shortcut to setup vector on a free list.  */
+
+#define SETUP_ON_FREE_LIST(v,nbytes,index) do { \
+  (v)->header.size = VECTOR_FREE_LIST_SIZE | (nbytes); \
+  eassert ((nbytes) % roundup_size == 0); \
+  (index) = (nbytes) / roundup_size; \
+  eassert ((index) < VECTOR_FREE_LISTS); \
+  (v)->header.next.vector = vector_free_lists[(index)]; \
+  vector_free_lists[(index)] = (v); } while (0)
+
+struct vector_block
+{
+  unsigned char data[VECTOR_BLOCK_BYTES];
+  unsigned char *bump;
+  struct vector_block *next;
+};
+
+/* Current vector block.  */
+
+static struct vector_block *vector_block;
+
+/* Vector free lists, where NTH item points to a chain
+   of free vectors of NTH * ROUNDUP_SIZE bytes.  */
+
+static struct Lisp_Vector *vector_free_lists[VECTOR_FREE_LISTS];
+
+/* Singly-linked list of large vectors.  */
+
+static struct Lisp_Vector *large_vectors;
+
+/* Get a new vector block.  */
+
+static struct vector_block *
+allocate_vector_block (void)
+{
+  struct vector_block *block;
+
+#ifdef DOUG_LEA_MALLOC
+  mallopt (M_MMAP_MAX, 0);
+#endif
+
+  block = xmalloc (VECTOR_BLOCK_SIZE);
+  if (!block)
+    memory_full (VECTOR_BLOCK_SIZE);
+
+#ifdef DOUG_LEA_MALLOC
+  mallopt (M_MMAP_MAX, MMAP_MAX_AREAS);
+#endif
+
+#if GC_MARK_STACK && !defined GC_MALLOC_CHECK
+  mem_insert (block->data, block->data + VECTOR_BLOCK_BYTES,
+	      MEM_TYPE_VECTORLIKE);
+#endif
+
+  block->bump = block->data;
+  block->next = vector_block;
+  vector_block = block;
+  return block;
+}
+
+/* Called once to initialize vector allocation.  */
+
+static void
+init_vectors (void)
+{
+  int i;
+
+  large_vectors = NULL;
+  vector_block = allocate_vector_block ();
+
+  for (i = 0; i < VECTOR_FREE_LISTS; i++)
+    vector_free_lists[i] = NULL;
+}
+
+/* Allocate vector from the vector block.  */
+
+static struct Lisp_Vector *
+allocate_vector_from_block (size_t nbytes)
+{
+  struct Lisp_Vector *vector;
+  struct vector_block *block;
+  size_t index, restbytes;
+
+  /* Vector with 0 slots for Lisp_Objects is allowed.  */
+  eassert (nbytes >= sizeof (struct vectorlike_header) &&
+	   nbytes <= VECTOR_BLOCK_BYTES);
+  eassert (nbytes % roundup_size == 0);
+  eassert (vector_block != NULL);
+
+  /* First, try to allocate from a free list
+     contains vectors of the requested size.  */
+  index = nbytes / roundup_size;
+  if (vector_free_lists[index])
+    {
+      vector = vector_free_lists[index];
+      vector_free_lists[index] = vector->header.next.vector;
+      vector->header.next.nbytes = nbytes;
+      return vector;
+    }
+
+  /* Next, check free lists contains larger vectors.  */
+  for (index = (nbytes + sizeof (struct Lisp_Vector)) / roundup_size;
+       index < VECTOR_FREE_LISTS; index++)
+    if (vector_free_lists[index])
+      {
+	struct Lisp_Vector *rest;
+
+	/* This vector is larger than it was requested.  */
+	vector = vector_free_lists[index];
+	vector_free_lists[index] = vector->header.next.vector;
+	vector->header.next.nbytes = nbytes;
+
+	/* Excessive bytes are used for the smaller vector,
+	   which should be set on an appropriate free list.  */
+	restbytes = index * roundup_size - nbytes;
+	eassert (restbytes % roundup_size == 0);
+	rest = ADVANCE (vector, nbytes);
+	SETUP_ON_FREE_LIST (rest, restbytes, index);
+	return vector;
+      }
+
+  /* Next, try to allocate from current vector block.  */
+  restbytes = vector_block->data + VECTOR_BLOCK_BYTES - vector_block->bump;
+  if (nbytes > restbytes)
+    {
+      /* Current block's free space isn't enough...  */
+      if (restbytes >= sizeof (struct Lisp_Vector))
+	{
+	  /* ...but it's enough to allocate smaller vector,
+	     which should be set on an appropriate free list.  */
+	  eassert (restbytes % roundup_size == 0);
+	  vector = (struct Lisp_Vector *) vector_block->bump;
+	  vector_block->bump += restbytes;
+	  SETUP_ON_FREE_LIST (vector, restbytes, index);
+	}
+      /* Anyway, need a new vector block.  */
+      block = allocate_vector_block ();
+    }
+  else
+    /* Allocate from current vector block.  */
+    block = vector_block;
+
+  /* Bump pointer allocation from current vector block.  */
+  vector = (struct Lisp_Vector *) block->bump;
+  block->bump += nbytes;
+  vector->header.next.nbytes = nbytes;
+  return vector;
+}
+
+/* Return amount of Lisp_Objects can be stored in that vector.  */
+
+#define VECTOR_SIZE(v) ((v)->header.size & PSEUDOVECTOR_FLAG ? \
+  (PSEUDOVECTOR_SIZE_MASK & (v)->header.size) : (v)->header.size)
+
+/* Reclaim space used by unmarked vectors.  */
+
+static void
+sweep_vectors (void)
+{
+  struct vector_block *block = vector_block, *bprev = NULL, *bnext;
+  struct Lisp_Vector *vector, *prev, *next;
+  int i;
+
+  total_vector_size = 0;
+  for (i = 0; i < VECTOR_FREE_LISTS; i++)
+    vector_free_lists[i] = NULL;
+
+  /* Looking through vector blocks.  */
+
+  while (block)
+    {
+      struct Lisp_Vector *free_vectors[VECTORS_PER_BLOCK_MAX];
+      int index, used;
+
+      for (index = 0, used = 0, vector = (struct Lisp_Vector *) block->data;
+	   (unsigned char *) vector < block->bump; vector = next)
+	{
+	  if (VECTOR_MARKED_P (vector))
+	    {
+	      VECTOR_UNMARK (vector), used = 1;
+	      total_vector_size += VECTOR_SIZE (vector);
+	      next = ADVANCE (vector, vector->header.next.nbytes);
+	    }
+	  else
+	    {
+	      EMACS_INT nbytes;
+
+	      if ((vector->header.size & VECTOR_FREE_LIST_SIZE) == 
+		  VECTOR_FREE_LIST_SIZE)
+		vector->header.next.nbytes =
+		  vector->header.size & (VECTOR_BLOCK_SIZE - 1);
+	      
+	      next = ADVANCE (vector, vector->header.next.nbytes);
+
+	      /* While NEXT is not marked, try to coalesce with VECTOR,
+		 thus making VECTOR of the largest possible size.  */
+
+	      while ((unsigned char *) next < block->bump)
+		{
+		  if (VECTOR_MARKED_P (next))
+		    break;
+		  if ((next->header.size & VECTOR_FREE_LIST_SIZE) == 
+		      VECTOR_FREE_LIST_SIZE)
+		    nbytes = next->header.size & (VECTOR_BLOCK_SIZE - 1);
+		  else
+		    nbytes = next->header.next.nbytes;
+		  vector->header.next.nbytes += nbytes;
+		  next = ADVANCE (next, nbytes);
+		}
+	      
+	      /* Collect a pointer to resulting vector.  */
+	      eassert (index < VECTORS_PER_BLOCK_MAX);
+	      free_vectors[index++] = vector;
+	    }
+	}
+
+      if (used)
+	{
+	  /* Setup collected vectors on a free lists.  */
+	  while (--index >= 0)
+	    {
+	      vector = free_vectors[index];
+	      eassert (vector->header.next.nbytes % roundup_size == 0);
+	      SETUP_ON_FREE_LIST (vector, vector->header.next.nbytes, i);
+	    }
+	  bprev = block, block = block->next;
+	}
+      else
+	{
+	  /* Nothing used in this block.  */
+	  if (bprev)
+	    bprev->next = block->next;
+	  else
+	    vector_block = block->next;
+	  bnext = block->next;
+#if GC_MARK_STACK && !defined GC_MALLOC_CHECK
+	  mem_delete (mem_find (block->data));
+#endif
+	  xfree (block);
+	  block = bnext;
+	}
+    }
+
+  /* Sweep large vectors.  */
+
+  vector = large_vectors, prev = NULL;
+
+  while (vector)
+    if (VECTOR_MARKED_P (vector))
+      {
+	VECTOR_UNMARK (vector);
+	total_vector_size += VECTOR_SIZE (vector);
+	prev = vector, vector = vector->header.next.vector;
+      }
+    else
+      {
+	if (prev)
+	  prev->header.next = vector->header.next;
+	else
+	  large_vectors = vector->header.next.vector;
+	next = vector->header.next.vector;
+	lisp_free (vector);
+	vector = next;
+      }
+}
+
 /* Value is a pointer to a newly allocated Lisp_Vector structure
    with room for LEN Lisp_Objects.  */
 
@@ -2929,7 +3221,15 @@
   /* eassert (!handling_signal); */
 
   nbytes = header_size + len * word_size;
-  p = (struct Lisp_Vector *) lisp_malloc (nbytes, MEM_TYPE_VECTORLIKE);
+  if (nbytes > VECTOR_BLOCK_BYTES)
+    {
+      p = (struct Lisp_Vector *) lisp_malloc (nbytes, MEM_TYPE_VECTORLIKE);
+      p->header.next.vector = large_vectors;
+      large_vectors = p;
+    }
+  else
+    /* Rounding is for 32-bit code to preserve alignment.  */
+    p = allocate_vector_from_block (roundup (nbytes, roundup_size));
 
 #ifdef DOUG_LEA_MALLOC
   /* Back to a reasonable maximum of mmap'ed areas.  */
@@ -2939,9 +3239,6 @@
   consing_since_gc += nbytes;
   vector_cells_consed += len;
 
-  p->header.next.vector = all_vectors;
-  all_vectors = p;
-
   MALLOC_UNBLOCK_INPUT;
 
   return p;
@@ -4016,7 +4313,41 @@
 static inline int
 live_vector_p (struct mem_node *m, void *p)
 {
-  return (p == m->start && m->type == MEM_TYPE_VECTORLIKE);
+  if (m->type == MEM_TYPE_VECTORLIKE)
+    {
+      if (m->end - m->start == VECTOR_BLOCK_BYTES)
+	{
+	  /* This memory node corresponds to a vector block.  */
+	  struct vector_block *block = (struct vector_block *) m->start;
+
+	  if (p < (void *) block->bump)
+	    {
+	      struct Lisp_Vector *vector = (struct Lisp_Vector *) block->data;
+
+	      /* P is in the block's allocation range. Scan the block
+		 up to P and see whether P points to the start of some
+		 vector which is not on a free list.  */
+	      while (vector <= (struct Lisp_Vector *) p)
+		{
+		  if ((vector->header.size & VECTOR_FREE_LIST_SIZE)
+		      == VECTOR_FREE_LIST_SIZE)
+		    vector = ADVANCE (vector, (vector->header.size & 
+					       (VECTOR_BLOCK_SIZE - 1)));
+		  else if (vector == p)
+		    return 1;
+		  else
+		    vector = ADVANCE (vector, vector->header.next.nbytes);
+		}
+	    }
+	}
+      else
+	{
+	  if (p == m->start)
+	    /* This memory node corresponds to a large vector.  */
+	    return 1;
+	}
+    }
+  return 0;
 }
 
 
@@ -6169,33 +6500,7 @@
 	}
   }
 
-  /* Free all unmarked vectors */
-  {
-    register struct Lisp_Vector *vector = all_vectors, *prev = 0, *next;
-    total_vector_size = 0;
-
-    while (vector)
-      if (!VECTOR_MARKED_P (vector))
-	{
-	  if (prev)
-	    prev->header.next = vector->header.next;
-	  else
-	    all_vectors = vector->header.next.vector;
-	  next = vector->header.next.vector;
-	  lisp_free (vector);
-	  vector = next;
-
-	}
-      else
-	{
-	  VECTOR_UNMARK (vector);
-	  if (vector->header.size & PSEUDOVECTOR_FLAG)
-	    total_vector_size += PSEUDOVECTOR_SIZE_MASK & vector->header.size;
-	  else
-	    total_vector_size += vector->header.size;
-	  prev = vector, vector = vector->header.next.vector;
-	}
-  }
+  sweep_vectors ();
 
 #ifdef GC_CHECK_STRING_BYTES
   if (!noninteractive)
@@ -6331,7 +6636,6 @@
   Vdead = make_pure_string ("DEAD", 4, 4, 0);
 #endif
 
-  all_vectors = 0;
   ignore_warnings = 1;
 #ifdef DOUG_LEA_MALLOC
   mallopt (M_TRIM_THRESHOLD, 128*1024); /* trim threshold */
@@ -6344,6 +6648,7 @@
   init_marker ();
   init_float ();
   init_intervals ();
+  init_vectors ();
   init_weak_hash_tables ();
 
 #ifdef REL_ALLOC

=== modified file 'src/lisp.h'
--- src/lisp.h	2011-12-05 00:15:15 +0000
+++ src/lisp.h	2011-12-09 11:46:15 +0000
@@ -868,12 +868,15 @@
 struct vectorlike_header
   {
     EMACS_INT size;
-
-    /* Pointer to the next vector-like object.  It is generally a buffer or a
+    /* When the vector is allocated from a vector block, NBYTES is used
+       if the vector is not on a free list, and VECTOR is used otherwise.
+       For large vector-like objects, BUFFER or VECTOR is used as a pointer
+       to the next vector-like object.  It is generally a buffer or a 
        Lisp_Vector alias, so for convenience it is a union instead of a
        pointer: this way, one can write P->next.vector instead of ((struct
        Lisp_Vector *) P->next).  */
     union {
+      EMACS_INT nbytes;
       struct buffer *buffer;
       struct Lisp_Vector *vector;
     } next;