mcache

#include "erl_nif.h"
#include <string.h>
#include "stdio.h"
#include "xxhash.h"

//#define DEBUG
#include "mcache.h"


ErlNifResourceType* mcache_t_handle;
static ERL_NIF_TERM atom_ok;
static ERL_NIF_TERM atom_undefined;
static ERL_NIF_TERM atom_overflow;


void print_entry(mc_entry_t *entry) {
  {};
  uint8_t i;
  if (!entry) {
    printf("\r\n");
    return;
  }
  printf("    @%ld:", entry->start);

  for (i=0; i < entry->count; i++) {
    printf(" %ld", entry->data[i]);
  };

  if (entry->next) {
    printf(" |");
    print_entry(entry->next);
  } else {
    printf("\r\n");
  };
  
}

void print_metric(mc_metric_t *metric) {
  uint16_t i;
  printf("  ");
  for(i = 0; i < metric->name_len; i++)
    printf("%c", (int) metric->name[i]);
  printf("[%zu]:\r\n", metric->alloc);
  print_entry(metric->head);
};

void init_buckets(mc_conf_t conf,/*@out@*/ mc_gen_t *gen) {
  dprint("init_buckets\r\n");
  gen->buckets = (mc_bucket_t *) mc_alloc(conf.buckets * sizeof(mc_bucket_t));
  for (int b = 0; b < conf.buckets; b++) {
    mc_bucket_t *bkt = &(gen->buckets[b]);

    for (int i = 0; i < LCOUNT; i++) {
      bkt->largest[i] = NULL;
    }

#ifdef TAGGED
    bkt->tag = TAG_BKT;
#endif

    for (int s = 0; s < SUBS; s++) {
      mc_sub_bucket_t *sub = &(bkt->subs[s]);
      sub->size = conf.initial_entries;
      sub->count = 0;
      sub->metrics = (mc_metric_t **) mc_alloc(sub->size * sizeof(mc_metric_t *));
#ifdef TAGGED
      sub->tag = TAG_SUB;
      for (int i = 0; i < sub->size; i++) {
        sub->metrics[i] = TAG_METRIC_L;
      }
#endif
    };
  }
}

void remove_largest(mc_bucket_t *bkt, mc_metric_t *metric) {
  int i = 0;
  dprint("[%p] remove largest: %llu\r\n",  bkt, metric->hash);
#ifdef DEBUG
  dprint("- >>");
  for(i = 0; i < LCOUNT; i++){
    if (! bkt->largest[i]) {
      break;
    };
    dprint(" %llu(%zu)", bkt->largest[i]->hash, bkt->largest[i]->alloc);
  };
  dprint("\r\n");
#endif
  
  i = 0;
  // short circuite when we have a last element
  if (bkt->largest[LCOUNT - 1]) {
    // if we are the last lement we can set it to null and return
    if (bkt->largest[LCOUNT - 1] == metric) {
      bkt->largest[LCOUNT - 1] = NULL;
      return;
    }
  }
  //skip all the ones we don't have
  while (i < LCOUNT && bkt->largest[i] != metric) {
    // if we found a null we can end our search
    if (!bkt->largest[i]) {
      dprint("return %d\r\n", i);
      return;
    }
    i++;
  };
  dprint("found %d\r\n", i);
  uint8_t moved = 0;
  while (i < LCOUNT - 1) {
    dprint("%d <- %d\r\n", i, i + 1);
    // found a null we don't need to go on.
    if (!bkt->largest[i]) {
      return;
    }
    moved = 1;
    bkt->largest[i] = bkt->largest[i+1];
    i++;
  }
  if (moved) {
    dprint("%d <- NULL\r\n", LCOUNT - 1);
    bkt->largest[LCOUNT - 1] = NULL;
  }
}

void insert_largest(mc_bucket_t *bkt, mc_metric_t *metric) {
  remove_largest(bkt, metric);
  dprint("[%p] insert largest: %llu\r\n", bkt, metric->hash);
#ifdef DEBUG
  dprint(">>");
  for(int j = 0; j < LCOUNT; j++){
    if (! bkt->largest[j]) {
      break;
    };
    dprint(" %llu(%zu)", bkt->largest[j]->hash, bkt->largest[j]->alloc);
  };
  dprint("\r\n");
#endif
  for (int i = 0; i < LCOUNT; i++) {
    if (!bkt->largest[i]) {
      dprint("[%d] <- %llu; %zu!\r\n", i, metric->hash, metric->alloc);
      bkt->largest[i] = metric;
      break;
    }
    if (metric == bkt->largest[i]) {
      break;
    }
    if (metric->alloc > bkt->largest[i]->alloc) {
      dprint("[%d] <- %llu: %zu\r\n", i, metric->hash, metric->alloc);
      mc_metric_t *t = bkt->largest[i];
      bkt->largest[i] = metric;
      metric = t;
    } else {
      dprint("[%d] | %llu: %zu\r\n", i, bkt->largest[i]->hash, bkt->largest[i]->alloc);
    }
  }
#ifdef DEBUG
  dprint("<<");
  for(int j = 0; j < LCOUNT; j++){
    if (! bkt->largest[j]) {
      break;
    };
    dprint(" %llu(%zu)", bkt->largest[j]->hash, bkt->largest[j]->alloc);
  };
  dprint("\r\n");
#endif

}

void age(mcache_t *cache) {
  dprint("age\r\n");
  for (int b = 0; b < cache->conf.buckets; b++) {
    // G1 -> G2
    for (int s = 0; s < SUBS; s++) {
      mc_sub_bucket_t *g2s = &(cache->g2.buckets[b].subs[s]);
      mc_sub_bucket_t *g1s = &(cache->g1.buckets[b].subs[s]);
      mc_metric_t **new_metrics = (mc_metric_t **) mc_alloc((g2s->count + g1s->count) * sizeof(mc_metric_t *));
      memcpy(new_metrics, g2s->metrics, g2s->count * sizeof(mc_metric_t *));
      mc_free(g2s->metrics);
      g2s->metrics = new_metrics;

      memcpy(
             // copy to the end of the current array
             &(g2s->metrics[g2s->count]),
             // copy the start ofn the lower gens array
             g1s->metrics,
             // Copy based on the sbuze of the old array
             g1s->count * sizeof(mc_metric_t *)
             );
      mc_free(g1s->metrics);
      g2s->count += g1s->count;
      g2s->size = g2s->count;
      // Free the G1 metric  list of thbs bucket as we copbed bt all out
    }
    // make sure to clear the largest
    for (int l = 0; l < LCOUNT; l++) {
      // move largest over when we have them
      if (cache->g1.buckets[b].largest[l]) {
        insert_largest(&(cache->g2.buckets[b]), cache->g1.buckets[b].largest[l]);
      } else {
        // we can stop our loop when we find the first null
        break;
      }
      // then set them to null
      cache->g1.buckets[b].largest[l] = NULL;
    }
  }
  // free g1 buckets (we copied the content to g2)
  mc_free(cache->g1.buckets);
  // move g0 buckets to g1
  cache->g1.buckets = cache->g0.buckets;

  cache->g2.alloc += cache->g1.alloc;
  cache->g1.alloc = cache->g0.alloc;
  cache->g0.alloc = 0;
  // reinitialize g0
  init_buckets(cache->conf, &(cache->g0));

}

static ERL_NIF_TERM serialize_entry(ErlNifEnv* env, mc_entry_t *entry) {
  ERL_NIF_TERM data;
  size_t to_copy = entry->count * sizeof(ErlNifUInt64);
  unsigned char *datap = enif_make_new_binary(env, to_copy, &data);
  memcpy(datap, entry->data, to_copy);
  return  enif_make_tuple2(env,
                           enif_make_uint64(env, entry->start),
                           data);

}

static ERL_NIF_TERM serialize_metric(ErlNifEnv* env, mc_metric_t *metric) {
  if (! metric) {
    return atom_undefined;
  }
  ERL_NIF_TERM result = enif_make_list(env, 0);
  ERL_NIF_TERM reverse;
  mc_entry_t *entry = metric->head;
  while (entry) {
    result = enif_make_list_cell(env, serialize_entry(env, entry), result);
    entry = entry->next;
  }
  enif_make_reverse_list(env, result, &reverse);
  return reverse;
}

static void free_entry(mc_entry_t *e) {
  if (e->next) {
    free_entry(e->next);
  }
  mc_free(e->data);
  mc_free(e);
}


static void free_metric(mc_metric_t *m) {
  if (m->head) {
    free_entry(m->head);
  }
  mc_free(m->name);
  mc_free(m);
}

static void free_gen(mc_conf_t conf, mc_gen_t gen) {
  for (int b = 0; b < conf.buckets; b++) {
    for (int s = 0; s < SUBS; s++) {
      for (int m = 0; m < gen.buckets[b].subs[s].count; m++) {
        free_metric(gen.buckets[b].subs[s].metrics[m]);
      };
      mc_free(gen.buckets[b].subs[s].metrics);
    }
  }
  mc_free(gen.buckets);
};

static void cache_dtor(ErlNifEnv* env, void* handle) {
  mcache_t * c = (mcache_t*) handle;
  free_gen(c->conf, c->g0);
  free_gen(c->conf, c->g1);
  free_gen(c->conf, c->g2);
};


static int
load(ErlNifEnv* env, void** priv, ERL_NIF_TERM load_info)
{
  ErlNifResourceFlags flags = (ErlNifResourceFlags)(ERL_NIF_RT_CREATE | ERL_NIF_RT_TAKEOVER);
  mcache_t_handle = enif_open_resource_type(env,
                                            "mcache_t",
                                            "cache",
                                            &cache_dtor,
                                            flags,
                                            0);
  atom_undefined = enif_make_atom(env, "undefined");
  atom_ok = enif_make_atom(env, "ok");
  atom_overflow = enif_make_atom(env, "overflow");
  return 0;
}

static int
upgrade(ErlNifEnv* env, void** priv, void** old_priv, ERL_NIF_TERM load_info)
{
  return 0;
}

static ERL_NIF_TERM
new_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {

  mcache_t *cache;
  ErlNifUInt64 max_alloc;
  ErlNifUInt64 buckets;
  ErlNifUInt64 age_cycle;
  ErlNifUInt64 initial_data_size;
  ErlNifUInt64 initial_entries;
  ErlNifUInt64 hash_seed;
  ErlNifUInt64 max_gap;
  if (argc != 7) {
    return enif_make_badarg(env);
  };
  if (!enif_get_uint64(env, argv[0], &max_alloc)) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[1], &buckets)) return enif_make_badarg(env);
  if (buckets <= 0) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[2], &age_cycle)) return enif_make_badarg(env);
  if (age_cycle <= 0) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[3], &initial_data_size)) return enif_make_badarg(env);
  if (initial_data_size <= 0) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[4], &initial_entries)) return enif_make_badarg(env);
  if (initial_entries <= 0) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[5], &hash_seed)) return enif_make_badarg(env);
  if (hash_seed <= 0) return enif_make_badarg(env);

  if (!enif_get_uint64(env, argv[6], &max_gap)) return enif_make_badarg(env);

  cache = (mcache_t *) enif_alloc_resource(mcache_t_handle, sizeof(mcache_t));

#ifdef TAGGED
  cache->tag = TAG_CACHE;
#endif

  // Set up the config
  cache->conf.max_alloc = max_alloc;
  cache->conf.buckets = buckets;
  cache->conf.age_cycle = age_cycle;

  cache->conf.initial_data_size = initial_data_size;
  cache->conf.initial_entries = initial_entries;
  cache->conf.hash_seed = hash_seed;
  cache->conf.max_gap = max_gap;

  // some cache wqide counters
  cache->inserts = 0;
  cache->age = 0;

  //now set up the tree genreations
  cache->g0.v = 0;
  cache->g0.alloc = 0;
#ifdef TAGGED
  cache->g0.tag = TAG_GEN;
#endif

  init_buckets(cache->conf, &(cache->g0));
  cache->g1.v = 1;
  cache->g1.alloc = 0;
#ifdef TAGGED
  cache->g1.tag = TAG_GEN;
#endif
  init_buckets(cache->conf, &(cache->g1));
  cache->g2.v = 2;
  cache->g2.alloc = 0;
#ifdef TAGGED
  cache->g2.tag = TAG_GEN;
#endif
  init_buckets(cache->conf, &(cache->g2));
  ERL_NIF_TERM term = enif_make_resource(env, cache);
  enif_release_resource(cache);
  return term;
};


mc_metric_t *find_metric_g(mc_conf_t conf, mc_gen_t gen, uint64_t hash, uint16_t name_len, uint8_t *name) {
  // Itterate over the existig metrics and see if we have already
  // seen this one.
  uint64_t bucket = hash % conf.buckets;
  uint8_t sub = subid(hash);
  for (int i = 0; i < gen.buckets[bucket].subs[sub].count; i++) {
    mc_metric_t *m = gen.buckets[bucket].subs[sub].metrics[i];
    if (m->name_len == name_len
        && m->hash == hash
        && memcmp(m->name, name, name_len) == 0) {
      return m;
    }
  }
  return NULL;
};

mc_metric_t *find_metric(mcache_t *cache, uint64_t hash, uint16_t name_len, uint8_t *name) {
  mc_metric_t *res = find_metric_g(cache->conf, cache->g0, hash, name_len, name);
  if(!res) {
    res = find_metric_g(cache->conf, cache->g1, hash, name_len, name);
    if(!res) {
      res = find_metric_g(cache->conf, cache->g2, hash, name_len, name);
    }
  }
  return res;
}


mc_metric_t *find_metric_and_remove_g(mc_conf_t conf, mc_gen_t *gen, uint64_t hash, uint16_t name_len, uint8_t *name) {
  int i = 0;
  uint64_t bucket = hash % conf.buckets;
  uint8_t sub = subid(hash);

  // Itterate over the existig metrics and see if we have already
  // seen this one.

  for (i = 0; i < gen->buckets[bucket].subs[sub].count; i++) {
    mc_metric_t *m = gen->buckets[bucket].subs[sub].metrics[i];
    if (m->name_len == name_len
        && m->hash == hash
        && memcmp(m->name, name, name_len) == 0) {
      if (i != gen->buckets[bucket].subs[sub].count - 1) {
        gen->buckets[bucket].subs[sub].metrics[i] = gen->buckets[bucket].subs[sub].metrics[gen->buckets[bucket].subs[sub].count - 1];
      }
      remove_largest(&(gen->buckets[bucket]), m);
      gen->buckets[bucket].subs[sub].count--;
      gen->alloc -= m->alloc;
      return m;
    }
  }
  return NULL;
};

uint64_t remove_prefix_g(mc_conf_t conf, mc_gen_t *gen, uint16_t pfx_len, uint8_t *pfx) {
  int i = 0;
  uint64_t counter = 0;
  // Itterate over the existig metrics and see if we have already
  // seen this one.
  for (int bucket = 0; bucket < conf.buckets; bucket++) {
    for (int sub = 0 ; sub < SUBS; sub++) {
      for (i = 0; i < gen->buckets[bucket].subs[sub].count; i++) {
        mc_metric_t *m = gen->buckets[bucket].subs[sub].metrics[i];
        if (m->name_len >= pfx_len
            && memcmp(m->name, pfx, pfx_len) == 0) {
          if (i != gen->buckets[bucket].subs[sub].count - 1) {
            gen->buckets[bucket].subs[sub].metrics[i] = gen->buckets[bucket].subs[sub].metrics[gen->buckets[bucket].subs[sub].count - 1];
            // we chear we move the last element in the current position and then go a step
            // back so we re-do it
            i--;
          }
          gen->buckets[bucket].subs[sub].count--;
          gen->alloc -= m->alloc;
          remove_largest(&(gen->buckets[bucket]), m);
          free_metric(m);
          counter++;
        }
      }
    }
  }
  return counter;
};

uint64_t remove_prefix(mcache_t *cache, uint16_t pfx_len, uint8_t *pfx) {
  return remove_prefix_g(cache->conf, &(cache->g0), pfx_len, pfx) +
    remove_prefix_g(cache->conf, &(cache->g1), pfx_len, pfx) +
    remove_prefix_g(cache->conf, &(cache->g2), pfx_len, pfx);
}

mc_metric_t *find_metric_and_remove(mcache_t *cache, uint64_t hash, uint16_t name_len, uint8_t *name) {
  mc_metric_t *res = find_metric_and_remove_g(cache->conf, &(cache->g0), hash, name_len, name);
  if(!res) {
    res = find_metric_and_remove_g(cache->conf, &(cache->g1), hash, name_len, name);
    if(!res) {
      res = find_metric_and_remove_g(cache->conf, &(cache->g2), hash, name_len, name);
    };
  };
  return res;
}


mc_metric_t *get_metric(mcache_t *cache, uint64_t hash, uint16_t name_len, uint8_t *name) {

  uint64_t bucket = hash % cache->conf.buckets;
  uint8_t sub = subid(hash);
  // Itterate over the existig metrics and see if we have already
  // seen this one.
  mc_metric_t *metric = find_metric_g(cache->conf, cache->g0, hash, name_len, name);
  if (metric) {
    return metric;
  }
  mc_bucket_t *b = &(cache->g0.buckets[bucket]);
  mc_sub_bucket_t *s = &(b->subs[sub]);
  // We start with a small cache and grow it as required, so we need to
  // make sure the new index doesn't exceet the count. If it does
  // double the size of the cache.
  if (s->count >= s->size) {
    mc_metric_t **new_metrics = mc_alloc(s->size * 2 * sizeof(mc_metric_t *));
#ifdef TAGGED
    for (int i = 0; i < s->size * 2; i++) {
      new_metrics[i] = TAG_METRIC_L;
    }
#endif
    memcpy(new_metrics, s->metrics, s->count * sizeof(mc_metric_t *));
    mc_free(s->metrics);
    s->metrics = new_metrics;
    s->size = s->size * 2;
  }
  // we havn't seen the metric so we'll create a new one.
  metric = find_metric_and_remove_g(cache->conf, &(cache->g1), hash, name_len, name);
  if (!metric) {
    metric = find_metric_and_remove_g(cache->conf, &(cache->g2), hash, name_len, name);
  }
  if (!metric) {
    metric = (mc_metric_t *) mc_alloc(sizeof(mc_metric_t));
#ifdef TAGGED
    metric->tag = TAG_METRIC;
#endif

    metric->alloc = name_len + sizeof(mc_metric_t);
    metric->name = mc_alloc(name_len * sizeof(uint8_t));
    metric->hash = hash;
    metric->name_len = name_len;
    memcpy(metric->name, name, name_len);
    metric->head = NULL;
    metric->tail = NULL;
  }
  cache->g0.alloc += metric->alloc;
  s->metrics[s->count] = metric;
  s->count++;
  insert_largest(b, metric);
  return metric;
};

void add_point(mc_conf_t conf, mc_gen_t *gen, mc_metric_t *metric, ErlNifUInt64 offset, size_t count, ErlNifUInt64* values) {
  // If eitehr we have no data yet or the current data is larger then
  // the offset we generate a new metric.
  // In both cases next will be the current head given that next might
  // be empty and it's needed to set next to empty for the first element.
  mc_entry_t *entry = NULL;
  if((!metric->head) || offset < metric->head->start) {
    size_t alloc = sizeof(ErlNifUInt64) * MAX(conf.initial_data_size, count * 2);
    entry = mc_alloc(sizeof(mc_entry_t));
    entry->start = offset;
    entry->size = conf.initial_data_size;
    entry->data = (ErlNifUInt64 *) mc_alloc(alloc);

#ifdef TAGGED
    entry->tag = TAG_ENTRY;
    for (int i = 0; i < entry->size; i++) {
      entry->data[i] = TAG_DATA_L;
    }
#endif

    entry->count = 0;
    entry->next = metric->head;
    metric->head = entry;
    metric->tail = entry;
    metric->alloc += alloc + sizeof(mc_entry_t);
    gen->alloc += alloc + sizeof(mc_entry_t);
  }
  if (offset > metric->tail->start) {
    entry = metric->tail;
  } else {
    entry = metric->head;
  }
  do {
    // if the offset is beyond the next chunks start
    // just go ahead and go for this chunk
    if(entry->next && (offset >= entry->next->start)) {
      entry = entry->next;
      continue;
    }

    //We know that we either have no next chunk or the data is before
    //the start of the next chunk

    // we allocate a new chunk behind this and continue with that.
    uint64_t internal_offset = offset - entry->start;

    // or we'd have gaps
    dprint("CHECK: %llu > (%u + %llu)!\r\n", internal_offset, entry->count, conf.max_gap);
    if (internal_offset > (entry->count + conf.max_gap)) {
      dprint("we got a gap!\r\n");
      // TODO: we could combine that with merge and dertermin if we can
      // just pull the start back but that remains for another day
      mc_entry_t *next = mc_alloc(sizeof(mc_entry_t));
      // we allocate at least as much memory as we need for our data
      // so we don't need to allocate twice for bigger blocks
      next->size = MAX(conf.initial_data_size, count);
      uint64_t alloc = next->size * sizeof(ErlNifUInt64);
      // create the new entry with the given offset
      next->start = offset;
      // reserve the data
      next->data = (ErlNifUInt64 *) mc_alloc(alloc);

#ifdef TAGGED
      next->tag = TAG_ENTRY;
      for (int i = 0; i < next->size; i++) {
        next->data[i] = TAG_DATA_L;
      }
#endif

      // we have not put in data yet
      next->count = 0;

      // insert this inbetween this and the next element
      next->next = entry->next;
      entry->next = next;
      // if next->next is null then we know our next is now the tail
      if (!next->next) {
        metric->tail = next;
      }
      // we continue with this element as entry

      entry = next;


      metric->alloc += alloc + sizeof(mc_entry_t);
      gen->alloc += alloc + sizeof(mc_entry_t);

      continue;
    }

    // if we would overlap with the next chunk we combine the two
    if (entry->next &&
        offset + count + conf.max_gap >= entry->next->start) {
      // the new size is the delta between our start and the total end of the next chunk
      mc_entry_t *next = entry->next;
      uint64_t new_size = next->start + next->size - entry->start;
      uint64_t new_count = next->start - entry->start + next->count;

      dprint("Asked to write %lu -> %lu\r\n", offset, offset + count);
      dprint("combining %ld->%ld(%ld) and %ld->%ld(%ld)\r\n",
             entry->start, entry->start + entry->count, entry->start + entry->size,
             next->start, next->start + next->count, next->start + next->size);
      dprint("new range %ld->%llu(%lld)\r\n", entry->start, entry->start + new_count, entry->start + new_size);


      ErlNifUInt64 *new_data = (ErlNifUInt64 *) mc_alloc(new_size * sizeof(ErlNifUInt64));
#ifdef TAGGED
      for (int i = 0; i < new_size; i++) {
        new_data[i] = TAG_DATA_L;
      }
#endif


      // recalculate the allocation
      metric->alloc -= (entry->size * sizeof(ErlNifUInt64));
      gen->alloc -= (entry->size * sizeof(ErlNifUInt64));

      metric->alloc -= (next->size * sizeof(ErlNifUInt64));
      gen->alloc -= (next->size * sizeof(ErlNifUInt64));

      metric->alloc += new_size * sizeof(ErlNifUInt64);
      gen->alloc += new_size * sizeof(ErlNifUInt64);


      entry->next = next->next;
      // copy, free and reassign old data
      memcpy(new_data, entry->data, entry->count * sizeof(ErlNifUInt64));
      mc_free(entry->data);
      entry->data = new_data;
      // set new size and count
      entry->size = new_size;

      // now we calculate the delta of old and new start to get the offset in the
      // new array to copy data to then free it
      dprint("2nd chunk offset: %ld\r\n", next->start - entry->start);
      dprint("copying points: %ld\r\n", next->start - entry->start);

      dprint("Filling between %ld and %ld\r\n", entry->count, next->start - entry->start);
      for (int i = entry->count; i < next->start - entry->start; i++) {
        entry->data[i] = 0;
      }

      entry->count = new_count;

      memcpy(entry->data + next->start - entry->start, next->data, next->count* sizeof(ErlNifUInt64));
      mc_free(next->data);
      mc_free(next);
      // if we don't have a next we are now the tail!
      if (!entry->next) {
        metric->tail = entry;
      }
      // we go back to the start of the loop in case we'd overlap multiple ones (oh my!)
      continue;
    };

    // the offset in our data array

    if (internal_offset + count >= entry->size) {
      uint64_t new_size = entry->size * 2;
      // keep growing untill we're sure we have the right size
      while (internal_offset + count >= new_size) {
        new_size *= 2;
      }

      // prevent oddmath we just  removethe oldsizeand then add
      // the new size
      metric->alloc -= (entry->size * sizeof(ErlNifUInt64));
      gen->alloc -= (entry->size * sizeof(ErlNifUInt64));

      ErlNifUInt64 *new_data = (ErlNifUInt64 *) mc_alloc(new_size * sizeof(ErlNifUInt64));
#ifdef TAGGED
      for (int i = 0; i < new_size; i++) {
        new_data[i] = TAG_DATA_L;
      }
#endif

      memcpy(new_data, entry->data, entry->size * sizeof(ErlNifUInt64));
      mc_free(entry->data);
      entry->data = new_data;
      entry->size = new_size;
      metric->alloc += (entry->size * sizeof(ErlNifUInt64));
      gen->alloc += (entry->size * sizeof(ErlNifUInt64));
    }

    // fill gap with zeros
    for (int i = entry->count; i < internal_offset; i++) {
      entry->data[i] = 0;
    }

    memcpy(entry->data + internal_offset, values, count * sizeof(ErlNifUInt64));

    entry->count = MAX(offset - entry->start + count, entry->count);

    if (!entry->next) {
      metric->tail = entry;
    }
    return;
  } while (entry);
}

mc_metric_t *
check_limit(ErlNifEnv* env, mcache_t *cache, uint64_t max_alloc, uint64_t bucket) {
  if (max_alloc > cache->g0.alloc +
      cache->g1.alloc +
      cache->g2.alloc) {
    return NULL;
  }
  // If we don't have g2 entries we age so g1 becomes
  // g2
  mc_gen_t *gen = &(cache->g2);

  if (gen->alloc == 0) {
    gen = &(cache->g1);
  }

  // If we still have no g2 entries we aga again,
  // this way g0 effectively becomes g0
  if (gen->alloc == 0) {
    gen = &(cache->g0);
  }

  // if we still have no g2 entries we know it's
  // all for nothing and just give up
  if (gen->alloc == 0) {
    return NULL;
  }

  //we know we have at least 1 entrie so we grab the alst one
  // and reduce the count and reduce the alloc;

  //TODO: This is not good!
  mc_metric_t *metric = NULL;
  mc_sub_bucket_t *largest_sub =  NULL;
  mc_bucket_t *largest_bkt =  NULL;
  int largest_idx = 0;
  // try to find a metric using the largest cache first
  for (int b = 0; b < cache->conf.buckets; b++) {
    if (gen->buckets[b].largest[0] &&
        (!metric || metric->alloc < gen->buckets[b].largest[0]->alloc)) {
      largest_bkt = &(gen->buckets[b]);
      metric = largest_bkt->largest[0];
      largest_sub = &(largest_bkt->subs[subid(metric->hash)]);
    }
  }
  if (metric) {
    largest_idx = 0;
    remove_largest(largest_bkt, metric);
    // find the index of the metric
    while (largest_idx < SUBS && largest_sub->metrics[largest_idx] != metric) {
      largest_idx++;
    }
    // We didn't found anything that is problematic so we pretend it never
    // happend, the metric is already removed from the largest index
    if (largest_idx == SUBS) {
      dprint("Oh my\r\n");
      largest_idx = 0;
      metric = NULL;
      largest_bkt = NULL;
      largest_sub = NULL;
    }
  }
  for (int i = 0; i < cache->conf.buckets; i++) {
    // we break if we found a sutiable metric either in this loop or the loop before
    if (metric) {
      break;
    }

    // We itterate through all buckets starting after the bucket we just edited
    // that way we avoid always changing the same buket over and over
    int b = (i + bucket + 1) % cache->conf.buckets;
    mc_bucket_t *bkt = &(gen->buckets[b]);
    for (int sub = 0; sub < SUBS; sub++) {
      // If we found a non empty bucket we find the largest metric in there to
      // evict, that way we can can free up the 'most sensible' thing;
      for (int j = 0; j < bkt->subs[sub].count; j++) {
        if (!metric || bkt->subs[sub].metrics[j]->alloc >= metric->alloc) {
          largest_idx = j;
          largest_sub = &(bkt->subs[sub]);
          metric = largest_sub->metrics[j];
        }
      }
    }

  }
  if (metric) {
    if (largest_idx != largest_sub->count - 1) {
      largest_sub->metrics[largest_idx] = largest_sub->metrics[largest_sub->count - 1];
    }
    largest_sub->count--;
    gen->alloc -= metric->alloc;
    return metric;
  }
  return NULL;
  // now we work on exporting the metric
}

static ERL_NIF_TERM
insert_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  mc_metric_t *metric;
  ErlNifUInt64 offset;
  ErlNifBinary value;
  ErlNifBinary name_bin;

  ERL_NIF_TERM data;
  ERL_NIF_TERM name;
  unsigned char *namep;

  uint64_t bucket;
  if (argc != 4) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };
  if (!enif_inspect_binary(env, argv[1], &name_bin)) {
    return enif_make_badarg(env);
  };
  if (!enif_get_uint64(env, argv[2], &offset)) {
    return enif_make_badarg(env);
  };
  if (!enif_inspect_binary(env, argv[3], &value)) {
    return enif_make_badarg(env);
  };
  if (value.size % sizeof(ErlNifUInt64) && value.size >= 8) {
    return enif_make_badarg(env);
  }
  uint64_t hash = XXH64(name_bin.data, name_bin.size, cache->conf.hash_seed) ;
  bucket = hash % cache->conf.buckets;
  dprint("INSERT[%llu]: %llu %lu@%lu\r\n", bucket, hash, value.size / 8, offset);

  metric = get_metric(cache, hash, name_bin.size, name_bin.data);
  // Add the datapoint
  add_point(cache->conf, &(cache->g0), metric, offset, value.size / 8, (ErlNifUInt64 *) value.data);
  // update largest
  insert_largest(&(cache->g0.buckets[bucket]), metric);

  cache->inserts++;
  if (cache->inserts > cache->conf.age_cycle) {
    //age(cache);
    cache->age++;
    cache->inserts = 0;
  }
  // We now check for overflow note that metric is re-used here!
  metric = check_limit(env, cache, cache->conf.max_alloc, bucket);
  if (metric) {
    data = serialize_metric(env, metric);
    namep = enif_make_new_binary(env, metric->name_len, &name);
    memcpy(namep, metric->name, metric->name_len);
    free_metric(metric);
    return  enif_make_tuple3(env,
                             atom_overflow,
                             name,
                             data);
  }
  return atom_ok;
};

static ERL_NIF_TERM
pop_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  mc_metric_t *metric = NULL;
  ERL_NIF_TERM data;
  ERL_NIF_TERM name;
  unsigned char *namep;

  if (argc != 1) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };

  // We cheat here, we can create a 'pop' by just checkig for a 0 limit.
  metric = check_limit(env, cache, 0, 0);
  if (metric) {
    data = serialize_metric(env, metric);
    namep = enif_make_new_binary(env, metric->name_len, &name);
    memcpy(namep, metric->name, metric->name_len);
    free_metric(metric);
    return  enif_make_tuple3(env,
                             atom_ok,
                             name,
                             data);
  }
  return atom_undefined;

};

static ERL_NIF_TERM
get_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  mc_metric_t *metric;
  ErlNifBinary name_bin;
  if (argc != 2) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };

  if (!enif_inspect_binary(env, argv[1], &name_bin)) {
    return enif_make_badarg(env);
  };

  uint64_t hash = XXH64(name_bin.data, name_bin.size, cache->conf.hash_seed) ;
  metric = find_metric(cache, hash, name_bin.size, name_bin.data);
  if (metric) {
    return  enif_make_tuple2(env,
                             atom_ok,
                             serialize_metric(env, metric));
  } else {
    return atom_undefined;
  }
};

static ERL_NIF_TERM
take_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  mc_metric_t *metric;
  ErlNifBinary name_bin;
  if (argc != 2) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };

  if (!enif_inspect_binary(env, argv[1], &name_bin)) {
    return enif_make_badarg(env);
  };

  uint64_t hash = XXH64(name_bin.data, name_bin.size, cache->conf.hash_seed) ;
  metric = find_metric_and_remove(cache, hash, name_bin.size, name_bin.data);
  if (metric) {
    ERL_NIF_TERM res = enif_make_tuple2(env,
                                        atom_ok,
                                        serialize_metric(env, metric));
    free_metric(metric);
    return res;
  } else {
    return atom_undefined;
  }
};

static ERL_NIF_TERM
remove_prefix_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  ErlNifBinary pfx_bin;
  if (argc != 2) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };

  if (!enif_inspect_binary(env, argv[1], &pfx_bin)) {
    return enif_make_badarg(env);
  };

  uint64_t count = remove_prefix(cache, pfx_bin.size, pfx_bin.data);
  return  enif_make_tuple2(env,
                           atom_ok,
                           enif_make_uint64(env, count));
};


void print_gen(mc_conf_t conf, mc_gen_t gen) {
  int size = 0;
  int count = 0;
  for (int i = 0; i < conf.buckets; i++) {
    for (int sub = 0; sub < SUBS; sub++) {
      size += gen.buckets[i].subs[sub].size;
      count += gen.buckets[i].subs[sub].count;
    }
  };
  printf("Cache: [c: %d |s: %d|a: %zu]:\r\n",  count, size, gen.alloc);
  for (int i = 0; i < conf.buckets; i++) {
    for (int sub = 0; sub < SUBS; sub++) {
      for(int j = 0; j < gen.buckets[i].subs[sub].count; j++) {
        print_metric(gen.buckets[i].subs[sub].metrics[j]);
      }
    }
  }
}

static ERL_NIF_TERM
print_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  if (argc != 1) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };
  print_gen(cache->conf, cache->g0);
  print_gen(cache->conf, cache->g1);
  print_gen(cache->conf, cache->g2);
  fflush(stdout);
  return  atom_ok;
};
static ERL_NIF_TERM
age_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  if (argc != 1) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };
  age(cache);
  return  atom_ok;
};

static ERL_NIF_TERM
gen_stats(ErlNifEnv* env, mc_conf_t conf, mc_gen_t gen) {
  int size = 0;
  int count = 0;
  for (int i = 0; i < conf.buckets; i++) {
    for (int sub = 0; sub < SUBS; sub++) {
      size += gen.buckets[i].subs[sub].size;
      count += gen.buckets[i].subs[sub].count;
    }
  };

  return enif_make_list3(env,
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "alloc"),
                                          enif_make_uint64(env, gen.alloc)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "count"),
                                          enif_make_uint64(env, count)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "size"),
                                          enif_make_uint64(env, size)));

}

static ERL_NIF_TERM
conf_info(ErlNifEnv* env, mc_conf_t conf) {
  return enif_make_list5(env,
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "buckets"),
                                          enif_make_uint64(env, conf.buckets)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "age_cycle"),
                                          enif_make_uint64(env, conf.age_cycle)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "initial_data_size"),
                                          enif_make_uint64(env, conf.initial_data_size)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "initial_entries"),
                                          enif_make_uint64(env, conf.initial_entries)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "max_alloc"),
                                          enif_make_uint64(env, conf.max_alloc)));
}



static ERL_NIF_TERM
stats_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  if (argc != 1) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };

  return enif_make_list7(env,
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "age"),
                                          enif_make_uint64(env, cache->age)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "inserts"),
                                          enif_make_uint64(env, cache->inserts)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "conf"),
                                          conf_info(env, cache->conf)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "total_alloc"),
                                          enif_make_uint64(env,
                                                           cache->g0.alloc +
                                                           cache->g1.alloc +
                                                           cache->g2.alloc)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "gen0"),
                                          gen_stats(env, cache->conf, cache->g0)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "gen1"),
                                          gen_stats(env, cache->conf, cache->g1)),
                         enif_make_tuple2(env,
                                          enif_make_atom(env, "gen2"),
                                          gen_stats(env, cache->conf, cache->g2)));
};
static ERL_NIF_TERM
is_empty_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) {
  mcache_t *cache;
  if (argc != 1) {
    return enif_make_badarg(env);
  };
  if (!enif_get_resource(env, argv[0], mcache_t_handle, (void **)&cache)) {
    return enif_make_badarg(env);
  };
  if (cache->g0.alloc == 0 && cache->g1.alloc == 0 && cache->g2.alloc == 0) {
    return enif_make_atom(env, "true");
  } else {
    return enif_make_atom(env, "false");
  }
};

static ErlNifFunc nif_funcs[] = {
  {"new", 7, new_nif},
  {"print", 1, print_nif},
  {"stats", 1, stats_nif},
  {"is_empty", 1, is_empty_nif},
  {"age", 1, age_nif},
  {"pop", 1, pop_nif},
  {"remove_prefix", 2, remove_prefix_nif},
  {"get", 2, get_nif},
  {"take", 2, take_nif},
  {"insert", 4, insert_nif},
};

// Initialize this NIF library.
//
// Args: (MODULE, ErlNifFunc funcs[], load, reload, upgrade, unload)
// Docs: http://erlang.org/doc/man/erl_nif.html#ERL_NIF_INIT

ERL_NIF_INIT(mcache, nif_funcs, &load, NULL, &upgrade, NULL);

/*
  {ok, H} = mcache:new(50*10*8).
  mcache:insert(H, <<"1">>,  1, <<1:64>>).
  mcache:insert(H, <<"2">>,  1, <<1:64>>).
  mcache:insert(H, <<"3">>,  1, <<1:64>>).
  mcache:insert(H, <<"4">>,  1, <<1:64>>).
  mcache:insert(H, <<"5">>,  1, <<1:64>>).
  mcache:insert(H, <<"6">>,  1, <<1:64>>).
  mcache:insert(H, <<"7">>,  1, <<1:64>>).
  mcache:insert(H, <<"8">>,  1, <<1:64>>).
  mcache:insert(H, <<"9">>,  1, <<1:64>>).
  mcache:age(H).

  mcache:insert(H, <<"test2">>, 1, <<2:64>>).
  mcache:age(H).
  mcache:insert(H, <<"test">>,  6, <<3:64>>).
  mcache:print(H).

  mcache:get(H, <<"test">>).

  mcache:insert(H, <<"test1">>, 1, <<2:64>>).
  mcache:insert(H, <<"test2">>, 1, <<2:64>>).
  mcache:insert(H, <<"test3">>, 1, <<2:64>>).
  mcache:insert(H, <<"test4">>, 1, <<2:64>>).
  mcache:insert(H, <<"test5">>, 1, <<2:64>>).
  mcache:insert(H, <<"test6">>, 1, <<2:64>>).
  mcache:insert(H, <<"test7">>, 1, <<2:64>>).
  mcache:insert(H, <<"test8">>, 1, <<2:64>>).
  mcache:insert(H, <<"test9">>, 1, <<2:64>>).
  mcache:insert(H, <<"testa">>, 1, <<2:64>>).
  mcache:insert(H, <<"testb">>, 1, <<2:64>>).
  mcache:insert(H, <<"testc">>, 1, <<2:64>>).
  mcache:get(H, <<"test">>).
*/


/*
  {ok, H} = mcache:new(10000).
  mcache:insert(H, <<"test">>,  1, <<2:64>>).
  mcache:age(H).
  mcache:age(H).
  mcache:insert(H, <<"test1">>,  2, <<2:64>>).
  mcache:print(H).
  mcache:age(H).
*/

/*
  {ok, H} =  mcache:new(726).
  mcache:insert(H, <<>>, 0, <<0,0,0,0,0,0,0,0>>).
  mcache:print(H).
  mcache:age(H).
  mcache:print(H).
  mcache:insert(H, <<>>, 0, <<0,0,0,0,0,0,0,0>>).
  mcache:print(H).
*/