getStream dont use keepalive feature

[fengmk2]

https://github.com/aliyun/oss-nodejs-sdk/blob/master/lib/object.js#L150

cc @zensh

[zensh]

Teambition 文件服务 https://striker.teambition.net/ 使用 toa 编写。OSS 是其存储源之一。完善了日志分析服务之后，我们发现该服务有一定概率出现异常：

 {"name":"ResponseError","message":"socket hang up (req \"error\")}

经过 @orangemi 同学的一番 debug 后终于找到了原因：

我们使用了 oss sdk 的 getStream 方法读取文件流，直接 pipe 给客户端的 response，代码大概如下：

var resp = yield store.getStream(uri)
this.body = resp.stream

但 toa 中有这样一个机制（koa 同样有）

if (typeof val.pipe === 'function') {
  onFinished(this.res, function () {
    destroy(val)
  })
}

也就是当上游数据源是 stream 时，response 结束后会强制 destroy 该 stream。

这回带来什么问题呢？因为 oss sdk 的 agent 默认会启用 keepAlive，共享 socket 连接池。我们设想一个 socket 接了一单把数据 pipe 给客户端后立马又接了另一单，但是前一单客户端的 response 完成后会 destroy socket，导致新单悲剧了~

那么我们是不是可以关闭 oss sdk 的 keepAlive 特性？我觉得这样不太好，毕竟 keepAlive 还是能节约很多开销的。所以目前采取的方案是修改了 toa：

if (typeof val.pipe === 'function') {
  onFinished(this.res, function (err) {
    // destroy stream only if error occured
    if (err) destroy(val)
  })
}

只有当 response 异常时才主动 destroy 上游的 stream，毕竟正常情况都是 stream 结束后触发 response 的结束，是没必要强行 destroy 该 stream 的。

大神们怎么看？ @fengmk2 @dead-horse

[fengmk2]

this.res error 后，去 destroy oss 的 stream，这个情况也会关闭 keepalive socket 的吧。然后这种场景就不会出现问题了？

[zensh]

会~，只是可能性极小了，得找个万全之策

[dead-horse]

关闭 keepalive 吧，毕竟读文件这种大小的传输是否 keepalive 感觉影响没那么大吧

[fengmk2]

@zensh 你们有测试 pipe 不加 keepalvie 慢多少么？

[luckydrq]

yy一下，能不能临时创建一个stream作为中介？

[zensh]

没有对比测过 keepalvie 对来的效益。想了下，response error 的情况下 destroy oss stream 是不会出现问题的，因为这个 socket 肯定还是在为这一单服务，不可能接其他单

[zensh]

搞个 passthrough 中介的话，且不说性能消耗，还要处理异常传递，不美~

[fengmk2]

@luckydrq 但是不 destroy ，oss stream 还是得消费完的，要不然会卡住。

@zensh 你们修复之后，就没有再出现过 socket hang up (req \"error\") 了？

[zensh]

昨晚上的，暂时没有了

[gxcsoccer]

三个不使用keepalive的理由，对oss场景来说比较match
（1）在短暂的故障期间，它们可能引起一个良好连接（good connection）被释放（dropped），
（2）它们消费了不必要的宽带，
（3）在以数据包计费的互联网上它们（额外）花费金钱

[gxcsoccer]

destroy 以后，keepaliveagent 不会自动补上吗？

[orangemi]

这是 NODE_DEBUG=http测试得到log

HTTP 95284: write ret = true
HTTP 95284: write ret = true
HTTP 95284: AGENT socket keep-alive
HTTP 95284: outgoing message end.
HTTP 95284: agent.on(free) oss-cn-hangzhou.aliyuncs.com:80:
HTTP 95284: removeSocket oss-cn-hangzhou.aliyuncs.com:80: destroyed: false
HTTP 95284: CLIENT socket onClose
HTTP 95284: removeSocket oss-cn-hangzhou.aliyuncs.com:80: destroyed: true
HTTP 95284: server socket close

第一次removeSocket是socket结束后keepalive
第二次removeSocket是destroy造成的，socket被destroy之后是不会keepalive的。

[zensh]

@gxcsoccer 因为都在阿里云上，走的是内网，所以没那些问题。destroy，keepaliveagent 会自动补上，问题是 destroy 时误杀了正常的 stream

[fengmk2]

@gxcsoccer 会自动补上

问题是这个

我们设想一个 socket 接了一单把数据 pipe 给客户端后立马又接了另一单，但是前一单客户端的 response 完成后会 destroy socket，导致新单悲剧了~

• agentkeepalive 是监听 socket 的 close 事件来将 socket 移除 sockets 队列。https://github.com/node-modules/agentkeepalive/blob/master/lib/_http_agent.js#L244
• onFinished(res) 只有 res 的 end, finish, error, close 其中一个时间触发了就会认为 finished 了。 https://github.com/jshttp/on-finished/blob/master/index.js#L114

---

出现问题的场景：

定义对象名称

• agentkeepalive.socket1
• oss.stream1, oss.stream2
• koa.res1, koa.res2

场景描述

正常场景：oss client 使用 agentkeepalive，创建了一个基于 agentkeepalive.socket1 的 oss.stream1。
然后将这个 oss.stream1 pipe 给 koa.res1。

• oss.stream1 会首先触发 end 事件
• 然后 agentkeepalive.socket1 会设置在 nextTick 触发 free socket
• 如果这个时候用户请求端的消费速度比较慢，那么 koa.res1 还是不会触发 end，而这个时候 nextTick 到了，agentkeepalive.socket1 会被归还到 agentkeepalive 的 freeSockets 队列中。
• 接着又一个新 koa.res2 => oss.stream2 请求过来，agentkeepalive 从 freeSockets 拿到刚才被 free 的 agentkeepalive.socket1，继续用。
• 最后，第一个 koa.res1 终于消费完了，触发了 end，然后会 destroy agentkeepalive.socket1，于是就出现了 socket hang up (req \"error\")

感觉比较乱，有空补个图。

[fengmk2]

@zensh 重新看了 http.js 和 agent.js ，确实，如果是 error 事件，必然会触发 close，然后这个 socket 是不会别重用的。你的 patch 理论是生效的。

[fengmk2]

@zensh 在并发比较大的情况下，按道理这种异常出现的概率非常高才对。。。

[gxcsoccer]

明白了，感觉那 koa/toa 都不用 destroy socket。由 agentkeepalive 自己去管理

[dead-horse]

koajs/koa#165 找到当时加这个的原因，本意是避免赋值给 this.body 的所有的 stream 都会销毁，不会泄露

[zensh]

@dead-horse 这也是个原因，不过也好处理了， 那么要考虑重复 set body 和错误响应，都需要确保 close stream

[fengmk2]

@dead-horse 感觉第二次还给 body 设置一个 stream，其实应该抛异常处理的。。。而不是现在这样兜底。

[dead-horse]

setter 不应该抛异常吧

[fengmk2]

// stream
    if (val instanceof Stream) {
      if (original instanceof Stream) {
        // same stream, do nothing
        if (original === val) return;

        // destroy original stream and clean up error listener
        original.removeListner('error', this.ctx.onerror);
        destroy(original);
      }

      onFinish(this.res, function(err) {
        if (err) {
          destroy(val);
        }
      });
      ensureErrorHandler(val, this.ctx.onerror);

改成判断，如果上一次也是 stream，就先清理它。

[fengmk2]

koajs/koa@fa63e25 这样修复如何？

[zensh]

我认为 set body 中不应该做这个事，否则考虑各种情况会变得很复杂，这个要放在 respond 和 error handle 中去做

[zensh]

toa 利用自有的逻辑修复了下：toajs/toa@2633f5e#diff-910eb6f57886ca16c136101fb1699231R737

主要有如下几点：

1. onerror 分两个层级，app.onerror 和 ctx.onerror，前者是顶级 error 监听，用于写日志，后者是 context 级 error 监听，拿到 error 后会尝试响应客户端，然后丢给 app.onerror 处理
2. ctx.catchStream 方法会对 stream 添加 ctx.onerror，执行后会卸载监听，如果存在 error 会 destroy stream，但卸载后会保证 app.onerror 监听。像 fs stream 可能在 destroy 后 emit error（有相关 test）。
3. set body 时如果是 stream 会 ctx.catchStream，没做 onFinished 处理，因为考虑可能多次 set body；重复 set body 时，如果之前的 body 是 stream，会卸载 ctx.catchStream，并 destroy 掉
4. request 进来初始化时做 onFinished 处理，对于 stream body，会确保卸载之前的 ctx.catchStream，如果有 error ，ctx.catchStream 的机制会 destroy stream。其它异常则 ctx.onerror 处理。
5. 处理过程中，对于 stream body，如果发生了异常或其它处理，均会触发 set body，从而确保ctx.catchStream 被卸载并被 destroy 掉。

总而言之，任何 stream，进来后会被添加ctx.onerror 级的 error listener，如果出错或弃之，会被 destroy 掉，然后error listener退化为 app.onerror。

所以，对于我们上面的需求，正常情况下 agent socket 不会被 destroy 掉，一旦出错或 body 重置（如 204），就会被 destroy 掉，但这种情况下 agent socket 肯定还是在为当前的 request 服务，不会出现误杀。

[fengmk2]

等 @dead-horse 的 koa fix

[dead-horse]

koajs/koa#612

[zhoujingchao]

yy下，大佬们@dead-horse 最后到底怎么解决的？设过timeout，设过maxSockets，也试过上面的toa机制，还是凉了。在线等，很急。

[ddzy]

有进展吗？上传文件会出现类似的错误（"ali-oss": "^6.10.0"）：

{
    "file": "./dist/img/scrollindicatorsprite.1dde3254.png",
    "err": {
      "code": "ResponseError",
      "message": "socket hang up, PUT http://xxx.com/official-website/img/scrollindicatorsprite.1dde3254.png -1 (connected: true, keepalive socket: true, agent status: {\"createSocketCount\":73,\"createSocketErrorCount\":0,\"closeSocketCount\":65,\"errorSocketCount\":0,\"timeoutSocketCount\":0,\"requestCount\":53,\"freeSockets\":{\"xxx.com:80:\":1},\"sockets\":{\"xxx.com:80:\":7},\"requests\":{}}, socketHandledRequests: 3, socketHandledResponses: 2)\nheaders: {}",
      "name": "ResponseError"
    }
  }