feat: first attempt to enable buffering signals during ownership handoff

CMakeLists.txt

@@ -112,6 +112,8 @@ set(SOURCES
   c_src/quicer_config.h
   c_src/quicer_queue.c
   c_src/quicer_queue.h
+  c_src/quicer_owner_queue.c
+  c_src/quicer_owner_queue.h
   c_src/quicer_ctx.c
   c_src/quicer_ctx.h
   c_src/quicer_listener.c

c_src/quicer_ctx.c

@@ -270,6 +270,7 @@ init_s_ctx()
   s_ctx->is_recv_pending = FALSE;
   s_ctx->is_closed = TRUE; // init
   s_ctx->event_mask = 0;
+  s_ctx->sig_queue = NULL;
   return s_ctx;
 }
 

c_src/quicer_ctx.h

@@ -18,6 +18,7 @@ limitations under the License.
 #define __QUICER_CTX_H_
 
 #include "quicer_nif.h"
+#include "quicer_owner_queue.h"
 #include "quicer_queue.h"
 #include <msquichelper.h>
 #include <openssl/x509.h>
@@ -137,6 +138,8 @@ typedef struct QuicerStreamCTX
   // Track lifetime of Stream handle
   CXPLAT_REF_COUNT ref_count;
   uint32_t event_mask;
+  // for ownership handoff
+  OWNER_SIGNAL_QUEUE *sig_queue;
   void *reserved1;
   void *reserved2;
   void *reserved3;

c_src/quicer_nif.c

@@ -1584,7 +1584,11 @@ static ErlNifFunc nif_funcs[] = {
   { "get_connections", 1, get_connectionsX, 0},
   { "get_conn_owner", 1, get_conn_owner1, 0},
   { "get_stream_owner", 1, get_stream_owner1, 0},
-  { "get_listener_owner", 1, get_listener_owner1, 0}
+  { "get_listener_owner", 1, get_listener_owner1, 0},
+  /* for testing */
+  {"buffer_sig", 3, buffer_sig, 0},
+  {"enable_sig_buffer", 1, enable_sig_buffer, 0},
+  {"flush_stream_buffered_sigs", 1, flush_stream_buffered_sigs, 0}
   // clang-format on
 };
 

c_src/quicer_owner_queue.c

@@ -0,0 +1,76 @@
+/*--------------------------------------------------------------------
+Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+
+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 "quicer_owner_queue.h"
+
+OWNER_SIGNAL_QUEUE *
+OwnerSignalQueueNew()
+{
+  OWNER_SIGNAL_QUEUE *sig
+      = CxPlatAlloc(sizeof(OWNER_SIGNAL_QUEUE), QUICER_OWNER_SIGNAL);
+  return sig;
+}
+
+void
+OwnerSignalQueueInit(OWNER_SIGNAL_QUEUE *queue)
+{
+  queue->env = enif_alloc_env();
+  CxPlatListInitializeHead(&queue->List);
+}
+
+void
+OwnerSignalQueueDestroy(OWNER_SIGNAL_QUEUE *queue)
+{
+  CXPLAT_DBG_ASSERT(CxPlatListIsEmpty(&queue->List));
+  enif_free_env(queue->env);
+  CxPlatFree(queue, QUICER_OWNER_SIGNAL);
+}
+
+OWNER_SIGNAL *
+OwnerSignalAlloc()
+{
+  OWNER_SIGNAL *sig = CxPlatAlloc(sizeof(OWNER_SIGNAL), QUICER_OWNER_SIGNAL);
+  sig->Link.Flink = NULL;
+  return sig;
+}
+
+void
+OwnerSignalFree(OWNER_SIGNAL *sig)
+{
+  CXPLAT_FREE(sig, QUICER_OWNER_SIGNAL);
+}
+
+void
+OwnerSignalEnqueue(_In_ OWNER_SIGNAL_QUEUE *queue, _In_ OWNER_SIGNAL *sig)
+{
+  CxPlatListInsertTail(&queue->List, &sig->Link);
+}
+
+OWNER_SIGNAL *
+OwnerSignalDequeue(_In_ OWNER_SIGNAL_QUEUE *queue)
+{
+  OWNER_SIGNAL *sig;
+  if (CxPlatListIsEmpty(&queue->List))
+    {
+      sig = NULL;
+    }
+  else
+    {
+      sig = CXPLAT_CONTAINING_RECORD(
+          CxPlatListRemoveHead(&queue->List), OWNER_SIGNAL, Link);
+    }
+
+  return sig;
+}

c_src/quicer_owner_queue.h

@@ -0,0 +1,50 @@
+/*--------------------------------------------------------------------
+Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+
+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.
+-------------------------------------------------------------------*/
+#ifndef QUICER_OWNER_QUEUE_H_
+#define QUICER_OWNER_QUEUE_H_
+
+#include <erl_nif.h>
+#include <quicer_internal.h>
+#include <msquic.h>
+#include <quic_platform.h>
+
+
+#define QUICER_OWNER_SIGNAL 'E0rQ' // 'Er0d'  QUICER_OWNER_SIGNAL
+
+// Owner Signal Queue
+typedef struct OWNER_SIGNAL_QUEUE
+{
+  ErlNifEnv *env;
+  CXPLAT_LIST_ENTRY List;
+} OWNER_SIGNAL_QUEUE;
+
+typedef struct OWNER_SIGNAL
+{
+  CXPLAT_LIST_ENTRY Link;
+  ERL_NIF_TERM msg;        // resides in `env` of OWNER_SIGNAL_QUEUE
+  ERL_NIF_TERM orig_owner; // owner when msg is generated
+} OWNER_SIGNAL;
+
+OWNER_SIGNAL_QUEUE *OwnerSignalQueueNew();
+OWNER_SIGNAL *OwnerSignalAlloc();
+void OwnerSignalQueueInit(OWNER_SIGNAL_QUEUE *queue);
+void OwnerSignalQueueDestroy(OWNER_SIGNAL_QUEUE *queue);
+void OwnerSignalFree(OWNER_SIGNAL *sig);
+void OwnerSignalEnqueue(_In_ OWNER_SIGNAL_QUEUE *queue,
+                        _In_ OWNER_SIGNAL *sig);
+OWNER_SIGNAL *OwnerSignalDequeue(_In_ OWNER_SIGNAL_QUEUE *queue);
+
+#endif // QUICER_OWNER_QUEUE_H_

c_src/quicer_stream.c

@@ -59,6 +59,9 @@ handle_stream_event_send_shutdown_complete(QuicerStreamCTX *s_ctx,
 
 static void reset_stream_recv(QuicerStreamCTX *s_ctx);
 
+static int
+signal_or_buffer(QuicerStreamCTX *s_ctx, ErlNifPid *owner, ERL_NIF_TERM sig);
+
 QUIC_STATUS
 ServerStreamCallback(HQUIC Stream, void *Context, QUIC_STREAM_EVENT *Event)
 {
@@ -1212,7 +1215,8 @@ handle_stream_event_send_shutdown_complete(QuicerStreamCTX *s_ctx,
                       ATOM_SEND_SHUTDOWN_COMPLETE,
                       enif_make_copy(env, s_ctx->eHandle),
                       ATOM_BOOLEAN(is_graceful));
-  enif_send(NULL, &(s_ctx->owner->Pid), NULL, report);
+
+  signal_or_buffer(s_ctx, &(s_ctx->owner->Pid), report);
   return QUIC_STATUS_SUCCESS;
 }
 
@@ -1268,6 +1272,136 @@ get_stream_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
   return res;
 }
 
+// s_ctx MUST be locked
+int
+signal_or_buffer(QuicerStreamCTX *s_ctx,
+                 ErlNifPid *owner_pid,
+                 ERL_NIF_TERM msg)
+{
+  if (s_ctx->sig_queue != NULL)
+    { // Ongoing handoff... buffering
+      CXPLAT_FRE_ASSERT(ACCEPTOR_RECV_MODE_PASSIVE == s_ctx->owner->active);
+      ErlNifEnv *q_env = s_ctx->sig_queue->env;
+      OWNER_SIGNAL *sig = OwnerSignalAlloc();
+      sig->msg = enif_make_copy(q_env, msg);
+      sig->orig_owner = enif_make_pid(q_env, owner_pid);
+      OwnerSignalEnqueue(s_ctx->sig_queue, sig);
+      return TRUE;
+    }
+  else
+    {
+      return enif_send(NULL, owner_pid, NULL, msg);
+    }
+}
+
+// s_ctx MUST be locked
+int
+flush_sig_buffer(__unused_parm__ ErlNifEnv *env, QuicerStreamCTX *s_ctx)
+{
+  OWNER_SIGNAL *sig = NULL;
+  if (!s_ctx->sig_queue)
+    {
+      return FALSE;
+    }
+
+  while ((sig = OwnerSignalDequeue(s_ctx->sig_queue)))
+    {
+      // if send failed, msg will be cleared in `OwnerSignalQueueDestroy`
+      enif_send(NULL, &(s_ctx->owner->Pid), NULL, sig->msg);
+
+      OwnerSignalFree(sig);
+    }
+  OwnerSignalQueueDestroy(s_ctx->sig_queue);
+  s_ctx->sig_queue = NULL;
+  return TRUE;
+}
+
+ERL_NIF_TERM
+buffer_sig(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerStreamCTX *s_ctx;
+  ErlNifPid orig_pid;
+  ERL_NIF_TERM res = ATOM_OK;
+
+  CXPLAT_FRE_ASSERT(argc == 3);
+
+  if (!enif_get_resource(env, argv[0], ctx_stream_t, (void **)&s_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+
+  if (!enif_get_local_pid(env, argv[1], &orig_pid))
+    {
+      return ERROR_TUPLE_2(ATOM_BAD_PID);
+    }
+
+  enif_mutex_lock(s_ctx->lock);
+  if (!s_ctx->sig_queue)
+    {
+      res = ERROR_TUPLE_2(ATOM_NONE);
+      goto Exit;
+    }
+
+  if (!signal_or_buffer(s_ctx, &orig_pid, argv[2]))
+    {
+      res = ERROR_TUPLE_2(ATOM_FALSE);
+    }
+Exit:
+  enif_mutex_unlock(s_ctx->lock);
+  return res;
+}
+
+ERL_NIF_TERM
+flush_stream_buffered_sigs(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerStreamCTX *s_ctx = NULL;
+  ERL_NIF_TERM res = ATOM_OK;
+
+  CXPLAT_FRE_ASSERT(argc == 1);
+
+  if (!enif_get_resource(env, argv[0], ctx_stream_t, (void **)&s_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+  enif_mutex_lock(s_ctx->lock);
+  if (!flush_sig_buffer(env, s_ctx))
+    {
+      res = ERROR_TUPLE_2(ATOM_NONE);
+    }
+  enif_mutex_unlock(s_ctx->lock);
+  return res;
+}
+
+/*
+** Enable signal buffering.
+** Signals are buffered instead of being sent to the owner.
+** call `flush_stream_buffered_sigs` to flush the buffer.
+*/
+ERL_NIF_TERM
+enable_sig_buffer(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerStreamCTX *s_ctx = NULL;
+  ERL_NIF_TERM res = ATOM_OK;
+
+  CXPLAT_FRE_ASSERT(argc == 1);
+
+  if (!enif_get_resource(env, argv[0], ctx_stream_t, (void **)&s_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+
+  enif_mutex_lock(s_ctx->lock);
+  if (!s_ctx->sig_queue)
+    {
+      s_ctx->owner->active = ACCEPTOR_RECV_MODE_PASSIVE;
+      s_ctx->sig_queue = OwnerSignalQueueNew();
+      OwnerSignalQueueInit(s_ctx->sig_queue);
+    }
+  enif_mutex_unlock(s_ctx->lock);
+
+  return res;
+}
+
 ///_* Emacs
 ///====================================================================
 /// Local Variables:

c_src/quicer_stream.h

@@ -66,3 +66,14 @@ get_stream_rid1(ErlNifEnv *env, int args, const ERL_NIF_TERM argv[]);
 
 ERL_NIF_TERM
 get_stream_owner1(ErlNifEnv *env, int args, const ERL_NIF_TERM argv[]);
+
+ERL_NIF_TERM
+buffer_sig(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
+
+ERL_NIF_TERM
+flush_stream_buffered_sigs(ErlNifEnv *env,
+                           int argc,
+                           const ERL_NIF_TERM argv[]);
+
+ERL_NIF_TERM
+enable_sig_buffer(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);

src/quicer_nif.erl

@@ -62,7 +62,11 @@
     get_connections/1,
     get_conn_owner/1,
     get_stream_owner/1,
-    get_listener_owner/1
+    get_listener_owner/1,
+    buffer_sig/3,
+    %% @TODO move to API
+    flush_stream_buffered_sigs/1,
+    enable_sig_buffer/1
 ]).
 
 -export([abi_version/0]).
@@ -373,6 +377,19 @@ get_connections() ->
 get_connections(_RegHandle) ->
     erlang:nif_error(nif_library_not_loaded).
 
+-spec enable_sig_buffer(stream_handle()) -> ok.
+enable_sig_buffer(_H) ->
+    erlang:nif_error(nif_library_not_loaded).
+
+-spec buffer_sig(stream_handle(), OrigOwner :: pid(), term()) ->
+    ok | {error, false | none | bad_pid | bad_arg}.
+buffer_sig(_StreamHandle, _OrigOwner, _Msg) ->
+    erlang:nif_error(nif_library_not_loaded).
+
+-spec flush_stream_buffered_sigs(stream_handle()) -> ok | {error, badarg | none}.
+flush_stream_buffered_sigs(_H) ->
+    erlang:nif_error(nif_library_not_loaded).
+
 %% Internals
 -spec locate_lib(file:name(), file:name()) ->
     {ok, file:filename()} | {error, not_found}.