fix(stream): free sigQ when destroy the stream

emqx · May 23, 2024 · dbcc82a · dbcc82a
1 parent 41b3716
commit dbcc82a
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 5 deletions.
diff --git a/c_src/quicer_owner_queue.h b/c_src/quicer_owner_queue.h
@@ -17,10 +17,12 @@ limitations under the License.
 #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
 

diff --git a/c_src/quicer_stream.c b/c_src/quicer_stream.c
@@ -62,6 +62,8 @@ static void reset_stream_recv(QuicerStreamCTX *s_ctx);
 static int
 signal_or_buffer(QuicerStreamCTX *s_ctx, ErlNifPid *owner, ERL_NIF_TERM sig);
 
+static int flush_sig_buffer(ErlNifEnv *env, QuicerStreamCTX *s_ctx);
+
 QUIC_STATUS
 ServerStreamCallback(HQUIC Stream, void *Context, QUIC_STREAM_EVENT *Event)
 {
@@ -132,6 +134,7 @@ ServerStreamCallback(HQUIC Stream, void *Context, QUIC_STREAM_EVENT *Event)
 
   if (is_destroy)
     {
+      flush_sig_buffer(NULL, s_ctx);
       s_ctx->is_closed = TRUE;
     }
 
@@ -232,6 +235,7 @@ _IRQL_requires_max_(DISPATCH_LEVEL)
   if (is_destroy)
     {
       s_ctx->is_closed = TRUE;
+      flush_sig_buffer(NULL, s_ctx);
       MsQuic->SetCallbackHandler(Stream, NULL, NULL);
     }
 
@@ -1296,7 +1300,7 @@ signal_or_buffer(QuicerStreamCTX *s_ctx,
 
 // s_ctx MUST be locked
 int
-flush_sig_buffer(__unused_parm__ ErlNifEnv *env, QuicerStreamCTX *s_ctx)
+flush_sig_buffer(ErlNifEnv *env, QuicerStreamCTX *s_ctx)
 {
   OWNER_SIGNAL *sig = NULL;
   if (!s_ctx->sig_queue)
@@ -1307,7 +1311,7 @@ flush_sig_buffer(__unused_parm__ ErlNifEnv *env, QuicerStreamCTX *s_ctx)
   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);
+      enif_send(env, &(s_ctx->owner->Pid), NULL, sig->msg);
 
       OwnerSignalFree(sig);
     }

diff --git a/test/prop_stream_sig_queue.erl b/test/prop_stream_sig_queue.erl
@@ -8,10 +8,11 @@
 %%%%%%%%%%%%%%%%%%
 prop_buffer_sig_err_none() ->
     ?FORALL(
-        {#prop_handle{handle = S}, Pid, Term},
+        {#prop_handle{handle = S, destructor = Destructor}, Pid, Term},
         {valid_stream(), pid(), term()},
         begin
             Res = quicer_nif:buffer_sig(S, Pid, Term),
+            Destructor(),
             Res == {error, none}
         end
     ).
@@ -97,7 +98,7 @@ receive_n(N, Ref, Acc) ->
             receive_n(N - 1, Ref, [X | Acc]);
         {quic, _, _, _} = _Drop ->
             receive_n(N, Ref, Acc)
-    after 1000 ->
+    after 500 ->
         {timeout, N}
     end.