build (refactor): Remove 'zip' opt from publish_spark

The slight convenience afforded by this opt doesn't seem worth the additional complexity. This script produces a self-contained application folder. Decisions on what to do with that folder (whether or not to zip it, etc.) are best left upstream.
fix: Prevent raw_pb_callback premature garbage collection
2022-12-23 11:40:35 -05:00 · 2022-12-23 11:25:07 -05:00
2 changed files with 216 additions and 208 deletions
@@ -1,192 +1,216 @@
-using Google.Protobuf.WellKnownTypes;
-using Google.Protobuf;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-using System.Reactive.Linq;
-using System.IO;
-using System.Reactive.Disposables;
-using mbts;
-using System.Threading;
-
-namespace bts.api.core {
-
-   internal class QuickbusClientChannel : ClientChannel {
-      IntPtr _client;
-
-      public QuickbusClientChannel(ServiceContext context, string type_name, string qualifier) : base(context) {
-         _client = ApiNativeMethods.api_client_create(context.native_context, type_name, qualifier);
-         System.Diagnostics.Trace.Assert(_client != IntPtr.Zero);
-      }
-
-      public override void Dispose() {
-         ApiNativeMethods.api_client_destroy(_client);
-         _client = IntPtr.Zero;
-      }
-
-
-      int _req_id = 0;
-      Dictionary<int, ApiNativeMethods.raw_pb_callback> _keepalive_cache = new Dictionary<int, ApiNativeMethods.raw_pb_callback>();
-
-      unsafe public override Task<Result<Response>> send<Request, Response>(int method_id, Request request) {
-         var id = Interlocked.Increment(ref _req_id);
-         var header = new RequestHeader() {
-            Method = method_id,
-            RequestId = id,
-            Token = _context.token.HasValue ? new AuthToken() { Token = _context.token?.Token } : null
-         };
-         var buffer = header.serialize(request);
-         var tcs = new TaskCompletionSource<Result<Response>>();
-         //ApiNativeMethods.api_client_log_error($"Call {id}");
-         ApiNativeMethods.raw_pb_callback callback = (rbuffer, rlength, _) => {
-            //ApiNativeMethods.api_client_log_error($"Reply {id}");
-            var res = ApiUtils.deserialize<ResponseHeader, Response>(rbuffer, (long)rlength);
-            if (res.header.Code == ApiError.Success) {
-               tcs.TrySetResult(new Result<Response>(res.body));
-            }
-            else {
-               tcs.TrySetResult(new Result<Response>(res.header.Code));
-            }
-            lock (_keepalive_cache) { _keepalive_cache.Remove(id); }
-         };
-         System.Diagnostics.Trace.Assert(_client != IntPtr.Zero, "Channel may have been disposed");
-         lock (_keepalive_cache) { _keepalive_cache.Add(id, callback); }
-         ApiNativeMethods.api_client_send(_client, buffer, buffer.Length, callback, IntPtr.Zero);
-         return tcs.Task;
-      }
-
-      public override IObservable<RawUpdate<TOut>> subscribe<TParse,TOut>(int method_id, string method, string qualifier, MessageParser<TParse> parser, Func<TParse,TOut> converter) {
-         var result = Observable.Create<RawUpdate<TOut>>(async observer => {
-            IntPtr temp_subscription = IntPtr.Zero;
-            void unsubscribe_temp_sub() {
-               // Unsubscribe late to keep qb refcount > 0 continuously
-               if (temp_subscription != IntPtr.Zero) ApiNativeMethods.api_client_unsubscribe(_client, temp_subscription);
-            }
-
-            // 'tcs' and 'temp_callback' are defined in the same scope as 'temp_subscription'
-            // (rather than in the 'while loop' scope) to prevent premature garbage collection
-            TaskCompletionSource<bool> tcs = null;
-            ApiNativeMethods.raw_subscription_callback temp_callback =
-               (event_type, changes, change_count, unused) => {
-               tcs?.TrySetResult(true);
-            };
-
-            while (true) {
-               var response = await check_stream_access(method_id, qualifier);
-               if (response.success) {
-                  if (response.value.Authorized) {
-                     // success. Break loop and subscribe
-                     break;
-                  }
-                  else {
-                     // Not authorized. return empty stream
-                     unsubscribe_temp_sub();
-                     observer.OnCompleted();
-                     return System.Reactive.Disposables.Disposable.Empty;
-                  }
-               }
-               else {
-                  // Wait for stream to fire once before trying again
-                  unsubscribe_temp_sub();
-                  tcs = new TaskCompletionSource<bool>();
-                  temp_subscription = ApiNativeMethods.api_client_subscribe(
-                     _client, method, qualifier, temp_callback, IntPtr.Zero);
-                  await tcs.Task;
-               }
-            }
-
-            // Authorized. Stream events
-            unsubscribe_temp_sub();
-            // Ensure 'temp_callback' and 'tcs' aren't Garbage Collected before 'temp_subscription'
-            // is disposed
-            GC.KeepAlive(temp_callback);
-            GC.KeepAlive(tcs);
-            return SubscribeNative(method, qualifier, parser, converter, observer);
-         });
-         Log.QBus.diag($"Subscribing for sub: '{method}/{qualifier}'");
-         return result;
-      }
-      private Task<Result<StreamAccessResponse>> check_stream_access(int method_id, string qualifier) {
-         var request = new StreamAccessRequest() { Qualifier = qualifier ?? "" };
-         return send<StreamAccessRequest, StreamAccessResponse>(method_id, request);
-      }
-
-      private unsafe IDisposable SubscribeNative<TParse, TOut>(string method, string qualifier, MessageParser<TParse> parser, Func<TParse, TOut> converter, IObserver<RawUpdate<TOut>> observer) where TParse : IMessage<TParse> {
-         bool first_update = true;
-         // Memory optimization. There is only one update per subscription.
-         RawUpdate<TOut> the_update = new RawUpdate<TOut>();
-         ApiNativeMethods.raw_subscription_callback cb = (event_type, changes, change_count, unused) => {
-            the_update.Clear();
-            try {
-               if (event_type == ApiNativeMethods.EventType.TABLE_UPDATED) {
-                  if (changes != null) {
-                     foreach (var change in changes) {
-                        switch (change.type) {
-                           case ApiNativeMethods.ChangeType.REMOVE:
-                              the_update.removed_keys.Add(change.key);
-                              break;
-                           case ApiNativeMethods.ChangeType.REPLACE:
-                              the_update.changed_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
-                              the_update.changed_keys.Add(change.key);
-                              break;
-                           case ApiNativeMethods.ChangeType.CREATE:
-                              the_update.added_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
-                              the_update.added_keys.Add(change.key);
-                              break;
-                        }
-                     }
-                  }
-                  observer.OnNext(the_update);
-               }
-               else if (event_type == ApiNativeMethods.EventType.TABLE_CREATED) {
-                  if (!first_update) {
-                     observer.OnNext(the_update.MarkAsReset());
-                  }
-                  first_update = false;
-               }
-            }
-            catch (Exception e) {
-               ApiNativeMethods.api_client_log_error($"Exception in subscription callback for sub: '{method}/{qualifier}' \n--\n '{e}'");
-               // Sleep to let the log message get asyncronously written to disk
-               Thread.Sleep(100);
-               Environment.Exit(-1);
-            }
-         };
-         var subscription = ApiNativeMethods.api_client_subscribe(_client, method, qualifier, cb, IntPtr.Zero);
-         return FinalizeableAnonymousDisposable.Create((disposing) => {
-            Log.QBus.diag($"Unsubscribing for sub: '{method}/{qualifier}'");
-            ApiNativeMethods.api_client_unsubscribe(_client, subscription);
-            subscription = IntPtr.Zero;
-            GC.KeepAlive(cb);
-            cb = null;
-         });
-      }
-
-      private class FinalizeableAnonymousDisposable : IDisposable {
-         private Action<bool> _action;
-         public FinalizeableAnonymousDisposable(Action<bool> dispose) {
-            _action = dispose;
-         }
-         public static FinalizeableAnonymousDisposable Create(Action<bool> dispose) { return new FinalizeableAnonymousDisposable(dispose); }
-         ~FinalizeableAnonymousDisposable() {
-            Dispose(false);
-         }
-         public void Dispose() {
-            GC.SuppressFinalize(this);
-            Dispose(true);
-         }
-
-         protected virtual void Dispose(bool disposing) {
-            var temp = _action;
-            _action = null;
-            temp?.Invoke(disposing);
-         }
-      }
-
-      public override bool is_sync_call_ok() {
-         return !ApiNativeMethods.api_client_is_on_qb_thread(_client);
-      }
-   }
-}
+using Google.Protobuf.WellKnownTypes;
+using Google.Protobuf;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using System.Reactive.Linq;
+using System.IO;
+using System.Reactive.Disposables;
+using mbts;
+using System.Threading;
+
+namespace bts.api.core {
+
+   internal class QuickbusClientChannel : ClientChannel {
+      IntPtr _client;
+
+      public QuickbusClientChannel(ServiceContext context, string type_name, string qualifier) : base(context) {
+         _client = ApiNativeMethods.api_client_create(context.native_context, type_name, qualifier);
+         System.Diagnostics.Trace.Assert(_client != IntPtr.Zero);
+      }
+
+      public override void Dispose() {
+         ApiNativeMethods.api_client_destroy(_client);
+         _client = IntPtr.Zero;
+      }
+
+
+      int _req_id = 0;
+      Dictionary<int, PBCallbackWrapper> _keepalive_cache = new();
+
+      unsafe public override Task<Result<Response>> send<Request, Response>(int method_id, Request request) {
+         var id = Interlocked.Increment(ref _req_id);
+         var header = new RequestHeader() {
+            Method = method_id,
+            RequestId = id,
+            Token = _context.token.HasValue ? new AuthToken() { Token = _context.token?.Token } : null
+         };
+         var buffer = header.serialize(request);
+         var tcs = new TaskCompletionSource<Result<Response>>();
+         //ApiNativeMethods.api_client_log_error($"Call {id}");
+         ApiNativeMethods.raw_pb_callback callback = (rbuffer, rlength, _) => {
+            //ApiNativeMethods.api_client_log_error($"Reply {id}");
+            var res = ApiUtils.deserialize<ResponseHeader, Response>(rbuffer, (long)rlength);
+            if (res.header.Code == ApiError.Success) {
+               tcs.TrySetResult(new Result<Response>(res.body));
+            }
+            else {
+               tcs.TrySetResult(new Result<Response>(res.header.Code));
+            }
+            lock (_keepalive_cache) {
+               _keepalive_cache.Remove(id);
+            }
+         };
+         System.Diagnostics.Trace.Assert(_client != IntPtr.Zero, "Channel may have been disposed");
+         var cb_wrapper = new PBCallbackWrapper(callback);
+         lock (_keepalive_cache) {
+            _keepalive_cache.Add(id, cb_wrapper);
+         }
+         ApiNativeMethods.api_client_send(
+            _client, buffer, buffer.Length, cb_wrapper.callback, IntPtr.Zero);
+         return tcs.Task;
+      }
+
+      public override IObservable<RawUpdate<TOut>> subscribe<TParse,TOut>(int method_id, string method, string qualifier, MessageParser<TParse> parser, Func<TParse,TOut> converter) {
+         var result = Observable.Create<RawUpdate<TOut>>(async observer => {
+            IntPtr temp_subscription = IntPtr.Zero;
+            void unsubscribe_temp_sub() {
+               // Unsubscribe late to keep qb refcount > 0 continuously
+               if (temp_subscription != IntPtr.Zero) ApiNativeMethods.api_client_unsubscribe(_client, temp_subscription);
+            }
+
+            // 'tcs' and 'temp_callback' are defined in the same scope as 'temp_subscription'
+            // (rather than in the 'while loop' scope) to prevent premature garbage collection
+            TaskCompletionSource<bool> tcs = null;
+            ApiNativeMethods.raw_subscription_callback temp_callback =
+               (event_type, changes, change_count, unused) => {
+               tcs?.TrySetResult(true);
+            };
+
+            while (true) {
+               var response = await check_stream_access(method_id, qualifier);
+               if (response.success) {
+                  if (response.value.Authorized) {
+                     // success. Break loop and subscribe
+                     break;
+                  }
+                  else {
+                     // Not authorized. return empty stream
+                     unsubscribe_temp_sub();
+                     observer.OnCompleted();
+                     return System.Reactive.Disposables.Disposable.Empty;
+                  }
+               }
+               else {
+                  // Wait for stream to fire once before trying again
+                  unsubscribe_temp_sub();
+                  tcs = new TaskCompletionSource<bool>();
+                  temp_subscription = ApiNativeMethods.api_client_subscribe(
+                     _client, method, qualifier, temp_callback, IntPtr.Zero);
+                  await tcs.Task;
+               }
+            }
+
+            // Authorized. Stream events
+            unsubscribe_temp_sub();
+            // Ensure 'temp_callback' and 'tcs' aren't Garbage Collected before 'temp_subscription'
+            // is disposed
+            GC.KeepAlive(temp_callback);
+            GC.KeepAlive(tcs);
+            return SubscribeNative(method, qualifier, parser, converter, observer);
+         });
+         Log.QBus.diag($"Subscribing for sub: '{method}/{qualifier}'");
+         return result;
+      }
+      private Task<Result<StreamAccessResponse>> check_stream_access(int method_id, string qualifier) {
+         var request = new StreamAccessRequest() { Qualifier = qualifier ?? "" };
+         return send<StreamAccessRequest, StreamAccessResponse>(method_id, request);
+      }
+
+      private unsafe IDisposable SubscribeNative<TParse, TOut>(string method, string qualifier, MessageParser<TParse> parser, Func<TParse, TOut> converter, IObserver<RawUpdate<TOut>> observer) where TParse : IMessage<TParse> {
+         bool first_update = true;
+         // Memory optimization. There is only one update per subscription.
+         RawUpdate<TOut> the_update = new RawUpdate<TOut>();
+         ApiNativeMethods.raw_subscription_callback cb = (event_type, changes, change_count, unused) => {
+            the_update.Clear();
+            try {
+               if (event_type == ApiNativeMethods.EventType.TABLE_UPDATED) {
+                  if (changes != null) {
+                     foreach (var change in changes) {
+                        switch (change.type) {
+                           case ApiNativeMethods.ChangeType.REMOVE:
+                              the_update.removed_keys.Add(change.key);
+                              break;
+                           case ApiNativeMethods.ChangeType.REPLACE:
+                              the_update.changed_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
+                              the_update.changed_keys.Add(change.key);
+                              break;
+                           case ApiNativeMethods.ChangeType.CREATE:
+                              the_update.added_items.Add(converter(parser.ParseFrom(new UnmanagedMemoryStream(change.buffer, change.buffer_len))));
+                              the_update.added_keys.Add(change.key);
+                              break;
+                        }
+                     }
+                  }
+                  observer.OnNext(the_update);
+               }
+               else if (event_type == ApiNativeMethods.EventType.TABLE_CREATED) {
+                  if (!first_update) {
+                     observer.OnNext(the_update.MarkAsReset());
+                  }
+                  first_update = false;
+               }
+            }
+            catch (Exception e) {
+               ApiNativeMethods.api_client_log_error($"Exception in subscription callback for sub: '{method}/{qualifier}' \n--\n '{e}'");
+               // Sleep to let the log message get asyncronously written to disk
+               Thread.Sleep(100);
+               Environment.Exit(-1);
+            }
+         };
+         var subscription = ApiNativeMethods.api_client_subscribe(_client, method, qualifier, cb, IntPtr.Zero);
+         return FinalizeableAnonymousDisposable.Create((disposing) => {
+            Log.QBus.diag($"Unsubscribing for sub: '{method}/{qualifier}'");
+            ApiNativeMethods.api_client_unsubscribe(_client, subscription);
+            subscription = IntPtr.Zero;
+            GC.KeepAlive(cb);
+            cb = null;
+         });
+      }
+
+      private class FinalizeableAnonymousDisposable : IDisposable {
+         private Action<bool> _action;
+         public FinalizeableAnonymousDisposable(Action<bool> dispose) {
+            _action = dispose;
+         }
+         public static FinalizeableAnonymousDisposable Create(Action<bool> dispose) { return new FinalizeableAnonymousDisposable(dispose); }
+         ~FinalizeableAnonymousDisposable() {
+            Dispose(false);
+         }
+         public void Dispose() {
+            GC.SuppressFinalize(this);
+            Dispose(true);
+         }
+
+         protected virtual void Dispose(bool disposing) {
+            var temp = _action;
+            _action = null;
+            temp?.Invoke(disposing);
+         }
+      }
+
+      public override bool is_sync_call_ok() {
+         return !ApiNativeMethods.api_client_is_on_qb_thread(_client);
+      }
+
+      /// <summary>
+      /// A simple wrapper that can be used to prevent the premature garbage collection of a
+      /// <see cref="ApiNativeMethods.raw_pb_callback"/> delegate.
+      /// </summary>
+      /// <remarks>
+      /// This is part of a workaround for an apparent bug in csharp's GarbageCollector, where the
+      /// final reference to a delegate instance mustn't be removed until any and all execution has
+      /// finished (even if the reference is removed by the delegate's last statemwent, a
+      /// garbage-collector race may yield the infamous error, "A callback was made on a garbage
+      /// collected delegate of type...").
+      /// </remarks>
+      class PBCallbackWrapper {
+         public readonly ApiNativeMethods.raw_pb_callback callback;
+
+         public PBCallbackWrapper(ApiNativeMethods.raw_pb_callback callback) =>
+            this.callback = callback;
+      }
+   }
+}
@@ -1,9 +1,6 @@
 import argparse
-import os
 from pathlib import Path
-import subprocess

-from core.argparse_utils import add_toggle_opt
 from build_spark import build
 from build_spark import SparkPaths

@@ -12,12 +9,6 @@ def _get_arg_parser():
    parser = argparse.ArgumentParser(
        description='Publish a self-contined spark client application')

-    add_toggle_opt(
-        parser, 'zip', help='Whether or not to zip the build folder')
-
-    parser.add_argument(
-        '--build-id', help='A unique identifier for this build')
-
    parser.add_argument(
        '-o',
        '--output-dir',
@@ -53,13 +44,6 @@ def _main():
        targets=['publish'],
        verbosity=args.verbosity)

-    if args.zip:
-        publish_path = output_dir.joinpath(
-            'Release', 'bin', 'net6.0-windows', 'win-x64', 'publish', '*')
-        subprocess.run(
-            f'7z a -mmt=30 spark_client_{args.build_id}.7z {publish_path}',
-            shell=True)
-

 if __name__ == '__main__':
    _main()