From cd36f1ef5f9b0e88373c5fe9bccd58ad9efbdc9b Mon Sep 17 00:00:00 2001
From: Calvin Rose <calsrose@gmail.com>
Date: Wed, 16 Aug 2023 14:26:52 -0500
Subject: [PATCH] Distinguish between threaded channels and non-threaded when
 marshalling.

Threaded channels _can_ be marshalled, just not for communication
between threads. This is a special case since the same abstract type
is used for both threaded and non-threaded channels.
---
 src/core/ev.c       | 9 ++++++++-
 src/core/marsh.c    | 6 ++++++
 src/include/janet.h | 1 +
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/src/core/ev.c b/src/core/ev.c
index 59a461d8..100f9332 100644
--- a/src/core/ev.c
+++ b/src/core/ev.c
@@ -1224,6 +1224,7 @@ static Janet janet_chanat_next(void *p, Janet key) {
 
 static void janet_chanat_marshal(void *p, JanetMarshalContext *ctx) {
     JanetChannel *channel = (JanetChannel *)p;
+    janet_marshal_byte(ctx, channel->is_threaded);
     janet_marshal_abstract(ctx, channel);
     janet_marshal_byte(ctx, channel->closed);
     janet_marshal_int(ctx, channel->limit);
@@ -1243,7 +1244,13 @@ static void janet_chanat_marshal(void *p, JanetMarshalContext *ctx) {
 }
 
 static void *janet_chanat_unmarshal(JanetMarshalContext *ctx) {
-    JanetChannel *abst = janet_unmarshal_abstract(ctx, sizeof(JanetChannel));
+    uint8_t is_threaded = janet_unmarshal_byte(ctx);
+    JanetChannel *abst;
+    if (is_threaded) {
+        abst = janet_unmarshal_abstract_threaded(ctx, sizeof(JanetChannel));
+    } else {
+        abst = janet_unmarshal_abstract(ctx, sizeof(JanetChannel));
+    }
     uint8_t is_closed = janet_unmarshal_byte(ctx);
     int32_t limit = janet_unmarshal_int(ctx);
     int32_t count = janet_unmarshal_int(ctx);
diff --git a/src/core/marsh.c b/src/core/marsh.c
index f81e91c0..b8f6de20 100644
--- a/src/core/marsh.c
+++ b/src/core/marsh.c
@@ -1245,6 +1245,12 @@ void *janet_unmarshal_abstract(JanetMarshalContext *ctx, size_t size) {
     return p;
 }
 
+void *janet_unmarshal_abstract_threaded(JanetMarshalContext *ctx, size_t size) {
+    void *p = janet_abstract_threaded(ctx->at, size);
+    janet_unmarshal_abstract_reuse(ctx, p);
+    return p;
+}
+
 static const uint8_t *unmarshal_one_abstract(UnmarshalState *st, const uint8_t *data, Janet *out, int flags) {
     Janet key;
     data = unmarshal_one(st, data, &key, flags + 1);
diff --git a/src/include/janet.h b/src/include/janet.h
index 6d0ef7ec..55ff974b 100644
--- a/src/include/janet.h
+++ b/src/include/janet.h
@@ -2072,6 +2072,7 @@ JANET_API uint8_t janet_unmarshal_byte(JanetMarshalContext *ctx);
 JANET_API void janet_unmarshal_bytes(JanetMarshalContext *ctx, uint8_t *dest, size_t len);
 JANET_API Janet janet_unmarshal_janet(JanetMarshalContext *ctx);
 JANET_API JanetAbstract janet_unmarshal_abstract(JanetMarshalContext *ctx, size_t size);
+JANET_API JanetAbstract janet_unmarshal_abstract_threaded(JanetMarshalContext *ctx, size_t size);
 JANET_API void janet_unmarshal_abstract_reuse(JanetMarshalContext *ctx, void *p);
 
 JANET_API void janet_register_abstract_type(const JanetAbstractType *at);