dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Add read/write stats to Socket and RandomAccessFile.

Browse Source 
BUG=
R=johnmccutchan@google.com

Review URL: https://codereview.chromium.org//298353002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@36738 260f80e4-7a28-3924-810f-c04153c831b5
This commit is contained in:
ajohnsen@google.com
2014-05-28 13:29:51 +00:00
parent f5d6c4106e
commit 8b612e8780
2 changed files with 83 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/bin/socket_patch.dart
+54 -128
View File
@@ -207,131 +207,6 @@ class _NetworkInterface implements NetworkInterface {
}
class _Rate {
final int buckets;
final data;
int lastValue = 0;
int nextBucket = 0;
_Rate(int buckets) : buckets = buckets, data = new List.filled(buckets, 0);
void update(int value) {
data[nextBucket] = value - lastValue;
lastValue = value;
nextBucket = (nextBucket + 1) % buckets;
}
int get rate {
int sum = data.fold(0, (prev, element) => prev + element);
return sum ~/ buckets;
}
}
// Statics information for the observatory.
class _SocketStat {
_Rate readRate = new _Rate(5);
_Rate writeRate = new _Rate(5);
void update(_NativeSocket socket) {
readRate.update(socket.totalRead);
writeRate.update(socket.totalWritten);
}
}
class _SocketsObservatory {
static int socketCount = 0;
static Map sockets = new Map<_NativeSocket, _SocketStat>();
static Timer timer;
static add(_NativeSocket socket) {
if (socketCount == 0) startTimer();
sockets[socket] = new _SocketStat();
socketCount++;
}
static remove(_NativeSocket socket) {
_SocketStat stats = sockets.remove(socket);
assert(stats != null);
socketCount--;
if (socketCount == 0) stopTimer();
}
static update(_) {
sockets.forEach((socket, stat) {
stat.update(socket);
});
}
static startTimer() {
if (timer != null) return;
// TODO(sgjesse): Enable the rate timer.
// timer = new Timer.periodic(new Duration(seconds: 1), update);
}
static stopTimer() {
if (timer == null) return;
timer.cancel();
timer = null;
}
static String generateResponse() {
var response = new Map();
response['type'] = 'SocketList';
var members = new List();
response['members'] = members;
sockets.forEach((socket, stat) {
var kind =
socket.isListening ? "LISTENING" :
socket.isPipe ? "PIPE" :
socket.isInternal ? "INTERNAL" : "NORMAL";
var protocol =
socket.isTcp ? "tcp" :
socket.isUdp ? "udp" : "";
var localAddress;
var localPort;
var remoteAddress;
var remotePort;
try {
localAddress = socket.address.address;
} catch (e) {
localAddress = "n/a";
}
try {
localPort = socket.port;
} catch (e) {
localPort = "n/a";
}
try {
remoteAddress = socket.remoteAddress.address;
} catch (e) {
remoteAddress = "n/a";
}
try {
remotePort = socket.remotePort;
} catch (e) {
remotePort = "n/a";
}
members.add({'type': 'Socket', 'kind': kind, 'protocol': protocol,
'localAddress': localAddress, 'localPort': localPort,
'remoteAddress': remoteAddress, 'remotePort': remotePort,
'totalRead': socket.totalRead,
'totalWritten': socket.totalWritten,
'readPerSec': stat.readRate.rate,
'writePerSec': stat.writeRate.rate});
});
return JSON.encode(response);;
}
static String toJSON() {
try {
return generateResponse();
} catch (e, s) {
return '{"type":"Error","text":"$e","stacktrace":"$s"}';
}
}
}
// The NativeFieldWrapperClass1 can not be used with a mixin, due to missing
// implicit constructor.
class _NativeSocketNativeWrapper extends NativeFieldWrapperClass1 {}
@@ -432,14 +307,21 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
bool writeEventIssued = false;
bool writeAvailable = false;
static final Stopwatch sw = new Stopwatch()..start();
// Statistics.
int totalRead = 0;
int totalWritten = 0;
int readCount = 0;
int writeCount = 0;
double lastRead;
double lastWrite;
// The owner object is the object that the Socket is being used by, e.g.
// a HttpServer, a WebSocket connection, a process pipe, etc.
Object owner;
static double get timestamp => sw.elapsedMicroseconds / 1000000.0;
static Future<List<InternetAddress>> lookup(
String host, {InternetAddressType type: InternetAddressType.ANY}) {
return _IOService.dispatch(_SOCKET_LOOKUP, [host, type._value])
@@ -644,6 +526,8 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
available -= result.length;
totalRead += result.length;
}
readCount++;
lastRead = timestamp;
return result;
}
@@ -662,7 +546,10 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
} else {
available -= result.data.length;
}
totalRead += result.data.length;
}
readCount++;
lastRead = timestamp;
return result;
}
@@ -702,6 +589,8 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
// Negate the result, as stated above.
if (result < 0) result = -result;
totalWritten += result;
writeCount++;
lastWrite = timestamp;
return result;
}
@@ -718,6 +607,9 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
scheduleMicrotask(() => reportError(result, "Send failed"));
result = 0;
}
totalWritten += result;
writeCount++;
lastWrite = timestamp;
return result;
}
@@ -733,6 +625,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
socket.localPort = localPort;
socket.address = address;
totalRead += 1;
lastRead = timestamp;
return socket;
}
@@ -1092,7 +985,26 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
r['readClosed'] = isClosedRead;
r['writeClosed'] = isClosedWrite;
r['closing'] = isClosing;
r['listening'] = isListening;
r['fd'] = nativeGetSocketId();
if (owner != null) {
r['owner'] = owner._toJSON(true);
}
return r;
}
Map _toJSONInternal(bool ref) {
var name = 'Internal';
var r = {
'id': _servicePath,
'type': _serviceType(ref),
'name': name,
'user_name': name,
'kind': _JSONKind(),
};
if (ref) {
return r;
}
r['closing'] = isClosing;
r['fd'] = nativeGetSocketId();
if (owner != null) {
r['owner'] = owner._toJSON(true);
@@ -1147,10 +1059,24 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject {
}
Map _toJSON(bool ref) {
var map;
if (isPipe) {
return _toJSONPipe(ref);
map = _toJSONPipe(ref);
} else if (isInternal) {
map = _toJSONInternal(ref);
} else {
map = _toJSONNetwork(ref);
}
return _toJSONNetwork(ref);
if (!ref) {
map['available'] = available;
map['totalRead'] = totalRead;
map['totalWritten'] = totalWritten;
map['readCount'] = totalWritten;
map['writeCount'] = writeCount;
map['lastRead'] = lastRead;
map['lastWrite'] = lastWrite;
}
return map;
}
void nativeSetSocketId(int id) native "Socket_SetSocketId";
sdk/lib/io/file_impl.dart
+29
View File
@@ -567,6 +567,11 @@ class _RandomAccessFile
bool _asyncDispatched = false;
SendPort _fileService;
int _totalRead = 0;
int _totalWritten = 0;
int _readCount = 0;
int _writeCount = 0;
_RandomAccessFile(this._id, this.path) {
_files[_serviceId] = this;
@@ -587,6 +592,10 @@ class _RandomAccessFile
}
r['asyncDispatched'] = _asyncDispatched;
r['fd'] = _getFD(_id);
r['totalRead'] = _totalRead;
r['totalWritten'] = _totalWritten;
r['readCount'] = _totalWritten;
r['writeCount'] = _writeCount;
return r;
}
@@ -627,6 +636,8 @@ class _RandomAccessFile
if (_isErrorResponse(response)) {
throw _exceptionFromResponse(response, "readByte failed", path);
}
_readCount++;
_totalRead++;
return response;
});
}
@@ -639,6 +650,8 @@ class _RandomAccessFile
if (result is OSError) {
throw new FileSystemException("readByte failed", path, result);
}
_readCount++;
_totalRead++;
return result;
}
@@ -650,6 +663,8 @@ class _RandomAccessFile
if (_isErrorResponse(response)) {
throw _exceptionFromResponse(response, "read failed", path);
}
_readCount++;
_totalRead += response[1].length;
return response[1];
});
}
@@ -665,6 +680,8 @@ class _RandomAccessFile
if (result is OSError) {
throw new FileSystemException("readSync failed", path, result);
}
_readCount++;
_totalRead += result.length;
return result;
}
@@ -684,6 +701,8 @@ class _RandomAccessFile
var read = response[1];
var data = response[2];
buffer.setRange(start, start + read, data);
_readCount++;
_totalRead += read;
return read;
});
}
@@ -713,6 +732,8 @@ class _RandomAccessFile
if (result is OSError) {
throw new FileSystemException("readInto failed", path, result);
}
_readCount++;
_totalRead += result;
return result;
}
@@ -724,6 +745,8 @@ class _RandomAccessFile
if (_isErrorResponse(response)) {
throw _exceptionFromResponse(response, "writeByte failed", path);
}
_writeCount++;
_totalWritten++;
return this;
});
}
@@ -739,6 +762,8 @@ class _RandomAccessFile
if (result is OSError) {
throw new FileSystemException("writeByte failed", path, result);
}
_writeCount++;
_totalWritten++;
return result;
}
@@ -765,6 +790,8 @@ class _RandomAccessFile
if (_isErrorResponse(response)) {
throw _exceptionFromResponse(response, "writeFrom failed", path);
}
_writeCount++;
_totalWritten += end - (start - result.start);
return this;
});
}
@@ -791,6 +818,8 @@ class _RandomAccessFile
if (result is OSError) {
throw new FileSystemException("writeFrom failed", path, result);
}
_writeCount++;
_totalWritten += end - (start - bufferAndStart.start);
}
Future<RandomAccessFile> writeString(String string,