buffer stream signals

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

@@ -1486,10 +1486,11 @@ stream_controlling_process(ErlNifEnv *env,
     {
       // rollback, must success
       enif_self(env, &s_ctx->owner->Pid);
+      flush_sig_buffer(env, s_ctx);
       enif_monitor_process(env, s_ctx, caller, &s_ctx->owner_mon);
       return ERROR_TUPLE_2(ATOM_OWNER_DEAD);
     }
-
+  flush_sig_buffer(env, s_ctx);
   TP_NIF_3(exit, (uintptr_t)s_ctx->Stream, (uintptr_t)&s_ctx->owner->Pid);
   return ATOM_OK;
 }
@@ -1575,6 +1576,8 @@ static ErlNifFunc nif_funcs[] = {
   { "setopt", 4, setopt4, 0},
   { "controlling_process", 2, controlling_process, 0},
   { "peercert", 1, peercert1, 0},
+  { "enable_sig_buffer", 1, enable_sig_buffer, 0},
+  { "flush_stream_buffered_sigs", 1, flush_stream_buffered_sigs, 0},
   /* for DEBUG */
   { "get_conn_rid", 1, get_conn_rid1, 1},
   { "get_stream_rid", 1, get_stream_rid1, 1},
@@ -1584,7 +1587,9 @@ 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 */
+  { "mock_buffer_sig", 3, mock_buffer_sig, 0}
   // clang-format on
 };
 

c_src/quicer_owner_queue.c

@@ -0,0 +1,75 @@
+/*--------------------------------------------------------------------
+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);
+  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,52 @@
+/*--------------------------------------------------------------------
+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>
+
+// clang-format off
+#include <quicer_internal.h>
+#include <msquic.h>
+#include <quic_platform.h>
+// clang-format on
+
+#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_