Skip to content

AHC holds onto its body stream writer even after the request is complete #663

@weissi

Description

@weissi

AHC seems to hold onto its body streaming delegate closure even after the request is complete. That means it's pretty much impossible to use it for bidi streaming without leaking it.

test case

    func testStuff() throws {
        let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
        defer {
            XCTAssertNoThrow(try group.syncShutdownGracefully())
        }

        let server = try ServerBootstrap(group: group)
            .childChannelInitializer { channel in
                channel.eventLoop.scheduleTask(in: .milliseconds(500)) {
                    channel.writeAndFlush(ByteBuffer(string: "HTTP/1.1 200 ok\r\ncontent-length: 2\r\n\r\nok")).map {
                        channel.close(promise: nil)
                    }.whenComplete { result in
                        print("server write complete", result)
                    }
                }
                return channel.eventLoop.makeSucceededFuture(())
            }
            .bind(host: "127.1", port: 0).wait()
        defer {
            XCTAssertNoThrow(try server.close().wait())
        }
        var request = try HTTPClient.Request(url: "http://\(server.localAddress!.ipAddress!):\(server.localAddress!.port!)/",
                                             method: .POST)
        let writerPromise = self.group.any().makePromise(of: HTTPClient.Body.StreamWriter.self)
        let donePromise = self.group.any().makePromise(of: Void.self)
        class LeakDetector {
            init() {
                print("INIT")
            }

            deinit {
                print("DEINIT")
            }
        }
        var leakDetector: LeakDetector? = LeakDetector()
        request.body = .stream { [leakDetector] writer in
            _ = leakDetector
            writerPromise.succeed(writer)
            return donePromise.futureResult
        }
        weak var weakLeak: LeakDetector? = leakDetector
        leakDetector = nil

        let resultFuture = self.httpClient.execute(request: request)
        request.body = nil // this is pointless but whatever
        writerPromise.futureResult.whenSuccess { writer in
            writer.write(.byteBuffer(ByteBuffer(string: "hello"))).map {
                print("written")
            }.cascade(to: donePromise)
        }
        print("HTTP sent")

        let result = try resultFuture.wait()

        XCTAssertEqual(.ok, result.status)
        XCTAssertEqual("ok", result.body.map { String(buffer: $0) } ?? "none")
        print("HTTP done", result)
        while weakLeak != nil {
            print("not nil")
            sleep(1)
        }
    }

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions