ios: fix leaking of http streams (#571)

At the moment, we leak all `EnvoyHTTPStreamImpl` and `EnvoyHTTPCallbacks` instances with every stream due to the deliberate retain cycle used to keep the stream in memory and pass callbacks through to the platform layer while the stream is active. This ends up also leaking the callback closures specified by the end consumer, along with anything captured by those closures.

To resolve this problem, we will release the strong reference cycle that the stream holds to itself when the stream completes.

The state/callback for stream completion has to be stored somewhere in the `ios_context` type and retained since the `ios_context` is a struct and doesn't retain its members.

We considered having the `EnvoyHTTPCallbacks` hold a reference to the `EnvoyHTTPStreamImpl` rather than having `EnvoyHTTPStreamImpl` hold a strong reference to itself. However, this posed a few problems:
- `EnvoyHTTPCallbacks` is designed to support being shared by multiple streams, and thus cannot have only 1 reference to a stream
- `EnvoyHTTPCallbacks` is instantiated by the Swift layer, which means we'd have to leak implementation details of how the stream is kept in memory and shift logic out of the stream implementation

With these in mind, we decided to have `EnvoyHTTPStreamImpl` manage its own lifecycle based on the state of the network stream.

Signed-off-by: Michael Rebello <[email protected]>
Signed-off-by: JP Simard <[email protected]>

mobile/library/objective-c/EnvoyEngine.h

@@ -112,6 +112,11 @@ typedef NSDictionary<NSString *, NSArray<NSString *> *> EnvoyHeaders;
  */
 - (int)cancel;
 
+/**
+ Clean up the stream after it's closed (by completion, cancellation, or error).
+ */
+- (void)cleanUp;
+
 @end
 
 #pragma mark - EnvoyHTTPStreamImpl

mobile/library/objective-c/EnvoyHTTPStreamImpl.m

@@ -8,7 +8,13 @@
 #pragma mark - Utility types and functions
 
 typedef struct {
-  EnvoyHTTPCallbacks *callbacks;
+  // The stream is kept in memory through a strong reference to itself. In order to free the
+  // stream when it finishes, it is stored in this context so that it can be called when it
+  // is safe to be cleaned up.
+  // This approach allows `EnvoyHTTPCallbacks` to be agnostic of associated streams, enabling
+  // instances to be reused with multiple streams if desired.
+  __unsafe_unretained EnvoyHTTPStreamImpl *stream;
+  __unsafe_unretained EnvoyHTTPCallbacks *callbacks;
   atomic_bool *canceled;
 } ios_context;
 
@@ -85,6 +91,7 @@ static void ios_on_headers(envoy_headers headers, bool end_stream, void *context
     if (atomic_load(c->canceled) || !callbacks.onHeaders) {
       return;
     }
+
     callbacks.onHeaders(to_ios_headers(headers), end_stream);
   });
 }
@@ -96,6 +103,7 @@ static void ios_on_data(envoy_data data, bool end_stream, void *context) {
     if (atomic_load(c->canceled) || !callbacks.onData) {
       return;
     }
+
     callbacks.onData(to_ios_data(data), end_stream);
   });
 }
@@ -107,6 +115,7 @@ static void ios_on_metadata(envoy_headers metadata, void *context) {
     if (atomic_load(c->canceled) || !callbacks.onMetadata) {
       return;
     }
+
     callbacks.onMetadata(to_ios_headers(metadata));
   });
 }
@@ -118,47 +127,59 @@ static void ios_on_trailers(envoy_headers trailers, void *context) {
     if (atomic_load(c->canceled) || !callbacks.onTrailers) {
       return;
     }
+
     callbacks.onTrailers(to_ios_headers(trailers));
   });
 }
 
 static void ios_on_complete(void *context) {
   ios_context *c = (ios_context *)context;
   EnvoyHTTPCallbacks *callbacks = c->callbacks;
+  EnvoyHTTPStreamImpl *stream = c->stream;
   dispatch_async(callbacks.dispatchQueue, ^{
-    // TODO: release stream
     if (atomic_load(c->canceled)) {
       return;
     }
+
+    assert(stream);
+    [stream cleanUp];
   });
 }
 
 static void ios_on_cancel(void *context) {
   ios_context *c = (ios_context *)context;
   EnvoyHTTPCallbacks *callbacks = c->callbacks;
-  // TODO: release stream
+  EnvoyHTTPStreamImpl *stream = c->stream;
   dispatch_async(callbacks.dispatchQueue, ^{
     // This call is atomically gated at the call-site and will only happen once.
-    if (!callbacks.onCancel) {
-      return;
+    if (callbacks.onCancel) {
+      callbacks.onCancel();
     }
-    callbacks.onCancel();
+
+    assert(stream);
+    [stream cleanUp];
   });
 }
 
 static void ios_on_error(envoy_error error, void *context) {
   ios_context *c = (ios_context *)context;
   EnvoyHTTPCallbacks *callbacks = c->callbacks;
+  EnvoyHTTPStreamImpl *stream = c->stream;
   dispatch_async(callbacks.dispatchQueue, ^{
-    // TODO: release stream
-    if (atomic_load(c->canceled) || !callbacks.onError) {
+    if (atomic_load(c->canceled)) {
       return;
     }
-    NSString *errorMessage = [[NSString alloc] initWithBytes:error.message.bytes
-                                                      length:error.message.length
-                                                    encoding:NSUTF8StringEncoding];
-    error.message.release(error.message.context);
-    callbacks.onError(error.error_code, errorMessage);
+
+    if (callbacks.onError) {
+      NSString *errorMessage = [[NSString alloc] initWithBytes:error.message.bytes
+                                                        length:error.message.length
+                                                      encoding:NSUTF8StringEncoding];
+      error.message.release(error.message.context);
+      callbacks.onError(error.error_code, errorMessage);
+    }
+
+    assert(stream);
+    [stream cleanUp];
   });
 }
 
@@ -179,13 +200,14 @@ - (instancetype)initWithHandle:(envoy_stream_t)handle
     return nil;
   }
 
-  _streamHandle = handle;
   // Retain platform callbacks
   _platformCallbacks = callbacks;
+  _streamHandle = handle;
 
   // Create callback context
   ios_context *context = safe_malloc(sizeof(ios_context));
   context->callbacks = callbacks;
+  context->stream = self;
   context->canceled = safe_malloc(sizeof(atomic_bool));
   atomic_store(context->canceled, NO);
 
@@ -197,7 +219,6 @@ - (instancetype)initWithHandle:(envoy_stream_t)handle
 
   // We need create the native-held strong ref on this stream before we call start_stream because
   // start_stream could result in a reset that would release the native ref.
-  // TODO: To be truly safe we probably need stronger guarantees of operation ordering on this ref
   _strongSelf = self;
   envoy_stream_options stream_options = {bufferForRetry};
   envoy_status_t result = start_stream(_streamHandle, native_callbacks, stream_options);
@@ -210,8 +231,7 @@ - (instancetype)initWithHandle:(envoy_stream_t)handle
 }
 
 - (void)dealloc {
-  envoy_http_callbacks native_callbacks = _nativeCallbacks;
-  ios_context *context = native_callbacks.context;
+  ios_context *context = _nativeCallbacks.context;
   free(context->canceled);
   free(context);
 }
@@ -247,4 +267,8 @@ - (int)cancel {
   }
 }
 
+- (void)cleanUp {
+  _strongSelf = nil;
+}
+
 @end

mobile/library/swift/test/MockEnvoyHTTPStream.swift

@@ -38,4 +38,6 @@ extension MockEnvoyHTTPStream: EnvoyHTTPStream {
   func cancel() -> Int32 {
     return 0
   }
+
+  func cleanUp() {}
 }