events: improve eventEmitter.once() performance

This commit improves once() performance by storing the event handler
directly instead of creating a wrapper function every time.

These changes bring ~150% increase in performance when simply adding
once() event handlers, ~220% increase in the included ee-emit-once
benchmark, and a ~50% increase in the included ee-add-remove-once
benchmark.

benchmark/events/ee-add-remove-once.js

@@ -0,0 +1,23 @@
+var common = require('../common.js');
+var EventEmitter = require('events').EventEmitter;
+
+var bench = common.createBenchmark(main, {n: [25e6]});
+
+function main(conf) {
+  var n = conf.n | 0;
+
+  var ee = new EventEmitter();
+  var listeners = [];
+
+  for (var k = 0; k < 10; k += 1)
+    listeners.push(function() {});
+
+  bench.start();
+  for (var i = 0; i < n; i += 1) {
+    for (var k = listeners.length; --k >= 0; /* empty */)
+      ee.once('dummy', listeners[k]);
+    for (var k = listeners.length; --k >= 0; /* empty */)
+      ee.removeListener('dummy', listeners[k]);
+  }
+  bench.end(n);
+}

benchmark/events/ee-emit-once.js

@@ -0,0 +1,19 @@
+var common = require('../common.js');
+var EventEmitter = require('events').EventEmitter;
+
+var bench = common.createBenchmark(main, {n: [25e6]});
+
+function main(conf) {
+  var n = conf.n | 0;
+
+  var ee = new EventEmitter();
+
+  function noop() {}
+
+  bench.start();
+  for (var i = 0; i < n; i += 1) {
+    ee.once('dummy', noop);
+    ee.emit('dummy');
+  }
+  bench.end(n);
+}

lib/_stream_readable.js

@@ -555,10 +555,16 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
   // is attached before any userland ones.  NEVER DO THIS.
   if (!dest._events || !dest._events.error)
     dest.on('error', onerror);
-  else if (Array.isArray(dest._events.error))
-    dest._events.error.unshift(onerror);
-  else
-    dest._events.error = [onerror, dest._events.error];
+  else {
+    var existingHandler = dest._events.error;
+    if (Array.isArray(existingHandler))
+      existingHandler.unshift(onerror);
+    else {
+      var handlers = [onerror, existingHandler];
+      handlers.onceCount = (existingHandler.once ? 1 : 0);
+      dest._events.error = handlers;
+    }
+  }
 
 
   // Both close and finish should trigger unpipe, but only once.
@@ -662,9 +668,7 @@ Readable.prototype.unpipe = function(dest) {
 
 // set up data events if they are asked for
 // Ensure readable listeners eventually get something
-Readable.prototype.on = function(ev, fn) {
-  var res = Stream.prototype.on.call(this, ev, fn);
-
+Readable.prototype._on = function(ev) {
   // If listening to data, and it has not explicitly been paused,
   // then call resume to start the flow of data on the next tick.
   if (ev === 'data' && false !== this._readableState.flowing) {
@@ -684,7 +688,15 @@ Readable.prototype.on = function(ev, fn) {
       }
     }
   }
-
+};
+Readable.prototype.on = function(ev, fn) {
+  var res = Stream.prototype.on.call(this, ev, fn);
+  this._on(ev);
+  return res;
+};
+Readable.prototype.once = function(ev, fn) {
+  var res = Stream.prototype.once.call(this, ev, fn);
+  this._on(ev);
   return res;
 };
 Readable.prototype.addListener = Readable.prototype.on;