#include "erl_nif.h" #include #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). */