Re: Proposal: block-based vector allocator

[Dmitry Antipov <dmantipov@yandex.ru>]

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

On 05/18/2012 09:40 PM, Stefan Monnier wrote:

> Here are some questions and comments about your code:
>
>> +#define VECTOR_BLOCK_SIZE (4096)
>
> No need to put parens, right?

OK

>> +    roundup_size = 8
>
> This deserves a comment explaining the choice of 8 and describing what
> this constant is used for.

OK

>> +/* 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))
>
> Why do we care about malloc overhead?

IIUC, if/when we will have mmap()ed Lisp_Objects someday, this should help
malloc() to do direct mmap()/munmap() of page-size vector blocks.

>> +/* Free lists for all possible block-allocated vectors.  */
>> +
>> +#define VECTOR_FREE_LISTS ((VECTOR_BLOCK_BYTES / roundup_size) + 1)
>
> Some comment here or earlier should make it clear that we have one free
> list per vector size.  Maybe just put this declaration together with the
> one for "vector_free_lists", after all they go together.
>
> And this macro should be called something like
> VECTOR_MAX_FREE_LIST_INDEX; its current names makes it sound like it
> returns the free lists (or a pointer to them).

OK

>> +/* 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)))
>
> The name doesn't sound right, it's not a real size, so it should be
> VECTOR_FREE_LIST_FLAG.

OK

> Also, do we really have to put this info in the `size' field?  E.g. for
> cons-cells we put Vdead in the `car' slot.  Of course having it in
> `size' is not terrible, but `size' is pretty complicated already.

May be. I'll try to consider some alternatives.

> BTW, if needed we could add a special case so there is only 1 vector of
> size 0 and (make-vector 0 ?) always returns that same vector (so it
> doesn't have to come from a vector_block).

Should we allocate it from pure space?

>> +/* Current vector block.  */
>> +static struct vector_block *vector_block;
>
> Do we actually want a "current vector block"?
> Here's what I mean by that: how often are we going to allocate from this
> "current vector block" as opposed to allocating from one of the
> free lists?
> It would be simpler, when we allocate a new vector block, to just carve
> out the vector we need from it and put all the rest directly on the
> appropriate free list, so there's never such a thing as a "current
> vector block".  That also should lets us get rid of the "bump" field.

Argh, yes.

>> +  /* Next, check free lists contains larger vectors.  */
>> +  for (index = (nbytes + sizeof (struct Lisp_Vector)) / roundup_size;
>
> Add a comment explaining that the "+ sizeof (struct Lisp_Vector)"
> eliminates the case where the "excess bytes" aren't sufficient to make
> them into a valid vector.

OK

>> +  while (block)
>> +    {
>> +      struct Lisp_Vector *free_vectors[VECTORS_PER_BLOCK_MAX];
>> +      int index, used;
>
> You should be able to avoid this two-step "put them in free_vectors,
> then loop over free_vectors to put them on the free lists": just at the
> end of the coalescing loop, if we bump into the end of the vector_block
> check if we're also the first vector in the block, and if so free the
> whole (otherwise push the vector on the free list).

OK

>> +  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));
>
> Maybe we should only allocate from vector blocks vectors that are
> smaller than (say) VECTOR_BLOCK_BYTES/2.  The reason is that allocating
> from a vector block incurs an overhead (memory overhead of "roundup (3 *
> sizeof (void *), roundup_size)", and CPU overhead of setting up the
> block, allocating the vector out of it, scanning the block, ...).
> This is shared among all the vectors in the block, so for smallish
> vectors it's worth the trouble (since we save other overheads) but if
> the vector is large, there won't be many vectors with which to share
> that overhead.

Hm... didn't checked that, most probably a very negligible difference.

>> +	      /* 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);
>> +		}
>> +	    }
>
> This deserves a FIXME mentioning that this could be a significant
> overhead.  I don't see how we can get rid of this overhead without using
> a different allocation technique, but it's always good to make this
> choice explicit.

OK

As for the roundup() check, it looks like GNU/Linux and *BSD systems
both has it in sys/param.h, but (Open)Solaris uses sys/sysmacros.h.
So I'm trying to check for the most common case and then fall back
to simple default.

Dmitry






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

=== modified file 'configure.in'
--- configure.in	2012-05-21 07:30:23 +0000
+++ configure.in	2012-05-21 10:39:57 +0000
@@ -1773,6 +1773,15 @@
 
 AC_CHECK_LIB(pthreads, cma_open)
 
+dnl Check whether roundup is available.
+if test "$ac_cv_header_sys_param_h"; then
+  AC_MSG_CHECKING([for roundup in sys/param.h])
+  AC_LINK_IFELSE([AC_LANG_PROGRAM([[#include <sys/param.h>]], [[roundup (5, 8);]])],
+    [AC_MSG_RESULT([yes])
+     AC_DEFINE(HAVE_ROUNDUP, 1, [Define to 1 if `roundup' is declared in <sys/param.h>.])],
+    [AC_MSG_RESULT([no])])
+fi
+
 ## Note: when using cpp in s/aix4.2.h, this definition depended on
 ## HAVE_LIBPTHREADS.  That was not defined earlier in configure when
 ## the system file was sourced.  Hence the value of LIBS_SYSTEM

=== modified file 'src/alloc.c'
--- src/alloc.c	2012-04-23 05:44:49 +0000
+++ src/alloc.c	2012-05-21 12:10:23 +0000
@@ -25,6 +25,12 @@
 
 #include <signal.h>
 
+#if defined HAVE_ROUNDUP
+#include <sys/param.h>
+#else
+#define roundup(x, y) ((((x) + ((y) - 1)) / (y)) * (y))
+#endif
+
 #ifdef HAVE_PTHREAD
 #include <pthread.h>
 #endif
@@ -2924,17 +2930,304 @@
 			   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),
+    /* On a 32-bit system, rounding up vector size (in bytes) up
+       to mult-of-8 helps to maintain mult-of-8 alignment.  */
+    roundup_size = 8
   };
 
+/* One pointer is malloc overhead, other is NEXT field of struct
+   vector_block.  Rounding helps to maintain alignment constraints.  */
+
+#define VECTOR_BLOCK_BYTES \
+  (VECTOR_BLOCK_SIZE - roundup (2 * 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))
+
+/* We maintain one free list for each possible block-allocated
+   vector size, and this is now much of the free lists we have.  */
+
+#define VECTOR_MAX_FREE_LIST_INDEX ((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_FLAG \
+  (((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_FLAG | (nbytes); \
+  eassert ((nbytes) % roundup_size == 0); \
+  (index) = (nbytes) / roundup_size; \
+  eassert ((index) < VECTOR_MAX_FREE_LIST_INDEX); \
+  (v)->header.next.vector = vector_free_lists[(index)]; \
+  vector_free_lists[(index)] = (v); } while (0)
+
+struct vector_block
+{
+  unsigned char data[VECTOR_BLOCK_BYTES];
+  struct vector_block *next;
+};
+
+/* Chain of vector blocks.  */
+
+static struct vector_block *vector_blocks;
+
+/* 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_MAX_FREE_LIST_INDEX];
+
+/* Singly-linked list of large vectors.  */
+
+static struct Lisp_Vector *large_vectors;
+
+/* The only vector with 0 slots, allocated from pure space.  */
+
+static struct Lisp_Vector *zero_vector;
+
+/* 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 (sizeof (struct vector_block));
+  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->next = vector_blocks;
+  vector_blocks = block;
+  return block;
+}
+
+/* Called once to initialize vector allocation.  */
+
+static void
+init_vectors (void)
+{
+  int i;
+
+  large_vectors = NULL;
+
+  zero_vector = (struct Lisp_Vector *)
+    pure_alloc (header_size, Lisp_Vectorlike);
+  zero_vector->header.size = 0;
+
+  for (i = 0; i < VECTOR_MAX_FREE_LIST_INDEX; i++)
+    vector_free_lists[i] = NULL;
+}
+
+/* Allocate vector from a vector block.  */
+
+static struct Lisp_Vector *
+allocate_vector_from_block (size_t nbytes)
+{
+  struct Lisp_Vector *vector, *rest;
+  struct vector_block *block;
+  size_t index, restbytes;
+
+  /* No vectors with 0 slots for Lisp_Objects here.  */
+  eassert (nbytes > sizeof (struct vectorlike_header) &&
+	   nbytes <= VECTOR_BLOCK_BYTES);
+  eassert (nbytes % roundup_size == 0);
+
+  /* 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.  Since we will
+     split the result, we should have remaining space large enough
+     to use for one-slot vector at least.  */
+  for (index = (nbytes + sizeof (struct Lisp_Vector)) / roundup_size;
+       index < VECTOR_MAX_FREE_LIST_INDEX; index++)
+    if (vector_free_lists[index])
+      {
+	/* 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;
+      }
+
+  /* Finally, need a new vector block.  */
+  block = allocate_vector_block ();
+
+  /* New vector will be at the beginning of this block.  */
+  vector = (struct Lisp_Vector *) block->data;
+  vector->header.next.nbytes = nbytes;
+
+  /* The rest of this block is set up on a free list.  */
+  rest = ADVANCE (vector, nbytes);
+  restbytes = VECTOR_BLOCK_BYTES - nbytes;
+  index = restbytes / roundup_size;
+  SETUP_ON_FREE_LIST (rest, restbytes, index);
+
+  return vector;
+ }
+
+/* Return amount of Lisp_Objects which 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_blocks, *bprev = NULL, *bnext;
+  struct Lisp_Vector *vector, *prev, *next;
+  int i;
+
+  total_vector_size = 0;
+  for (i = 0; i < VECTOR_MAX_FREE_LIST_INDEX; i++)
+    vector_free_lists[i] = NULL;
+
+  /* Looking through vector blocks.  */
+
+  while (block)
+    {
+      struct Lisp_Vector *free_vectors[VECTORS_PER_BLOCK_MAX];
+      int index, free_block_p;
+
+      for (index = 0, vector = (struct Lisp_Vector *) block->data;
+	   (unsigned char *) vector < block->data + VECTOR_BLOCK_BYTES;
+	   vector = next)
+	{
+	  free_block_p = 0;
+
+	  if (VECTOR_MARKED_P (vector))
+	    {
+	      VECTOR_UNMARK (vector);
+	      total_vector_size += VECTOR_SIZE (vector);
+	      next = ADVANCE (vector, vector->header.next.nbytes);
+	    }
+	  else
+	    {
+	      EMACS_INT nbytes;
+
+	      if ((vector->header.size & VECTOR_FREE_LIST_FLAG) == 
+		  VECTOR_FREE_LIST_FLAG)
+		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->data + VECTOR_BLOCK_BYTES)
+		{
+		  if (VECTOR_MARKED_P (next))
+		    break;
+		  if ((next->header.size & VECTOR_FREE_LIST_FLAG) == 
+		      VECTOR_FREE_LIST_FLAG)
+		    nbytes = next->header.size & (VECTOR_BLOCK_SIZE - 1);
+		  else
+		    nbytes = next->header.next.nbytes;
+		  vector->header.next.nbytes += nbytes;
+		  next = ADVANCE (next, nbytes);
+		}
+	      
+	      eassert (index < VECTORS_PER_BLOCK_MAX);
+	      eassert (vector->header.next.nbytes % roundup_size == 0);
+
+	      if (vector == (struct Lisp_Vector *) block->data &&
+		  (unsigned char *) next >= block->data + VECTOR_BLOCK_BYTES)
+		/* This block should be freed because all of it's
+		   space was coalesced into the only free vector.  */
+		free_block_p = 1;
+	      else
+		SETUP_ON_FREE_LIST (vector, vector->header.next.nbytes, i);
+	    }
+	}
+
+      if (free_block_p)
+	{
+	  if (bprev)
+	    bprev->next = block->next;
+	  else
+	    vector_blocks = block->next;
+	  bnext = block->next;
+#if GC_MARK_STACK && !defined GC_MALLOC_CHECK
+	  mem_delete (mem_find (block->data));
+#endif
+	  xfree (block);
+	  block = bnext;
+	}
+      else
+	bprev = block, block = block->next;
+    }
+
+  /* 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.  */
 
@@ -2956,8 +3249,20 @@
   /* This gets triggered by code which I haven't bothered to fix.  --Stef  */
   /* eassert (!handling_signal); */
 
+  if (len == 0)
+    return zero_vector;
+
   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 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.  */
@@ -2967,9 +3272,6 @@
   consing_since_gc += nbytes;
   vector_cells_consed += len;
 
-  p->header.next.vector = all_vectors;
-  all_vectors = p;
-
   MALLOC_UNBLOCK_INPUT;
 
   return p;
@@ -4068,7 +4370,39 @@
 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;
+	  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.  FIXME: check whether
+	     some allocation patterns (probably a lot of short vectors)
+	     may cause a substantial overhead of this loop.  */
+	  while (vector <= (struct Lisp_Vector *) p)
+	    {
+	      if ((vector->header.size & VECTOR_FREE_LIST_FLAG)
+		  == VECTOR_FREE_LIST_FLAG)
+		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;
 }
 
 
@@ -6237,33 +6571,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)
@@ -6400,7 +6708,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 */
@@ -6413,6 +6720,7 @@
   init_marker ();
   init_float ();
   init_intervals ();
+  init_vectors ();
   init_weak_hash_tables ();
 
 #ifdef REL_ALLOC

=== modified file 'src/lisp.h'
--- src/lisp.h	2012-05-09 17:51:30 +0000
+++ src/lisp.h	2012-05-21 07:51:40 +0000
@@ -899,12 +899,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;