Skip to content

Commit 888b1a7

Browse files
ryry
committed
experimental: light ev_io binding for net.Server
1 parent 2ec4cd5 commit 888b1a7

2 files changed

Lines changed: 214 additions & 66 deletions

File tree

‎lib/net.js‎

Lines changed: 88 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,13 @@ var binding = process.binding('net');
2727
var FreeList = require('freelist').FreeList;
2828

2929
var IOWatcher = process.binding('io_watcher').IOWatcher;
30+
31+
var ioAlloc = process.binding('io_watcher').ioAlloc;
32+
var ioFree = process.binding('io_watcher').ioFree;
33+
var ioStop = process.binding('io_watcher').ioStop;
34+
var ioStart = process.binding('io_watcher').ioStart;
35+
var ioSet = process.binding('io_watcher').ioSet;
36+
3037
var constants = process.binding('constants');
3138
var assert = process.assert;
3239

@@ -820,73 +827,74 @@ function Server(/* [ options, ] listener */) {
820827
self.connections = 0;
821828

822829
self.allowHalfOpen = options.allowHalfOpen || false;
830+
}
831+
util.inherits(Server, events.EventEmitter);
832+
exports.Server = Server;
823833

824-
self.watcher = new IOWatcher();
825-
self.watcher.host = self;
826-
self.watcher.callback = function() {
827-
// Just in case we don't have a dummy fd.
828-
getDummyFD();
829834

830-
if (self._pauseTimer) {
831-
// Somehow the watcher got started again. Need to wait until
832-
// the timer finishes.
833-
self.watcher.stop();
834-
}
835+
exports.createServer = function() {
836+
return new Server(arguments[0], arguments[1]);
837+
};
835838

836-
while (self.fd) {
837-
try {
838-
var peerInfo = accept(self.fd);
839-
} catch (e) {
840-
if (e.errno != EMFILE) throw e;
841839

842-
// Gracefully reject pending clients by freeing up a file
843-
// descriptor.
844-
rescueEMFILE(function() {
845-
self._rejectPending();
846-
});
847-
return;
848-
}
849-
if (!peerInfo) return;
840+
Server.prototype._onIO = function(watcher, readable, writable) {
841+
var self = this;
850842

851-
if (self.maxConnections && self.connections >= self.maxConnections) {
852-
// Close the connection we just had
853-
close(peerInfo.fd);
854-
// Reject all other pending connectins.
855-
self._rejectPending();
856-
return;
857-
}
843+
// Just in case we don't have a dummy fd.
844+
getDummyFD();
858845

859-
self.connections++;
846+
if (self._pauseTimer) {
847+
// Somehow the watcher got started again. Need to wait until
848+
// the timer finishes.
849+
ioStop(watcher);
850+
}
860851

861-
var options = { fd: peerInfo.fd,
862-
type: self.type,
863-
allowHalfOpen: self.allowHalfOpen };
864-
var s = new Socket(options);
865-
s.remoteAddress = peerInfo.address;
866-
s.remotePort = peerInfo.port;
867-
s.type = self.type;
868-
s.server = self;
869-
s.resume();
852+
while (self.fd) {
853+
try {
854+
var peerInfo = accept(self.fd);
855+
} catch (e) {
856+
if (e.errno != EMFILE) throw e;
870857

871-
self.emit('connection', s);
858+
// Gracefully reject pending clients by freeing up a file
859+
// descriptor.
860+
rescueEMFILE(function() {
861+
self._rejectPending();
862+
});
863+
return;
864+
}
865+
if (!peerInfo) return;
872866

873-
// The 'connect' event probably should be removed for server-side
874-
// sockets. It's redundant.
875-
try {
876-
s.emit('connect');
877-
} catch (e) {
878-
s.destroy(e);
879-
return;
880-
}
867+
if (self.maxConnections && self.connections >= self.maxConnections) {
868+
// Close the connection we just had
869+
close(peerInfo.fd);
870+
// Reject all other pending connectins.
871+
self._rejectPending();
872+
return;
881873
}
882-
};
883-
}
884-
util.inherits(Server, events.EventEmitter);
885-
exports.Server = Server;
886874

875+
self.connections++;
887876

888-
exports.createServer = function() {
889-
return new Server(arguments[0], arguments[1]);
877+
var options = { fd: peerInfo.fd,
878+
type: self.type,
879+
allowHalfOpen: self.allowHalfOpen };
880+
var s = new Socket(options);
881+
s.remoteAddress = peerInfo.address;
882+
s.remotePort = peerInfo.port;
883+
s.type = self.type;
884+
s.server = self;
885+
s.resume();
886+
887+
self.emit('connection', s);
888+
889+
// The 'connect' event probably should be removed for server-side
890+
// sockets. It's redundant.
891+
try {
892+
s.emit('connect');
893+
} catch (e) {
894+
s.destroy(e);
895+
return;
896+
}
897+
}
890898
};
891899

892900

@@ -899,15 +907,18 @@ Server.prototype.pause = function(msecs) {
899907
var self = this;
900908
msecs = msecs || 1000;
901909

902-
this.watcher.stop();
910+
assert(this.watcher);
911+
ioStop(this.watcher);
903912

904913
// Wait a second before accepting more.
905914
this._pauseTimer = setTimeout(function() {
906915
// Our fd should still be there. If someone calls server.close() then
907916
// the pauseTimer should be cleared.
908917
assert(parseInt(self.fd) >= 0);
909918
self._pauseTimer = null;
910-
self.watcher.start();
919+
920+
assert(self.watcher);
921+
ioStart(self.watcher);
911922
}, msecs);
912923
};
913924

@@ -931,6 +942,16 @@ Server.prototype._rejectPending = function() {
931942
};
932943

933944

945+
Server.prototype._getFD = function(type) {
946+
assert(!this.fd);
947+
assert(!this.watcher);
948+
949+
this.type = type;
950+
this.fd = socket(type);
951+
this.watcher = ioAlloc(this);
952+
};
953+
954+
934955
// Listen on a UNIX socket
935956
// server.listen('/tmp/socket');
936957
//
@@ -953,13 +974,11 @@ Server.prototype.listen = function() {
953974
if (arguments.length == 0 || typeof arguments[0] == 'function') {
954975
// Don't bind(). OS will assign a port with INADDR_ANY.
955976
// The port can be found with server.address()
956-
self.type = 'tcp4';
957-
self.fd = socket(self.type);
977+
self._getFD('tcp4');
958978
self._doListen(port);
959979
} else if (port === false) {
960980
// the first argument specifies a path
961-
self.fd = socket('unix');
962-
self.type = 'unix';
981+
self._getFD('unix');
963982
var path = arguments[0];
964983
self.path = path;
965984
// unlink sockfile if it exists
@@ -987,8 +1006,7 @@ Server.prototype.listen = function() {
9871006
if (err) {
9881007
self.emit('error', err);
9891008
} else {
990-
self.type = addressType == 4 ? 'tcp4' : 'tcp6';
991-
self.fd = socket(self.type);
1009+
self._getFD(addressType == 4 ? 'tcp4' : 'tcp6');
9921010
self._doListen(port, ip);
9931011
}
9941012
});
@@ -1002,12 +1020,14 @@ Server.prototype.listenFD = function(fd, type) {
10021020

10031021
this.fd = fd;
10041022
this.type = type || null;
1023+
this.watcher = ioAlloc(this);
10051024
this._startWatcher();
10061025
};
10071026

10081027
Server.prototype._startWatcher = function() {
1009-
this.watcher.set(this.fd, true, false);
1010-
this.watcher.start();
1028+
assert(this.watcher);
1029+
ioSet(this.watcher, this.fd, true, false);
1030+
ioStart(this.watcher);
10111031
this.emit('listening');
10121032
};
10131033

@@ -1053,7 +1073,9 @@ Server.prototype.close = function() {
10531073
var self = this;
10541074
if (!self.fd) throw new Error('Not running');
10551075

1056-
self.watcher.stop();
1076+
ioStop(self.watcher);
1077+
ioFree(self.watcher);
1078+
self.watcher = null;
10571079

10581080
close(self.fd);
10591081
self.fd = null;

‎src/node_io_watcher.cc‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,130 @@
55
#include <v8.h>
66

77
#include <assert.h>
8+
#include <stdint.h> // intptr_t
9+
#include <stdlib.h>
10+
11+
extern char etext;
812

913
namespace node {
1014

1115
using namespace v8;
1216

1317
Persistent<FunctionTemplate> IOWatcher::constructor_template;
1418
Persistent<String> callback_symbol;
19+
Persistent<String> on_io_symbol;
20+
21+
22+
static inline Local<Integer> AddressToJS(void* address) {
23+
intptr_t diff = (intptr_t)address - (intptr_t)&etext;
24+
assert(diff > 0 && diff < 0x7FFFFFFF);
25+
return Integer::New(diff);
26+
}
27+
28+
29+
template <class T>
30+
static inline T* JSToAddress(Local<Value> v) {
31+
return static_cast<T*>((void*)(v->IntegerValue() + &etext));
32+
}
33+
34+
35+
static void IOCallback(EV_P_ ev_io *watcher, int revents) {
36+
HandleScope scope;
37+
38+
Persistent<Object> obj = *static_cast<Persistent<Object>*>(watcher->data);
39+
40+
Local<Value> callback_v = obj->Get(on_io_symbol);
41+
if (!callback_v->IsFunction()) return;
42+
43+
Local<Function> callback = Local<Function>::Cast(callback_v);
44+
45+
TryCatch try_catch;
46+
47+
Local<Value> argv[4];
48+
argv[0] = AddressToJS(watcher);
49+
argv[1] = Local<Value>::New(revents & EV_READ ? True() : False());
50+
argv[2] = Local<Value>::New(revents & EV_WRITE ? True() : False());
51+
52+
callback->Call(obj, 3, argv);
53+
54+
if (try_catch.HasCaught()) {
55+
FatalException(try_catch);
56+
}
57+
}
58+
59+
60+
static Handle<Value> IOAlloc(const Arguments& args) {
61+
HandleScope scope;
62+
63+
ev_io *io = new ev_io;
64+
ev_init(io, IOCallback);
65+
66+
assert(args[0]->IsObject());
67+
68+
Persistent<Object> *obj_p = new Persistent<Object>();
69+
*obj_p = Persistent<Object>::New(args[0]->ToObject());
70+
io->data = obj_p;
71+
72+
return scope.Close(AddressToJS(io));
73+
}
74+
75+
76+
static Handle<Value> IOFree(const Arguments& args) {
77+
HandleScope scope;
78+
ev_io *io = JSToAddress<ev_io>(args[0]);
79+
delete io;
80+
return Undefined();
81+
}
82+
83+
84+
static Handle<Value> IOSet(const Arguments& args) {
85+
HandleScope scope;
86+
87+
ev_io *io = JSToAddress<ev_io>(args[0]);
88+
89+
if (!args[1]->IsInt32()) {
90+
return ThrowException(Exception::TypeError(
91+
String::New("First arg should be a file descriptor.")));
92+
}
93+
94+
int fd = args[1]->Int32Value();
95+
96+
if (!args[2]->IsBoolean()) {
97+
return ThrowException(Exception::TypeError(
98+
String::New("Second arg should boolean (readable).")));
99+
}
100+
101+
int events = 0;
102+
103+
if (args[2]->IsTrue()) events |= EV_READ;
104+
105+
if (!args[3]->IsBoolean()) {
106+
return ThrowException(Exception::TypeError(
107+
String::New("Third arg should boolean (writable).")));
108+
}
109+
110+
if (args[3]->IsTrue()) events |= EV_WRITE;
111+
112+
ev_io_set(io, fd, events);
113+
114+
return Undefined();
115+
}
116+
117+
118+
static Handle<Value> IOStart(const Arguments& args) {
119+
HandleScope scope;
120+
ev_io *io = JSToAddress<ev_io>(args[0]);
121+
ev_io_start(EV_DEFAULT_UC_ io);
122+
return Undefined();
123+
}
124+
125+
126+
static Handle<Value> IOStop(const Arguments& args) {
127+
HandleScope scope;
128+
ev_io *io = JSToAddress<ev_io>(args[0]);
129+
ev_io_stop(EV_DEFAULT_UC_ io);
130+
return Undefined();
131+
}
15132

16133

17134
void IOWatcher::Initialize(Handle<Object> target) {
@@ -28,7 +145,16 @@ void IOWatcher::Initialize(Handle<Object> target) {
28145

29146
target->Set(String::NewSymbol("IOWatcher"), constructor_template->GetFunction());
30147

148+
// New style IOWatchers don't use an ObjectWrap, but rather return a raw
149+
// pointer in javascript to the ev_io.
150+
NODE_SET_METHOD(target, "ioAlloc", IOAlloc);
151+
NODE_SET_METHOD(target, "ioSet", IOSet);
152+
NODE_SET_METHOD(target, "ioStop", IOStop);
153+
NODE_SET_METHOD(target, "ioStart", IOStart);
154+
NODE_SET_METHOD(target, "ioFree", IOFree);
155+
31156
callback_symbol = NODE_PSYMBOL("callback");
157+
on_io_symbol = NODE_PSYMBOL("_onIO");
32158
}
33159

34160

0 commit comments

Comments
 (0)