After stress testing, the RSS memory usage does not decrease.

[yufeifan]

nodejs version:22
platform: Ubuntu 24.04.3 LTS

import { Http3Server} from "@fails-components/webtransport";
import { WebTransportSessionImpl } from "@fails-components/webtransport/dist/lib/types";
import { WebTransportSocket } from './webtransportSocket';
import ... others


export interface WebtransportConnectorOptions extends HybridSwitcherOptions {
    useDict?: boolean;
    useProtobuf?: boolean;

    cert: string
    privateKey: string
}

let curId = 1;

export class WebTransportConnector extends EventEmitter implements IConnector {
    opts: WebtransportConnectorOptions;
    port: number;
    host: string;
    useDict: boolean;
    useProtobuf: boolean;
    handshake: HandshakeCommand;
    heartbeat: HeartbeatCommand;
    cert: string;
    privateKey: string;

    connector: IConnector;
    dictionary: DictionaryComponent;
    protobuf: ProtobufComponent;
    decodeIO_protobuf: IComponent;

    listeningServer: Http3Server;
    // 持久流管理

    constructor(port: number, host: string, opts: WebtransportConnectorOptions) {
        super();
        this.opts = opts;
        this.port = port;
        this.host = host;
        this.useDict = opts.useDict;
        this.useProtobuf = opts.useProtobuf;
        this.handshake = new HandshakeCommand(opts);
        this.heartbeat = new HeartbeatCommand(opts);
        this.cert = opts.cert;
        this.privateKey = opts.privateKey;
    }

    /**
     * Start connector to listen the specified port
     */
    start(cb: () => void) {
        let app = pinus.app
        let self = this;

        let gensocket = function (socketId: number, session: WebTransportSessionImpl, clientInfo: { ip: string, port: string }) {
            let wTsocket = new WebTransportSocket(socketId, session, clientInfo);
            wTsocket.on('handshake', self.handshake.handle.bind(self.handshake, wTsocket));
            wTsocket.on('heartbeat', self.heartbeat.handle.bind(self.heartbeat, wTsocket));
            wTsocket.on('disconnect', self.heartbeat.clear.bind(self.heartbeat, wTsocket.id));
            wTsocket.on('closing', Kick.handle.bind(null, wTsocket));
            self.emit('connection', wTsocket);
            return wTsocket
        };

        this.connector = app.components.__connector__.connector;
        this.dictionary = app.components.__dictionary__;
        this.protobuf = app.components.__protobuf__;
        this.decodeIO_protobuf = app.components.__decodeIO__protobuf__;

        let server = new Http3Server({
            port: this.port,
            host: '0.0.0.0',
            secret: 'mysecret',
            cert: this.cert,
            privKey: this.privateKey,
            reliability: 'reliableOnly',
            defaultDatagramsReadableMode: 'bytes',
        });
        this.listeningServer = server;

        server.ready.then(async () => {
            server.setRequestCallback(async (args: any) => {
                const url = args.header[':path']
                const [path] = url.split('?')

                if (server.sessionController[path] == null) {
                    return {
                        ...args,
                        path,
                        status: 404
                    }
                }
                const protocols = args.header['wt-available-protocols']
                    ? args.header['wt-available-protocols']
                        .split(',')
                        .map((el: string) => el.trim())
                    : undefined
                let selectedProtocol = protocols && protocols[protocols.length - 1]
                if (selectedProtocol === 'noprot') selectedProtocol = undefined
                return {
                    ...args,
                    path,
                    userData: {
                        search: url.substring(path.length)
                    },
                    header: {
                        ...args.header,
                        ':path': path
                    },
                    status: 200,
                    selectedProtocol
                }
            })

            for await (const session of getReaderStream(server.sessionStream('/bidirectional'))) {
                try {
                    await session.ready
                    let socketId = curId++
                    //@ts-ignore
                    const clientInfo = parseClientAddress(session.peerAddress)
                    let wTsocket = gensocket(socketId, session, clientInfo)
                    logger.info('WebTransportConnector session opened:', socketId)

                    wTsocket.establishPersistentStream()
                    session.closed.then((reason) => {
                        logger.info('WebTransportConnector session closed:', reason)
                        wTsocket.emit('disconnect', JSON.stringify(reason))
                        // 清理持久流
                        wTsocket.closePersistentStream()
                        // 释放所有事件监听，避免引用链存活
                        wTsocket.removeAllListeners()
                        wTsocket = null;
                    })
                } catch (err) {
                    logger.error('WebTransportConnector session error:', err);
                    session.close({
                        closeCode: 1002,
                        reason: 'session error'
                    })
                }

            }
        }).catch((err) => {
            logger.error('WebTransportConnector server error:', err);
        })

        server.startServer()

        process.nextTick(cb);
    }

    stop(force: boolean, cb: () => void) {
        this.listeningServer.stopServer()
        process.nextTick(cb);
    }
    decode = coder.decode;

    encode = coder.encode;
}

function parseClientAddress(address: string | undefined): { ip: string, port: string } {
    if (!address) {
        return { ip: '0', port: '0' }
    }

    // 检查是否包含端口号 (格式: IP:PORT)
    const colonIndex = address.lastIndexOf(':')
    if (colonIndex !== -1) {
        const ip = address.substring(0, colonIndex)
        const port = address.substring(colonIndex + 1)
        return { ip, port }
    }

    // 只有IP地址
    return { ip: address, port: '0' }
}

webtransportSocket.ts

import * as path from 'path';
import { EventEmitter } from "stream";
import { ISocket } from "../interfaces/ISocket";
import { WebTransportBidirectionalStream, WebTransportSession } from "@fails-components/webtransport";
import { getLogger } from 'pinus-logger';
import { Message, Package } from 'pinus-protocol';
import { WebTransportSessionImpl } from '@fails-components/webtransport/dist/lib/types';
import handler from './common/handler';
let logger = getLogger('pinus', path.basename(__filename));

let ST_INITED = 0;
let ST_WAIT_ACK = 1;
let ST_WORKING = 2;
let ST_CLOSED = 3;

export class WebTransportSocket extends EventEmitter implements ISocket {
    id: number;
    remoteAddress: { ip: string, port: number };
    state: number;
    session: WebTransportSessionImpl;


    private streamWriter: WritableStreamDefaultWriter<Uint8Array> | null = null
    private streamReader: ReadableStreamDefaultReader<Uint8Array> | null = null

    private _datagramReader: ReadableStreamDefaultReader<Uint8Array> | null = null;
    private _datagramWriter: WritableStreamDefaultWriter<Uint8Array> | null = null;


    private messageQueue: Buffer[] = [];
    private isSending: boolean = false;

    constructor(id: number, session: WebTransportSessionImpl, clientInfo: { ip: string, port: string }) {
        super();
        this.id = id;
        this.session = session;
        this.state = ST_INITED
        this.remoteAddress = { ip: clientInfo.ip, port: parseInt(clientInfo.port) }
    }

    async establishPersistentStream() {

        try {
            // this.initDatagramStream();
            this.initBidirectionalStream();

        } catch (error) {
            logger.error('Failed to establish persistent stream:', error);
        }
    }

    async initBidirectionalStream() {
        const streamIterator = this.session.incomingBidirectionalStreams[Symbol.asyncIterator]();
        const firstStream = await streamIterator.next();
        if (!firstStream.done) {
            const persistentStream: WebTransportBidirectionalStream = firstStream.value;
            this.streamWriter = persistentStream.writable.getWriter();
            this.streamWriter.closed.then(() => {
                logger.info('Persistent writer stream closed gracefully:', this.id);
            }).catch((error) => {
                logger.error('Persistent writer stream closed abruptly:', error);
            });
            logger.info('Persistent stream set for socket:', this.id);


            this.streamReader = persistentStream.readable.getReader();
            this.streamReader.closed.then(() => {
                logger.info('Persistent reader stream closed gracefully:', this.id);
            }).catch((error) => {
                logger.error('Persistent reader stream closed abruptly:', error);
            });
            this.handlePersistentStreamReading(this.streamReader);
        }

    }

    async initDatagramStream() {
        this._datagramReader = this.session.datagrams.readable.getReader();
        this._datagramReader.closed.catch((error) => {
            logger.error('Datagram stream closed:', error);
        });

        this._datagramWriter = this.session.datagrams.createWritable().getWriter();
        this._datagramWriter.closed.catch((error) => {
            logger.error('Datagram stream closed:', error);
        });

        this.handlePersistentStreamReading(this._datagramReader);
    }

    async handlePersistentStreamReading(reader: ReadableStreamDefaultReader<Uint8Array>) {
        let valueArr: Uint8Array[] = []

        try {
            while (this.session.state !== 'closed' && this.session.state !== 'failed') {
                const { done, value } = await reader.read();
                if (done) {
                    logger.info('Persistent stream ended for socket:', this.id);
                    break;
                }

                if (value[value.length - 1] != 0) {
                    valueArr.push(value)
                    continue
                } else {
                    valueArr.push(value.subarray(0, value.length - 1))
                }
                let dataUint8Array = concatUint8Array(valueArr)
                valueArr = []
                if (value) {
                    try {
                        const msg = Package.decode(Buffer.from(dataUint8Array));
                        this.onPersistentMessage(msg);
                    } catch (decodeError) {
                        logger.error('Failed to decode message:', decodeError);
                    }
                }
            }
            logger.info('Persistent stream reading ended for socket:', this.id, " state:", this.session.state);
        } catch (error) {
            logger.error('Persistent stream reading error:', error);
        } finally {
            try {
                reader.releaseLock();
            } catch (releaseError) {
                logger.warn('Error releasing reader lock:', releaseError);
            }
        }
    }



    onPersistentMessage(pkg: ReturnType<typeof Package.decode>) {
        if (pkg instanceof Array) {
            handler(this, pkg, true);
            return;
        }
        handler(this, pkg, true);
    }


    private enqueueMessage(msg: Buffer) {
        this.messageQueue.push(msg);
        logger.debug(`Message enqueued, queue length: ${this.messageQueue.length}`);

        this.processQueue();
    }

    private dequeueMessage(): Buffer | undefined {
        return this.messageQueue.shift();
    }


    private async processQueue() {
        if (this.session.state === 'closed') {
            return;
        }


        if (this.isSending) {
            return;
        }


        if (this.messageQueue.length === 0) {
            return;
        }

        this.isSending = true;
        try {
            //@ts-ignore
            while (this.messageQueue.length > 0 && this.session.state !== 'closed') {
                const msg = this.dequeueMessage();
                if (!msg) {
                    break;
                }
                await this.sendMessageToStream(msg);
                logger.debug('Message sent from queue');

                await new Promise(resolve => setTimeout(resolve, 100));
            }
        } catch (error) {
            logger.error('Error sending message from queue:', error);
            this.disconnect();
        } finally {
            this.isSending = false;
            //@ts-ignore
            if (this.messageQueue.length > 0 && this.session.state !== 'closed') {
                Promise.resolve().then(() => this.processQueue());
            }
        }
    }


    private async sendMessageToStream(msg: Buffer): Promise<void> {
        let writer = this.streamWriter || this._datagramWriter;
        if (!writer) {
            throw new Error('No persistent stream writer available');
        }
        // logger.info('writer.desiredSize:', writer.desiredSize);
        await writer.ready;
        await writer.write(msg);
    }

    async sendRaw(msg: any) {
        if (this.session.state == "closed") {
            return;
        }

        let writer = this.streamWriter || this._datagramWriter;
        if (!writer) {
            logger.error('No persistent stream writer available');
            return;
        }

        const encodedMsg = encode(msg);
        this.enqueueMessage(encodedMsg);

        logger.debug('Message queued for sending');
    }

    async closePersistentStream() {
        this.messageQueue = [];
        this.isSending = false;

        if (this.session.state == "closed") {
            logger.info("WebTransportSocket session already closed for closePersistentStream:", this.id);
            return;
        }

        if (this.streamReader) {
            try {
                await this.streamReader.cancel();
            } catch (error) {
                logger.error('Error canceling persistent stream reader:', error);
            }
            this.streamReader = null;
        }
        if (this._datagramReader) {
            try {
                await this._datagramReader.cancel();
            } catch (error) {
                logger.error('Error canceling datagram reader:', error);
            }
            this._datagramReader = null;
        }

        if (this.streamWriter) {
            try {
                logger.info('Persistent stream closed for socket:', this.id);
                await this.streamWriter.close();
            } catch (error) {
                logger.error('Error closing persistent stream:', error);
            }
            this.streamWriter = null;
        }
        if (this._datagramWriter) {
            try {
                logger.info('Datagram stream closed for socket:', this.id);
                await this._datagramWriter.close();
            } catch (error) {
                logger.error('Error closing datagram stream:', error);
            }
            this._datagramWriter = null;
        }
    }

    send(msg: any) {
        if (msg instanceof String) {
            msg = Buffer.from(msg as string);
        } else if (!(msg instanceof Buffer)) {
            msg = Buffer.from(JSON.stringify(msg));
        }
        this.sendRaw(Package.encode(Package.TYPE_DATA, msg));
    }

    sendBatch(msgs: any[]) {
        let rs: Buffer[] = [];
        for (let i = 0; i < msgs.length; i++) {
            let src = Package.encode(Package.TYPE_DATA, msgs[i]);
            rs.push(src);
        }
        this.sendRaw(rs);
    }

    sendForce(msg: any) {
        this.sendRaw(msg);
    }


    /**
     * Response handshake request
     *
     * @api private
     */
    handshakeResponse(resp: any) {
        this.sendRaw(resp);
        this.state = ST_WAIT_ACK;
    }


    disconnect() {
        logger.info("WebTransportSocket disconnect:", this.id);
        this.state = ST_CLOSED;

        this.closePersistentStream();

        if (this.session.state == "closed") {
            logger.info("WebTransportSocket session already closed:", this.id);
            return;
        }

        this.session.close({
            closeCode: 1002,
            reason: 'disconnect'
        })
    }

}

function encode(msg: Buffer) {
    //add asscii 0 to last one
    return Buffer.concat([msg, Buffer.from([0])])
}

function concatUint8Array(arr: Uint8Array[]) {
    if (arr.length === 0) {
        return new Uint8Array(0);
    }
    if (arr.length === 1) {
        return arr[0];
    }

    let totalLength = 0;
    for (const item of arr) {
        totalLength += item.length;
    }

    const result = new Uint8Array(totalLength);
    let offset = 0;
    for (const item of arr) {
        result.set(item, offset);
        offset += item.length;
    }
    return result;
}

once stress testing is 20 minute,after tesing,heap mem can down initial level(about 45M),but rss can not decrease,
after Several 5-hour stress tests, the rss up to 3300M, one of webtransport process be killed in k8s pod.
The remaining process is shown in the following figure.

Each process handles approximately 4.1 million connections, with each connection ending and being disconnected by the client in about one minute. Each process has approximately 2,500 concurrent connections, processing 500 requests per second (QPS) simultaneously, with CPU utilization at 80%.

I change the mem alloc to libjemalloc.so.2,but no use.

[martenrichter]

So far I only tested the library for video streaming applications, so such a high number of connections did not appear. And I did not see exceptional memory usage. (But rather small user group).

Did you try to analyse first using a node debugger memory inspector, if it is a js side problem of memory build up?
(If not, one has to look if there is memory leaking on the C++ side....).

[yufeifan]

So far I only tested the library for video streaming applications, so such a high number of connections did not appear. And I did not see exceptional memory usage. (But rather small user group).

Did you try to analyse first using a node debugger memory inspector, if it is a js side problem of memory build up? (If not, one has to look if there is memory leaking on the C++ side....).

Initially, I would open a bidirectional or unidirectional stream for every communication event during a game session. Memory consumption rose rapidly. After exporting and analyzing the heap memory, I discovered it was dominated by stream objects. I then modified the approach to use a single bidirectional stream per player for all communications within a game session. This prevented heap memory growth, but the RSS memory remained unreleased.

When I implemented the same functionality using uwebsockets.js, RSS memory did not grow indefinitely.

How do I debug a memory leak in C++?

[martenrichter]

Did you close the streams?

How do I debug a memory leak in C++?

Valgrind would be a way.
But we can also do something else.
If you can provide a minimal example (no external packages) for client and server, that I can easily run. I would try it in my dev setup and see if I can debug it. (But it must be very minimal).
I will then probably add log messages to constructor and destructor of the objects.

[yufeifan]

Valgrind

Connection ending and being disconnected by the client,
Server receive session close event, and then close write stream.but 98.9% of writer close errors: "Persistent writer stream closed abruptly". Error is {“name”:“WebTransportError”}

I will try close writer before session close and the results will follow later.

            this.streamWriter.closed.then(() => {
                logger.info('Persistent writer stream closed gracefully:', this.id);
            }).catch((error) => {
                logger.error('Persistent writer stream closed abruptly:', error);
            });

[martenrichter]

Ok, this can happen, if it is running to slowly. Then it works only with counters added to the C++ code.

[yufeifan]

After the writer close gracefully,the RSS memory still cannot be reduced.
Then I write sample case,Simulate 2,500 online and 500 concurrent requests, with matches concluding within 1 minute. run 12 hours.Server RSS memory increased from 500M to over 700M. After stopping the load testing, it cannot decrease.

the case:

server:

import { Http3Server, WebTransportSession, quicheLoaded } from '@fails-components/webtransport';
import { getReaderStream } from './util/stream';
import * as fs from 'fs'
import * as path from 'path'

let receiveCnt = 0;
let parseErrCnt = 0;
let playerCnt = 0;

async function main() {
    await quicheLoaded

    const cert = fs.readFileSync(path.join(__dirname, '../cert/cert.pem'))
    const privateKey = fs.readFileSync(path.join(__dirname, '../cert/private.pem'))

    console.log('Creating server instance...')
    const server: Http3Server = new Http3Server({
        port: 8083,
        host: '0.0.0.0',
        secret: 'mysecret',
        cert: cert.toString(),
        privKey: privateKey.toString(),
        reliability: 'reliableOnly',
        defaultDatagramsReadableMode: 'bytes',
    })

    console.log('Starting server...')

    console.log('Listening for bidirectional streams...')

    server.ready.then(async () => {
        server.setRequestCallback(async (args: any) => {
            // This is a bidirectional stream
            // console.log('Received request:', args)

            const url = args.header[':path']
            const [path] = url.split('?')

            if (server.sessionController[path] == null) {
                console.log('sessionController[path] == null')
                return {
                    ...args,
                    path,
                    status: 404
                }
            }
            const protocols = args.header['wt-available-protocols']
                ? args.header['wt-available-protocols']
                    .split(',')
                    .map((el: string) => el.trim())
                : undefined
            let selectedProtocol = protocols && protocols[protocols.length - 1]
            if (selectedProtocol === 'noprot') selectedProtocol = undefined
            return {
                ...args,
                path,
                userData: {
                    search: url.substring(path.length)
                },
                header: {
                    ...args.header,
                    ':path': path
                },
                status: 200,
                selectedProtocol
            }
        })
        let sessions: AsyncGenerator<WebTransportSession, void, unknown> = getReaderStream(
            server.sessionStream('/bidirectional')
        )
        for await (const session of sessions) {
            try {
                await session.ready
                playerCnt++;

                OnMessage(session)


            } catch {
            }
        }




    })

    async function OnMessage(session: WebTransportSession) {
        let interval: NodeJS.Timeout | null = null

        try {
            session.closed.then((reason) => {
                playerCnt--;
            }).catch((err) => {
                console.log('session closed2 error', err)
            })

        } catch (err) {
            console.log('session closed3 error', err)
        }


        const streamIterator = session.incomingBidirectionalStreams[Symbol.asyncIterator]();
        const firstStream = await streamIterator.next();

        if (!firstStream.done) {
            const persistentStream = firstStream.value;
            let stream = persistentStream;
            const reader = stream.readable.getReader();
            let writer = stream.writable.getWriter();


            writer.closed.then(() => {
                // console.log('writer closed')
            }).catch((err) => {
                console.log('writer close error', err)
            })

            let dataArr: Uint8Array[] = []

            try {
                while (session.state != "closed" && session.state != "failed") {
                    const { done, value } = await reader.read();
                    if (done) {
                        break;
                    }

                    dataArr.push(value.slice(0, value.length - 1))


                    // Calculate total length
                    const totalLength = dataArr.reduce((acc, arr) => acc + arr.length, 0);

                    // Create new Uint8Array and merge all data
                    let data = new Uint8Array(totalLength);
                    let offset = 0;
                    for (const arr of dataArr) {
                        data.set(arr, offset);
                        offset += arr.length;
                    }

                    // Reset array, ready to receive next batch of data
                    dataArr = [];

                    let receivedData = _decode(data);
                    // console.log('receivedData', receivedData);

                    switch (receivedData.path) {
                        case "connector.entryHandler.entry": {
                            let resBody = {
                                isOk: true,
                                roomId: 123
                            }
                            pushDataUseBidirectionalStream(genRes(receivedData.reqId, receivedData.path, 0, resBody), false, writer)
                            break;
                        }

                        case "connector.entryHandler.joinRoom": {
                            let resBody = {
                                isOk: true,
                            }
                            pushDataUseBidirectionalStream(genRes(receivedData.reqId, receivedData.path, 0, resBody), false, writer)
                            break;
                        }

                        case "battle.battleHandler.operate": {
                            receiveCnt++;
                            let resBody = {
                                isOk: true,
                            }
                            pushDataUseBidirectionalStream(genRes(receivedData.reqId, receivedData.path, 0, resBody), false, writer)
                            break;
                        }
                        case "battle.battleHandler.endBattle": {
                            let resBody = {
                                isOk: true,
                            }
                            pushDataUseBidirectionalStream(genRes(receivedData.reqId, receivedData.path, 0, resBody), true, writer)
                            break;
                        }

                        default: {
                            break;
                        }
                    }
                }
            } catch (err) {
                if (err && err.constructor && err.constructor.name === 'WebTransportError') {
                    // console.log('Session closed, stop reading data')
                } else {
                    console.error('Error reading data:', err)
                }
            }
        }
    }


    async function pushDataUseBidirectionalStream(data: any, close: boolean = false, writer: WritableStreamDefaultWriter) {
        await writer.write(_encode(data))
        if (close) {
            writer.close()
        }
    }

    server.startServer()

    try {
        console.log('Waiting for server readiness...')
        await server.ready
        console.log('Server ready', server.address())
    } catch (err) {
        console.error('Failed to start server:', err)
        return
    }

    console.log('Server configuration complete, listening for connections...')

    function genRes(reqId: number, Path: string, errorCode: number, data: any) {
        return {
            errorCode: errorCode,
            reqId: reqId,
            Path: Path,
            data: data
        }
    }

    function _encode(data) {
        let string = JSON.stringify(data);
        let len = string.length;
        let res = new Uint8Array(len);
        for (let i = 0; i < string.length; i++) {
            res[i] = string.charCodeAt(i);
        }
        return res;
    }

    function _decode(data) {
        let string = '';
        if (!data) {
            return null;
        }
        for (let i = 0; i < data.length; i++) {
            string += String.fromCharCode(data[i]);
        }
        try {
            return JSON.parse(string);
        } catch (err) {
            console.log('decode error:', string.split('\0').length);
            parseErrCnt++;

            return string;
        }
    }
}



main().catch(err => {
    console.error('Program execution error:', err)
})

setInterval(() => {
    // console.log('Server is running and waiting for connections...');
    console.log('receiveCnt', receiveCnt, " parseErrCnt", parseErrCnt, " playerCnt", playerCnt);
    receiveCnt = 0;
    parseErrCnt = 0;
}, 1000);

client.mjs

import CCCWebTransport from './cccwebtransport.mjs'
import * as fs from 'fs'


export const MTBattlePath = {
    Entry: "connector.entryHandler.entry",
    StartMatch: "match.matchHandler.startMatch",
    CancelMatch: "match.matchHandler.cancelMatch",
    JoinRoom: "connector.entryHandler.joinRoom",
    LoadingOk: "battle.battleHandler.loadingOk",
    Operate: "battle.battleHandler.operate",
    EndBattle: "battle.battleHandler.endBattle",
}


function getHashFromFile(){
    const hash = fs.readFileSync("./cert/hash", 'utf-8')
    return hash.toString().split(':').map((i) => parseInt(i, 16))
}
let cnt = 0;
let cntret = 0;
let playerNeed = 2500;
let playerNum = 0;

export async function startOne() {
    const transport = new CCCWebTransport()
    let certhash = getHashFromFile()

    let roleId = 123331 + playerNum++;
    await transport.init('https://127.0.0.1:8083/bidirectional', certhash, async () => {
        transport.request(MTBattlePath.Entry, { roleId: roleId, token: '' }, (code, msg, data) => {
            if (code == 0) {
                const { isOk, roomId } = data;
                transport.request(MTBattlePath.JoinRoom, { roomId }, (code, msg, data) => {

                    let thisOpCnt = 0;
                    let interval = null;
                    interval = setInterval(() => {
                        thisOpCnt++;

                        if (thisOpCnt> 12){
                            transport.request(MTBattlePath.EndBattle, { opCnt: thisOpCnt }, (code, msg, data) => {
                                cntret++;
                            })

                            clearInterval(interval);
                            transport.close()
                            playerNum--;
                        }else{
                            transport.request(MTBattlePath.Operate, { opCnt: thisOpCnt }, (code, msg, data) => {
                                cntret++;
                            })
                        }
                    }, 5000)
                })
            }
        })


    }, (error) => {
        console.log('init error', error)
    })
}




setInterval(() => {
    console.log('operate cntret', cntret);
    cntret = 0;
}, 1000);


async function startPlayers() {
    let oneSecondLimit = 50;

    setInterval(() => {
        if (playerNum < playerNeed) {
            for (let i = 0; i < oneSecondLimit; i++) {
                startOne();
            }
        }    
    }, 1000)
}

// start()

startPlayers()

cccwebtransport.mjs

import { WebTransport, quicheLoaded } from '@fails-components/webtransport'


export const MTBattleNotify = {
    OnMatchSuccess: "onMatchSuccess",
    OnJoin: "onJoin",
    OnLoadingProgress: "onLoadingProgress",
    OnRoomStatusChange: "onRoomStatusChange",
    OnFrameUpdate: "onFrameUpdate",
    OnServerError: "onServerError",
}

export default class CCCWebTransport {
    static _isSupported = true;


     _transport = null;

     _reqId = 0;
     _resMap = new Map();

    constructor () {
        this._resMap = new Map()
    }
    callbacks = {}
    wtStream = null;

    writer = null;
    reader= null

    async init (url,certhash, successCb, errorCb) {
        await quicheLoaded

        try {
            this._transport = new WebTransport(url, {
                "serverCertificateHashes": [{
                    "algorithm": "sha-256",
                    "value": new Uint8Array(certhash)
                }]
            });
            await this._transport.ready;

            this.wtStream = await this._transport.createBidirectionalStream();
            
            this.writer = this.wtStream.writable.getWriter();
            this.reader = this.wtStream.readable.getReader();
            await this.writer.ready;

            this._startReading();
            successCb && successCb();
        } catch (e) {
            console.error(e);
            errorCb && errorCb();
        }        
    }

    async request (path, msg, cb) {
        
        this.callbacks[this._reqId] = cb;
        
        let sendData = { reqId: this._reqId, path, data: JSON.stringify(msg) };
        let encodedData = this._encode(sendData);
        this._resMap.set(this._reqId, cb);

        if (encodedData.length > 1000) {
            let chunks = [];
            for (let i = 0; i < encodedData.length; i += 1000) {
                chunks.push(encodedData.slice(i, i + 1000));
            }
            for (let chunk of chunks) {
                await this.writer.write(chunk);
            }
        }else{
            await this.writer.write(encodedData);
        }

        
        this._reqId = (this._reqId + 1) % 256;

    }


    async _startReading () {
        try {
            // const reader = this.wtStream.readable.getReader();
            const reader = this.reader;
            while(this._transport.state != "closed" && this._transport.state != "failed") {
                const { done, value } = await reader.read();
                if (done) {
                    break;
                }
                let receivedData = this._decode(value);
                if (!receivedData) continue;

                let { reqId, errorCode, data, path } = receivedData;
                if (path == MTBattleNotify.OnFrameUpdate) {
                    console.log('received OnFrameUpdate', receivedData.frameId, data);
                    continue;
                }

                let cb = this.callbacks[reqId];
                if (cb) {
                    cb(errorCode, "", data);
                }
                delete this.callbacks[reqId];
            }
        }catch (error) {
            console.error('WebTransport stream error:', error);
        }

    }

    _encode (data) {
        let string = JSON.stringify(data);
        let len = string.length;
        let res = new Uint8Array(len +1);
        let i = 0
        for (; i < string.length; i++) {
            res[i] = string.charCodeAt(i);
        }
        res[i] = 0
        return res;
    }

    _decode (data) {
        let string = '';
        if (!data){
            return null;
        }
        for (let i = 0; i < data.length; i++) {
            string += String.fromCharCode(data[i]);
        }
        try {
            return JSON.parse(string);
        } catch (err) {
            console.log( 'decode error:', string);
            console.log('[CCCWebTransport]', 'decode error:', err);
            return string;
        }
    }

    close (reason) {
        this._transport.close(reason)
    }

}

[martenrichter]

Thanks for providing the code.
I could reproduce, that no C++ side sessions and streams are ever freed,
Unfortunetaly my free weekend time is almost over, so it is probably a task for the next weekend.
But there is definitely an issue, I am not sure it it is in your code or mine. We will see.

[martenrichter]

Correction, I did something wrong (I waited for time outs), now I see destructor calls.

[martenrichter]

And everything is always 0 to 100 over the 2500, so the object lifetime works well, for the streaming and session objects. So it must be somewhere else. (P.S: I can't run it for hours, as debugging an hour long run would not be an option.)

[martenrichter]

TBC next weekend.
Maybe you want to check, if you are opening streams on a gone session. I may have seen such behaviour. Of course the lib should error. Just one idea.

[yufeifan]

Maybe you want to check, if you are opening streams on a gone session. I may have seen such behaviour. Of course the lib should error.

I'll check it later, probably in about two weeks

[martenrichter]

Not really a smoking gun:
#429
and please note that I have seen sessions to take a while to time out.
(I have shutdown the client manually.) Though I do not saw a memory problem after the changes.