redis 源码和情景分析

-Thread 1-
y.store (20, memory_order_relaxed)
x.store (10, memory_order_relaxed)

-Thread 2-
if (x.load (memory_order_relaxed) == 10)
  {
  assert (y.load(memory_order_relaxed) == 20) /* assert A */
  y.store (10, memory_order_relaxed)
  }

-Thread 3-
if (y.load (memory_order_relaxed) == 10)
  assert (x.load(memory_order_relaxed) == 10) /* assert B */

-Thread 1-
y.store (20, memory_order_release);

-Thread 2-
x.store (10, memory_order_release);

-Thread 3-
assert (y.load (memory_order_acquire) == 20 && x.load (memory_order_acquire) == 0)

-Thread 4-
assert (y.load (memory_order_acquire) == 0 && x.load (memory_order_acquire) == 10)

atomicIncr(var,count) -- Increment the atomic counter
atomicGetIncr(var,oldvalue_var,count) -- Get and increment the atomic counter
atomicDecr(var,count) -- Decrement the atomic counter
atomicGet(var,dstvar) -- Fetch the atomic counter value
atomicSet(var,value)  -- Set the atomic counter value

long myvar;
pthread_mutex_t myvar_mutex;
atomicSet(myvar,12345);

#define __ATOMIC_VAR_FORCE_SYNC_MACROS

#if !defined(__ATOMIC_VAR_FORCE_SYNC_MACROS) && defined(__ATOMIC_RELAXED) && !defined(__sun) && (!defined(__clang__) || !defined(__APPLE__) || __apple_build_version__ > 4210057)
  // use __atomic ...
#elif defined(HAVE_ATOMIC)
  // use __sync ...
#else
  // use pthread mutex ...
#endif

#if (__i386 || __amd64 || __powerpc__) && __GNUC__
 #define GNUC_VERSION (__GNUC__ * 10000 + __GNUC_MINOR__ * 100 + __GNUC_PATCHLEVEL__)
 #if defined(__clang__)
  #define HAVE_ATOMIC
 #endif
 #if (defined(__GLIBC__) && defined(__GLIBC_PREREQ))
  #if (GNUC_VERSION >= 40100 && __GLIBC_PREREQ(2, 6))
  #define HAVE_ATOMIC
  #endif
 #endif
#endif

#define atomicIncr(var,count) __atomic_add_fetch(&var,(count),__ATOMIC_RELAXED)
#define atomicGetIncr(var,oldvalue_var,count) do { \
  oldvalue_var = __atomic_fetch_add(&var,(count),__ATOMIC_RELAXED); \
} while(0)
#define atomicDecr(var,count) __atomic_sub_fetch(&var,(count),__ATOMIC_RELAXED)
#define atomicGet(var,dstvar) do { \
  dstvar = __atomic_load_n(&var,__ATOMIC_RELAXED); \
} while(0)
#define atomicSet(var,value) __atomic_store_n(&var,value,__ATOMIC_RELAXED)
#define REDIS_ATOMIC_API "atomic-builtin"

#define ASIGN_TOTAL(a, b, c) \
  val1 = a;  \
  val2 = b;  \
  val3 = c;
// ...

if (condition)
  ASIGN_TOTAL(10, 20, 30);

/*
 *上面的代码将被预处理器扩展成
 *  if (condition)
 *  val1 = 10;
 *  val2 = 20;
 *  val3 = 30;
 * val2 和 val3 都在分支条件之外，这显然不是代码的意图。
 */

#define atomicIncr(var,count) __sync_add_and_fetch(&var,(count))
#define atomicGetIncr(var,oldvalue_var,count) do { \
  oldvalue_var = __sync_fetch_and_add(&var,(count)); \
} while(0)
#define atomicDecr(var,count) __sync_sub_and_fetch(&var,(count))
#define atomicGet(var,dstvar) do { \
  dstvar = __sync_sub_and_fetch(&var,0); \
} while(0)
#define atomicSet(var,value) do { \
  while(!__sync_bool_compare_and_swap(&var,var,value)); \
} while(0)
#define REDIS_ATOMIC_API "sync-builtin"

#define atomicIncr(var,count) do { \
  pthread_mutex_lock(&var ## _mutex); \
  var += (count); \
  pthread_mutex_unlock(&var ## _mutex); \
} while(0)
#define atomicGetIncr(var,oldvalue_var,count) do { \
  pthread_mutex_lock(&var ## _mutex); \
  oldvalue_var = var; \
  var += (count); \
  pthread_mutex_unlock(&var ## _mutex); \
} while(0)
#define atomicDecr(var,count) do { \
  pthread_mutex_lock(&var ## _mutex); \
  var -= (count); \
  pthread_mutex_unlock(&var ## _mutex); \
} while(0)
#define atomicGet(var,dstvar) do { \
  pthread_mutex_lock(&var ## _mutex); \
  dstvar = var; \
  pthread_mutex_unlock(&var ## _mutex); \
} while(0)
#define atomicSet(var,value) do { \
  pthread_mutex_lock(&var ## _mutex); \
  var = value; \
  pthread_mutex_unlock(&var ## _mutex); \
} while(0)
#define REDIS_ATOMIC_API "pthread-mutex"

int __failed_tests = 0;
int __test_num = 0;
#define test_cond(descr,_c) do { \
  __test_num++; printf("%d - %s: ", __test_num, descr); \
  if(_c) printf("PASSED\n"); else {printf("FAILED\n"); __failed_tests++;} \
} while(0);
#define test_report() do { \
  printf("%d tests, %d passed, %d failed\n", __test_num, \
  __test_num-__failed_tests, __failed_tests); \
  if (__failed_tests) { \
  printf("=== WARNING === We have failed tests here...\n"); \
  exit(1); \
  } \
} while(0);

sds x = sdsnew("foo"), y;

test_cond("Create a string and obtain the length",
sdslen(x) == 3 && memcmp(x,"foo\0",4) == 0)

typedef char *sds;

/* Note: sdshdr5 is never used, we just access the flags byte directly.
 * However is here to document the layout of type 5 SDS strings. */
struct __attribute__ ((__packed__)) sdshdr5 {
  unsigned char flags; /* 3 lsb of type, and 5 msb of string length */
  char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {
  uint8_t len; /* used */
  uint8_t alloc; /* excluding the header and null terminator */
  unsigned char flags; /* 3 lsb of type, 5 unused bits */
  char buf[];
};
struct __attribute__ ((__packed__)) sdshdr16 {
  uint16_t len; /* used */
  uint16_t alloc; /* excluding the header and null terminator */
  unsigned char flags; /* 3 lsb of type, 5 unused bits */
  char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {
  uint32_t len; /* used */
  uint32_t alloc; /* excluding the header and null terminator */
  unsigned char flags; /* 3 lsb of type, 5 unused bits */
  char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {
  uint64_t len; /* used */
  uint64_t alloc; /* excluding the header and null terminator */
  unsigned char flags; /* 3 lsb of type, 5 unused bits */
  char buf[];
};

char * str = "200 length of string ...";
sdshdr8 *hr = (sdshdr8*)malloc(sizeof(sdshdr8) + 200 + 1);
hr->len = 200;
hr->alloc = 200;
strncpy(hr->buf, str, 200);

sds my_sds_string = hr->buf;

/*
 * flags 类型
 */
#define SDS_TYPE_5  0
#define SDS_TYPE_8  1
#define SDS_TYPE_16 2
#define SDS_TYPE_32 3
#define SDS_TYPE_64 4

/*
 * 头部数据结构使用 flags 的低三位表示头部类型
 * header.flags & SDS_TYPE_MASK 就是数据结构的类型
 */
#define SDS_TYPE_MASK 7
#define SDS_TYPE_BITS 3

/*
 * 从 sds 变量获得其对应的头部，其中 T 是 5,8,16,32 或 64，必须是字面常量；
 * s 是一个 sds 变量
 */
#define SDS_HDR_VAR(T,s) struct sdshdr##T *sh = (struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T)));
#define SDS_HDR(T,s) ((struct sdshdr##T *)((s)-(sizeof(struct sdshdr##T))))

/*
 * sdshdr5 使用 flags 的高 5 位表示长度
 */
#define SDS_TYPE_5_LEN(f) ((f)>>SDS_TYPE_BITS)

static inline size_t sdslen(const sds s) {
  unsigned char flags = s[-1];
  switch(flags&SDS_TYPE_MASK) {
  case SDS_TYPE_5:
  return SDS_TYPE_5_LEN(flags);
  case SDS_TYPE_8:
  return SDS_HDR(8,s)->len;
  case SDS_TYPE_16:
  return SDS_HDR(16,s)->len;
  case SDS_TYPE_32:
  return SDS_HDR(32,s)->len;
  case SDS_TYPE_64:
  return SDS_HDR(64,s)->len;
  }
  return 0;
}

static inline char sdsReqType(size_t string_size) {
  if (string_size < 32)
  return SDS_TYPE_5;
  if (string_size < 0xff)
  return SDS_TYPE_8;
  if (string_size < 0xffff)
  return SDS_TYPE_16;
  if (string_size < 0xffffffff)
  return SDS_TYPE_32;
  return SDS_TYPE_64;
}

sds mystring = sdsnewlen("abc", 3);

sds sdsnewlen(const void *init, size_t initlen) {
  void *sh;
  sds s;
  char type = sdsReqType(initlen);
  /* Empty strings are usually created in order to append. Use type 8
  * since type 5 is not good at this. */
  if (type == SDS_TYPE_5 && initlen == 0) type = SDS_TYPE_8;
  int hdrlen = sdsHdrSize(type);
  unsigned char *fp; /* flags pointer. */

  sh = s_malloc(hdrlen+initlen+1);
  if (sh == NULL) return NULL;
  if (!init)
  memset(sh, 0, hdrlen+initlen+1);
  s = (char*)sh+hdrlen;
  fp = ((unsigned char*)s)-1;
  switch(type) {
  case SDS_TYPE_5: {
  *fp = type | (initlen << SDS_TYPE_BITS);
  break;
  }
  case SDS_TYPE_8: {
  SDS_HDR_VAR(8,s);
  sh->len = initlen;
  sh->alloc = initlen;
  *fp = type;
  break;
  }
  case SDS_TYPE_16: {
  SDS_HDR_VAR(16,s);
  sh->len = initlen;
  sh->alloc = initlen;
  *fp = type;
  break;
  }
  case SDS_TYPE_32: {
  SDS_HDR_VAR(32,s);
  sh->len = initlen;
  sh->alloc = initlen;
  *fp = type;
  break;
  }
  case SDS_TYPE_64: {
  SDS_HDR_VAR(64,s);
  sh->len = initlen;
  sh->alloc = initlen;
  *fp = type;
  break;
  }
  }
  if (initlen && init)
  memcpy(s, init, initlen);
  s[initlen] = '\0';
  return s;
}

void sdsfree(sds s) {
  if (s == NULL) return;
  s_free((char*)s-sdsHdrSize(s[-1]));
}

sds sdsMakeRoomFor(sds s, size_t addlen) {
  void *sh, *newsh;
  size_t avail = sdsavail(s);
  size_t len, newlen;
  char type, oldtype = s[-1] & SDS_TYPE_MASK;
  int hdrlen;

  /* Return ASAP if there is enough space left. */
  if (avail >= addlen) return s;

  len = sdslen(s);
  sh = (char*)s-sdsHdrSize(oldtype);
  newlen = (len+addlen);
  if (newlen < SDS_MAX_PREALLOC)
  newlen *= 2;
  else
  newlen += SDS_MAX_PREALLOC;

  type = sdsReqType(newlen);

  /* Don't use type 5: the user is appending to the string and type 5 is
  * not able to remember empty space, so sdsMakeRoomFor() must be called
  * at every appending operation. */
  if (type == SDS_TYPE_5) type = SDS_TYPE_8;

  hdrlen = sdsHdrSize(type);
  if (oldtype==type) {
  newsh = s_realloc(sh, hdrlen+newlen+1);
  if (newsh == NULL) return NULL;
  s = (char*)newsh+hdrlen;
  } else {
  /* Since the header size changes, need to move the string forward,
  * and can't use realloc */
  newsh = s_malloc(hdrlen+newlen+1);
  if (newsh == NULL) return NULL;
  memcpy((char*)newsh+hdrlen, s, len+1);
  s_free(sh);
  s = (char*)newsh+hdrlen;
  s[-1] = type;
  sdssetlen(s, len);
  }
  sdssetalloc(s, newlen);
  return s;
}

sds sdsRemoveFreeSpace(sds s) {
  void *sh, *newsh;
  char type, oldtype = s[-1] & SDS_TYPE_MASK;
  int hdrlen;
  size_t len = sdslen(s);
  sh = (char*)s-sdsHdrSize(oldtype);

  type = sdsReqType(len);
  hdrlen = sdsHdrSize(type);
  if (oldtype==type) {
  newsh = s_realloc(sh, hdrlen+len+1);
  if (newsh == NULL) return NULL;
  s = (char*)newsh+hdrlen;
  } else {
  newsh = s_malloc(hdrlen+len+1);
  if (newsh == NULL) return NULL;
  memcpy((char*)newsh+hdrlen, s, len+1);
  s_free(sh);
  s = (char*)newsh+hdrlen;
  s[-1] = type;
  sdssetlen(s, len);
  }
  sdssetalloc(s, len);
  return s;
}

00010110001011010011110110110011
 0 1 1 1 0 1 2 1 0 2 2 1 1 2 0 2
  1  2  1  3  2  3  3  2
  3  4  5  5
  7  10
  17

const uint64_t m1  = 0x5555555555555555; //binary: 0101...
const uint64_t m2  = 0x3333333333333333; //binary: 00110011..
const uint64_t m4  = 0x0f0f0f0f0f0f0f0f; //binary:  4 zeros,  4 ones ...
const uint64_t m8  = 0x00ff00ff00ff00ff; //binary:  8 zeros,  8 ones ...
const uint64_t m16 = 0x0000ffff0000ffff; //binary: 16 zeros, 16 ones ...
const uint64_t m32 = 0x00000000ffffffff; //binary: 32 zeros, 32 ones
const uint64_t hff = 0xffffffffffffffff; //binary: all ones
const uint64_t h01 = 0x0101010101010101; //the sum of 256 to the power of 0,1,2,3...

int popcount64a(uint64_t x)
{
  x = (x & m1) + ((x >>  1) & m1); //put count of each  2 bits into those  2 bits
  x = (x & m2) + ((x >>  2) & m2); //put count of each  4 bits into those  4 bits
  x = (x & m4) + ((x >>  4) & m4); //put count of each  8 bits into those  8 bits
  x = (x & m8) + ((x >>  8) & m8); //put count of each 16 bits into those 16 bits
  x = (x & m16) + ((x >> 16) & m16); //put count of each 32 bits into those 32 bits
  x = (x & m32) + ((x >> 32) & m32); //put count of each 64 bits into those 64 bits
  return x;
}

int popcount64b(uint64_t x)
{
  x -= (x >> 1) & m1;  //put count of each 2 bits into those 2 bits
  x = (x & m2) + ((x >> 2) & m2); //put count of each 4 bits into those 4 bits
  x = (x + (x >> 4)) & m4;  //put count of each 8 bits into those 8 bits
  x += x >>  8;  //put count of each 16 bits into their lowest 8 bits
  x += x >> 16;  //put count of each 32 bits into their lowest 8 bits
  x += x >> 32;  //put count of each 64 bits into their lowest 8 bits
  return x & 0x7f;
}

int popcount64c(uint64_t x)
{
  x -= (x >> 1) & m1;  //put count of each 2 bits into those 2 bits
  x = (x & m2) + ((x >> 2) & m2); //put count of each 4 bits into those 4 bits
  x = (x + (x >> 4)) & m4;  //put count of each 8 bits into those 8 bits
  return (x * h01) >> 56;  //returns left 8 bits of x + (x<<8) + (x<<16) + (x<<24) + ...
}

size_t redisPopcount(void *s, long count) {
  size_t bits = 0;
  unsigned char *p = s;
  uint32_t *p4;
  static const unsigned char bitsinbyte[256] = {0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8};

  /* Count initial bytes not aligned to 32 bit. */
  while((unsigned long)p & 3 && count) {
  bits += bitsinbyte[*p++];
  count--;
  }

  /* Count bits 28 bytes at a time */
  p4 = (uint32_t*)p;
  while(count>=28) {
  uint32_t aux1, aux2, aux3, aux4, aux5, aux6, aux7;

  aux1 = *p4++;
  aux2 = *p4++;
  aux3 = *p4++;
  aux4 = *p4++;
  aux5 = *p4++;
  aux6 = *p4++;
  aux7 = *p4++;
  count -= 28;

  aux1 = aux1 - ((aux1 >> 1) & 0x55555555);
  aux1 = (aux1 & 0x33333333) + ((aux1 >> 2) & 0x33333333);
  aux2 = aux2 - ((aux2 >> 1) & 0x55555555);
  aux2 = (aux2 & 0x33333333) + ((aux2 >> 2) & 0x33333333);
  aux3 = aux3 - ((aux3 >> 1) & 0x55555555);
  aux3 = (aux3 & 0x33333333) + ((aux3 >> 2) & 0x33333333);
  aux4 = aux4 - ((aux4 >> 1) & 0x55555555);
  aux4 = (aux4 & 0x33333333) + ((aux4 >> 2) & 0x33333333);
  aux5 = aux5 - ((aux5 >> 1) & 0x55555555);
  aux5 = (aux5 & 0x33333333) + ((aux5 >> 2) & 0x33333333);
  aux6 = aux6 - ((aux6 >> 1) & 0x55555555);
  aux6 = (aux6 & 0x33333333) + ((aux6 >> 2) & 0x33333333);
  aux7 = aux7 - ((aux7 >> 1) & 0x55555555);
  aux7 = (aux7 & 0x33333333) + ((aux7 >> 2) & 0x33333333);
  bits += ((((aux1 + (aux1 >> 4)) & 0x0F0F0F0F) +
  ((aux2 + (aux2 >> 4)) & 0x0F0F0F0F) +
  ((aux3 + (aux3 >> 4)) & 0x0F0F0F0F) +
  ((aux4 + (aux4 >> 4)) & 0x0F0F0F0F) +
  ((aux5 + (aux5 >> 4)) & 0x0F0F0F0F) +
  ((aux6 + (aux6 >> 4)) & 0x0F0F0F0F) +
  ((aux7 + (aux7 >> 4)) & 0x0F0F0F0F))* 0x01010101) >> 24;
  }
  /* Count the remaining bytes. */
  p = (unsigned char*)p4;
  while(count--) bits += bitsinbyte[*p++];
  return bits;
}

int __builtin_popcount (unsigned int x);
int __builtin_popcountll (unsigned long long x);

/* Return the position of the first bit set to one (if 'bit' is 1) or
 * zero (if 'bit' is 0) in the bitmap starting at 's' and long 'count' bytes.
 *
 * The function is guaranteed to return a value >= 0 if 'bit' is 0 since if
 * no zero bit is found, it returns count*8 assuming the string is zero
 * padded on the right. However if 'bit' is 1 it is possible that there is
 * not a single set bit in the bitmap. In this special case -1 is returned. */
long redisBitpos(void *s, unsigned long count, int bit) {
  unsigned long *l;
  unsigned char *c;
  unsigned long skipval, word = 0, one;
  long pos = 0; /* Position of bit, to return to the caller. */
  unsigned long j;
  int found;

  /* Process whole words first, seeking for first word that is not
  * all ones or all zeros respectively if we are lookig for zeros
  * or ones. This is much faster with large strings having contiguous
  * blocks of 1 or 0 bits compared to the vanilla bit per bit processing.
  *
  * Note that if we start from an address that is not aligned
  * to sizeof(unsigned long) we consume it byte by byte until it is
  * aligned. */

  /* Skip initial bits not aligned to sizeof(unsigned long) byte by byte. */
  skipval = bit ? 0 : UCHAR_MAX;
  c = (unsigned char*) s;
  found = 0;
  while((unsigned long)c & (sizeof(*l)-1) && count) {
  if (*c != skipval) {
  found = 1;
  break;
  }
  c++;
  count--;
  pos += 8;
  }

  /* Skip bits with full word step. */
  l = (unsigned long*) c;
  if (!found) {
  skipval = bit ? 0 : ULONG_MAX;
  while (count >= sizeof(*l)) {
  if (*l != skipval) break;
  l++;
  count -= sizeof(*l);
  pos += sizeof(*l)*8;
  }
  }

  /* Load bytes into "word" considering the first byte as the most significant
  * (we basically consider it as written in big endian, since we consider the
  * string as a set of bits from left to right, with the first bit at position
  * zero.
  *
  * Note that the loading is designed to work even when the bytes left
  * (count) are less than a full word. We pad it with zero on the right. */
  c = (unsigned char*)l;
  for (j = 0; j < sizeof(*l); j++) {
  word <<= 8;
  if (count) {
  word |= *c;
  c++;
  count--;
  }
  }

  /* Special case:
  * If bits in the string are all zero and we are looking for one,
  * return -1 to signal that there is not a single "1" in the whole
  * string. This can't happen when we are looking for "0" as we assume
  * that the right of the string is zero padded. */
  if (bit == 1 && word == 0) return -1;

  /* Last word left, scan bit by bit. The first thing we need is to
  * have a single "1" set in the most significant position in an
  * unsigned long. We don't know the size of the long so we use a
  * simple trick. */
  one = ULONG_MAX; /* All bits set to 1.*/
  one >>= 1;  /* All bits set to 1 but the MSB. */
  one = ~one;  /* All bits set to 0 but the MSB. */

  while(one) {
  if (((one & word) != 0) == bit) return pos;
  pos++;
  one >>= 1;
  }

  /* If we reached this point, there is a bug in the algorithm, since
  * the case of no match is handled as a special case before. */
  serverPanic("End of redisBitpos() reached.");
  return 0; /* Just to avoid warnings. */
}

(f) ""
\
(o) "f"
\
(o) "fo"
\
[t  b] "foo"
/  \
"foot" (e)  (a) "foob"
/  \
"foote" (r)  (r) "fooba"
/  \
"footer" []  [] "foobar"

["foo"] ""
|
[t  b] "foo"
/  \
"foot" ("er")  ("ar") "foob"
/  \
"footer" []  [] "foobar"

(f) ""
/
(i o) "f"
/  \
"firs"  ("rst")  (o) "fo"
/  \
"first" []  [t  b] "foo"
/  \
"foot" ("er")  ("ar") "foob"
/  \
"footer" []  [] "foobar"

typedef struct raxNode {
  uint32_t iskey:1;  /* Does this node contain a key? */
  uint32_t isnull:1;  /* Associated value is NULL (don't store it). */
  uint32_t iscompr:1;  /* Node is compressed. */
  uint32_t size:29;  /* Number of children, or compressed string len. */
  /* Data layout is as follows:
  *
  * If node is not compressed we have 'size' bytes, one for each children
  * character, and 'size' raxNode pointers, point to each child node.
  * Note how the character is not stored in the children but in the
  * edge of the parents:
  *
  * [header strlen=0][abc][a-ptr][b-ptr][c-ptr](value-ptr?)
  *
  * if node is compressed (strlen != 0) the node has 1 children.
  * In that case the 'size' bytes of the string stored immediately at
  * the start of the data section, represent a sequence of successive
  * nodes linked one after the other, for which only the last one in
  * the sequence is actually represented as a node, and pointed to by
  * the current compressed node.
  *
  * [header strlen=3][xyz][z-ptr](value-ptr?)
  *
  * Both compressed and not compressed nodes can represent a key
  * with associated data in the radix tree at any level (not just terminal
  * nodes).
  *
  * If the node has an associated key (iskey=1) and is not NULL
  * (isnull=0), then after the raxNode pointers poiting to the
  * childen, an additional value pointer is present (as you can see
  * in the representation above as "value-ptr" field).
  */
  unsigned char data[];
} raxNode;

typedef struct rax {
  raxNode *head;
  uint64_t numele;
  uint64_t numnodes;
} rax;

typedef struct raxStack {
  void **stack; /* Points to static_items or an heap allocated array. */
  size_t items, maxitems; /* Number of items contained and total space. */
  /* Up to RAXSTACK_STACK_ITEMS items we avoid to allocate on the heap
  * and use this static array of pointers instead. */
  void *static_items[RAX_STACK_STATIC_ITEMS];
  int oom; /* True if pushing into this stack failed for OOM at some point. */
} raxStack;

typedef struct raxIterator {
  int flags;
  rax *rt;  /* Radix tree we are iterating. */
  unsigned char *key;  /* The current string. */
  void *data;  /* Data associated to this key. */
  size_t key_len;  /* Current key length. */
  size_t key_max;  /* Max key len the current key buffer can hold. */
  unsigned char key_static_string[RAX_ITER_STATIC_LEN];
  raxNode *node;  /* Current node. Only for unsafe iteration. */
  raxStack stack;  /* Stack used for unsafe iteration. */
} raxIterator;

/* Initialize the stack. */
static inline void raxStackInit(raxStack *ts) {
  ts->stack = ts->static_items;
  ts->items = 0;
  ts->maxitems = RAX_STACK_STATIC_ITEMS;
  ts->oom = 0;
}

/* Push an item into the stack, returns 1 on success, 0 on out of memory. */
static inline int raxStackPush(raxStack *ts, void *ptr) {
  if (ts->items == ts->maxitems) {
  if (ts->stack == ts->static_items) {
  ts->stack = rax_malloc(sizeof(void*)*ts->maxitems*2);
  if (ts->stack == NULL) {
  ts->stack = ts->static_items;
  ts->oom = 1;
  errno = ENOMEM;
  return 0;
  }
  memcpy(ts->stack,ts->static_items,sizeof(void*)*ts->maxitems);
  } else {
  void **newalloc = rax_realloc(ts->stack,sizeof(void*)*ts->maxitems*2);
  if (newalloc == NULL) {
  ts->oom = 1;
  errno = ENOMEM;
  return 0;
  }
  ts->stack = newalloc;
  }
  ts->maxitems *= 2;
  }
  ts->stack[ts->items] = ptr;
  ts->items++;
  return 1;
}

/* Pop an item from the stack, the function returns NULL if there are no
 * items to pop. */
static inline void *raxStackPop(raxStack *ts) {
  if (ts->items == 0) return NULL;
  ts->items--;
  return ts->stack[ts->items];
}

/* Return the stack item at the top of the stack without actually consuming
 * it. */
static inline void *raxStackPeek(raxStack *ts) {
  if (ts->items == 0) return NULL;
  return ts->stack[ts->items-1];
}

/* Free the stack in case we used heap allocation. */
static inline void raxStackFree(raxStack *ts) {
  if (ts->stack != ts->static_items) rax_free(ts->stack);
}

/* Allocate a new non compressed node with the specified number of children.
 * If datafiled is true, the allocation is made large enough to hold the
 * associated data pointer.
 * Returns the new node pointer. On out of memory NULL is returned. */
raxNode *raxNewNode(size_t children, int datafield) {
  size_t nodesize = sizeof(raxNode)+children+
  sizeof(raxNode*)*children;
  if (datafield) nodesize += sizeof(void*);
  raxNode *node = rax_malloc(nodesize);
  if (node == NULL) return NULL;
  node->iskey = 0;
  node->isnull = 0;
  node->iscompr = 0;
  node->size = children;
  return node;
}

/* Allocate a new rax and return its pointer. On out of memory the function
 * returns NULL. */
rax *raxNew(void) {
  rax *rax = rax_malloc(sizeof(*rax));
  if (rax == NULL) return NULL;
  rax->numele = 0;
  rax->numnodes = 1;
  rax->head = raxNewNode(0,0);
  if (rax->head == NULL) {
  rax_free(rax);
  return NULL;
  } else {
  return rax;
  }
}

/* Return the current total size of the node. */
#define raxNodeCurrentLength(n) (\
  sizeof(raxNode)+(n)->size+ \
  ((n)->iscompr ? sizeof(raxNode*) : sizeof(raxNode*)*(n)->size)+ \
  (((n)->iskey && !(n)->isnull)*sizeof(void*)) \)

/* realloc the node to make room for auxiliary data in order
 * to store an item in that node. On out of memory NULL is returned. */
raxNode *raxReallocForData(raxNode *n, void *data) {
  if (data == NULL) return n; /* No reallocation needed, setting isnull=1 */
  size_t curlen = raxNodeCurrentLength(n);
  return rax_realloc(n,curlen+sizeof(void*));
}

/* Set the node auxiliary data to the specified pointer. */
void raxSetData(raxNode *n, void *data) {
  n->iskey = 1;
  if (data != NULL) {
  void **ndata = (void**)
  ((char*)n+raxNodeCurrentLength(n)-sizeof(void*));
  memcpy(ndata,&data,sizeof(data));
  n->isnull = 0;
  } else {
  n->isnull = 1;
  }
}

/* Get the node auxiliary data. */
void *raxGetData(raxNode *n) {
  if (n->isnull) return NULL;
  void **ndata =(void**)((char*)n+raxNodeCurrentLength(n)-sizeof(void*));
  void *data;
  memcpy(&data,ndata,sizeof(data));
  return data;
}

/* Add a new child to the node 'n' representing the character 'c' and return
 * its new pointer, as well as the child pointer by reference. Additionally
 * '***parentlink' is populated with the raxNode pointer-to-pointer of where
 * the new child was stored, which is useful for the caller to replace the
 * child pointer if it gets reallocated.
 *
 * On success the new parent node pointer is returned (it may change because
 * of the realloc, so the caller should discard 'n' and use the new value).
 * On out of memory NULL is returned, and the old node is still valid. */
raxNode *raxAddChild(raxNode *n, unsigned char c, raxNode **childptr, raxNode ***parentlink) {
  assert(n->iscompr == 0);

  size_t curlen = sizeof(raxNode)+
  n->size+
  sizeof(raxNode*)*n->size;
  size_t newlen;

  /* Alloc the new child we will link to 'n'. */
  raxNode *child = raxNewNode(0,0);
  if (child == NULL) return NULL;

  /* Make space in the original node. */
  if (n->iskey) curlen += sizeof(void*);
  newlen = curlen+sizeof(raxNode*)+1; /* Add 1 char and 1 pointer. */
  raxNode *newn = rax_realloc(n,newlen);
  if (newn == NULL) {
  rax_free(child);
  return NULL;
  }
  n = newn;

  /* After the reallocation, we have 5/9 (depending on the system
  * pointer size) bytes at the end, that is, the additional char
  * in the 'data' section, plus one pointer to the new child:
  *
  * [numc][abx][ap][bp][xp]|auxp|.....
  *
  * Let's find where to insert the new child in order to make sure
  * it is inserted in-place lexicographically. */
  int pos;
  for (pos = 0; pos < n->size; pos++) {
  if (n->data[pos] > c) break;
  }

  /* Now, if present, move auxiliary data pointer at the end
  * so that we can mess with the other data without overwriting it.
  * We will obtain something like that:
  *
  * [numc][abx][ap][bp][xp].....|auxp| */
  unsigned char *src;
  if (n->iskey && !n->isnull) {
  src = n->data+n->size+sizeof(raxNode*)*n->size;
  memmove(src+1+sizeof(raxNode*),src,sizeof(void*));
  }

  /* Now imagine we are adding a node with edge 'c'. The insertion
  * point is between 'b' and 'x', so the 'pos' variable value is
  * To start, move all the child pointers after the insertion point
  * of 1+sizeof(pointer) bytes on the right, to obtain:
  *
  * [numc][abx][ap][bp].....[xp]|auxp| */
  src = n->data+n->size+sizeof(raxNode*)*pos;
  memmove(src+1+sizeof(raxNode*),src,sizeof(raxNode*)*(n->size-pos));

  /* Now make the space for the additional char in the data section,
  * but also move the pointers before the insertion point in the right
  * by 1 byte, in order to obtain the following:
  *
  * [numc][ab.x][ap][bp]....[xp]|auxp| */
  src = n->data+pos;
  memmove(src+1,src,n->size-pos+sizeof(raxNode*)*pos);

  /* We can now set the character and its child node pointer to get:
  *
  * [numc][abcx][ap][bp][cp]....|auxp|
  * [numc][abcx][ap][bp][cp][xp]|auxp| */
  n->data[pos] = c;
  n->size++;
  raxNode **childfield = (raxNode**)(n->data+n->size+sizeof(raxNode*)*pos);
  memcpy(childfield,&child,sizeof(child));
  *childptr = child;
  *parentlink = childfield;
  return n;
}

/* Return the pointer to the last child pointer in a node. For the compressed
 * nodes this is the only child pointer. */
#define raxNodeLastChildPtr(n) ((raxNode**) (\
  ((char*)(n)) + \
  raxNodeCurrentLength(n) - \
  sizeof(raxNode*) - \
  (((n)->iskey && !(n)->isnull) ? sizeof(void*) : 0) \))

/* Return the pointer to the first child pointer. */
#define raxNodeFirstChildPtr(n) ((raxNode**)((n)->data+(n)->size))

/* Turn the node 'n', that must be a node without any children, into a
 * compressed node representing a set of nodes linked one after the other
 * and having exactly one child each. The node can be a key or not: this
 * property and the associated value if any will be preserved.
 *
 * The function also returns a child node, since the last node of the
 * compressed chain cannot be part of the chain: it has zero children while
 * we can only compress inner nodes with exactly one child each. */
raxNode *raxCompressNode(raxNode *n, unsigned char *s, size_t len, raxNode **child) {
  assert(n->size == 0 && n->iscompr == 0);
  void *data = NULL; /* Initialized only to avoid warnings. */
  size_t newsize;

  debugf("Compress node: %.*s\n", (int)len,s);

  /* Allocate the child to link to this node. */
  *child = raxNewNode(0,0);
  if (*child == NULL) return NULL;

  /* Make space in the parent node. */
  newsize = sizeof(raxNode)+len+sizeof(raxNode*);
  if (n->iskey) {
  data = raxGetData(n); /* To restore it later. */
  if (!n->isnull) newsize += sizeof(void*);
  }
  raxNode *newn = rax_realloc(n,newsize);
  if (newn == NULL) {
  rax_free(*child);
  return NULL;
  }
  n = newn;

  n->iscompr = 1;
  n->size = len;
  memcpy(n->data,s,len);
  if (n->iskey) raxSetData(n,data);
  raxNode **childfield = raxNodeLastChildPtr(n);
  memcpy(childfield,child,sizeof(*child));
  return n;
}

static inline size_t raxLowWalk(rax *rax, unsigned char *s, size_t len, raxNode **stopnode, raxNode ***plink, int *splitpos, raxStack *ts) {
  raxNode *h = rax->head;
  raxNode **parentlink = &rax->head;

  size_t i = 0; /* Position in the string. */
  size_t j = 0; /* Position in the node children (or bytes if compressed).*/
  while(h->size && i < len) {
  debugnode("Lookup current node",h);
  unsigned char *v = h->data;

  if (h->iscompr) {
  for (j = 0; j < h->size && i < len; j++, i++) {
  if (v[j] != s[i]) break;
  }
  if (j != h->size) break;
  } else {
  /* Even when h->size is large, linear scan provides good
  * performances compared to other approaches that are in theory
  * more sounding, like performing a binary search. */
  for (j = 0; j < h->size; j++) {
  if (v[j] == s[i]) break;
  }
  if (j == h->size) break;
  i++;
  }

  if (ts) raxStackPush(ts,h); /* Save stack of parent nodes. */
  raxNode **children = raxNodeFirstChildPtr(h);
  if (h->iscompr) j = 0; /* Compressed node only child is at index 0. */
  memcpy(&h,children+j,sizeof(h));
  parentlink = children+j;
  j = 0; /* If the new node is compressed and we do not
  iterate again (since i == l) set the split
  position to 0 to signal this node represents
  the searched key. */
  }
  if (stopnode) *stopnode = h;
  if (plink) *plink = parentlink;
  if (splitpos && h->iscompr) *splitpos = j;
  return i;
}

  "ANNIBALE" -> "SCO" -> []

1) Inserting "ANNIENTARE"

  |B| -> "ALE" -> "SCO" -> []
  "ANNI" -> |-|
  |E| -> (... continue algo ...) "NTARE" -> []

2) Inserting "ANNIBALI"

  |E| -> "SCO" -> []
  "ANNIBAL" -> |-|
  |I| -> (... continue algo ...) []

3) Inserting "AGO" (Like case 1, but set iscompr = 0 into original node)

  |N| -> "NIBALE" -> "SCO" -> []
  |A| -> |-|
  |G| -> (... continue algo ...) |O| -> []

4) Inserting "CIAO"

  |A| -> "NNIBALE" -> "SCO" -> []
  |-|
  |C| -> (... continue algo ...) "IAO" -> []

5) Inserting "ANNI"

  "ANNI" -> "BALE" -> "SCO" -> []

/* Insert the element 's' of size 'len', setting as auxiliary data
 * the pointer 'data'. If the element is already present, the associated
 * data is updated, and 0 is returned, otherwise the element is inserted
 * and 1 is returned. On out of memory the function returns 0 as well but
 * sets errno to ENOMEM, otherwise errno will be set to 0. */
int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old) {
  size_t i;
  int j = 0; /* Split position. If raxLowWalk() stops in a compressed
  node, the index 'j' represents the char we stopped within the
  compressed node, that is, the position where to split the
  node for insertion. */
  raxNode *h, **parentlink;

  debugf("### Insert %.*s with value %p\n", (int)len, s, data);
  i = raxLowWalk(rax,s,len,&h,&parentlink,&j,NULL);

  /* If i == len we walked following the whole string. If we are not
  * in the middle of a compressed node, the string is either already
  * inserted or this middle node is currently not a key, but can represent
  * our key. We have just to reallocate the node and make space for the
  * data pointer. */
  if (i == len && (!h->iscompr || j == 0 /* not in the middle if j is 0 */)) {
  if (h->iskey) {
  if (old) *old = raxGetData(h);
  raxSetData(h,data);
  errno = 0;
  return 0; /* Element already exists. */
  }
  h = raxReallocForData(h,data);
  if (h == NULL) {
  errno = ENOMEM;
  return 0;
  }
  memcpy(parentlink,&h,sizeof(h));
  raxSetData(h,data);
  rax->numele++;
  return 1; /* Element inserted. */
  }

  /* ------------------------- ALGORITHM 1 --------------------------- */
  if (h->iscompr && i != len) {
  debugf("ALGO 1: Stopped at compressed node %.*s (%p)\n",
  h->size, h->data, (void*)h);
  debugf("Still to insert: %.*s\n", (int)(len-i), s+i);
  debugf("Splitting at %d: '%c'\n", j, ((char*)h->data)[j]);
  debugf("Other (key) letter is '%c'\n", s[i]);

  /* 1: Save next pointer. */
  raxNode **childfield = raxNodeLastChildPtr(h);
  raxNode *next;
  memcpy(&next,childfield,sizeof(next));
  debugf("Next is %p\n", (void*)next);
  debugf("iskey %d\n", h->iskey);
  if (h->iskey) {
  debugf("key value is %p\n", raxGetData(h));
  }

  /* Set the length of the additional nodes we will need. */
  size_t trimmedlen = j;
  size_t postfixlen = h->size - j - 1;
  int split_node_is_key = !trimmedlen && h->iskey && !h->isnull;
  size_t nodesize;

  /* 2: Create the split node. Also allocate the other nodes we'll need
  *  ASAP, so that it will be simpler to handle OOM. */
  raxNode *splitnode = raxNewNode(1, split_node_is_key);
  raxNode *trimmed = NULL;
  raxNode *postfix = NULL;

  if (trimmedlen) {
  nodesize = sizeof(raxNode)+trimmedlen+sizeof(raxNode*);
  if (h->iskey && !h->isnull) nodesize += sizeof(void*);
  trimmed = rax_malloc(nodesize);
  }

  if (postfixlen) {
  nodesize = sizeof(raxNode)+postfixlen+
  sizeof(raxNode*);
  postfix = rax_malloc(nodesize);
  }

  /* OOM? Abort now that the tree is untouched. */
  if (splitnode == NULL ||
  (trimmedlen && trimmed == NULL) ||
  (postfixlen && postfix == NULL))
  {
  rax_free(splitnode);
  rax_free(trimmed);
  rax_free(postfix);
  errno = ENOMEM;
  return 0;
  }
  splitnode->data[0] = h->data[j];

  if (j == 0) {
  /* 3a: Replace the old node with the split node. */
  if (h->iskey) {
  void *ndata = raxGetData(h);
  raxSetData(splitnode,ndata);
  }
  memcpy(parentlink,&splitnode,sizeof(splitnode));
  } else {
  /* 3b: Trim the compressed node. */
  trimmed->size = j;
  memcpy(trimmed->data,h->data,j);
  trimmed->iscompr = j > 1 ? 1 : 0;
  trimmed->iskey = h->iskey;
  trimmed->isnull = h->isnull;
  if (h->iskey && !h->isnull) {
  void *ndata = raxGetData(h);
  raxSetData(trimmed,ndata);
  }
  raxNode **cp = raxNodeLastChildPtr(trimmed);
  memcpy(cp,&splitnode,sizeof(splitnode));
  memcpy(parentlink,&trimmed,sizeof(trimmed));
  parentlink = cp; /* Set parentlink to splitnode parent. */
  rax->numnodes++;
  }

  /* 4: Create the postfix node: what remains of the original
  * compressed node after the split. */
  if (postfixlen) {
  /* 4a: create a postfix node. */
  postfix->iskey = 0;
  postfix->isnull = 0;
  postfix->size = postfixlen;
  postfix->iscompr = postfixlen > 1;
  memcpy(postfix->data,h->data+j+1,postfixlen);
  raxNode **cp = raxNodeLastChildPtr(postfix);
  memcpy(cp,&next,sizeof(next));
  rax->numnodes++;
  } else {
  /* 4b: just use next as postfix node. */
  postfix = next;
  }

  /* 5: Set splitnode first child as the postfix node. */
  raxNode **splitchild = raxNodeLastChildPtr(splitnode);
  memcpy(splitchild,&postfix,sizeof(postfix));

  /* 6. Continue insertion: this will cause the splitnode to
  * get a new child (the non common character at the currently
  * inserted key). */
  rax_free(h);
  h = splitnode;
  } else if (h->iscompr && i == len) {
  /* ------------------------- ALGORITHM 2 --------------------------- */
  debugf("ALGO 2: Stopped at compressed node %.*s (%p) j = %d\n",
  h->size, h->data, (void*)h, j);

  /* Allocate postfix & trimmed nodes ASAP to fail for OOM gracefully. */
  size_t postfixlen = h->size - j;
  size_t nodesize = sizeof(raxNode)+postfixlen+sizeof(raxNode*);
  if (data != NULL) nodesize += sizeof(void*);
  raxNode *postfix = rax_malloc(nodesize);

  nodesize = sizeof(raxNode)+j+sizeof(raxNode*);
  if (h->iskey && !h->isnull) nodesize += sizeof(void*);
  raxNode *trimmed = rax_malloc(nodesize);

  if (postfix == NULL || trimmed == NULL) {
  rax_free(postfix);
  rax_free(trimmed);
  errno = ENOMEM;
  return 0;
  }

  /* 1: Save next pointer. */
  raxNode **childfield = raxNodeLastChildPtr(h);
  raxNode *next;
  memcpy(&next,childfield,sizeof(next));

  /* 2: Create the postfix node. */
  postfix->size = postfixlen;
  postfix->iscompr = postfixlen > 1;
  postfix->iskey = 1;
  postfix->isnull = 0;
  memcpy(postfix->data,h->data+j,postfixlen);
  raxSetData(postfix,data);
  raxNode **cp = raxNodeLastChildPtr(postfix);
  memcpy(cp,&next,sizeof(next));
  rax->numnodes++;

  /* 3: Trim the compressed node. */
  trimmed->size = j;
  trimmed->iscompr = j > 1;
  trimmed->iskey = 0;
  trimmed->isnull = 0;
  memcpy(trimmed->data,h->data,j);
  memcpy(parentlink,&trimmed,sizeof(trimmed));
  if (h->iskey) {
  void *aux = raxGetData(h);
  raxSetData(trimmed,aux);
  }

  /* Fix the trimmed node child pointer to point to
  * the postfix node. */
  cp = raxNodeLastChildPtr(trimmed);
  memcpy(cp,&postfix,sizeof(postfix));

  /* Finish! We don't need to contine with the insertion
  * algorithm for ALGO 2. The key is already inserted. */
  rax->numele++;
  rax_free(h);
  return 1; /* Key inserted. */
  }

  /* We walked the radix tree as far as we could, but still there are left
  * chars in our string. We need to insert the missing nodes.
  * Note: while loop never entered if the node was split by ALGO2,
  * since i == len. */
  while(i < len) {
  raxNode *child;

  /* If this node is going to have a single child, and there
  * are other characters, so that that would result in a chain
  * of single-childed nodes, turn it into a compressed node. */
  if (h->size == 0 && len-i > 1) {
  debugf("Inserting compressed node\n");
  size_t comprsize = len-i;
  if (comprsize > RAX_NODE_MAX_SIZE)
  comprsize = RAX_NODE_MAX_SIZE;
  raxNode *newh = raxCompressNode(h,s+i,comprsize,&child);
  if (newh == NULL) goto oom;
  h = newh;
  memcpy(parentlink,&h,sizeof(h));
  parentlink = raxNodeLastChildPtr(h);
  i += comprsize;
  } else {
  debugf("Inserting normal node\n");
  raxNode **new_parentlink;
  raxNode *newh = raxAddChild(h,s[i],&child,&new_parentlink);
  if (newh == NULL) goto oom;
  h = newh;
  memcpy(parentlink,&h,sizeof(h));
  parentlink = new_parentlink;
  i++;
  }
  rax->numnodes++;
  h = child;
  }
  raxNode *newh = raxReallocForData(h,data);
  if (newh == NULL) goto oom;
  h = newh;
  if (!h->iskey) rax->numele++;
  raxSetData(h,data);
  memcpy(parentlink,&h,sizeof(h));
  return 1; /* Element inserted. */

oom:
  /* This code path handles out of memory after part of the sub-tree was
  * already modified. Set the node as a key, and then remove it. However we
  * do that only if the node is a terminal node, otherwise if the OOM
  * happened reallocating a node in the middle, we don't need to free
  * anything. */
  if (h->size == 0) {
  h->isnull = 1;
  h->iskey = 1;
  rax->numele++; /* Compensate the next remove. */
  assert(raxRemove(rax,s,i,NULL) != 0);
  }
  errno = ENOMEM;
  return 0;
}

/* Find a key in the rax, returns raxNotFound special void pointer value
 * if the item was not found, otherwise the value associated with the
 * item is returned. */
void *raxFind(rax *rax, unsigned char *s, size_t len) {
  raxNode *h;

  debugf("### Lookup: %.*s\n", (int)len, s);
  int splitpos = 0;
  size_t i = raxLowWalk(rax,s,len,&h,NULL,&splitpos,NULL);
  if (i != len || (h->iscompr && splitpos != 0) || !h->iskey)
  return raxNotFound;
  return raxGetData(h);
}

/* Return the memory address where the 'parent' node stores the specified
 * 'child' pointer, so that the caller can update the pointer with another
 * one if needed. The function assumes it will find a match, otherwise the
 * operation is an undefined behavior (it will continue scanning the
 * memory without any bound checking). */
raxNode **raxFindParentLink(raxNode *parent, raxNode *child) {
  raxNode **cp = raxNodeFirstChildPtr(parent);
  raxNode *c;
  while(1) {
  memcpy(&c,cp,sizeof(c));
  if (c == child) break;
  cp++;
  }
  return cp;
}

/* Low level child removal from node. The new node pointer (after the child
 * removal) is returned. Note that this function does not fix the pointer
 * of the parent node in its parent, so this task is up to the caller.
 * The function never fails for out of memory. */
raxNode *raxRemoveChild(raxNode *parent, raxNode *child) {
  debugnode("raxRemoveChild before", parent);
  /* If parent is a compressed node (having a single child, as for definition
  * of the data structure), the removal of the child consists into turning
  * it into a normal node without children. */
  if (parent->iscompr) {
  void *data = NULL;
  if (parent->iskey) data = raxGetData(parent);
  parent->isnull = 0;
  parent->iscompr = 0;
  parent->size = 0;
  if (parent->iskey) raxSetData(parent,data);
  debugnode("raxRemoveChild after", parent);
  return parent;
  }

  /* Otherwise we need to scan for the children pointer and memmove()
  * accordingly.
  *
  * 1. To start we seek the first element in both the children
  *  pointers and edge bytes in the node. */
  raxNode **cp = raxNodeFirstChildPtr(parent);
  raxNode **c = cp;
  unsigned char *e = parent->data;

  /* 2. Search the child pointer to remove inside the array of children
  *  pointers. */
  while(1) {
  raxNode *aux;
  memcpy(&aux,c,sizeof(aux));
  if (aux == child) break;
  c++;
  e++;
  }

  /* 3. Remove the edge and the pointer by memmoving the remaining children
  *  pointer and edge bytes one position before. */
  int taillen = parent->size - (e - parent->data) - 1;
  debugf("raxRemoveChild tail len: %d\n", taillen);
  memmove(e,e+1,taillen);

  /* Since we have one data byte less, also child pointers start one byte
  * before now. */
  memmove(((char*)cp)-1,cp,(parent->size-taillen-1)*sizeof(raxNode**));

  /* Move the remaining "tail" pointer at the right position as well. */
  memmove(((char*)c)-1,c+1,taillen*sizeof(raxNode**)+parent->iskey*sizeof(void*));

  /* 4. Update size. */
  parent->size--;

  /* realloc the node according to the theoretical memory usage, to free
  * data if we are over-allocating right now. */
  raxNode *newnode = rax_realloc(parent,raxNodeCurrentLength(parent));
  if (newnode) {
  debugnode("raxRemoveChild after", newnode);
  }
  /* Note: if rax_realloc() fails we just return the old address, which
  * is valid. */
  return newnode ? newnode : parent;
}

/* Remove the specified item. Returns 1 if the item was found and
 * deleted, 0 otherwise. */
int raxRemove(rax *rax, unsigned char *s, size_t len, void **old) {
  raxNode *h;
  raxStack ts;

  debugf("### Delete: %.*s\n", (int)len, s);
  raxStackInit(&ts);
  int splitpos = 0;
  size_t i = raxLowWalk(rax,s,len,&h,NULL,&splitpos,&ts);
  if (i != len || (h->iscompr && splitpos != 0) || !h->iskey) {
  raxStackFree(&ts);
  return 0;
  }
  if (old) *old = raxGetData(h);
  h->iskey = 0;
  rax->numele--;

  /* If this node has no children, the deletion needs to reclaim the
  * no longer used nodes. This is an iterative process that needs to
  * walk the three upward, deleting all the nodes with just one child
  * that are not keys, until the head of the rax is reached or the first
  * node with more than one child is found. */

  int trycompress = 0; /* Will be set to 1 if we should try to optimize the
  tree resulting from the deletion. */

  if (h->size == 0) {
  debugf("Key deleted in node without children. Cleanup needed.\n");
  raxNode *child = NULL;
  while(h != rax->head) {
  child = h;
  debugf("Freeing child %p [%.*s] key:%d\n", (void*)child,
  (int)child->size, (char*)child->data, child->iskey);
  rax_free(child);
  rax->numnodes--;
  h = raxStackPop(&ts);
  /* If this node has more then one child, or actually holds
  * a key, stop here. */
  if (h->iskey || (!h->iscompr && h->size != 1)) break;
  }
  if (child) {
  debugf("Unlinking child %p from parent %p\n",
  (void*)child, (void*)h);
  raxNode *new = raxRemoveChild(h,child);
  if (new != h) {
  raxNode *parent = raxStackPeek(&ts);
  raxNode **parentlink;
  if (parent == NULL) {
  parentlink = &rax->head;
  } else {
  parentlink = raxFindParentLink(parent,h);
  }
  memcpy(parentlink,&new,sizeof(new));
  }

  /* If after the removal the node has just a single child
  * and is not a key, we need to try to compress it. */
  if (new->size == 1 && new->iskey == 0) {
  trycompress = 1;
  h = new;
  }
  }
  } else if (h->size == 1) {
  /* If the node had just one child, after the removal of the key
  * further compression with adjacent nodes is pontentially possible. */
  trycompress = 1;
  }

  /* Don't try node compression if our nodes pointers stack is not
  * complete because of OOM while executing raxLowWalk() */
  if (trycompress && ts.oom) trycompress = 0;

  if (trycompress) {
  debugf("After removing %.*s:\n", (int)len, s);
  debugnode("Compression may be needed",h);
  debugf("Seek start node\n");

  /* Try to reach the upper node that is compressible.
  * At the end of the loop 'h' will point to the first node we
  * can try to compress and 'parent' to its parent. */
  raxNode *parent;
  while(1) {
  parent = raxStackPop(&ts);
  if (!parent || parent->iskey ||
  (!parent->iscompr && parent->size != 1)) break;
  h = parent;
  debugnode("Going up to",h);
  }
  raxNode *start = h; /* Compression starting node. */

  /* Scan chain of nodes we can compress. */
  size_t comprsize = h->size;
  int nodes = 1;
  while(h->size != 0) {
  raxNode **cp = raxNodeLastChildPtr(h);
  memcpy(&h,cp,sizeof(h));
  if (h->iskey || (!h->iscompr && h->size != 1)) break;
  /* Stop here if going to the next node would result into
  * a compressed node larger than h->size can hold. */
  if (comprsize + h->size > RAX_NODE_MAX_SIZE) break;
  nodes++;
  comprsize += h->size;
  }
  if (nodes > 1) {
  /* If we can compress, create the new node and populate it. */
  size_t nodesize =
  sizeof(raxNode)+comprsize+sizeof(raxNode*);
  raxNode *new = rax_malloc(nodesize);
  /* An out of memory here just means we cannot optimize this
  * node, but the tree is left in a consistent state. */
  if (new == NULL) {
  raxStackFree(&ts);
  return 1;
  }
  new->iskey = 0;
  new->isnull = 0;
  new->iscompr = 1;
  new->size = comprsize;
  rax->numnodes++;

  /* Scan again, this time to populate the new node content and
  * to fix the new node child pointer. At the same time we free
  * all the nodes that we'll no longer use. */
  comprsize = 0;
  h = start;
  while(h->size != 0) {
  memcpy(new->data+comprsize,h->data,h->size);
  comprsize += h->size;
  raxNode **cp = raxNodeLastChildPtr(h);
  raxNode *tofree = h;
  memcpy(&h,cp,sizeof(h));
  rax_free(tofree); rax->numnodes--;
  if (h->iskey || (!h->iscompr && h->size != 1)) break;
  }
  debugnode("New node",new);

  /* Now 'h' points to the first node that we still need to use,
  * so our new node child pointer will point to it. */
  raxNode **cp = raxNodeLastChildPtr(new);
  memcpy(cp,&h,sizeof(h));

  /* Fix parent link. */
  if (parent) {
  raxNode **parentlink = raxFindParentLink(parent,start);
  memcpy(parentlink,&new,sizeof(new));
  } else {
  rax->head = new;
  }

  debugf("Compressed %d nodes, %d total bytes\n",
  nodes, (int)comprsize);
  }
  }
  raxStackFree(&ts);
  return 1;
}