Extend the integration test harness to track FDs (#2411)

* Extend the integration test harness to track FDs

Motivation

This patch extends the NIO integration test harness to track
file descriptors, in particular to search for leaks. This
change has been validated on Linux and Darwin, and in both cases
correctly diagnoses FD leaks.

The goal is to enable us to regression test for things like

Modifications

- Add support for hooking socket and close calls.
- Wire up this support into the test harness.
- Extend the test harness to handle the logging.
- Add new regression test for #2047.

Results

We can write regression tests for FD leaks.

* Disable FD checking in most builds.

I'm doing this for speed reasons

* Always print the leaked fds number

IntegrationTests/allocation-counter-tests-framework/template/AtomicCounter/Sources/AtomicCounter/include/atomic-counter.h

@@ -29,4 +29,14 @@ void add_malloc_bytes_counter(intptr_t v);
 void reset_malloc_bytes_counter(void);
 intptr_t read_malloc_bytes_counter(void);
 
+typedef struct {
+    size_t count;
+    int *leaked;
+} LeakedFDs;
+
+void begin_tracking_fds(void);
+void track_open_fd(int fd);
+void track_closed_fd(int fd);
+LeakedFDs stop_tracking_fds(void);
+
 #endif

IntegrationTests/allocation-counter-tests-framework/template/AtomicCounter/Sources/AtomicCounter/src/atomic-counter.c

@@ -2,7 +2,7 @@
 //
 // This source file is part of the SwiftNIO open source project
 //
-// Copyright (c) 2017-2018 Apple Inc. and the SwiftNIO project authors
+// Copyright (c) 2017-2023 Apple Inc. and the SwiftNIO project authors
 // Licensed under Apache License v2.0
 //
 // See LICENSE.txt for license information
@@ -12,7 +12,13 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include <atomic-counter.h>
+#include <assert.h>
+#include <pthread.h>
 #include <stdatomic.h>
+#include <stdbool.h>
+#include <stdlib.h>
+#include <string.h>
 
 #define MAKE_COUNTER(name) /*
 */ _Atomic long g_ ## name ## _counter = ATOMIC_VAR_INIT(0); /*
@@ -34,3 +40,127 @@
 MAKE_COUNTER(free)
 MAKE_COUNTER(malloc)
 MAKE_COUNTER(malloc_bytes)
+
+// This section covers tracking leaked FDs.
+//
+// We do this by recording which FD has been set in a queue. A queue is a bad data structure here,
+// but using a better one requires writing too much code, and the performance impact here is not
+// going to be too bad.
+typedef struct {
+    size_t capacity;
+    size_t count;
+    int *allocatedFDs;
+} FDTracker;
+
+static _Bool FDTracker_search_fd(const FDTracker *tracker, int fd, size_t *foundIndex) {
+    if (tracker == NULL) { return false; }
+
+    for (size_t i = 0; i < tracker->count; i++) {
+        if (tracker->allocatedFDs[i] == fd) {
+            if (foundIndex != NULL) { 
+                *foundIndex = i;
+            }
+            return true;
+        }
+    }
+
+    return false;
+}
+
+static void FDTracker_remove_at_index(FDTracker *tracker, size_t index) {
+    assert(tracker != NULL);
+    assert(index < tracker->count);
+
+    // Shuffle everything down by 1 from index onwards.
+    const size_t lastValidTargetIndex = tracker->count - 1;
+    for (size_t i = index; i < lastValidTargetIndex; i++) {
+        tracker->allocatedFDs[i] = tracker->allocatedFDs[i + 1];
+    }
+    tracker->count--;
+}
+
+_Atomic _Bool is_tracking = ATOMIC_VAR_INIT(false);
+pthread_mutex_t tracker_lock = PTHREAD_MUTEX_INITIALIZER;
+FDTracker tracker = { 0 };
+
+void begin_tracking_fds(void) {
+    int rc = pthread_mutex_lock(&tracker_lock);
+    assert(rc == 0);
+
+    assert(tracker.capacity == 0);
+    assert(tracker.count == 0);
+    assert(tracker.allocatedFDs == NULL);
+
+    tracker.allocatedFDs = calloc(1024, sizeof(int));
+    tracker.capacity = 1024;
+
+    atomic_store_explicit(&is_tracking, true, memory_order_release);
+    rc = pthread_mutex_unlock(&tracker_lock);
+    assert(rc == 0);
+}
+
+void track_open_fd(int fd) {
+    bool should_track = atomic_load_explicit(&is_tracking, memory_order_acquire);
+    if (!should_track) { return; }
+
+    int rc = pthread_mutex_lock(&tracker_lock);
+    assert(rc == 0);
+
+    // We need to not be tracking this FD already, or there's a correctness error.
+    assert(!FDTracker_search_fd(&tracker, fd, NULL));
+
+    // We want to append to the queue.
+    if (tracker.capacity == tracker.count) {
+        // Wuh-oh, resize. We do this by doubling.
+        assert((tracker.capacity * sizeof(int)) < (SIZE_MAX / 2));
+        size_t newCapacity = tracker.capacity * 2;
+        int *new = realloc(tracker.allocatedFDs, newCapacity * sizeof(int));
+        assert(new != NULL);
+        tracker.allocatedFDs = new;
+        tracker.capacity = newCapacity;
+    }
+
+    tracker.allocatedFDs[tracker.count] = fd;
+    tracker.count++;
+
+    rc = pthread_mutex_unlock(&tracker_lock);
+    assert(rc == 0);
+}
+
+void track_closed_fd(int fd) {
+    bool should_track = atomic_load_explicit(&is_tracking, memory_order_acquire);
+    if (!should_track) { return; }
+
+    int rc = pthread_mutex_lock(&tracker_lock);
+    assert(rc == 0);
+
+    size_t index;
+    if (FDTracker_search_fd(&tracker, fd, &index)) {
+        // We're tracking this FD, let's remove it.
+        FDTracker_remove_at_index(&tracker, index);
+    }
+
+    rc = pthread_mutex_unlock(&tracker_lock);
+    assert(rc == 0);
+}
+
+LeakedFDs stop_tracking_fds(void) {
+    int rc = pthread_mutex_lock(&tracker_lock);
+    assert(rc == 0);
+
+    LeakedFDs result = {
+        .count = tracker.count,
+        .leaked = tracker.allocatedFDs
+    };
+
+    // Clear the tracker.
+    tracker.allocatedFDs = NULL;
+    tracker.capacity = 0;
+    tracker.count = 0;
+
+    atomic_store_explicit(&is_tracking, false, memory_order_release);
+    rc = pthread_mutex_unlock(&tracker_lock);
+    assert(rc == 0);
+
+    return result;
+}

IntegrationTests/allocation-counter-tests-framework/template/HookedFunctionsDoHook/Sources/HookedFunctions/include/hooked-functions.h

@@ -16,6 +16,8 @@
 #define HOOKED_FREE
 
 #include <stdlib.h>
+#include <sys/socket.h>
+#include <sys/types.h>
 #if __APPLE__
 #  include <malloc/malloc.h>
 #endif
@@ -27,6 +29,10 @@ void *replacement_realloc(void *ptr, size_t size);
 void *replacement_reallocf(void *ptr, size_t size);
 void *replacement_valloc(size_t size);
 int replacement_posix_memalign(void **memptr, size_t alignment, size_t size);
+int replacement_socket(int domain, int type, int protocol);
+int replacement_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len);
+int replacement_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
+int replacement_close(int fildes);
 
 #if __APPLE__
 void *replacement_malloc_zone_malloc(malloc_zone_t *zone, size_t size);

IntegrationTests/allocation-counter-tests-framework/template/HookedFunctionsDoHook/Sources/HookedFunctions/src/hooked-functions-darwin.c

@@ -12,6 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include <assert.h>
 #if __APPLE__
 
 #define _GNU_SOURCE
@@ -23,6 +24,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/mman.h>
+#include <sys/socket.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ -148,6 +150,37 @@ int replacement_posix_memalign(void **memptr, size_t alignment, size_t size) {
     JUMP_INTO_LIBC_FUN(posix_memalign, memptr, alignment, size);
 }
 
+int replacement_socket(int domain, int type, int protocol) {
+    int fd = socket(domain, type, protocol);
+    if (fd < 0) {
+        return fd;
+    }
+
+    track_open_fd(fd);
+    return fd;
+}
+
+int replacement_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+    int fd = accept(socket, address, address_len);
+
+    if (fd < 0) {
+        return fd;
+    }
+
+    track_open_fd(fd);
+    return fd;
+}
+
+int replacement_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
+    // Should never be called.
+    assert(false);
+}
+
+int replacement_close(int fildes) {
+    track_closed_fd(fildes);
+    JUMP_INTO_LIBC_FUN(close, fildes);
+}
+
 DYLD_INTERPOSE(replacement_free, free)
 DYLD_INTERPOSE(replacement_malloc, malloc)
 DYLD_INTERPOSE(replacement_realloc, realloc)
@@ -161,4 +194,7 @@ DYLD_INTERPOSE(replacement_malloc_zone_valloc, malloc_zone_valloc)
 DYLD_INTERPOSE(replacement_malloc_zone_realloc, malloc_zone_realloc)
 DYLD_INTERPOSE(replacement_malloc_zone_memalign, malloc_zone_memalign)
 DYLD_INTERPOSE(replacement_malloc_zone_free, malloc_zone_free)
+DYLD_INTERPOSE(replacement_socket, socket)
+DYLD_INTERPOSE(replacement_accept, accept)
+DYLD_INTERPOSE(replacement_close, close)
 #endif

IntegrationTests/allocation-counter-tests-framework/template/HookedFunctionsDoHook/Sources/HookedFunctions/src/hooked-functions-unix.c

@@ -41,16 +41,28 @@ static _Atomic ptrdiff_t g_recursive_malloc_next_free_ptr = ATOMIC_VAR_INIT(0);
 static __thread bool g_in_malloc = false;
 static __thread bool g_in_realloc = false;
 static __thread bool g_in_free = false;
+static __thread bool g_in_socket = false;
+static __thread bool g_in_accept = false;
+static __thread bool g_in_accept4 = false;
+static __thread bool g_in_close = false;
 
 /* The types of the variables holding the libc function pointers. */
 typedef void *(*type_libc_malloc)(size_t);
 typedef void *(*type_libc_realloc)(void *, size_t);
 typedef void  (*type_libc_free)(void *);
+typedef int   (*type_libc_socket)(int, int, int);
+typedef int   (*type_libc_accept)(int, struct sockaddr*, socklen_t *);
+typedef int   (*type_libc_accept4)(int, struct sockaddr *, socklen_t *, int);
+typedef int   (*type_libc_close)(int);
 
 /* The (atomic) globals holding the pointer to the original libc implementation. */
 _Atomic type_libc_malloc g_libc_malloc;
 _Atomic type_libc_realloc g_libc_realloc;
 _Atomic type_libc_free g_libc_free;
+_Atomic type_libc_socket g_libc_socket;
+_Atomic type_libc_accept g_libc_accept;
+_Atomic type_libc_accept4 g_libc_accept4;
+_Atomic type_libc_close g_libc_close;
 
 // this is called if malloc is called whilst trying to resolve libc's realloc.
 // we just vend out pointers to a large block in the BSS (which we never free).
@@ -93,6 +105,30 @@ static void recursive_free(void *ptr) {
     abort();
 }
 
+// this is called if socket is called whilst trying to resolve libc's socket.
+static int recursive_socket(int domain, int type, int protocol) {
+    // not possible
+    abort();
+}
+
+// this is called if accept is called whilst trying to resolve libc's accept.
+static int recursive_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+    // not possible
+    abort();
+}
+
+// this is called if accept4 is called whilst trying to resolve libc's accept4.
+static int recursive_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
+    // not possible
+    abort();
+}
+
+// this is called if close is called whilst trying to resolve libc's close.
+static int recursive_close(int fildes) {
+    // not possible
+    abort();
+}
+
 /* On Apple platforms getting to the original libc function from a hooked
  * function is easy.  On other UNIX systems this is slightly harder because we
  * have to look up the function with the dynamic linker.  Because that isn't
@@ -192,4 +228,53 @@ int replacement_posix_memalign(void **memptr, size_t alignment, size_t size) {
     }
 }
 
+static int socket_thunk(int domain, int type, int protocol) {
+    JUMP_INTO_LIBC_FUN(socket, domain, type, protocol);
+}
+
+int replacement_socket(int domain, int type, int protocol) {
+    int fd = socket_thunk(domain, type, protocol);
+    if (fd < 0) {
+        return fd;
+    }
+
+    track_open_fd(fd);
+    return fd;
+}
+
+static int accept_thunk(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+    JUMP_INTO_LIBC_FUN(accept, socket, address, address_len);
+}
+
+int replacement_accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+    int fd = accept_thunk(socket, address, address_len);
+
+    if (fd < 0) {
+        return fd;
+    }
+
+    track_open_fd(fd);
+    return fd;
+}
+
+static int accept4_thunk(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
+    JUMP_INTO_LIBC_FUN(accept4, sockfd, addr, addrlen, flags);
+}
+
+int replacement_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
+    int fd = accept4_thunk(sockfd, addr, addrlen, flags);
+
+    if (fd < 0) {
+        return fd;
+    }
+
+    track_open_fd(fd);
+    return fd;
+}
+
+int replacement_close(int fildes) {
+    track_closed_fd(fildes);
+    JUMP_INTO_LIBC_FUN(close, fildes);
+}
+
 #endif

IntegrationTests/allocation-counter-tests-framework/template/Sources/bootstrapDoHook/main.c

@@ -41,6 +41,18 @@ void *valloc(size_t size) {
 int posix_memalign(void **memptr, size_t alignment, size_t size) {
     return replacement_posix_memalign(memptr, alignment, size);
 }
+int socket(int domain, int type, int protocol) {
+    return replacement_socket(domain, type, protocol);
+}
+int accept(int socket, struct sockaddr *restrict address, socklen_t *restrict address_len) {
+    return replacement_accept(socket, address, address_len);
+}
+int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
+    return replacement_accept4(sockfd, addr, addrlen, flags);
+}
+int close(int fildes) {
+    return replacement_close(fildes);
+}
 #endif
 
 void swift_main(void);

IntegrationTests/allocation-counter-tests-framework/template/scaffolding.swift

@@ -47,11 +47,47 @@ func waitForThreadsToQuiesce(shouldReachZero: Bool) {
     } while true
 }
 
-func measureAll(_ fn: () -> Int) -> [[String: Int]] {
+struct Measurement {
-    func measureOne(throwAway: Bool = false, _ fn: () -> Int) -> [String: Int]? {
+    var totalAllocations: Int
+    var totalAllocatedBytes: Int
+    var remainingAllocations: Int
+    var leakedFDs: [CInt]
+}
+
+extension Array where Element == Measurement {
+    private func printIntegerMetric(_ keyPath: KeyPath<Measurement, Int>, description desc: String, metricName k: String) {
+        let vs = self.map { $0[keyPath: keyPath] }
+        print("\(desc).\(k): \(vs.min() ?? -1)")
+    }
+
+    func printTotalAllocations(description: String) {
+        self.printIntegerMetric(\.totalAllocations, description: description, metricName: "total_allocations")
+    }
+
+    func printTotalAllocatedBytes(description: String) {
+        self.printIntegerMetric(\.totalAllocatedBytes, description: description, metricName: "total_allocated_bytes")
+    }
+
+    func printRemainingAllocations(description: String) {
+        self.printIntegerMetric(\.remainingAllocations, description: description, metricName: "remaining_allocations")
+    }
+
+    func printLeakedFDs(description desc: String) {
+        let vs = self.map { $0.leakedFDs }.filter { !$0.isEmpty }
+        print("\(desc).leaked_fds: \(vs.first.map { $0.count } ?? 0)")
+    }
+}
+
+func measureAll(trackFDs: Bool, _ fn: () -> Int) -> [Measurement] {
+    func measureOne(throwAway: Bool = false, trackFDs: Bool, _ fn: () -> Int) -> Measurement? {
         AtomicCounter.reset_free_counter()
         AtomicCounter.reset_malloc_counter()
         AtomicCounter.reset_malloc_bytes_counter()
+
+        if trackFDs {
+            AtomicCounter.begin_tracking_fds()
+        }
+
 #if os(macOS) || os(iOS) || os(watchOS) || os(tvOS)
         autoreleasepool {
             _ = fn()
@@ -63,46 +99,56 @@ func measureAll(_ fn: () -> Int) -> [[String: Int]] {
         let frees = AtomicCounter.read_free_counter()
         let mallocs = AtomicCounter.read_malloc_counter()
         let mallocedBytes = AtomicCounter.read_malloc_bytes_counter()
+        var leakedFDs: [CInt] = []
+        if trackFDs {
+            let leaks = AtomicCounter.stop_tracking_fds()
+            defer {
+                free(leaks.leaked)
+            }
+            leakedFDs = Array(UnsafeBufferPointer(start: leaks.leaked, count: leaks.count))
+        }
         if mallocs - frees < 0 {
             print("WARNING: negative remaining allocation count, skipping.")
             return nil
         }
-        return [
+        return Measurement(
-            "total_allocations": mallocs,
+            totalAllocations: mallocs,
-            "total_allocated_bytes": mallocedBytes,
+            totalAllocatedBytes: mallocedBytes,
-            "remaining_allocations": mallocs - frees
+            remainingAllocations: mallocs - frees,
-        ]
+            leakedFDs: leakedFDs
+        )
     }
 
-    _ = measureOne(throwAway: true, fn) /* pre-heat and throw away */
+    _ = measureOne(throwAway: true, trackFDs: trackFDs, fn) /* pre-heat and throw away */
 
-    var measurements: [[String: Int]] = []
+    var measurements: [Measurement] = []
     for _ in 0..<10 {
-        if let results = measureOne(fn) {
+        if let results = measureOne(trackFDs: trackFDs, fn) {
             measurements.append(results)
         }
     }
     return measurements
 }
 
-func measureAndPrint(desc: String, fn: () -> Int) -> Void {
+func measureAndPrint(desc: String, trackFDs: Bool, fn: () -> Int) -> Void {
-    let measurements = measureAll(fn)
+    let measurements = measureAll(trackFDs: trackFDs, fn)
-    for k in measurements[0].keys {
+    measurements.printTotalAllocations(description: desc)
-        let vs = measurements.map { $0[k]! }
+    measurements.printRemainingAllocations(description: desc)
-        print("\(desc).\(k): \(vs.min() ?? -1)")
+    measurements.printTotalAllocatedBytes(description: desc)
-    }
+    measurements.printLeakedFDs(description: desc)
+    
     print("DEBUG: \(measurements)")
 }
 
-public func measure(identifier: String, _ body: () -> Int) {
+public func measure(identifier: String, trackFDs: Bool = false, _ body: () -> Int) {
-    measureAndPrint(desc: identifier) {
+    measureAndPrint(desc: identifier, trackFDs: trackFDs) {
         return body()
     }
 }
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-func measureAll(_ fn: @escaping () async -> Int) -> [[String: Int]] {
+func measureAll(trackFDs: Bool, _ fn: @escaping () async -> Int) -> [Measurement] {
-    func measureOne(throwAway: Bool = false, _ fn: @escaping () async -> Int) -> [String: Int]? {
+    func measureOne(throwAway: Bool = false, trackFDs: Bool, _ fn: @escaping () async -> Int) -> Measurement? {
         func run(_ fn: @escaping () async -> Int) {
             let group = DispatchGroup()
             group.enter()
@@ -112,9 +158,15 @@ func measureAll(_ fn: @escaping () async -> Int) -> [[String: Int]] {
             }
             group.wait()
         }
+        
+        if trackFDs {
+            AtomicCounter.begin_tracking_fds()
+        }
+
         AtomicCounter.reset_free_counter()
         AtomicCounter.reset_malloc_counter()
         AtomicCounter.reset_malloc_bytes_counter()
+
 #if os(macOS) || os(iOS) || os(watchOS) || os(tvOS)
         autoreleasepool {
             run(fn)
@@ -126,22 +178,31 @@ func measureAll(_ fn: @escaping () async -> Int) -> [[String: Int]] {
         let frees = AtomicCounter.read_free_counter()
         let mallocs = AtomicCounter.read_malloc_counter()
         let mallocedBytes = AtomicCounter.read_malloc_bytes_counter()
+        var leakedFDs: [CInt] = []
+        if trackFDs {
+            let leaks = AtomicCounter.stop_tracking_fds()
+            defer {
+                free(leaks.leaked)
+            }
+            leakedFDs = Array(UnsafeBufferPointer(start: leaks.leaked, count: leaks.count))
+        }
         if mallocs - frees < 0 {
             print("WARNING: negative remaining allocation count, skipping.")
             return nil
         }
-        return [
+        return Measurement(
-            "total_allocations": mallocs,
+            totalAllocations: mallocs,
-            "total_allocated_bytes": mallocedBytes,
+            totalAllocatedBytes: mallocedBytes,
-            "remaining_allocations": mallocs - frees
+            remainingAllocations: mallocs - frees,
-        ]
+            leakedFDs: leakedFDs
+        )
     }
 
-    _ = measureOne(throwAway: true, fn) /* pre-heat and throw away */
+    _ = measureOne(throwAway: true, trackFDs: trackFDs, fn) /* pre-heat and throw away */
 
-    var measurements: [[String: Int]] = []
+    var measurements: [Measurement] = []
     for _ in 0..<10 {
-        if let results = measureOne(fn) {
+        if let results = measureOne(trackFDs: trackFDs, fn) {
             measurements.append(results)
         }
     }
@@ -149,16 +210,15 @@ func measureAll(_ fn: @escaping () async -> Int) -> [[String: Int]] {
 }
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-func measureAndPrint(desc: String, fn: @escaping () async -> Int) -> Void {
+func measureAndPrint(desc: String, trackFDs: Bool, fn: @escaping () async -> Int) -> Void {
-    let measurements = measureAll(fn)
+    let measurements = measureAll(trackFDs: trackFDs, fn)
-    for k in measurements[0].keys {
+    measurements.printTotalAllocations(description: desc)
-        let vs = measurements.map { $0[k]! }
+    measurements.printRemainingAllocations(description: desc)
-        print("\(desc).\(k): \(vs.min() ?? -1)")
+    measurements.printTotalAllocatedBytes(description: desc)
-    }
+    measurements.printLeakedFDs(description: desc)
-    print("DEBUG: \(measurements)")
 }
 
 @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
-public func measure(identifier: String, _ body: @escaping () async -> Int) {
+public func measure(identifier: String, trackFDs: Bool = false, _ body: @escaping () async -> Int) {
-    measureAndPrint(desc: identifier, fn: body)
+    measureAndPrint(desc: identifier, trackFDs: trackFDs, fn: body)
 }

IntegrationTests/tests_04_performance/test_01_allocation_counts.sh

@@ -35,13 +35,16 @@ for test in "${all_tests[@]}"; do
         test_case=${test_case#test_*}
         total_allocations=$(grep "^test_$test_case.total_allocations:" "$tmp/output" | cut -d: -f2 | sed 's/ //g')
         not_freed_allocations=$(grep "^test_$test_case.remaining_allocations:" "$tmp/output" | cut -d: -f2 | sed 's/ //g')
+        leaked_fds=$(grep "^test_$test_case.leaked_fds:" "$tmp/output" | cut -d: -f2 | sed 's/ //g')
         max_allowed_env_name="MAX_ALLOCS_ALLOWED_$test_case"
 
         info "$test_case: allocations not freed: $not_freed_allocations"
         info "$test_case: total number of mallocs: $total_allocations"
+        info "$test_case: leaked fds: $leaked_fds"
 
         assert_less_than "$not_freed_allocations" 5     # allow some slack
         assert_greater_than "$not_freed_allocations" -5 # allow some slack
+        assert_less_than "$leaked_fds" 1  # No slack allowed here though
         if [[ -z "${!max_allowed_env_name+x}" ]]; then
             if [[ -z "${!max_allowed_env_name+x}" ]]; then
                 warn "no reference number of allocations set (set to \$$max_allowed_env_name)"

IntegrationTests/tests_04_performance/test_01_resources/test_1000_rst_connections.swift

@@ -0,0 +1,44 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2020-2021 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import NIOPosix
+
+func run(identifier: String) {
+    let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
+    defer {
+        try! group.syncShutdownGracefully()
+    }
+
+    let serverConnection = try! ServerBootstrap(group: group)
+        .bind(host: "localhost", port: 0)
+        .wait()
+    
+    let serverAddress = serverConnection.localAddress!
+    let clientBootstrap = ClientBootstrap(group: group)
+
+    measure(identifier: identifier, trackFDs: true) {
+        let iterations = 1000
+        for _ in 0..<iterations {
+            let conn = clientBootstrap.connect(to: serverAddress)
+            
+            let _: Void? = try? conn.flatMap { channel in
+                (channel as! SocketOptionProvider).setSoLinger(linger(l_onoff: 1, l_linger: 0)).flatMap {
+                    channel.close()
+                }
+            }.wait()
+        }
+        return iterations
+    }
+}