はじめに

Dart VMのObservatoryという機能について、簡単にご紹介したいと思います。
Observatory機能は、VMのmonitoringとprofiling用の機能で、Java VMにも同様の機能が存在しています。

Java VMの場合、専用のツールを用意したり、JMX経由でremoteアクセスしたりと少し面倒なところはあります。
Dart VMの場合、起動時に追加のオプションを指定し、開いたポートにブラウザからアクセスするだけでOKですので、大変便利です。
簡単な機能ですので、手元で一度試してみることをおすすめします。

詳細は公式サイトでも紹介されています。
https://www.dartlang.org/tools/observatory/

起動オプション

dartコマンドのオプションを指定する場合は、以下のいずれかのオプションです。
待ち受けaddress/portを指定して、localhost外から参照することも可能です。
dart enable-vm-service xxx.dart
もしくは
dart --observe xxx.dart

その後下記へアクセス
http://localhost:8181

Dart EditorからRunした場合、起動時に下記のログが出ました。
Observatory listening on http://127.0.0.1:41539

試しにブラウザからアクセスしてみたら、上記ポートでobservatoryが有効になってました。

どんな機能なのか

機能の説明書くかなーっって思って公式サイトぶらついてたら、詳細な説明をたくさんみつけてしまったので、あまり書くことがないです。。
https://www.dartlang.org/tools/observatory/

というのもあれなので、その中でも拡張機能的なtagsについて試してみました。
https://www.dartlang.org/tools/observatory/tags.html

profile.dart

import 'dart:profiler';

var customTag = new UserTag('MyTag');

// Save the previous tag when installing the custom tag.
var previousTag = customTag.makeCurrent();
// your code here
// Restore the previous tag.
previousTag.makeCurrent();

tagの名称をつけて、ステートメントをmakeCurrent();とpreviousTag.makeCurrent();で囲んであげるだけです。
囲んだプログラムをobservertoryのcpu profileで参照すると、囲んだステートメントを実行した際のプロファイルを参照できます。

VMのidle時間は取り除かれた結果が表示されますので、大変便利です。
測定に使用した関数はこちらになります。

dedup.dart

import 'dart:io';
import 'dart:async';
import 'package:crypto/crypto.dart';
import 'dart:profiler';
import 'dart:math';

var sha1Tag = new UserTag('sha1');
var listTag = new UserTag('list');
var fileTag = new UserTag('file');
var hashTag = new UserTag('hash');
var readTag = new UserTag("read");

Stream<String> dedupStream(Directory parentDir) {
  var dedupFileStream = new StreamController<String>();
  var dedupDict = new Map<String, String>();
  try {
    parentDir.list(recursive: true).forEach((FileSystemEntity entity) {
      FileSystemEntity.isFile(entity.path).then((isFile) {
        if (isFile) {
          SHA1 sha1Hash = new SHA1();
          var previousTag = fileTag.makeCurrent();  /* fileのopen/readまでプロファイル対象にしたいが、、 */
//          tagQueue.add(readTag.makeCurrent());
          new File(entity.path)..openRead().take(10).listen((List<int> readData) {
            var previousTag = sha1Tag.makeCurrent();  /* listenの中のSHA1Hashの計算は囲めるので、正しく計測できる */
            sha1Hash.add(readData);
            previousTag.makeCurrent();
          }).onDone(() {
            var previousTag = hashTag.makeCurrent();  /* Mapへの挿入、参照も、tagで囲めるので、正しく計測できる */
            String hash = sha1Hash.close().toString();
            if (dedupDict.containsKey(hash)) {
              dedupFileStream.add("stream:"+entity.path);
            } else {
              dedupDict.putIfAbsent(hash, () => entity.path);
            }
            previousTag.makeCurrent();
//            tagQueue.removeAt(0)..makeCurrent();
          });
          previousTag.makeCurrent();  /* このパスが実行されるのは、上記listenをすべて消化後ではない。 */
        }
      });
    });
  } catch (e) {
    print(e);
  }
  return dedupFileStream.stream;
}

対象のディレクトリから再帰的にファイルを探し、
ファイルの先頭から数バイト読み込んで、SHA1のdigestを取得し、Mapに突っ込んでおく。
同じdigestが見つかったら重複とみなす。重複とみなしたfilepathをstreamで返す関数です。
ディスクアクセスで負荷をかけつつ、一時的に消費するメモリはStreamのChunkのみ、結果を格納するMapは少しずつ膨らんできます。

プロファイル結果と照らし合わせて見ると、sha1の計算に50%なのはいいとして、Defaultってなんだよって話になります。
Defaultで計測されちゃっているのは、主にここになります。

new File(entity.path)..openRead().take(10).listen((List readData) {

openRead().take(10)がFutureとStreamになっているため、

var previousTag = fileTag.makeCurrent();
xxx
previousTag.makeCurrent();

では測定できていません。Fileを生成して、MicroTaskをququeに積むところしか測定されていません。

かといって、コメントアウトしたpreviousをqueueに積むように書けば、
そこそこ正確なデータがとれますが、問題はあります。
異なるZoneのTree間でTaskSchedulerのcontext switchが発生すると、正確な値がとれません。

Profile用のTagは便利だと感じましたが、FutureやStreamを対象とした測定方法はいろいろと考える必要がありそうです。

注意書き

(1) observatoryモードで起動する場合、実行時のエラーや例外も、observeのほうに出力されます。
また、killするまでVMがidleします。
そのため、try&errorする際には、--observe外しておいた方がよいです。

(2) take10にしているのは、そこそこの時間回すためです。普通はtake1で十分だと思います。

new File(entity.path)..openRead().take(10).listen((List readData) {

openReadしたChunkであるList readDataは、最大64KByteになります。