You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2019/01/30 13:35:44 UTC
[couchdb-hqueue] branch master created (now 151c9bc)
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/couchdb-hqueue.git.
at 151c9bc Initial import
This branch includes the following new commits:
new 151c9bc Initial import
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[couchdb-hqueue] 01/01: Initial import
Posted by rn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rnewson pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/couchdb-hqueue.git
commit 151c9bc8de819b5cdb90b3cc57786c37c4991f00
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Jan 30 12:56:06 2019 +0000
Initial import
---
.gitignore | 8 +
Makefile | 28 +++
README | 0
c_src/hqueue.c | 318 +++++++++++++++++++++++++
c_src/hqueue.h | 60 +++++
c_src/hqueue_nif.c | 601 ++++++++++++++++++++++++++++++++++++++++++++++++
c_src/valgrind_sample.c | 72 ++++++
rebar.config | 13 ++
src/hqueue.app.src | 18 ++
src/hqueue.erl | 160 +++++++++++++
test/hqueue_proper.erl | 33 +++
test/hqueue_statem.erl | 115 +++++++++
test/hqueue_tests.erl | 253 ++++++++++++++++++++
13 files changed, 1679 insertions(+)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..b54f306
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,8 @@
+.hqueue.dev
+*.o
+*.beam
+*.so
+.eunit
+erl_crash.dump
+rebar
+
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..0116693
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,28 @@
+REBAR?=rebar
+
+all: build
+
+
+clean:
+ $(REBAR) clean
+ rm -rf .eunit
+ rm -f test/*.beam
+ rm -rf priv/*.so
+ rm -f c_src/valgrind_sample
+
+
+distclean: clean
+ git clean -fxd
+
+
+build:
+ $(REBAR) compile
+
+
+check: build
+ $(REBAR) eunit
+
+check-valgrind:
+ cc -I c_src/ -g -Wall -Werror c_src/hqueue.c c_src/valgrind_sample.c -o c_src/valgrind_sample
+ valgrind --leak-check=yes c_src/valgrind_sample
+
diff --git a/README b/README
new file mode 100644
index 0000000..e69de29
diff --git a/c_src/hqueue.c b/c_src/hqueue.c
new file mode 100644
index 0000000..f02f251
--- /dev/null
+++ b/c_src/hqueue.c
@@ -0,0 +1,318 @@
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations under
+// the License.
+
+#include <assert.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "hqueue.h"
+
+
+struct hqueue
+{
+ int version;
+ uint32_t idx;
+ uint32_t max_elems;
+ uint32_t heap_size;
+ hqnode_t* heap; // one based index
+};
+
+
+struct hqnode
+{
+ double priority;
+ void* value;
+};
+
+
+static inline void
+hqueue_exchange(hqueue_t* hqueue, int i, int j)
+{
+ hqnode_t tmp;
+
+ tmp = hqueue->heap[i];
+ hqueue->heap[i] = hqueue->heap[j];
+ hqueue->heap[j] = tmp;
+ return;
+}
+
+
+static inline int
+hqueue_less(hqueue_t* hqueue, int i, int j)
+{
+ return hqueue->heap[i].priority < hqueue->heap[j].priority;
+}
+
+
+static void
+hqueue_fix_up(hqueue_t* hqueue, int k)
+{
+ while(k > 1 && hqueue_less(hqueue, k/2, k)) {
+ hqueue_exchange(hqueue, k/2, k);
+ k = k/2;
+ }
+ return;
+}
+
+
+static void
+hqueue_fix_down(hqueue_t* hqueue, int k)
+{
+ int j;
+ int n = hqueue->idx;
+
+ while(2*k <= n) {
+ j = 2*k;
+ if(j < n && hqueue_less(hqueue, j, j+1)) {
+ j++;
+ }
+ if(!hqueue_less(hqueue, k, j)) {
+ break;
+ }
+ hqueue_exchange(hqueue, k, j);
+ k = j;
+ }
+ return;
+}
+
+
+hqueue_t*
+hqueue_new(uint32_t max_elems, uint32_t heap_size)
+{
+ hqueue_t* hqueue = NULL;
+ size_t total_heap_size;
+
+ if(max_elems == 0 || heap_size == 0) {
+ return NULL;
+ }
+
+ if(max_elems < heap_size) {
+ heap_size = max_elems;
+ }
+
+ hqueue = HQUEUE_ALLOC(sizeof(hqueue_t));
+ if(hqueue == NULL) {
+ return NULL;
+ }
+
+ memset(hqueue, '\0', sizeof(hqueue_t));
+ hqueue->version = HQ_VERSION;
+ hqueue->max_elems = max_elems;
+ hqueue->heap_size = heap_size;
+ hqueue->idx = 0;
+
+ total_heap_size = sizeof(hqnode_t) * (hqueue->heap_size+1);
+
+ hqueue->heap = (hqnode_t*) HQUEUE_ALLOC(total_heap_size);
+
+ if(hqueue->heap == NULL ) {
+ HQUEUE_FREE(hqueue);
+ return NULL;
+ }
+
+ memset(hqueue->heap, '\0', total_heap_size);
+
+ return hqueue;
+}
+
+
+void
+hqueue_free(hqueue_t* hqueue)
+{
+ HQUEUE_FREE(hqueue->heap);
+ HQUEUE_FREE(hqueue);
+
+ return;
+}
+
+
+void
+hqueue_free2(hqueue_t* hqueue, void (*free_node)(void* node))
+{
+ uint32_t i;
+
+ for(i = 1; i < hqueue->heap_size + 1; i++) {
+ if(i <= hqueue->idx) {
+ free_node(hqueue->heap[i].value);
+ } else {
+ assert(hqueue->heap[i].value == NULL && "inactive elements must be NULL");
+ }
+ }
+
+ hqueue_free(hqueue);
+
+ return;
+}
+
+
+// Extraction order is undefined for entries with duplicate priorities
+int
+hqueue_extract_max(hqueue_t* hqueue, double* priority, void** value)
+{
+ if(hqueue->idx <= 0) {
+ return 0;
+ }
+
+ hqueue_exchange(hqueue, 1, hqueue->idx);
+
+ *priority = hqueue->heap[hqueue->idx].priority;
+ *value = hqueue->heap[hqueue->idx].value;
+
+ hqueue->heap[hqueue->idx].value = NULL;
+
+ hqueue->idx--; // heap uses one based index, so we decrement after
+ hqueue_fix_down(hqueue, 1);
+
+ return 1;
+}
+
+
+void
+hqueue_get_elem(hqueue_t* hqueue, uint32_t idx, double *priority, void** value)
+{
+ *priority = hqueue->heap[idx].priority;
+ *value = hqueue->heap[idx].value;
+
+ return;
+}
+
+
+static int
+hqueue_maybe_resize(hqueue_t* hqueue)
+{
+ uint32_t min_resize;
+
+ if(hqueue->idx + 1 > hqueue->heap_size) {
+ if(hqueue->idx * HQ_SCALE_FACTOR > hqueue->max_elems) {
+ min_resize = hqueue->max_elems;
+ } else {
+ min_resize = hqueue->idx * HQ_SCALE_FACTOR;
+ }
+ return hqueue_resize_heap(hqueue, min_resize);
+ }
+
+ return 1;
+}
+
+
+int
+hqueue_insert(hqueue_t* hqueue, double priority, void* value)
+{
+ if(hqueue->idx >= hqueue->max_elems) {
+ return 0;
+ }
+
+ if(!hqueue_maybe_resize(hqueue)) {
+ return 0;
+ }
+
+ hqueue->idx++; // heap uses one based index, so we increment first
+ hqueue->heap[hqueue->idx].priority = priority;
+ hqueue->heap[hqueue->idx].value = value;
+
+ hqueue_fix_up(hqueue, hqueue->idx);
+
+ return 1;
+}
+
+
+uint32_t
+hqueue_size(hqueue_t* hqueue)
+{
+ return hqueue->idx;
+}
+
+
+uint32_t
+hqueue_heap_size(hqueue_t* hqueue)
+{
+ return hqueue->heap_size;
+}
+
+
+uint32_t
+hqueue_max_elems(hqueue_t* hqueue)
+{
+ return hqueue->max_elems;
+}
+
+
+void
+hqueue_scale_by(hqueue_t* hqueue, double factor)
+{
+ uint32_t i;
+
+ for(i = 1; i <= hqueue->idx && i <= hqueue->heap_size; i++) {
+ hqueue->heap[i].priority *= factor;
+ }
+
+ return;
+}
+
+
+uint32_t
+hqueue_resize_heap(hqueue_t* hqueue, uint32_t new_heap_size)
+{
+ uint32_t old_heap_size;
+ size_t total_heap_size;
+ hqnode_t* tmp_heap;
+ uint32_t i;
+
+ if(hqueue->idx > new_heap_size) {
+ return 0;
+ }
+
+ total_heap_size = sizeof(hqnode_t) * (new_heap_size+1);
+ old_heap_size = hqueue->heap_size;
+
+ if((tmp_heap = (hqnode_t*) HQUEUE_ALLOC(total_heap_size)) == NULL) {
+ return 0;
+ }
+
+ memset(tmp_heap, '\0', total_heap_size);
+
+ for(i = 1; i <= hqueue->idx && i <= old_heap_size; i++) {
+ if(i <= hqueue->idx) {
+ tmp_heap[i] = hqueue->heap[i];
+ hqueue->heap[i].value = NULL;
+ } else {
+ assert(hqueue->heap[i].value == NULL &&
+ "unexpected NULL element during heap resize");
+ }
+ }
+
+ HQUEUE_FREE(hqueue->heap);
+ hqueue->heap = tmp_heap;
+ hqueue->heap_size = new_heap_size;
+
+ return old_heap_size;
+}
+
+
+int
+hqueue_set_max_elems(hqueue_t* hqueue, uint32_t new_max_elems)
+{
+ uint32_t old_max_elems;
+
+ if(hqueue->heap_size > new_max_elems) {
+ if(!hqueue_resize_heap(hqueue, new_max_elems)) {
+ return 0;
+ }
+ }
+
+ old_max_elems = hqueue->max_elems;
+ hqueue->max_elems = new_max_elems;
+
+ return old_max_elems;
+}
diff --git a/c_src/hqueue.h b/c_src/hqueue.h
new file mode 100644
index 0000000..4e422e4
--- /dev/null
+++ b/c_src/hqueue.h
@@ -0,0 +1,60 @@
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations under
+// the License.
+
+#pragma once
+
+
+#include <stdint.h>
+
+#define HQ_VERSION 0
+#define HQ_SCALE_FACTOR 2 // heap expansion scale factor
+
+
+// Override the default memory allocator to use the Erlang versions.
+// This bubbles up memory usage for the NIF into Erlang stats.
+#ifdef HQ_ENIF_ALLOC
+
+#include "erl_nif.h"
+
+#define HQUEUE_ALLOC enif_alloc
+#define HQUEUE_FREE enif_free
+
+#else
+
+#define HQUEUE_ALLOC malloc
+#define HQUEUE_FREE free
+
+#endif
+
+
+typedef struct hqnode hqnode_t;
+typedef struct hqueue hqueue_t;
+
+
+hqueue_t* hqueue_new(uint32_t max_elems, uint32_t heap_size);
+
+void hqueue_free(hqueue_t* hqueue);
+void hqueue_free2(hqueue_t* hqueue, void (*free_node)(void* node));
+
+int hqueue_insert(hqueue_t* hqueue, double priority, void* val);
+int hqueue_extract_max(hqueue_t* hqueue, double* priority, void** value);
+void hqueue_get_elem(hqueue_t* hqueue, uint32_t idx, double *priority,
+ void** value);
+
+uint32_t hqueue_size(hqueue_t* hqueue);
+uint32_t hqueue_heap_size(hqueue_t* hqueue);
+
+uint32_t hqueue_max_elems(hqueue_t* hqueue);
+int hqueue_set_max_elems(hqueue_t* hqueue, uint32_t new_max_elems);
+
+void hqueue_scale_by(hqueue_t* hqueue, double factor);
+uint32_t hqueue_resize_heap(hqueue_t* hqueue, uint32_t new_heap_size);
diff --git a/c_src/hqueue_nif.c b/c_src/hqueue_nif.c
new file mode 100644
index 0000000..7cbc5e2
--- /dev/null
+++ b/c_src/hqueue_nif.c
@@ -0,0 +1,601 @@
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations under
+// the License.
+
+#include <assert.h>
+#include <string.h>
+#include <stdio.h>
+
+#include "hqueue.h"
+
+
+typedef struct
+{
+ ERL_NIF_TERM atom_ok;
+ ERL_NIF_TERM atom_error;
+ ERL_NIF_TERM atom_value;
+ ERL_NIF_TERM atom_empty;
+ ERL_NIF_TERM atom_full;
+ ERL_NIF_TERM atom_max_elems;
+ ERL_NIF_TERM atom_heap_size;
+ ERL_NIF_TERM atom_too_small;
+ ErlNifResourceType* res_hqueue;
+} hqueue_priv;
+
+
+typedef struct
+{
+ ErlNifEnv* env;
+ ERL_NIF_TERM value;
+} hqnode_nif_t;
+
+
+typedef struct
+{
+ int version;
+ uint64_t gen;
+ hqueue_t* hqueue;
+ ErlNifPid p;
+} hqueue_nif_t;
+
+
+static const uint32_t default_max_elems = UINT32_MAX-1;
+static const uint32_t default_heap_size = 1024;
+
+
+static inline ERL_NIF_TERM
+make_atom(ErlNifEnv* env, const char* name)
+{
+ ERL_NIF_TERM ret;
+ if(enif_make_existing_atom(env, name, &ret, ERL_NIF_LATIN1)) {
+ return ret;
+ }
+ return enif_make_atom(env, name);
+}
+
+
+static inline ERL_NIF_TERM
+make_ok(ErlNifEnv* env, hqueue_priv* priv, ERL_NIF_TERM value)
+{
+ return enif_make_tuple2(env, priv->atom_ok, value);
+}
+
+
+static inline ERL_NIF_TERM
+make_error(ErlNifEnv* env, hqueue_priv* priv, ERL_NIF_TERM reason)
+{
+ return enif_make_tuple2(env, priv->atom_error, reason);
+}
+
+
+static inline int
+check_pid(ErlNifEnv* env, hqueue_nif_t* hqueue_nif)
+{
+ ErlNifPid pid;
+ enif_self(env, &pid);
+
+ if(enif_compare(pid.pid, hqueue_nif->p.pid) == 0) {
+ return 1;
+ }
+
+ return 0;
+}
+
+
+void
+hqueue_nif_node_free(hqnode_nif_t* hqnode_nif)
+{
+ enif_free_env(hqnode_nif->env);
+ enif_free(hqnode_nif);
+
+ return;
+}
+
+
+void
+hqueue_nif_node_free_ext(void* node)
+{
+ hqueue_nif_node_free((hqnode_nif_t*) node);
+
+ return;
+}
+
+
+hqnode_nif_t*
+hqueue_nif_node_alloc()
+{
+ hqnode_nif_t* node = (hqnode_nif_t*) enif_alloc(sizeof(hqnode_nif_t*));
+
+ memset(node, 0, sizeof(hqnode_nif_t));
+
+ node->env = enif_alloc_env();
+
+ return node;
+}
+
+
+static int
+get_uint_param(ErlNifEnv* env, ERL_NIF_TERM value, ERL_NIF_TERM atom, uint32_t* p)
+{
+ const ERL_NIF_TERM* tuple;
+ int arity;
+
+ if(!enif_get_tuple(env, value, &arity, &tuple)) {
+ return 0;
+ }
+
+ if(arity != 2) {
+ return 0;
+ }
+
+ if(enif_compare(tuple[0], atom) != 0) {
+ return 0;
+ }
+
+ if(!enif_get_uint(env, tuple[1], p)) {
+ return 0;
+ }
+
+ return 1;
+}
+
+
+static inline hqueue_nif_t*
+hqueue_nif_create_int(ErlNifEnv* env, hqueue_priv* priv, uint32_t max_elems,
+ uint32_t heap_size)
+{
+ hqueue_nif_t* hqueue_nif = NULL;
+
+ assert(priv != NULL && "missing private data member");
+
+ hqueue_nif = (hqueue_nif_t*) enif_alloc_resource(
+ priv->res_hqueue, sizeof(hqueue_nif_t));
+ memset(hqueue_nif, 0, sizeof(hqueue_nif_t));
+ hqueue_nif->version = HQ_VERSION;
+
+ hqueue_nif->hqueue = hqueue_new(max_elems, heap_size);
+
+ if(hqueue_nif->hqueue == NULL ) {
+ enif_release_resource(hqueue_nif);
+ return NULL;
+ }
+
+ enif_self(env, &(hqueue_nif->p));
+
+ return hqueue_nif;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_new(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+ ERL_NIF_TERM opts;
+ ERL_NIF_TERM value;
+ uint32_t max_elems = default_max_elems;
+ uint32_t heap_size = default_heap_size;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ opts = argv[0];
+ if(!enif_is_list(env, opts)) {
+ return enif_make_badarg(env);
+ }
+
+ while(enif_get_list_cell(env, opts, &value, &opts)) {
+ if(get_uint_param(env, value, priv->atom_max_elems, &max_elems)) {
+ continue;
+ } else if(get_uint_param(env, value, priv->atom_heap_size, &heap_size)) {
+ continue;
+ } else {
+ return enif_make_badarg(env);
+ }
+ }
+
+ hqueue_nif = hqueue_nif_create_int(env, priv, max_elems, heap_size);
+ if(hqueue_nif == NULL) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_resource(env, hqueue_nif);
+ enif_release_resource(hqueue_nif);
+
+ return make_ok(env, priv, ret);
+}
+
+
+static void
+hqueue_nif_free(ErlNifEnv* env, void* obj)
+{
+ hqueue_nif_t* hqueue_nif = (hqueue_nif_t*) obj;
+
+ hqueue_free2(hqueue_nif->hqueue, hqueue_nif_node_free_ext);
+
+ return;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_extract_max(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ hqnode_nif_t* hqnode_nif;
+ double tmp_priority;
+ ERL_NIF_TERM ret;
+ ERL_NIF_TERM priority;
+ ERL_NIF_TERM value;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if (!hqueue_extract_max(hqueue_nif->hqueue, &tmp_priority, (void**) &hqnode_nif)) {
+ return make_error(env, priv, priv->atom_empty);
+ }
+
+ priority = enif_make_double(env, tmp_priority);
+ value = enif_make_copy(env, hqnode_nif->value);
+ ret = enif_make_tuple2(env, priority, value);
+
+ hqueue_nif_node_free(hqnode_nif);
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_insert(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ hqnode_nif_t* hqnode_nif;
+ ERL_NIF_TERM ret;
+ double priority;
+
+ if(argc != 3) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_double(env, argv[1], &priority)) {
+ return enif_make_badarg(env);
+ }
+
+ if(priority < 0.0) {
+ return enif_make_badarg(env);
+ }
+
+ hqnode_nif = hqueue_nif_node_alloc();
+ hqnode_nif->value = enif_make_copy(hqnode_nif->env, argv[2]);
+
+ if (!hqueue_insert(hqueue_nif->hqueue, priority, (void*) hqnode_nif)) {
+ return make_error(env, priv, priv->atom_full);
+ }
+
+ ret = priv->atom_ok;
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_size(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_uint64(env, hqueue_size(hqueue_nif->hqueue));
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_heap_size(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_uint64(env, hqueue_heap_size(hqueue_nif->hqueue));
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_max_elems(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_uint64(env, hqueue_max_elems(hqueue_nif->hqueue));
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_to_list(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ hqueue_t* hqueue;
+ hqnode_nif_t* hqnode_nif;
+ double tmp_priority;
+ ERL_NIF_TERM ret = enif_make_list(env, 0);
+ ERL_NIF_TERM priority;
+ ERL_NIF_TERM value;
+ ERL_NIF_TERM tuple;
+ uint32_t i;
+
+ if(argc != 1) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ hqueue = hqueue_nif->hqueue;
+
+ for (i = 1; i <= hqueue_size(hqueue); i++) {
+ hqueue_get_elem(hqueue, i, &tmp_priority, (void **) &hqnode_nif);
+ priority = enif_make_double(env, tmp_priority);
+ value = enif_make_copy(env, hqnode_nif->value);
+ tuple = enif_make_tuple2(env, priority, value);
+ ret = enif_make_list_cell(env, tuple, ret);
+ }
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_scale_by(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+ double factor;
+
+ if(argc != 2) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_double(env, argv[1], &factor)) {
+ return enif_make_badarg(env);
+ }
+
+ if(factor < 0.0) {
+ return enif_make_badarg(env);
+ }
+
+ hqueue_scale_by(hqueue_nif->hqueue, factor);
+
+ ret = priv->atom_ok;
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_resize_heap(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+ uint32_t new_heap_size;
+ uint32_t old_heap_size;
+
+ if(argc != 2) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_uint(env, argv[1], &new_heap_size)) {
+ return enif_make_badarg(env);
+ }
+
+ if(hqueue_size(hqueue_nif->hqueue) > new_heap_size) {
+ return make_error(env, priv, priv->atom_too_small);
+ }
+
+ if((old_heap_size = hqueue_resize_heap(hqueue_nif->hqueue, new_heap_size)) == 0) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_uint64(env, old_heap_size);
+
+ return ret;
+}
+
+
+static ERL_NIF_TERM
+hqueue_nif_set_max_elems(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+ hqueue_priv* priv = enif_priv_data(env);
+ hqueue_nif_t* hqueue_nif;
+ ERL_NIF_TERM ret;
+ uint32_t new_max_elems;
+ uint32_t old_max_elems;
+
+ if(argc != 2) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_resource(env, argv[0], priv->res_hqueue, (void**) &hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!check_pid(env, hqueue_nif)) {
+ return enif_make_badarg(env);
+ }
+
+ if(!enif_get_uint(env, argv[1], &new_max_elems)) {
+ return enif_make_badarg(env);
+ }
+
+ if(hqueue_size(hqueue_nif->hqueue) > new_max_elems) {
+ return make_error(env, priv, priv->atom_too_small);
+ }
+
+ if ((old_max_elems = hqueue_set_max_elems(hqueue_nif->hqueue, new_max_elems)) == 0) {
+ return enif_make_badarg(env);
+ }
+
+ ret = enif_make_uint64(env, old_max_elems);
+
+ return ret;
+}
+
+
+static int
+load(ErlNifEnv* env, void** priv, ERL_NIF_TERM info)
+{
+ int flags = ERL_NIF_RT_CREATE | ERL_NIF_RT_TAKEOVER;
+ ErlNifResourceType* res;
+
+ hqueue_priv* new_priv = (hqueue_priv*) enif_alloc(sizeof(hqueue_priv));
+ if(new_priv == NULL) {
+ return 1;
+ }
+
+ res = enif_open_resource_type(
+ env, NULL, "hqueue", hqueue_nif_free, flags, NULL);
+ if(res == NULL) {
+ enif_free(new_priv);
+ return 1;
+ }
+ new_priv->res_hqueue = res;
+
+ new_priv->atom_ok = make_atom(env, "ok");
+ new_priv->atom_error = make_atom(env, "error");
+ new_priv->atom_value = make_atom(env, "value");
+ new_priv->atom_empty = make_atom(env, "empty");
+ new_priv->atom_full = make_atom(env, "full");
+ new_priv->atom_max_elems = make_atom(env, "max_elems");
+ new_priv->atom_heap_size = make_atom(env, "heap_size");
+ new_priv->atom_too_small = make_atom(env, "too_small");
+
+ *priv = (void*) new_priv;
+
+ return 0;
+}
+
+
+static int
+upgrade(ErlNifEnv* env, void** priv, void** old_priv, ERL_NIF_TERM info)
+{
+ return load(env, priv, info);
+}
+
+
+static void
+unload(ErlNifEnv* env, void* priv)
+{
+ enif_free(priv);
+ return;
+}
+
+
+static ErlNifFunc funcs[] = {
+ {"new", 1, hqueue_nif_new},
+ {"extract_max", 1, hqueue_nif_extract_max},
+ {"insert", 3, hqueue_nif_insert},
+ {"size", 1, hqueue_nif_size},
+ {"heap_size", 1, hqueue_nif_heap_size},
+ {"max_elems", 1, hqueue_nif_max_elems},
+ {"set_max_elems", 2, hqueue_nif_set_max_elems},
+ {"to_list", 1, hqueue_nif_to_list},
+ {"scale_by", 2, hqueue_nif_scale_by},
+ {"resize_heap", 2, hqueue_nif_resize_heap}
+};
+
+
+ERL_NIF_INIT(hqueue, funcs, &load, NULL, &upgrade, &unload);
diff --git a/c_src/valgrind_sample.c b/c_src/valgrind_sample.c
new file mode 100644
index 0000000..3c78da5
--- /dev/null
+++ b/c_src/valgrind_sample.c
@@ -0,0 +1,72 @@
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not
+// use this file except in compliance with the License. You may obtain a copy of
+// the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations under
+// the License.
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+
+#include "hqueue.h"
+
+
+// Simple test script to stress the public HQueue API.
+// Primary use case is for running this under Valgrind.
+int main(void)
+{
+ int str_len = 100;
+ int iterations = 1000;
+ uint32_t max_elems = 1024;
+ uint32_t heap_size = 64;
+ hqueue_t* hq = hqueue_new(max_elems, heap_size);
+ double priority;
+ double priority_res;
+ char* val;
+ char* val_res;
+ int i;
+
+ assert(max_elems == hqueue_max_elems(hq));
+ assert(heap_size == hqueue_heap_size(hq));
+
+ for(i = 0; i < iterations; i++) {
+ priority = 1234.4321 * i;
+ val = (char*) malloc(str_len + 1);
+
+ if(val == NULL) {
+ return 1;
+ }
+
+ assert(hqueue_size(hq) == i);
+
+ if(snprintf(val, str_len + 1, "Fun string #%d\n", i)) {
+ if(!hqueue_insert(hq, priority, val)) {
+ return 1;
+ }
+ } else {
+ return 1;
+ }
+ }
+
+ hqueue_scale_by(hq, 3.7);
+
+ // Added 1000 elements, so heap size should have expanded to 1024
+ assert(max_elems == hqueue_max_elems(hq));
+ assert(max_elems == hqueue_heap_size(hq));
+
+ if(!hqueue_extract_max(hq, &priority_res, (void**) &val_res)) {
+ return 1;
+ }
+ free(val_res);
+
+ hqueue_free2(hq, free);
+
+ return 0;
+}
+
diff --git a/rebar.config b/rebar.config
new file mode 100644
index 0000000..4c4da01
--- /dev/null
+++ b/rebar.config
@@ -0,0 +1,13 @@
+{deps, [
+ {proper, ".*", {git, "https://github.com/manopapad/proper.git", "master"}}
+]}.
+
+{port_specs, [
+ {"priv/hqueue.so", ["c_src/hqueue*.c"]}
+]}.
+{port_env, [
+ {".*", "CFLAGS", "$CFLAGS -g -Wall -Werror -DHQ_ENIF_ALLOC -O3"}
+ %% {".*", "CFLAGS", "$CFLAGS -g -Wall -Werror -Wextra"}
+]}.
+{eunit_opts, [verbose]}.
+{erl_opts, [debug_info]}.
diff --git a/src/hqueue.app.src b/src/hqueue.app.src
new file mode 100644
index 0000000..a2315ac
--- /dev/null
+++ b/src/hqueue.app.src
@@ -0,0 +1,18 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http:%www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+{application, hqueue, [
+ {description, "Heap based priority queue NIF"},
+ {vsn, git},
+ {registered, []},
+ {applications, [kernel, stdlib]}
+]}.
diff --git a/src/hqueue.erl b/src/hqueue.erl
new file mode 100644
index 0000000..eec8b98
--- /dev/null
+++ b/src/hqueue.erl
@@ -0,0 +1,160 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http:%www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(hqueue).
+
+
+-on_load(init/0).
+
+
+-export([
+ new/0,
+ new/1,
+
+ extract_max/1,
+ insert/3,
+
+ from_list/1,
+ from_list/2,
+ to_list/1,
+
+ heap_size/1,
+ info/1,
+ is_empty/1,
+ max_elems/1,
+ size/1,
+
+ resize_heap/2,
+ scale_by/2,
+ set_max_elems/2
+]).
+
+
+-define(NOT_LOADED, not_loaded(?LINE)).
+
+
+-type hqueue() :: term().
+-type hqueue_priority() :: float(). %% this should be non_neg_float()
+-type hqueue_val() :: term().
+-type hqueue_elem() :: {hqueue_priority(), hqueue_val()}.
+-type hqueue_option() :: {max_elems, pos_integer()}
+ | {heap_size, pos_integer()}.
+-type hqueue_stat() :: {max_elems, pos_integer()}
+ | {heap_size, pos_integer()}
+ | {size, non_neg_integer()}.
+
+-export_type([hqueue/0]).
+
+
+-spec new() -> {ok, hqueue()}.
+new() ->
+ new([]).
+
+
+-spec new([hqueue_option()]) -> {ok, hqueue()}.
+new(_Options) ->
+ ?NOT_LOADED.
+
+
+%% Extraction order is undefined for entries with duplicate priorities
+-spec extract_max(hqueue()) -> hqueue_elem() | {error, empty}.
+extract_max(_HQ) ->
+ ?NOT_LOADED.
+
+
+-spec insert(hqueue(), hqueue_priority(), hqueue_val()) -> ok | {error, full}.
+insert(_HQ, _Priority, _Val) ->
+ ?NOT_LOADED.
+
+
+-spec size(hqueue()) -> integer().
+size(_HQ) ->
+ ?NOT_LOADED.
+
+
+-spec max_elems(hqueue()) -> integer().
+max_elems(_HQ) ->
+ ?NOT_LOADED.
+
+
+%% Returns old max elems or error if NewMaxElems < size(HQ)
+-spec set_max_elems(hqueue(), pos_integer()) -> pos_integer()
+ | {error, too_small}.
+set_max_elems(_HQ, _NewMaxElems) ->
+ ?NOT_LOADED.
+
+
+-spec is_empty(hqueue()) -> boolean().
+is_empty(HQ) ->
+ hqueue:size(HQ) =:= 0.
+
+
+-spec to_list(hqueue()) -> [hqueue_elem()].
+to_list(_HQ) ->
+ ?NOT_LOADED.
+
+
+-spec from_list([hqueue_elem()]) -> {ok, hqueue()}.
+from_list(Elems) ->
+ from_list(Elems, []).
+
+
+-spec from_list([hqueue_elem()], [hqueue_option()]) -> {ok, hqueue()}.
+from_list(Elems, Options) ->
+ {ok, HQ} = ?MODULE:new(Options),
+ lists:foreach(fun({Priority, Val}) ->
+ ?MODULE:insert(HQ, Priority, Val)
+ end, Elems),
+ {ok, HQ}.
+
+
+-spec scale_by(hqueue(), float()) -> ok.
+scale_by(_HQ, _Factor) ->
+ ?NOT_LOADED.
+
+
+%% Returns old heap size or error if NewHeapSize < size(HQ)
+-spec resize_heap(hqueue(), pos_integer()) -> pos_integer()
+ | {error, too_small}.
+resize_heap(_HQ, _NewHeapSize) ->
+ ?NOT_LOADED.
+
+
+-spec heap_size(hqueue()) -> pos_integer().
+heap_size(_HQ) ->
+ ?NOT_LOADED.
+
+
+-spec info(hqueue()) -> [hqueue_stat()].
+info(HQ) ->
+ [
+ {heap_size, hqueue:heap_size(HQ)},
+ {max_elems, hqueue:max_elems(HQ)},
+ {size, hqueue:size(HQ)}
+ ].
+
+
+
+init() ->
+ PrivDir = case code:priv_dir(?MODULE) of
+ {error, _} ->
+ EbinDir = filename:dirname(code:which(?MODULE)),
+ AppPath = filename:dirname(EbinDir),
+ filename:join(AppPath, "priv");
+ Path ->
+ Path
+ end,
+ erlang:load_nif(filename:join(PrivDir, "hqueue"), 0).
+
+
+not_loaded(Line) ->
+ erlang:nif_error({not_loaded, [{module, ?MODULE}, {line, Line}]}).
diff --git a/test/hqueue_proper.erl b/test/hqueue_proper.erl
new file mode 100644
index 0000000..0337b01
--- /dev/null
+++ b/test/hqueue_proper.erl
@@ -0,0 +1,33 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http:%www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(hqueue_proper).
+
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+
+-define(QC(Prop), proper:quickcheck(Prop, [{to_file, user}])).
+
+
+prop_simple() ->
+ ?FORALL({P, V}, {non_neg_float(), nat()},
+ begin
+ {ok, HQ} = hqueue:new(),
+ hqueue:insert(HQ, P, V),
+ {P, V} == hqueue:extract_max(HQ)
+ end).
+
+
+simple_test_() ->
+ ?_assertEqual(true, ?QC(prop_simple())).
+
diff --git a/test/hqueue_statem.erl b/test/hqueue_statem.erl
new file mode 100644
index 0000000..027d65d
--- /dev/null
+++ b/test/hqueue_statem.erl
@@ -0,0 +1,115 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http:%www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(hqueue_statem).
+
+-include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+
+-behaviour(proper_statem).
+
+
+-export([
+ hqueue_works/0
+]).
+-export([
+ command/1,
+ initial_state/0,
+ next_state/3,
+ postcondition/3,
+ precondition/2
+]).
+
+
+-type priority() :: float().
+-type val() :: integer().
+-type job() :: {priority(), val()}.
+
+
+-record(state, {
+ queue :: [job()]
+}).
+
+
+hqueue_works_test_() ->
+ {
+ timeout,
+ 100000,
+ ?_assertEqual(
+ true,
+ proper:quickcheck(
+ ?MODULE:hqueue_works(),
+ [{to_file, user}, {numtests, 100}]))
+ }.
+
+
+hqueue_works() ->
+ ?FORALL(Cmds, commands(?MODULE),
+ ?TRAPEXIT(
+ begin
+ {ok, HQ} = hqueue:new(),
+ {History,State,Result} = run_commands(?MODULE, Cmds, [{hq, HQ}]),
+ ?WHENFAIL(io:format("History: ~w\nState: ~w\nResult: ~w\n",
+ [History,State,Result]),
+ aggregate(command_names(Cmds), Result =:= ok))
+
+ end)).
+
+
+initial_state() ->
+ #state{queue=[]}.
+
+
+command(_) ->
+ frequency([
+ {30, {call, hqueue, insert, [{var, hq}, non_neg_float(), integer()]}},
+ {30, {call, hqueue, extract_max, [{var, hq}]}},
+ {1, {call, hqueue, size, [{var, hq}]}},
+ {1, {call, hqueue, is_empty, [{var, hq}]}},
+ {1, {call, hqueue, max_elems, [{var, hq}]}}
+ ]).
+
+
+precondition(_, _) ->
+ true.
+
+
+next_state(#state{queue=Q0}=S, _RV, {call, _, insert, [_, P, V]}) ->
+ Q1 = lists:reverse(lists:keysort(1, [{P, V} | Q0])),
+ S#state{queue=Q1};
+next_state(#state{queue=[]}=S, _RV, {call, _, extract_max, [_]}) ->
+ S;
+next_state(#state{queue=[_|Q]}=S, _RV, {call, _, extract_max, [_]}) ->
+ S#state{queue=Q};
+next_state(S, _RV, {call, _, size, [_]}) ->
+ S;
+next_state(S, _RV, {call, _, is_empty, [_]}) ->
+ S;
+next_state(S, _RV, {call, _, max_elems, [_]}) ->
+ S.
+
+
+postcondition(_S, {call, _, insert, _}, Result) ->
+ ok =:= Result;
+postcondition(#state{queue=[]}, {call, _, extract_max, [_]}, {error, empty}) ->
+ true;
+postcondition(#state{queue=[{_P1,V}|_]}, {call, _, extract_max, [_]},
+ {_P2, Result}) ->
+ V =:= Result;
+postcondition(#state{queue=Q}, {call, _, size, [_]}, Result) ->
+ length(Q) =:= Result;
+postcondition(#state{queue=Q}, {call, _, is_empty, [_]}, Result) ->
+ (length(Q) =:= 0) =:= Result;
+postcondition(_S, {call, _, max_elems, [_]}, Result) ->
+ 0 < Result.
+
diff --git a/test/hqueue_tests.erl b/test/hqueue_tests.erl
new file mode 100644
index 0000000..35bfa67
--- /dev/null
+++ b/test/hqueue_tests.erl
@@ -0,0 +1,253 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http:%www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(hqueue_tests).
+
+
+-include_lib("eunit/include/eunit.hrl").
+
+
+simple_test() ->
+ ?assertMatch({ok, _}, hqueue:new()).
+
+
+empty_extract_max_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertMatch({error, empty}, hqueue:extract_max(HQ)).
+
+
+simple_insert_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertEqual(ok, hqueue:insert(HQ, 1.1, foo)).
+
+
+simple_insert_extract_max_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertEqual(ok, hqueue:insert(HQ, 1.1, foo)),
+ ?assertEqual({1.1, foo}, hqueue:extract_max(HQ)).
+
+
+negative_priority_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertError(badarg, hqueue:insert(HQ, -1.2345, foo)).
+
+
+insert_extract_max_test() ->
+ {ok, HQ} = hqueue:new(),
+ Elems = [{1.5, foo}, {1.1, bar}, {0.4, baz}],
+ [?assertEqual(ok, hqueue:insert(HQ, P, E)) || {P,E} <- Elems],
+ [?assertEqual({P,E}, hqueue:extract_max(HQ)) || {P,E} <- Elems].
+
+
+check_pid_test() ->
+ Fun = fun() -> receive Parent -> Parent ! {self(), hqueue:new()} end end,
+ Pid = spawn(Fun),
+ Pid ! self(),
+ {ok, HQ} = receive
+ {Pid, Resp} ->
+ Resp
+ after 2000 ->
+ timeout
+ end,
+ ?assertError(badarg, hqueue:extract_max(HQ)).
+
+
+size_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertEqual(0, hqueue:size(HQ)),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,10)],
+ ?assertEqual(10, hqueue:size(HQ)),
+ [hqueue:extract_max(HQ) || _ <- lists:seq(1,10)],
+ ?assertEqual(0, hqueue:size(HQ)).
+
+
+is_empty_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertEqual(true, hqueue:is_empty(HQ)),
+ hqueue:insert(HQ, 1.0, foo),
+ ?assertEqual(false, hqueue:is_empty(HQ)).
+
+
+full_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 5}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,5)],
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 6)),
+ ?assertEqual(5, hqueue:size(HQ)),
+ {1.0, _} = hqueue:extract_max(HQ),
+ ?assertEqual(4, hqueue:size(HQ)),
+ ?assertEqual(ok, hqueue:insert(HQ, 1.0, 6)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 6)),
+ ?assertEqual(5, hqueue:size(HQ)).
+
+
+max_elems_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 1024}]),
+ ?assertEqual(1024, hqueue:max_elems(HQ)).
+
+
+empty_to_list_test() ->
+ {ok, HQ} = hqueue:new(),
+ ?assertEqual([], hqueue:to_list(HQ)).
+
+
+to_list_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 3}]),
+ Elems = [{1.1, foo}, {2.2, bar}, {3.3, baz}],
+ [hqueue:insert(HQ, P, E) || {P, E} <- Elems],
+ ?assertEqual(Elems, lists:keysort(1, hqueue:to_list(HQ))).
+
+
+empty_from_list_test() ->
+ Elems = [],
+ {ok, HQ} = hqueue:from_list(Elems),
+ ?assertEqual(Elems, hqueue:to_list(HQ)).
+
+
+from_list_test() ->
+ Elems = [{1.1, foo}, {2.2, bar}, {3.3, baz}],
+ {ok, HQ} = hqueue:from_list(Elems),
+ ?assertEqual(Elems, lists:keysort(1, hqueue:to_list(HQ))).
+
+
+scale_test() ->
+ {ok, HQ} = hqueue:new(),
+ Elems = [{1.1, foo}, {2.2, bar}, {3.3, baz}],
+ Scale = 1.7,
+ [hqueue:insert(HQ, P, E) || {P, E} <- Elems],
+ ?assertEqual(Elems, lists:keysort(1, hqueue:to_list(HQ))),
+ ?assertEqual(ok, hqueue:scale_by(HQ, Scale)),
+ Elems1 = [{P*Scale, E} || {P, E} <- Elems],
+ ?assertEqual(Elems1, lists:keysort(1, hqueue:to_list(HQ))).
+
+
+small_heap_size_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 8}]),
+ ?assertEqual(8, hqueue:max_elems(HQ)),
+ ?assertEqual(8, hqueue:heap_size(HQ)).
+
+
+heap_size_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 2048}, {heap_size, 1024}]),
+ ?assertEqual(2048, hqueue:max_elems(HQ)),
+ ?assertEqual(1024, hqueue:heap_size(HQ)).
+
+
+init_heap_size_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 2048}, {heap_size, 16}]),
+ ?assertEqual(2048, hqueue:max_elems(HQ)),
+ ?assertEqual(16, hqueue:heap_size(HQ)).
+
+
+small_heap_resize_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 29}, {heap_size, 4}]),
+ ?assertEqual(0, hqueue:size(HQ)),
+ ?assertEqual(29, hqueue:max_elems(HQ)),
+ ?assertEqual(4, hqueue:heap_size(HQ)),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,5)],
+ ?assertEqual(5, hqueue:size(HQ)),
+ ?assertEqual(8, hqueue:heap_size(HQ)),
+ ?assertEqual(29, hqueue:max_elems(HQ)),
+ [?assertEqual(ok, hqueue:insert(HQ, 1.0, E)) || E <- lists:seq(1,7)],
+ ?assertEqual(12, hqueue:size(HQ)),
+ ?assertEqual(16, hqueue:heap_size(HQ)),
+ ?assertEqual(29, hqueue:max_elems(HQ)),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,9)],
+ ?assertEqual(21, hqueue:size(HQ)),
+ ?assertEqual(29, hqueue:heap_size(HQ)),
+ ?assertEqual(29, hqueue:max_elems(HQ)),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,8)],
+ ?assertEqual(29, hqueue:size(HQ)),
+ ?assertEqual(29, hqueue:heap_size(HQ)),
+ ?assertEqual(29, hqueue:max_elems(HQ)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 1.4)).
+
+
+resize_heap_test() ->
+ Max = 4,
+ {ok, HQ} = hqueue:new([{max_elems, Max}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,Max)],
+ ?assertEqual(Max, hqueue:max_elems(HQ)),
+ ?assertEqual(Max, hqueue:heap_size(HQ)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 5)),
+ ?assertEqual(Max, hqueue:resize_heap(HQ, Max*2)),
+ ?assertEqual(Max*2, hqueue:heap_size(HQ)),
+ ?assertEqual(Max, hqueue:size(HQ)),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,Max)],
+ ?assertEqual(Max*2, hqueue:heap_size(HQ)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 5)).
+
+
+resize_heap_too_small_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 8}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,8)],
+ ?assertEqual({error, too_small}, hqueue:resize_heap(HQ, 4)).
+
+
+simple_set_max_elems_test() ->
+ Max = 4,
+ {ok, HQ} = hqueue:new([{max_elems, Max}]),
+ ?assertEqual(Max, hqueue:max_elems(HQ)),
+ ?assertEqual(Max, hqueue:set_max_elems(HQ, Max*2)),
+ ?assertEqual(Max*2, hqueue:max_elems(HQ)).
+
+
+set_max_elems_test() ->
+ Max = 8,
+ NewMax = Max div 2,
+ {ok, HQ} = hqueue:new([{max_elems, Max}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,Max)],
+ ?assertEqual(Max, hqueue:max_elems(HQ)),
+ ?assertEqual(Max, hqueue:size(HQ)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 9)),
+ [hqueue:extract_max(HQ) || _ <- lists:seq(1,NewMax)],
+ ?assertEqual(Max, hqueue:set_max_elems(HQ, NewMax)),
+ ?assertEqual(NewMax, hqueue:max_elems(HQ)),
+ ?assertEqual(NewMax, hqueue:size(HQ)),
+ ?assertEqual({error, full}, hqueue:insert(HQ, 1.0, 5)).
+
+
+set_max_elems_too_small_test() ->
+ {ok, HQ} = hqueue:new([{max_elems, 8}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,8)],
+ ?assertEqual({error, too_small}, hqueue:set_max_elems(HQ, 4)).
+
+
+simple_info_test() ->
+ MaxElems = 256,
+ HeapSize = 64,
+ {ok, HQ} = hqueue:new([{max_elems, MaxElems}, {heap_size, HeapSize}]),
+ ?assertEqual(
+ [{heap_size, HeapSize}, {max_elems, MaxElems}, {size, 0}],
+ hqueue:info(HQ)
+ ).
+
+
+size_info_test() ->
+ MaxElems = 256,
+ HeapSize = 64,
+ {ok, HQ} = hqueue:new([{max_elems, MaxElems}, {heap_size, HeapSize}]),
+ [hqueue:insert(HQ, 1.0, E) || E <- lists:seq(1,10)],
+ ?assertEqual(
+ [{heap_size, HeapSize}, {max_elems, MaxElems}, {size, 10}],
+ hqueue:info(HQ)
+ ).
+
+
+duplicates_test() ->
+ {ok, HQ} = hqueue:new(),
+ {P,V} = {1.9, foo},
+ ok = hqueue:insert(HQ, P, V),
+ ok = hqueue:insert(HQ, P, V),
+ ?assertEqual({P, V}, hqueue:extract_max(HQ)),
+ ?assertEqual({P, V}, hqueue:extract_max(HQ)),
+ ?assertEqual({error, empty}, hqueue:extract_max(HQ)).
+