Standardizing ".destroy()"

@mafintosh

This is a behavior that streams used to have (as of streams 1), but was never really fully specified, and was removed in streams2+. The result is that there's no standard way to prematurely end a stream from the outside, and the effects of prematurely ended streams are undefined (or surprising to our users).

@mafintosh may have more words about this!

FWIW: in WHATWG streams, we have readableStream.cancel() and writableStream.abort(). The former makes the readable stream look closed, whereas the latter makes the writable stream look errored. I can go into more detail on the reasoning if desired.

I think an overridable ._destroy(cb) would be good way to do this and then have all streams have a public .destroy that would call ._destroy once and only once. When the callback is called the stream would emit close and possibly error if an error is passed.

myStream._destroy = function (cb) {
  closeResources(function (err) {
    cb(err)
  })
}

myStream.destroy() // calls the above method
myStream.destroy() // doesn't do anything

+1! I always end up bike-shedding my own destroy method, eg. for cleaning up the underlying source of a Readable. However I've never really been sure in what state I leave the stream...

@domenic

@domenic I'd definitely be interested in why prematurely ending a writable implies an error.

@mafintosh So, to make sure I've got this correct, a few of the potential event sequences are:

`A.pipe(B).pipe(C); A.destroy()` – streams2+, default _destroy

User calls "A.destroy()".
"A" emits "close".

A is still piped to B, which is still piped to C.

`A.pipe(B).pipe(C); B.destroy()` – streams2+, default _destroy

User calls "B.destroy()".
"B" emits "close".
"A.unpipe(B)"
1. "B" emits "unpipe" A.
2. If "B" is the only destination of "A", "A" is implicitly paused.
3. Else if "B" awaitsDrain, then "A" may start flowing.

B is still piped to C. A is still available to be re-piped, and may be paused or flowing.

`A.pipe(B).pipe(C); B.destroy()` – A is streams1, B and C are streams2+, default _destroy

User calls "B.destroy()".
"B" emits "close".
"A" calls "B.destroy()"
"A" calls cleanup and removes all listeners.

B is still piped to C. A is available for re-piping.

@chrisdickinson

@chrisdickinson there's like three things you can do:

close() normally, which will let any enqueued data finish
abort(err), which will error the stream and throw away any enqueued data, saying nobody should ever use it again
a third thing we haven't given a verb for, because it seemed uncommon and kind of bad, where you flush any enqueued data but still act as if the stream was closed normally. It seems bad especially to pretend things are OK if you've thrown away enqueued data, so we didn't add this.

@chrisdickinson

@chrisdickinson the scenarios look right to me! at some point we should consider if .pipe should return to streams1 behavior and call .destroy on close as well (but that should be a different issue)

@domenic

@domenic Ah, cool. That makes sense. Do errors travel up pipelines? (i.e., would something like http.createServer((req, res) => { fs.createReadStream(file).pipe(res) }) still leak descriptors when res aborts?)

@chrisdickinson

@chrisdickinson yeah piping by default propagates errors backward and forward. If res becomes errored then fs.createReadStream(file) gets canceled so no leaked descriptors. Similarly if fs.createReadStream(file) becomes errored res would get aborted.

@domenic

@domenic Ooh, jealous. :)

@mafintosh Cool. Starting to collect .destroy usage throughout core here:

`net.Socket`

                       <Readable impl>
     ____________             |
    /            \            ↓
   |  onSocketEnd |  .end   "end"
   |         |    |     \   /
   |         |    |      \ /
   |         |  onread    Y
   |         ⅄   / _______/
   |        / \ | /
   |       |   \|/
   |       |    ⩛
   |       |    |      <Writable impl>   onSocketFinish_
   |       |    ↓           /                           \
   |       | maybeDestroy  /             connect _______ |
   |       ↓       \      /                             \|
   |   destroySoon  |  "finish"          .connect ______ |
    \________    \  |  /                                \|
             \    \ | /                  afterConnect___ |
              |    \|/                                  \|
              |     ⩛                    ._read ________ |
              |     |                                   \|
              |     ↓                    ._writeGeneric_ |
              | destroy([exception])                    \|
               \____|                    afterWrite_____ |
                    ↓                                   \|
                _destroy([exception][, callback])        |
                    | ^                                  |
                    ⅄  \________________                 |
         __________/|\_________         \_______________/
        /           |          \
       |            ↓           |
       |     timers.unenroll    |
       ↓                        ↓
    fireErrorCallbacks      handle.close
                                |
                                ↓
                             "close"

*Note: http.IncomingMessage and http.OutgoingMessage both exclusively
call .destroy().

`fs.ReadStream`

                 _read
                  |
                  ↓
         .open   fs.read
           |      |
           ↓      ↓
  "end" fs.open  onread
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

`fs.WriteStream`

         .open   _write
           |      |
           ↓      ↓
  .end  fs.open  fs.write
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

`fs.SyncWriteStream`

    .end -> .destroy -> "close"

@chrisdickinson

I'm +1. awesome ascii flow diagrams @chrisdickinson

I think it is possible to have a method of propagating this kind of stream state up and down the pipeline that would be awesome too. At the moment the amount if error handling needed to properly prevent against potential leaks and errors is quite verbose and easily overlooked from the slick stream .pipe examples (particularly in the case of fs & http streams.

Interestingly http.ClientRequest also has a similar abort method (https://github.com/iojs/io.js/blob/v1.x/lib/_http_client.js#L168), which calls net.Socket#destroy (and emits an error).

This is being worked on in nodejs/node#12925.

This has landed 🎉 !

landed and will be in 8.0.0

@chrisdickinson

`net.Socket`

                       <Readable impl>
     ____________             |
    /            \            ↓
   |  onSocketEnd |  .end   "end"
   |         |    |     \   /
   |         |    |      \ /
   |         |  onread    Y
   |         ⅄   / _______/
   |        / \ | /
   |       |   \|/
   |       |    ⩛
   |       |    |      <Writable impl>   onSocketFinish_
   |       |    ↓           /                           \
   |       | maybeDestroy  /             connect _______ |
   |       ↓       \      /                             \|
   |   destroySoon  |  "finish"          .connect ______ |
    \________    \  |  /                                \|
             \    \ | /                  afterConnect___ |
              |    \|/                                  \|
              |     ⩛                    ._read ________ |
              |     |                                   \|
              |     ↓                    ._writeGeneric_ |
              | destroy([exception])                    \|
               \____|                    afterWrite_____ |
                    ↓                                   \|
                _destroy([exception][, callback])        |
                    | ^                                  |
                    ⅄  \________________                 |
         __________/|\_________         \_______________/
        /           |          \
       |            ↓           |
       |     timers.unenroll    |
       ↓                        ↓
    fireErrorCallbacks      handle.close
                                |
                                ↓
                             "close"

*Note: http.IncomingMessage and http.OutgoingMessage both exclusively
call .destroy().

`fs.ReadStream`

                 _read
                  |
                  ↓
         .open   fs.read
           |      |
           ↓      ↓
  "end" fs.open  onread
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

`fs.WriteStream`

         .open   _write
           |      |
           ↓      ↓
  .end  fs.open  fs.write
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

`fs.SyncWriteStream`

    .end -> .destroy -> "close"

@chrisdickinson Unrelated* what did you use to generate this? Looking for something similar.

Navigation Menu

Repository navigation

Standardizing ".destroy()" #124

`A.pipe(B).pipe(C); A.destroy()` – streams2+, default _destroy

`A.pipe(B).pipe(C); B.destroy()` – streams2+, default _destroy

`A.pipe(B).pipe(C); B.destroy()` – A is streams1, B and C are streams2+, default _destroy

`net.Socket`

`fs.ReadStream`

`fs.WriteStream`

`fs.SyncWriteStream`

`net.Socket`

`fs.ReadStream`

`fs.WriteStream`

`fs.SyncWriteStream`

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Standardizing ".destroy()" #124

Description

Activity

domenic commented on Apr 3, 2015

mafintosh commented on Apr 3, 2015

emilbayes commented on Apr 3, 2015

chrisdickinson commented on Apr 3, 2015

A.pipe(B).pipe(C); A.destroy() – streams2+, default _destroy

A.pipe(B).pipe(C); B.destroy() – streams2+, default _destroy

A.pipe(B).pipe(C); B.destroy() – A is streams1, B and C are streams2+, default _destroy

domenic commented on Apr 3, 2015

mafintosh commented on Apr 3, 2015

chrisdickinson commented on Apr 3, 2015

domenic commented on Apr 3, 2015

chrisdickinson commented on Apr 4, 2015

net.Socket

fs.ReadStream

fs.WriteStream

fs.SyncWriteStream

sonewman commented on Apr 4, 2015

mcollina commented on May 19, 2017

mcollina commented on May 24, 2017

jasnell commented on May 24, 2017

BobbyBakes commented on Dec 11, 2018

net.Socket

fs.ReadStream

fs.WriteStream

fs.SyncWriteStream

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

`A.pipe(B).pipe(C); A.destroy()` – streams2+, default _destroy

`A.pipe(B).pipe(C); B.destroy()` – streams2+, default _destroy

`A.pipe(B).pipe(C); B.destroy()` – A is streams1, B and C are streams2+, default _destroy

`net.Socket`

`fs.ReadStream`

`fs.WriteStream`

`fs.SyncWriteStream`

`net.Socket`

`fs.ReadStream`

`fs.WriteStream`

`fs.SyncWriteStream`