Heap snapshot visualizations

runtime/observatory/lib/src/service/object.dart

 // Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 part of service;
 
 /// An RpcException represents an exceptional event that happened
 /// while invoking an rpc.
 abstract class RpcException implements Exception {
   RpcException(this.message);
@@ skipping 809 matching lines @@
     totalCollectionTimeInSeconds = heapMap['time'];
     averageCollectionPeriodInMillis = heapMap['avgCollectionPeriodMillis'];
   }
 }
 
 class HeapSnapshot {
   final ObjectGraph graph;
   final DateTime timeStamp;
   final Isolate isolate;
 
-  HeapSnapshot(this.isolate, ByteData data) :
-      graph = new ObjectGraph(new ReadStream(data)),
-      timeStamp = new DateTime.now() {
-  }
+  HeapSnapshot(this.isolate, chunks, nodeCount) :
+      graph = new ObjectGraph(chunks, nodeCount),
+      timeStamp = new DateTime.now();
 
   List<Future<ServiceObject>> getMostRetained({int classId, int limit}) {
     var result = [];
     for (var v in graph.getMostRetained(classId: classId, limit: limit)) {
       var address = v.addressForWordSize(isolate.vm.architectureBits ~/ 8);
       result.add(isolate.getObjectByAddress(address.toRadixString(16)).then((obj) {
         obj.retainedSize = v.retainedSize;
         return new Future(() => obj);
       }));
     }
     return result;
   }
-
-
 }
 
 /// State for a running isolate.
 class Isolate extends ServiceObjectOwner with Coverage {
   @reflectable VM get vm => owner;
   @reflectable Isolate get isolate => this;
   @observable int number;
   @observable DateTime startTime;
   @observable Duration get upTime =>
       (new DateTime.now().difference(startTime));
@@ skipping 40 matching lines @@
   }
 
   /// Fetches and builds the class hierarchy for this isolate. Returns the
   /// Object class object.
   Future<Class> getClassHierarchy() {
     return invokeRpc('getClassList', {})
         .then(_loadClasses)
         .then(_buildClassHierarchy);
   }
 
+  Future<List<Class>> getClassRefs() async {
+    ServiceMap classList = await invokeRpc('getClassList', {});
+    assert(classList.type == 'ClassList');
+    var classRefs = [];
+    for (var cls in classList['classes']) {
+      // Skip over non-class classes.
+      if (cls is Class) {
+        _classesByCid[cls.vmCid] = cls;
+        classRefs.add(cls);
+      }
+    }
+    return classRefs;
+  }
+
   /// Given the class list, loads each class.
   Future<List<Class>> _loadClasses(ServiceMap classList) {
     assert(classList.type == 'ClassList');
     var futureClasses = [];
     for (var cls in classList['classes']) {
       // Skip over non-class classes.
       if (cls is Class) {
+        _classesByCid[cls.vmCid] = cls;
         futureClasses.add(cls.load());
       }
     }
     return Future.wait(futureClasses);
   }
 
   /// Builds the class hierarchy and returns the Object class.
   Future<Class> _buildClassHierarchy(List<Class> classes) {
     rootClasses.clear();
     objectClass = null;
     for (var cls in classes) {
       if (cls.superclass == null) {
         rootClasses.add(cls);
       }
       if ((cls.vmName == 'Object') && (cls.isPatch == false)) {
         objectClass = cls;
       }
     }
     assert(objectClass != null);
     return new Future.value(objectClass);
   }
 
+  Class getClassByCid(int cid) => _classesByCid[cid];
+
   ServiceObject getFromMap(ObservableMap map) {
     if (map == null) {
       return null;
     }
     var mapType = _stripRef(map['type']);
     if (mapType == 'Isolate') {
       // There are sometimes isolate refs in ServiceEvents.
       return vm.getFromMap(map);
     }
     String mapId = map['id'];
@@ skipping 32 matching lines @@
     };
     return isolate.invokeRpc('getObject', params);
   }
 
   Future<ObservableMap> _fetchDirect() {
     return invokeRpcNoUpgrade('getIsolate', {});
   }
 
   @observable Class objectClass;
   @observable final rootClasses = new ObservableList<Class>();
+  Map<int, Class> _classesByCid = new Map<int, Class>();
 
   @observable Library rootLibrary;
   @observable ObservableList<Library> libraries =
       new ObservableList<Library>();
   @observable ObservableMap topFrame;
 
   @observable String name;
   @observable String vmName;
   @observable ServiceFunction entry;
 
   @observable final Map<String, double> timers =
       toObservable(new Map<String, double>());
 
   final HeapSpace newSpace = new HeapSpace();
   final HeapSpace oldSpace = new HeapSpace();
 
   @observable String fileAndLine;
 
   @observable DartError error;
   @observable HeapSnapshot latestSnapshot;
-  Completer<HeapSnapshot> _snapshotFetch;
+  StreamController _snapshotFetch;
+
+  List<ByteData> _chunksInProgress;
 
   void _loadHeapSnapshot(ServiceEvent event) {
-    latestSnapshot = new HeapSnapshot(this, event.data);
+    if (_snapshotFetch == null || _snapshotFetch.isClosed) {
+      // No outstanding snapshot request. Presumably another client asked for a
+      // snapshot.
+      Logger.root.info("Dropping unsolicited heap snapshot chunk");
+      return;
+    }
+
+    // Occasionally these actually arrive out of order.
+    var chunkIndex = event.chunkIndex;
+    var chunkCount = event.chunkCount;
+    if (_chunksInProgress == null) {
+      _chunksInProgress = new List(chunkCount);
+    }
+    _chunksInProgress[chunkIndex] = event.data;
+    _snapshotFetch.add("Receiving snapshot chunk ${chunkIndex + 1}"
+                       " of $chunkCount...");
+
+    for (var i = 0; i < chunkCount; i++) {
+      if (_chunksInProgress[i] == null) return;
+    }
+
+    var loadedChunks = _chunksInProgress;
+    _chunksInProgress = null;
+
+    latestSnapshot = new HeapSnapshot(this, loadedChunks, event.nodeCount);
     if (_snapshotFetch != null) {
-      _snapshotFetch.complete(latestSnapshot);
+      latestSnapshot.graph.process(_snapshotFetch).then((graph) {
+        _snapshotFetch.add(latestSnapshot);
+        _snapshotFetch.close();
+      });
     }
   }
 
-  Future<HeapSnapshot> fetchHeapSnapshot() {
-    if (_snapshotFetch == null || _snapshotFetch.isCompleted) {
-      _snapshotFetch = new Completer<HeapSnapshot>();
+  Stream fetchHeapSnapshot() {
+    if (_snapshotFetch == null || _snapshotFetch.isClosed) {
+      _snapshotFetch = new StreamController();
+      isolate.vm.streamListen('_Graph');
       isolate.invokeRpcNoUpgrade('requestHeapSnapshot', {});
     }
-    return _snapshotFetch.future;
+    return _snapshotFetch.stream;
   }
 
   void updateHeapsFromMap(ObservableMap map) {
     newSpace.update(map['new']);
     oldSpace.update(map['old']);
   }
 
   void _update(ObservableMap map, bool mapIsRef) {
     name = map['name'];
     vmName = map['name'];
@@ skipping 417 matching lines @@
   }
 
   @observable String eventType;
   @observable Breakpoint breakpoint;
   @observable ServiceMap topFrame;
   @observable ServiceMap exception;
   @observable ServiceObject inspectee;
   @observable ByteData data;
   @observable int count;
   @observable String reason;
+  int chunkIndex, chunkCount, nodeCount;
 
   @observable bool get isPauseEvent {
     return (eventType == kPauseStart ||
             eventType == kPauseExit ||
             eventType == kPauseBreakpoint ||
             eventType == kPauseInterrupted ||
             eventType == kPauseException);
   }
 
   void _update(ObservableMap map, bool mapIsRef) {
@@ skipping 12 matching lines @@
     }
     if (map['exception'] != null) {
       exception = map['exception'];
     }
     if (map['inspectee'] != null) {
       inspectee = map['inspectee'];
     }
     if (map['_data'] != null) {
       data = map['_data'];
     }
+    if (map['chunkIndex'] != null) {
+      chunkIndex = map['chunkIndex'];
+    }
+    if (map['chunkCount'] != null) {
+      chunkCount = map['chunkCount'];
+    }
+    if (map['nodeCount'] != null) {
+      nodeCount = map['nodeCount'];
+    }
     if (map['count'] != null) {
       count = map['count'];
     }
   }
 
   String toString() {
     if (data == null) {
       return "ServiceEvent(owner='${owner.id}', type='${eventType}')";
     } else {
       return "ServiceEvent(owner='${owner.id}', type='${eventType}', "
@@ skipping 1641 matching lines @@
     var v = list[i];
     if ((v is ObservableMap) && _isServiceMap(v)) {
       list[i] = owner.getFromMap(v);
     } else if (v is ObservableList) {
       _upgradeObservableList(v, owner);
     } else if (v is ObservableMap) {
       _upgradeObservableMap(v, owner);
     }
   }
 }