IOSink and many small .adds is slow

[jensjoha]

On current head (not that I think it has changed lately) I see the following:

Having many small .adds on an IOList is very slow, and I seem to have to buffer it myself. E.g. the following example:

import 'dart:io';
import 'dart:convert';
import 'dart:typed_data';

Utf8Encoder utf8Encoder = const Utf8Encoder();

main() async {
  final int count = 100000;
  // Warmup
  await unbuffered(10000);

  Stopwatch stopwatch = new Stopwatch()..start();
  await buffered(count, 10 * 1024 /* 10 KB buffer */);
  print("Buffered => ${stopwatch.elapsedMilliseconds} ms");

  stopwatch.reset();
  await unbuffered(count);
  print("Unbuffered => ${stopwatch.elapsedMilliseconds} ms");
}

void unbuffered(int count) async {
  File f = new File("io_unbuffered_test");
  IOSink sink = f.openWrite();
  for (int i = 0; i < count; i++) {
    String s = "foo_$i";
    List<int> data = utf8Encoder.convert(s);
    sink.add(data);
  }
  await sink.flush();
  await sink.close();
}

void buffered(int count, int bufferSize) async {
  File f = new File("io_buffered_test");
  IOSink sink = f.openWrite();
  int size = 0;
  Uint8List buffer = new Uint8List(bufferSize);
  for (int i = 0; i < count; i++) {
    String s = "foo_$i";
    List<int> data = utf8Encoder.convert(s);
    if (size + data.length > bufferSize || data.length > bufferSize) {
      sink.add(buffer.sublist(0, size));
      size = 0;
      buffer = new Uint8List(bufferSize);
    }
    if (data.length > bufferSize) {
      sink.add(data);
    } else {
      buffer.setRange(size, size + data.length, data);
      size += data.length;
    }
  }
  sink.add(buffer.sublist(0, size));
  await sink.flush();
  await sink.close();
}

has the following output on my machine:

Buffered => 22 ms
Unbuffered => 2383 ms

So as a thanks for writing 24 lines of code instead of 11 I get a 100x+ speedup...
Is this intentional, and am I really supposed to write the buffering code myself?

/cc @lrhn /cc @zanderso

[lucalooz]

Is there a better way in the SDK to have Sink that facilitates buffering with automatic flush when the internal buffer is full?

I'm facing the same kind of issue by writing myself the buffering code while on other platforms usually I've always found dedicated tools.

[speaking-in-code]

I just hit this. It would be nice if the buffering happened by default.

[TimWhiting]

I also just hit this recently, writing large csv logs.

[brianquinlan]

I have a fix out for review: https://dart-review.googlesource.com/c/sdk/+/285420

The only issue is that the buffering logic reduces the performance of large writes:

Benchmarks (benchmarks/FileIOSink/dart/FileIOSink.dart):

 @before:

 FileIOSink.Add.AlternatingAddSize(RunTime): 548.6213033953998 us.
 FileIOSink.Add.ManySmall(RunTime): 2572584.0 us. (THIS ISSUE)
 FileIOSink.Add.OneLarge(RunTime): 614.477325732055 us.

 @after

 FileIOSink.Add.AlternatingAddSize(RunTime): 1383.766598220397 us.
 FileIOSink.Add.ManySmall(RunTime): 4135.545454545455 us. (YEAH, IT IS FIXED)
 FileIOSink.Add.OneLarge(RunTime): 1022.5904954499495 us.

So the relative performance changes are:

 FileIOSink.Add.AlternatingAddSize(RunTime): 2.52x (slower)
 FileIOSink.Add.ManySmall(RunTime): 0.00161x (MUCH faster)
 FileIOSink.Add.OneLarge(RunTime): 1.665x (slower)

[whesse]

I think that buffering many small writes yourself, if they are generated very quickly, is the right way to fix this. In most cases, StringBuffer, not a hand-coded Uint8List buffer, would be the most appropriate thing to use, and is shorter and simpler.

If a helper utility should be added, then it could be a new method added (a breaking change), called "bufferedWrite" or something like that. It is important that not all writes to a file get buffered at a high level, and don't even get to the OS writing (and buffering) level until flush or close is called. Logging, and large writes, should not get buffered (and copies made) by default.

Proposed changes to File IO writes should be then made in a design proposal, and reviewed by all stakeholders.

[lrhn]

I'd compare this to the Java OutputStream and BufferedOutputStream.

The BufferedOutputStream is a wrapper around an OutputStream, with the same interface, which is guaranteed to buffer data in fixed size buffer until it's full, or you flush or close it.

We can create something similar, say:

class BufferingIOSink implements IOSink {
  static const int _defaultBufferSize = 2048;
  final Sink<List<int>> _sink;
  Uint8List _buffer;
  // Current live bytes in buffer.
  int _start = 0;
  int _end = 0;
  Encoding _encoding;
  BufferingIOSink(Sink<List<int>> output, {Encoding? encoding, Uint8List? buffer, int? bufferSize})
     : _sink = output, 
       _encoding = encoding ?? (output is IOSink ? output.encoding : utf8),
       _buffer = buffer ?? Uint8List(bufferSize ?? _defaultBufferSize);

  void _flushBuffer() {
    if (_start >= _end) return;
    _sink.add(Uint8List.sublistView(_buffer, _start, _end));
    _start = _end;
    var capacity = _buffer.length;
    if ((capacity - _end) * 4 < capacity) { // Reallocate if less that 25% free after flush.
      _buffer = Uint8List(capacity); 
      _start = _end = 0;
    }
  }

  Future<void> flush() {
    _flushBuffer();
    var sink = _sink;
    if (sink is IOSink) return sink.flush();
    return Future.value(null);
  }

  void add(List<int> bytes) {
    var totalCapacity = _buffer.length;
    var capacity = totalCapacity - _end;
    var bytesCursor = 0;
    while (bytesCursor + capacity < bytes.length) {
      _buffer.setRange(_end, totalCapacity, bytes, bytesCursor);
      _bytesCursor += capacity;
      _flushBuffer();
      capacity = totalCapacity - _end;
    }
    _buffer.setRange(_end, _end + bytes.length - bytesCursor, bytes, bytesCursor);
  }
  // All the other methods.

}

[brianquinlan]

The change that I'm proposing is a performance win in all of the scenarios that I've tested i.e. big writes, small writes and mixed writes. It also does not weaken the existing eager write behavior of IOSink.

I don't have a problem with us having a new BufferedIOSink class but, in the case of the RandomAccessFile-backed IOSink found in dart:io, we can increase performance without semantic changes.

[whesse]

OK, I agree with this change, that only combines small writes in the important case that they were building up in a queue inside the existing implementation already. It should not change the semantics at all, and is a performance improvement in all cases.

[rmcloughlin]

@brianquinlan I'm curious: is this issue fixed? It appears that the commit that you linked to was merged over a year ago.