René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Pipe to multiple writable streams doesn't throttle and may blow the memory (v4)","articleBody":"\u003c!--\nThanks for wanting to report an issue you've found in Node.js. Please fill in\nthe template below by replacing the html comments with an appropriate answer.\nIf unsure about something, just do as best as you're able.\n\nversion: usually output of `node -v`\nplatform:  either `uname -a` output, or if Windows, version and 32 or 64-bit.\nsubsystem:  optional -- if known please specify affected core module name.\n\nIt will be much easier for us to fix the issue if a test case that reproduces\nthe problem is provided. Ideally this test case should not have any external\ndependencies. We understand that it is not always possible to reduce your code\nto a small test case, but we would appreciate to have as\nmuch data as possible.\n\nThank you!\n--\u003e\n- **Version**: v4.4.3\n- **Platform**: Darwin 15.4.0 Darwin Kernel Version 15.4.0\n- **Subsystem**: stream\n\n\u003c!-- Enter your issue details below this comment. --\u003e\n\nHi\nSee below the code to reproduce and the output.\nWith node v0.10 or \u003e=v5.11 the issue doesn't happen.\nI think this is the commit that fixed it - https://github.com/nodejs/node/pull/6023\nShould it be backported to v4 LTS?\n\nThe test case is when piping from single readable to multiple writable streams.\nOne writable stream is a realistic one that has some delay to process the data. \nThe other writable stream is used for progress reporting, so calls its callback immediately.\nWith nodejs v4 the behavior is that adding a fast writable pipe will consume the reader completely into memory which can blow the process.\n\nThanks!\nGuy\n\nThe code to reproduce:\n\n```\n'use strict';\n\nvar stream = require('stream');\n\nif (require.main === module) {\n    main();\n}\n\nfunction main() {\n\n    console.log('');\n    console.log('===========');\n    console.log('NO PROGRESS');\n    console.log('===========');\n    test('none', function() {\n\n        console.log('');\n        console.log('=======================');\n        console.log('PROGRESS WITH TRANSFORM');\n        console.log('=======================');\n        test('transform', function() {\n\n            console.log('');\n            console.log('======================');\n            console.log('PROGRESS WITH WRITABLE');\n            console.log('======================');\n            test('writable', function() {});\n        });\n    });\n}\n\nfunction test(progress_mode, callback) {\n    var nr = 0;\n    var nw = 0;\n    var HIDDEN_KEY = 'tralala##';\n    var CHUNK_SIZE = 16 * 1024;\n\n    var input = new stream.Readable({\n        highWaterMark: CHUNK_SIZE\n    });\n    input._read = function() {\n        if (nr \u003e= 20) {\n            console.log('Read: done');\n            this.push(null);\n            return;\n        }\n        // set a special property on the buffer so writer can verify if copied\n        var buf = new Buffer(CHUNK_SIZE);\n        buf[HIDDEN_KEY + nr] = buf;\n        buf.fill(nr % 256);\n        this.push(buf);\n        nr += 1;\n    };\n\n    var output = new stream.Writable();\n    output._write = function(data, encoding, callback) {\n        // check that the buffer has our special property set by the reader\n        if (data[HIDDEN_KEY + nw] !== data) {\n            console.error('DATA GOT COPIED... AAAAAAAAAAAAHHHHH !!!', nw);\n        }\n        // check how much readhead occured\n        var readahead = nr - nw;\n        if (readahead \u003e 3) {\n            console.error('TOO MUCH READAHEAD', readahead);\n        } else {\n            console.log('Readahead', readahead);\n        }\n        nw += 1;\n        // slow down the writes\n        setTimeout(callback, 10);\n    };\n\n    if (progress_mode === 'transform') {\n        var progress_transform = new stream.Transform({\n            highWaterMark: 0\n        });\n        progress_transform._transform = function(data, encoding, callback) {\n            callback(null, data);\n        };\n        input.pipe(progress_transform).pipe(output);\n    } else if (progress_mode === 'writable') {\n        var progress_writable = new stream.Writable({\n            highWaterMark: 0\n        });\n        progress_writable._write = function(data, encoding, callback) {\n            callback();\n        };\n        input.pipe(progress_writable);\n        input.pipe(output);\n    } else {\n        input.pipe(output);\n    }\n\n    output.on('finish', callback);\n}\n```\n\nHere is the output:\n\n```\n$ node progress_stream.js \n\n===========\nNO PROGRESS\n===========\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nReadahead 2\nRead: done\nReadahead 1\n\n=======================\nPROGRESS WITH TRANSFORM\n=======================\nReadahead 2\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nReadahead 3\nRead: done\nReadahead 2\nReadahead 1\n\n======================\nPROGRESS WITH WRITABLE\n======================\nReadahead 2\nRead: done\nTOO MUCH READAHEAD 19\nTOO MUCH READAHEAD 18\nTOO MUCH READAHEAD 17\nTOO MUCH READAHEAD 16\nTOO MUCH READAHEAD 15\nTOO MUCH READAHEAD 14\nTOO MUCH READAHEAD 13\nTOO MUCH READAHEAD 12\nTOO MUCH READAHEAD 11\nTOO MUCH READAHEAD 10\nTOO MUCH READAHEAD 9\nTOO MUCH READAHEAD 8\nTOO MUCH READAHEAD 7\nTOO MUCH READAHEAD 6\nTOO MUCH READAHEAD 5\nTOO MUCH READAHEAD 4\nReadahead 3\nReadahead 2\nReadahead 1\n```\n","author":{"url":"https://github.com/guymguym","@type":"Person","name":"guymguym"},"datePublished":"2016-04-30T20:34:52.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":9},"url":"https://github.com/6491/node/issues/6491"}