Version 1.1.0

* Added "fastpath" and "slowpath" branching for faster filtering of smaller messages.
* I/O request and reply are dropped by default now (may be allowed by enabled `untrusted` mode with a custom handler).

Author: potatosalad

.github/workflows/ci.yml

@@ -14,9 +14,9 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - elixir: 1.15.4
-            otp: 26.0.2
-            rebar3: 3.22.0
+          - elixir: 1.15.6
+            otp: 26.1.1
+            rebar3: 3.22.1
             lint: true
 
           - elixir: 1.14.5
@@ -56,7 +56,7 @@ jobs:
         run: mix format --check-formatted
 
       - name: Elixir Tests
-        run: mix test
+        run: mix test --trace
 
       - name: Logs
         uses: actions/upload-artifact@v3

apps/erldist_filter/CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 1.1.0 (2023-10-04)
+
+* Added "fastpath" and "slowpath" branching for faster filtering of smaller messages.
+* I/O request and reply are dropped by default now (may be allowed by enabled `untrusted` mode with a custom handler).
+
 ## 1.0.0 (2023-09-07)
 
 * Initial release.

apps/erldist_filter/c_src/Makefile

@@ -81,8 +81,8 @@ ifeq ($(PLATFORM),msys2)
 else ifeq ($(PLATFORM),darwin)
 	CC ?= cc
 	CXX ?= c++
-	CFLAGS ?= -Og -g -ggdb -std=c2x -arch x86_64 -Wall -Wmissing-prototypes
-	CXXFLAGS ?= -Og -g -ggdb -std=c++20 -arch x86_64 -Wall
+	CFLAGS ?= -O2 -g -ggdb -std=c2x -arch x86_64 -Wall -Wmissing-prototypes
+	CXXFLAGS ?= -O2 -g -ggdb -std=c++20 -arch x86_64 -Wall
 	LDFLAGS ?= -arch x86_64 -flat_namespace -undefined suppress
 else ifeq ($(PLATFORM),freebsd)
 	CC ?= cc

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

@@ -11,8 +11,10 @@
 #include "edf_channel_inspect.h"
 #include "edf_channel_recv.h"
 
+#include "../config/edf_config.h"
 #include "../core/simd.h"
 #include "../erts/dist.h"
+#include "../uterm/uterm.h"
 
 ERL_NIF_TERM
 erldist_filter_nif_channel_open_5(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
@@ -212,33 +214,73 @@ erldist_filter_nif_channel_list_1(ErlNifEnv *env, int argc, const ERL_NIF_TERM a
     return list;
 }
 
+static int channel_recv_2_fast(ErlNifEnv *env, edf_channel_t *channel, ERL_NIF_TERM input_term, ErlNifBinary *input_binary,
+                               ERL_NIF_TERM *out_termp, ERL_NIF_TERM *err_termp);
+
 ERL_NIF_TERM
 erldist_filter_nif_channel_recv_2(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
 {
-    ERL_NIF_TERM out_term;
+    ERL_NIF_TERM out_term = THE_NON_VALUE;
+    ERL_NIF_TERM err_term = THE_NON_VALUE;
     ERL_NIF_TERM trap_term;
     edf_channel_resource_t *resource = NULL;
     edf_channel_t *channel = NULL;
     int flags = (EDF_CHANNEL_RESOURCE_FLAG_OWNER_REQUIRED | EDF_CHANNEL_RESOURCE_FLAG_WRITE_LOCK);
+    bool eligible_for_fast_recv = true;
+    unsigned int iovec_length;
+    ERL_NIF_TERM iovec_tail = THE_NON_VALUE;
+    ERL_NIF_TERM input_term = THE_NON_VALUE;
+    ErlNifBinary input_binary;
 
     if (argc != 2) {
         return EXCP_BADARG(env, "argc must be 2");
     }
 
-    if (!edf_channel_resource_acquire(env, argv[0], &resource, &channel, &out_term, flags)) {
-        return out_term;
+    if (!edf_channel_resource_acquire(env, argv[0], &resource, &channel, &err_term, flags)) {
+        return err_term;
+    }
+
+    if (ioq_size(&channel->rx.ioq) > 0 || edf_config_is_untrusted_enabled()) {
+        eligible_for_fast_recv = false;
+    }
+
+    if (eligible_for_fast_recv && enif_is_list(env, argv[1]) && enif_get_list_length(env, argv[1], &iovec_length) &&
+        iovec_length == 1 && enif_get_list_cell(env, argv[1], &input_term, &iovec_tail) && enif_is_binary(env, input_term) &&
+        enif_is_empty_list(env, iovec_tail)) {
+        eligible_for_fast_recv = true;
+    } else {
+        input_term = argv[1];
+        eligible_for_fast_recv = (eligible_for_fast_recv && enif_is_binary(env, argv[1])) ? true : false;
+    }
+
+    if (eligible_for_fast_recv && enif_is_binary(env, input_term) && enif_inspect_binary(env, input_term, &input_binary)) {
+        int retval = channel_recv_2_fast(env, channel, input_term, &input_binary, &out_term, &err_term);
+        if (retval == 0) {
+            (void)edf_channel_resource_release(&resource, &channel, flags);
+            return err_term;
+        } else if (retval == 1) {
+            (void)edf_channel_resource_release(&resource, &channel, flags);
+            return out_term;
+        } else {
+            eligible_for_fast_recv = false;
+        }
     }
 
-    if (!ioq_inspect_iovec_and_enqv(env, ~(0ULL), argv[1], &out_term, &channel->rx.ioq)) {
+    if (enif_is_binary(env, input_term)) {
+        input_term = enif_make_list1(env, input_term);
+    }
+
+    if (!ioq_inspect_iovec_and_enqv(env, ~(0ULL), input_term, &out_term, &channel->rx.ioq)) {
         (void)edf_channel_resource_release(&resource, &channel, flags);
-        return EXCP_BADARG(env, "Iodata is invalid");
+        return EXCP_BADARG(env, "Input is invalid");
     }
 
     if (!enif_is_empty_list(env, out_term)) {
         (void)edf_channel_resource_release(&resource, &channel, flags);
         return EXCP_BADARG(env, "Call to enif_ioq_enqv() failed: expected tail to be an empty list");
     }
 
+    CHANNEL_RX_STATS_COUNT(channel, slowpath, 1);
     trap_term = edf_channel_recv_trap_open(env, resource, channel, NULL);
     if (enif_is_exception(env, trap_term)) {
         (void)edf_channel_resource_release(&resource, &channel, flags);
@@ -248,6 +290,203 @@ erldist_filter_nif_channel_recv_2(ErlNifEnv *env, int argc, const ERL_NIF_TERM a
     return edf_trap_schedule_from_term(env, trap_term);
 }
 
+int
+channel_recv_2_fast(ErlNifEnv *env, edf_channel_t *channel, ERL_NIF_TERM input_term, ErlNifBinary *input_binary,
+                    ERL_NIF_TERM *out_termp, ERL_NIF_TERM *err_termp)
+{
+    uint8_t dist_header_type;
+    uint64_t fragment_id;
+    const uint8_t *p = NULL;
+    const uint8_t *pend = NULL;
+    size_t packet_length = 0;
+
+    if (input_binary->size == 0) {
+        *out_termp = enif_make_list(env, 0);
+        return 1;
+    }
+
+    p = (void *)input_binary->data;
+    pend = p + input_binary->size;
+
+    if (channel->rx.packet_size != 0) {
+        if ((pend - p) < channel->rx.packet_size) {
+            goto try_decode_slow;
+        }
+        switch (channel->rx.packet_size) {
+        case 1:
+            packet_length = (size_t)(*p++);
+            break;
+        case 2:
+            packet_length = (size_t)(be16toh(*((uint16_t *)(void *)(p))));
+            p += 2;
+            break;
+        case 4:
+            packet_length = (size_t)(be32toh(*((uint32_t *)(void *)(p))));
+            p += 4;
+            break;
+        case 8:
+            packet_length = (size_t)(be64toh(*((uint64_t *)(void *)(p))));
+            p += 8;
+            break;
+        default:
+            goto try_decode_slow;
+        }
+        if ((pend - p) != packet_length) {
+            goto try_decode_slow;
+        }
+    } else {
+        packet_length = input_binary->size;
+    }
+
+    if ((channel->dflags & (DFLAG_DIST_HDR_ATOM_CACHE | DFLAG_FRAGMENTS)) == 0) {
+        // If none of these distribution flags are set, then we fallback to "pass-through" mode.
+        // We expect a PASS_THROUGH byte to be present.
+        if (*(p++) != PASS_THROUGH) {
+            *err_termp = EXCP_ERROR_F(env, "Dist Frame in pass-through mode must have an 8-bit type field equal to %u (was %u)\n",
+                                      PASS_THROUGH, *(p - 1));
+            return 0;
+        }
+        dist_header_type = PASS_THROUGH;
+        goto try_decode_slow;
+    } else {
+        // Otherwise, we expect a VERSION_MAGIC byte to be present.
+        if (*(p++) != VERSION_MAGIC) {
+            *err_termp = EXCP_ERROR_F(env, "Dist Frame in pass-through mode must have an 8-bit type field equal to %u (was %u)\n",
+                                      PASS_THROUGH, *(p - 1));
+            return 0;
+        }
+        if (*p == DIST_HEADER) {
+            p++;
+            if ((channel->dflags & (DFLAG_DIST_HDR_ATOM_CACHE)) == 0) {
+                *err_termp = EXCP_ERROR(env, "Dist Frame tagged with 'DIST_HEADER' not support by current "
+                                             "distribution flags (missing: DFLAG_DIST_HDR_ATOM_CACHE).\n");
+                return 0;
+            }
+            dist_header_type = DIST_HEADER;
+            goto try_decode_fast;
+        } else if (*p == DIST_FRAG_HEADER) {
+            p++;
+            if ((channel->dflags & (DFLAG_FRAGMENTS)) == 0) {
+                *err_termp = EXCP_ERROR(env, "Dist Frame tagged with 'DIST_FRAG_HEADER' not support by current "
+                                             "distribution flags (missing: DFLAG_FRAGMENTS).\n");
+                return 0;
+            }
+            if ((pend - p) < 16) {
+                *err_termp = EXCP_ERROR_F(
+                    env, "Dist Frame tagged with 'DIST_FRAG_HEADER' must be at least 16-bytes in size (was %u-bytes).\n",
+                    (pend - p));
+                return 0;
+            }
+            // skip sequence_id
+            p += 8;
+            fragment_id = be64toh(*((uint64_t *)(void *)(p)));
+            p += 8;
+            dist_header_type = DIST_FRAG_HEADER;
+            if (fragment_id != 1) {
+                goto try_decode_slow;
+            }
+            goto try_decode_fast;
+        } else if (*p == DIST_FRAG_CONT) {
+            dist_header_type = DIST_FRAG_CONT;
+            goto try_decode_slow;
+        } else {
+            *err_termp = EXCP_ERROR_F(env, "Dist Frame tagged with unrecognized tag '%u' is not supported.\n", *p);
+            return 0;
+        }
+    }
+try_decode_fast: {
+    vec_t vec[1];
+    vec_t control_vec[1];
+    vec_reader_t vr[1];
+    edf_atom_translation_table_t attab[1];
+    int external_flags = 0;
+    slice_t headers[1];
+    slice_t control[1];
+    slice_t payload[1];
+    vterm_env_t vtenv[1];
+    udist_t up[1];
+    (void)edf_atom_translation_table_init(attab);
+    (void)vec_init_free(vec);
+    (void)vec_create_from_slice(vec, p, pend);
+    (void)vec_reader_create(vr, vec, 0);
+    if (!etf_fast_decode_dist_header(env, channel, attab, vr, &external_flags, headers, err_termp)) {
+        return 0;
+    }
+    control->head = vec_reader_raw_bytes(vr);
+    if (!etf_fast_skip_terms(env, false, vr, 1, err_termp)) {
+        return 0;
+    }
+    control->tail = vec_reader_raw_bytes(vr);
+    vtenv->nif_env = env;
+    vtenv->attab = attab;
+    (void)vec_init_free(control_vec);
+    (void)vec_create_from_slice(control_vec, control->head, control->tail);
+    (void)udist_init(up);
+    if (!etf_decode_udist_control(env, vtenv, false, true, control_vec, up, err_termp)) {
+        return 0;
+    }
+    if (udist_control_is_send_to_name(up) &&
+        (up->control.data.send_to == ATOM(net_kernel) || up->control.data.send_to == ATOM(rex))) {
+        goto try_decode_slow;
+    }
+    if (up->info.payload && udist_control_is_send(up)) {
+        payload->head = vec_reader_raw_bytes(vr);
+        if (!etf_fast_skip_terms(env, false, vr, 1, err_termp)) {
+            return 0;
+        }
+        payload->tail = vec_reader_raw_bytes(vr);
+        if (!udist_classify(env, vtenv, up, false, false, payload, err_termp)) {
+            return 0;
+        }
+        if (up->flags == UDIST_CLASSIFY_FLAG_EMIT) {
+            if (packet_length < input_binary->size) {
+                input_term = enif_make_sub_binary(env, input_term, channel->rx.packet_size, packet_length);
+            }
+            *out_termp = enif_make_list1(env, enif_make_tuple2(env, ATOM(emit), input_term));
+            {
+                edf_channel_stats_dop_t *statsdop = NULL;
+                edf_world_slot_t *slot = NULL;
+                CHANNEL_RX_STATS_COUNT(channel, fastpath, 1);
+                CHANNEL_RX_STATS_COUNT(channel, packet_count, 1);
+                CHANNEL_RX_STATS_COUNT(channel, emit_count, 1);
+                switch (dist_header_type) {
+                case PASS_THROUGH:
+                    CHANNEL_RX_STATS_COUNT(channel, dist_pass_through_count, 1);
+                    break;
+                case DIST_HEADER:
+                    CHANNEL_RX_STATS_COUNT(channel, dist_header_count, 1);
+                    break;
+                case DIST_FRAG_HEADER:
+                    CHANNEL_RX_STATS_COUNT(channel, dist_frag_header_count, 1);
+                    break;
+                case DIST_FRAG_CONT:
+                    CHANNEL_RX_STATS_COUNT(channel, dist_frag_cont_count, 1);
+                    break;
+                default:
+                    break;
+                }
+                statsdop = NULL;
+                if (udist_get_channel_stats_dop(up, &channel->rx.stats, &statsdop) && statsdop != NULL) {
+                    statsdop->seen += 1;
+                    statsdop->emit += 1;
+                }
+                statsdop = NULL;
+                slot = edf_world_get();
+                if (udist_get_channel_stats_dop(up, &slot->stats.channel.rx_stats, &statsdop) && statsdop != NULL) {
+                    statsdop->seen += 1;
+                    statsdop->emit += 1;
+                }
+            };
+            return 1;
+        }
+    }
+    goto try_decode_slow;
+}
+try_decode_slow: {
+    return -1;
+}
+}
+
 ERL_NIF_TERM
 erldist_filter_nif_channel_set_controlling_process_2(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
 {

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

@@ -108,7 +108,7 @@ static int edf_channel_inspect_stats_dop(ErlNifEnv *env, edf_channel_stats_dop_t
 int
 edf_channel_inspect_stats(ErlNifEnv *env, edf_channel_stats_t *stats, ERL_NIF_TERM *termp)
 {
-#define RET_MAP_SIZE (47)
+#define RET_MAP_SIZE (49)
 
     ERL_NIF_TERM keys[RET_MAP_SIZE];
     ERL_NIF_TERM vals[RET_MAP_SIZE];
@@ -151,6 +151,10 @@ edf_channel_inspect_stats(ErlNifEnv *env, edf_channel_stats_t *stats, ERL_NIF_TE
     vals[v++] = enif_make_uint64(env, (ErlNifUInt64)stats->payload_has_export_ext);
     keys[k++] = ATOM(payload_has_new_fun_ext);
     vals[v++] = enif_make_uint64(env, (ErlNifUInt64)stats->payload_has_new_fun_ext);
+    keys[k++] = ATOM(fastpath);
+    vals[v++] = enif_make_uint64(env, (ErlNifUInt64)stats->fastpath);
+    keys[k++] = ATOM(slowpath);
+    vals[v++] = enif_make_uint64(env, (ErlNifUInt64)stats->slowpath);
     keys[k++] = ATOM(dop_link);
     if (!edf_channel_inspect_stats_dop(env, &stats->dop_link, &vals[v++])) {
         return vals[v - 1];