Support OTP 26, add routers/handlers, and prepare for version 1.0.0 release

Author: potatosalad

.formatter.exs

@@ -6,6 +6,6 @@
 
 # Used by "mix format"
 [
-  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"],
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}", "apps/erldist_filter/{config,lib,test}/**/*.{ex,exs}"],
   line_length: 132
 ]

CHANGELOG.md

@@ -0,0 +1 @@
+apps/erldist_filter/CHANGELOG.md

CHANGELOG.md

@@ -0,0 +1,5 @@
+# Changelog
+
+## 1.0.0 (2023-09-07)
+
+* Initial release.

apps/erldist_filter/CHANGELOG.md

@@ -77,7 +77,7 @@ action_into_term_emit(ErlNifEnv *env, action_t *action, ERL_NIF_TERM *termp)
     action_term = vec_into_binary_term(env, &action->data.emit.vec);
     action_term = enif_make_tuple2(env, ATOM(emit), action_term);
     *termp = action_term;
-    (void)action_init_free(action);
+    (void)action_destroy(action);
     return 1;
 }
 

apps/erldist_filter/c_src/nif/action.c

@@ -19,8 +19,6 @@
 ErlNifResourceType *edf_channel_resource_type = NULL;
 static edf_channel_resource_table_t edf_channel_resource_table_internal = {._link = {.next = NULL, .prev = NULL}};
 edf_channel_resource_table_t *edf_channel_resource_table = &edf_channel_resource_table_internal;
-static edf_channel_index_table_t edf_channel_index_table_internal = {._link = {.next = NULL, .prev = NULL}};
-edf_channel_index_table_t *edf_channel_index_table = &edf_channel_index_table_internal;
 
 /* Static Declarations */
 
@@ -30,8 +28,6 @@ static void edf_channel_resource_type_down(ErlNifEnv *env, void *obj, ErlNifPid
 static int edf_channel_create(ErlNifEnv *env, size_t packet_size, ERL_NIF_TERM sysname, uint32_t creation, uint32_t connection_id,
                               uint64_t dflags, edf_channel_resource_t *resource, edf_channel_t **channelp,
                               ERL_NIF_TERM *error_term);
-static void edf_channel_stats_init_empty(edf_channel_stats_t *stats);
-static void edf_channel_stats_dop_init_empty(edf_channel_stats_dop_t *dop);
 
 /* Function Definitions */
 
@@ -59,15 +55,6 @@ edf_channel_load(ErlNifEnv *env)
         (void)linklist_init_anchor(&edf_channel_resource_table->_link);
     }
 
-    if (!linklist_is_linked(&edf_channel_index_table->_link)) {
-        retval = enif_tsd_key_create("erldist_filter.channel_index_table_tsd_key", &edf_channel_index_table->key);
-        if (retval != 0) {
-            return retval;
-        }
-        (void)core_rwlock_create(&edf_channel_index_table->rwlock, "erldist_filter.channel_index_table_rwlock");
-        (void)linklist_init_anchor(&edf_channel_index_table->_link);
-    }
-
     return retval;
 }
 
@@ -114,7 +101,7 @@ edf_channel_resource_open(ErlNifEnv *env, size_t packet_size, ERL_NIF_TERM sysna
     // Don't release the resource here, so a reference is kept on the root table.
     // (void)enif_release_resource((void *)resource);
 
-    (void)atomic_fetch_add(&edf_world->channels_created, 1);
+    WORLD_STATS_COUNT(channel, create, 1);
 
     if (resourcep != NULL) {
         *resourcep = resource;
@@ -198,6 +185,8 @@ edf_channel_create(ErlNifEnv *env, size_t packet_size, ERL_NIF_TERM sysname, uin
     channel->creation = creation;
     channel->connection_id = connection_id;
     channel->dflags = dflags;
+    channel->rx.router_name = edf_channel_router_name(env, channel->sysname);
+    channel->rx.sort = 0;
     channel->rx.packet_size = packet_size;
     channel->rx.state = EDF_CHANNEL_RX_STATE_PACKET_HEADER;
     (void)ioq_init_free(&channel->rx.ioq);
@@ -287,93 +276,16 @@ edf_channel_destroy(ErlNifEnv *env, edf_channel_resource_t *resource, edf_channe
     }
     resource->inner = NULL;
     (void)enif_free((void *)channel);
-    (void)atomic_fetch_add(&edf_world->channels_destroyed, 1);
+    WORLD_STATS_COUNT(channel, destroy, 1);
     return;
 }
 
-edf_channel_index_slot_t *
-edf_channel_index_get_slow(void)
-{
-    edf_channel_index_slot_t *slot = NULL;
-    slot = (void *)enif_tsd_get(edf_channel_index_table->key);
-    if (slot == NULL) {
-        slot = enif_alloc(sizeof(edf_channel_index_slot_t));
-        if (slot == NULL) {
-            unreachable();
-            (void)perror("Too many processors or threads on this machine: OOM, unable to allocate edf_channel_index_slot_t!");
-            abort();
-            return NULL;
-        }
-        slot->_link.prev = NULL;
-        slot->_link.next = NULL;
-        (void)edf_channel_stats_init_empty(&slot->rx_stats);
-        (void)core_rwlock_write_lock(&edf_channel_index_table->rwlock);
-        (void)linklist_insert(&edf_channel_index_table->_link, &slot->_link);
-        (void)core_rwlock_write_unlock(&edf_channel_index_table->rwlock);
-        (void)enif_tsd_set(edf_channel_index_table->key, (void *)slot);
-        return slot;
-    }
-    return slot;
-}
-
-void
-edf_channel_stats_init_empty(edf_channel_stats_t *stats)
+ERL_NIF_TERM
+edf_channel_router_name(ErlNifEnv *env, ERL_NIF_TERM sysname)
 {
-    stats->packet_count = 0;
-    stats->emit_count = 0;
-    stats->drop_count = 0;
-    stats->dist_header_count = 0;
-    stats->dist_frag_header_count = 0;
-    stats->dist_frag_cont_count = 0;
-    stats->dist_pass_through_count = 0;
-    stats->atom_cache_read_count = 0;
-    stats->atom_cache_write_count = 0;
-    stats->atom_cache_overwrite_count = 0;
-    stats->rewrite_fragment_header_count = 0;
-    stats->rollback_atom_cache_count = 0;
-    stats->compact_external_count = 0;
-    stats->compact_fragment_count = 0;
-    stats->control_has_export_ext = 0;
-    stats->control_has_new_fun_ext = 0;
-    stats->payload_has_export_ext = 0;
-    stats->payload_has_new_fun_ext = 0;
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_link);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_send);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_exit);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_unlink);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_reg_send);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_group_leader);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_exit2);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_send_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_exit_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_reg_send_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_exit2_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_monitor_p);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_demonitor_p);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_monitor_p_exit);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_send_sender);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_send_sender_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_payload_exit);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_payload_exit_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_payload_exit2);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_payload_exit2_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_payload_monitor_p_exit);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_spawn_request);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_spawn_request_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_spawn_reply);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_spawn_reply_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_alias_send);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_alias_send_tt);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_unlink_id);
-    (void)edf_channel_stats_dop_init_empty(&stats->dop_unlink_id_ack);
-    return;
-}
+    ErlNifUInt64 router_number;
 
-inline void
-edf_channel_stats_dop_init_empty(edf_channel_stats_dop_t *dop)
-{
-    dop->seen = 0;
-    dop->emit = 0;
-    dop->drop = 0;
-    return;
+    router_number = enif_hash(ERL_NIF_INTERNAL_HASH, sysname, 0);
+    router_number %= erldist_filter_router_count;
+    return erldist_filter_router_names[router_number];
 }

apps/erldist_filter/c_src/nif/channel/edf_channel.c

@@ -24,93 +24,24 @@ extern "C" {
 #include "edf_atom_cache.h"
 #include "edf_external.h"
 #include "edf_external_sequence.h"
+#include "edf_channel_stats.h"
+#include "../world/edf_world.h"
 
 /* Macro Definitions */
 
-#define CHANNEL_RX_STATS_COUNT(channel, field, inc)                                                                                \
+#define CHANNEL_RX_STATS_COUNT(channelp, field, inc)                                                                               \
     do {                                                                                                                           \
-        (channel)->rx.stats.field += inc;                                                                                          \
-        (edf_channel_index_get())->rx_stats.field += inc;                                                                          \
+        (channelp)->rx.stats.field += inc;                                                                                         \
+        (edf_world_get())->stats.channel.rx_stats.field += inc;                                                                    \
     } while (0)
 
 /* Type Definitions */
 
 typedef struct edf_channel_s edf_channel_t;
-typedef struct edf_channel_stats_s edf_channel_stats_t;
-typedef struct edf_channel_stats_dop_s edf_channel_stats_dop_t;
-typedef struct edf_channel_index_table_s edf_channel_index_table_t;
-typedef struct edf_channel_index_slot_s edf_channel_index_slot_t;
 typedef enum edf_channel_rx_state_t edf_channel_rx_state_t;
 typedef struct edf_channel_resource_s edf_channel_resource_t;
 typedef struct edf_channel_resource_table_s edf_channel_resource_table_t;
 
-struct edf_channel_stats_dop_s {
-    uint64_t seen;
-    uint64_t emit;
-    uint64_t drop;
-};
-
-struct edf_channel_stats_s {
-    uint64_t packet_count;
-    uint64_t emit_count;
-    uint64_t drop_count;
-    uint64_t dist_header_count;
-    uint64_t dist_frag_header_count;
-    uint64_t dist_frag_cont_count;
-    uint64_t dist_pass_through_count;
-    uint64_t atom_cache_read_count;
-    uint64_t atom_cache_write_count;
-    uint64_t atom_cache_overwrite_count;
-    uint64_t rewrite_fragment_header_count;
-    uint64_t rollback_atom_cache_count;
-    uint64_t compact_external_count;
-    uint64_t compact_fragment_count;
-    uint64_t control_has_export_ext;
-    uint64_t control_has_new_fun_ext;
-    uint64_t payload_has_export_ext;
-    uint64_t payload_has_new_fun_ext;
-    edf_channel_stats_dop_t dop_link;
-    edf_channel_stats_dop_t dop_send;
-    edf_channel_stats_dop_t dop_exit;
-    edf_channel_stats_dop_t dop_unlink;
-    edf_channel_stats_dop_t dop_reg_send;
-    edf_channel_stats_dop_t dop_group_leader;
-    edf_channel_stats_dop_t dop_exit2;
-    edf_channel_stats_dop_t dop_send_tt;
-    edf_channel_stats_dop_t dop_exit_tt;
-    edf_channel_stats_dop_t dop_reg_send_tt;
-    edf_channel_stats_dop_t dop_exit2_tt;
-    edf_channel_stats_dop_t dop_monitor_p;
-    edf_channel_stats_dop_t dop_demonitor_p;
-    edf_channel_stats_dop_t dop_monitor_p_exit;
-    edf_channel_stats_dop_t dop_send_sender;
-    edf_channel_stats_dop_t dop_send_sender_tt;
-    edf_channel_stats_dop_t dop_payload_exit;
-    edf_channel_stats_dop_t dop_payload_exit_tt;
-    edf_channel_stats_dop_t dop_payload_exit2;
-    edf_channel_stats_dop_t dop_payload_exit2_tt;
-    edf_channel_stats_dop_t dop_payload_monitor_p_exit;
-    edf_channel_stats_dop_t dop_spawn_request;
-    edf_channel_stats_dop_t dop_spawn_request_tt;
-    edf_channel_stats_dop_t dop_spawn_reply;
-    edf_channel_stats_dop_t dop_spawn_reply_tt;
-    edf_channel_stats_dop_t dop_alias_send;
-    edf_channel_stats_dop_t dop_alias_send_tt;
-    edf_channel_stats_dop_t dop_unlink_id;
-    edf_channel_stats_dop_t dop_unlink_id_ack;
-};
-
-struct edf_channel_index_slot_s {
-    linklist_t _link;
-    edf_channel_stats_t rx_stats;
-};
-
-struct edf_channel_index_table_s {
-    linklist_t _link;
-    ErlNifTSDKey key;
-    core_rwlock_t rwlock;
-};
-
 enum edf_channel_rx_state_t {
     EDF_CHANNEL_RX_STATE_PACKET_HEADER = 0,
     EDF_CHANNEL_RX_STATE_PACKET_DATA,
@@ -135,6 +66,8 @@ struct edf_channel_s {
     uint32_t connection_id;
     uint64_t dflags;
     struct {
+        ERL_NIF_TERM router_name;
+        uint64_t sort;
         size_t packet_size;
         edf_channel_rx_state_t state;
         ioq_t ioq;
@@ -160,7 +93,6 @@ struct edf_channel_resource_table_s {
 
 extern ErlNifResourceType *edf_channel_resource_type;
 extern edf_channel_resource_table_t *edf_channel_resource_table;
-extern edf_channel_index_table_t *edf_channel_index_table;
 
 /* Function Declarations */
 
@@ -170,9 +102,8 @@ extern ERL_NIF_TERM edf_channel_resource_open(ErlNifEnv *env, size_t packet_size
                                               uint32_t connection_id, uint64_t dflags, edf_channel_resource_t **resourcep,
                                               edf_channel_t **channelp);
 extern void edf_channel_destroy(ErlNifEnv *env, edf_channel_resource_t *resource, edf_channel_t *channel);
-static edf_channel_index_slot_t *edf_channel_index_get(void);
-extern edf_channel_index_slot_t *edf_channel_index_get_slow(void);
 static int edf_channel_is_tracing_enabled(const edf_channel_t *channel);
+extern ERL_NIF_TERM edf_channel_router_name(ErlNifEnv *env, ERL_NIF_TERM sysname);
 static int edf_channel_tracing_send(edf_channel_t *channel, ErlNifEnv *caller_env, ErlNifEnv *msg_env, ERL_NIF_TERM msg);
 
 #ifdef ERLDIST_FILTER_NIF_INTERNAL_API
@@ -187,16 +118,6 @@ static void edf_channel_resource_release(edf_channel_resource_t **resourcep, edf
 
 /* Inline Function Definitions */
 
-inline edf_channel_index_slot_t *
-edf_channel_index_get(void)
-{
-    edf_channel_index_slot_t *slot = (void *)enif_tsd_get(edf_channel_index_table->key);
-    if (slot == NULL) {
-        return edf_channel_index_get_slow();
-    }
-    return slot;
-}
-
 inline int
 edf_channel_is_tracing_enabled(const edf_channel_t *channel)
 {

apps/erldist_filter/c_src/nif/channel/edf_channel.h

@@ -92,56 +92,6 @@ erldist_filter_nif_channel_close_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM
     return out_term;
 }
 
-ERL_NIF_TERM
-erldist_filter_nif_channel_index_get_0(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
-{
-#define RET_MAP_SIZE (2)
-
-    ERL_NIF_TERM out_term;
-    ERL_NIF_TERM keys[RET_MAP_SIZE];
-    ERL_NIF_TERM vals[RET_MAP_SIZE];
-    size_t k = 0;
-    size_t v = 0;
-    ErlNifUInt64 slots;
-    edf_channel_stats_t acc[1];
-    edf_channel_index_slot_t *root = NULL;
-    edf_channel_index_slot_t *slot = NULL;
-
-    if (argc != 0) {
-        return EXCP_BADARG(env, "argc must be 0");
-    }
-
-    (void)memset((void *)acc, 0, sizeof(edf_channel_stats_t));
-
-    slots = 0;
-    (void)core_rwlock_read_lock(&edf_channel_index_table->rwlock);
-    root = (void *)&edf_channel_index_table->_link;
-    slot = (void *)root->_link.next;
-    while (slot != root) {
-        (void)core_simd_add_vec_u64((uint64_t *)acc, (const uint64_t *)&slot->rx_stats,
-                                    (sizeof(edf_channel_stats_t) / sizeof(uint64_t)));
-        slots += 1;
-        slot = (void *)slot->_link.next;
-    }
-    (void)core_rwlock_read_unlock(&edf_channel_index_table->rwlock);
-    keys[k++] = ATOM(rx_stats);
-    if (!edf_channel_inspect_stats(env, acc, &out_term)) {
-        return out_term;
-    }
-    vals[v++] = out_term;
-
-    keys[k++] = ATOM(slots);
-    vals[v++] = enif_make_uint64(env, slots);
-
-    if (!enif_make_map_from_arrays(env, keys, vals, RET_MAP_SIZE, &out_term)) {
-        return EXCP_BADARG(env, "Call to enif_make_map_from_arrays() failed: duplicate keys detected");
-    }
-
-    return out_term;
-
-#undef RET_MAP_SIZE
-}
-
 ERL_NIF_TERM
 erldist_filter_nif_channel_inspect_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
 {

apps/erldist_filter/c_src/nif/channel/edf_channel_impl.c

@@ -25,7 +25,6 @@ extern "C" {
 
 extern ERL_NIF_TERM erldist_filter_nif_channel_open_5(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 extern ERL_NIF_TERM erldist_filter_nif_channel_close_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
-extern ERL_NIF_TERM erldist_filter_nif_channel_index_get_0(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 extern ERL_NIF_TERM erldist_filter_nif_channel_inspect_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 extern ERL_NIF_TERM erldist_filter_nif_channel_list_0(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 extern ERL_NIF_TERM erldist_filter_nif_channel_list_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);