"listChanges" Event Stream of ObservableList

DartのobserveパッケージにあるObservableListクラスにはlistChangesというEvent Streamがあって、このStreamはObservableListに加えられた変化を外部に通知する役割を担ってる。このlistChangesは仕組みがちょっとわかりづらいのでここにまとめておく。

listChanges

listChangesのドキュメントに記載されているコードを下記に転載する。

var model = new ObservableList.from(['a', 'b']);
model.listChanges.listen((records) => records.forEach(print));
model.removeAt(1);
model.insertAll(0, ['c', 'd', 'e']);
model.removeRange(1, 3);
model.insert(1, 'f');

で、これを実行すると次のように表示される。

#<ListChangeRecord index: 0, removed: [], addedCount: 2>
#<ListChangeRecord index: 3, removed: [b], addedCount: 0>

この出力、見るとちょっと良くわからない点が多い。例えばコード上ではもっとremoveしてるはずなのにbしかremoveされたことになってないところとか。

最初と最後の差分

実はさっきの出力はリストの最初の状態（['a','b']）と最後の状態（['c','f','a']）の差分を一回のイベント通知で知らせていて、ListChangeRecord間の状態を書き足すとと次のようになる。

['a','b']
#<ListChangeRecord index: 0, removed: [], addedCount: 2>
['??','??','a','b']
#<ListChangeRecord index: 3, removed: [b], addedCount: 0>
['??','??','a']

この'??'に当てはまるのが何かは、イベント通知で送られてくる情報と現在のリストの状態から逆算して実はcとfだった、ということがわかるようになっている。そのほかに追加してすぐ消したdとeかはどうなるかというと、このデフォルトの動作だとそもそも追加されたことすら（listChangesからでは）検知できない。

非同期なバースト型イベント

更に注意しなきゃいけないのは、このイベントが非同期に、しかもバーストでまとめて送られてくるという点にある。例えば元のコードを下記のように書き換えてみる。

var model = new ObservableList.from(['a', 'b']);
model.listChanges.listen((records) {
  print("got ${records.length} changes");
  for (var r in records) {
    print(r);
    print(r.object);
  }
});
model.removeAt(1);
model.insertAll(0, ['c', 'd', 'e']);
model.removeRange(1, 3);
model.insert(1, 'f');
print("end state: $model");

これの出力は次のようになる。

end state: [c, f, a]
got 2 changes
#<ListChangeRecord index: 0, removed: [], addedCount: 2>
[c, f, a]
#<ListChangeRecord index: 3, removed: [b], addedCount: 0>
[c, f, a]

最初にend state...って表示されているところを見ると、リストに対する操作の都度ではなく全部操作が終わってからlistChangesへイベントが一回にまとめて送られてきてるのがわかる。

scheduleMicrotask()

この辺の仕組みがどうなってるかを調べるためにobservable_list.dartのソースを直接見たところ、内部ではscheduleMicrotask()という関数を使っていた。このscheduleMicrotaskの説明は長くなるので公式の立派な文書に任せるとして、ObservableList的には次のような動作となっている。

1. addやremoveのようにObservableListの状態が変わる関数を呼ぶと、その変更を通知する非同期タスク（microtask）がエンキューされる
2. microtaskが実行されるまで以降のaddやremoveで追加・削除された項目は全て「変更された項目リスト」に追加されていく
3. microtaskが実行される段階で「変更された項目リスト」は最小化される。例えばaddで追加された項目が直後にremoveされた場合は相殺されてそもそも最初から追加されなかったことにされる
4. 最小化された「変更された項目リスト」はlistChangesへ通知されて、ObservableList内では消える

同期的な通知

ちなみに同期的に通知する方法ももちろん用意されていて、それをやっているのが次のコード。

var model = new ObservableList.from(['a', 'b']);
model.listChanges.listen((records) {
  print("got ${records.length} changes");
  for (var r in records) {
    print(r);
    print(r.object);
  }
});
model.removeAt(1);
model.deliverListChanges();
model.insertAll(0, ['c', 'd', 'e']);
model.deliverListChanges();
model.removeRange(1, 3);
model.deliverListChanges();
model.insert(1, 'f');
model.deliverListChanges();
print("end state: $model");

やっているのはリストへの操作の度にdeliverListChangesという通知用の関数を呼んでいること。これを実行すると次のように表示される。これは本当に手順通りの結果なのでわかりやすいと思う。

got 1 changes
#<ListChangeRecord index: 1, removed: [b], addedCount: 0>
[a]
got 1 changes
#<ListChangeRecord index: 0, removed: [], addedCount: 3>
[c, d, e, a]
got 1 changes
#<ListChangeRecord index: 1, removed: [d, e], addedCount: 0>
[c, a]
got 1 changes
#<ListChangeRecord index: 1, removed: [], addedCount: 1>
[c, f, a]
end state: [c, f, a]

感想

最初このlistChangesのドキュメントを読んだときは全然理解できなくて、サンプルを書き換えていてやっと理解できた。

ただこのようにObservableListが非同期かつバーストなイベント通知をしている理由は推測できて、おそらく何らかのリスト操作の度にイベントが飛ぶのはパフォーマンス的に避けたかった、ということなのだと思う。 例えば数万の要素をObservableListに追加するようなループを書いたとして、要素が追加されるたびにイベントが飛んでたらさすがにパフォーマンスへの影響は大きいはず。なので、可能な限り少ないイベントで大量の差分を通知できる今の方式になっているのだと考えられる。

改めて考えると多分因果が逆で、Streamていう"非同期で発行されることが前提な通知インフラ"でObservableListの変化を通知するには前記のような仕組みにするのが最適だった、というのが本当のところかもしれない。それが結果的に"変化の度に通知されてしまうよパフォーマンス問題"も解消してる、と。

ちなみにDartでのObservableListに相当するWPFのObservableCollectionはまさに操作の度にイベントを通知するようになっている。これはこれでわかりやすくていいと思うんだけど、ただ今度は大量に追加するときのパフォーマンス問題が出てくるので、一度に大量の要素を追加する可能性の高いチャート用のライブラリとかは一時的にイベント通知を切る仕組みが用意されていたりする。知る限りではDynamicDataDisplayのObservableDataSourceとか。