| OLD | NEW |
| 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file |
| 2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
| 3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
| 4 | 4 |
| 5 // VMOptions=--old_gen_heap_size=32 | 5 // VMOptions=--old_gen_heap_size=64 |
| 6 | 6 |
| 7 library slow_consumer_test; | 7 library slow_consumer_test; |
| 8 | 8 |
| 9 import 'dart:async'; | 9 import 'dart:async'; |
| 10 import 'dart:isolate'; | 10 import 'dart:isolate'; |
| 11 | 11 |
| 12 const int KB = 1024; | 12 const int KB = 1024; |
| 13 const int MB = KB * KB; | 13 const int MB = KB * KB; |
| 14 const int GB = KB * KB * KB; | 14 const int GB = KB * KB * KB; |
| 15 | 15 |
| (...skipping 37 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 53 int targetCount; | 53 int targetCount; |
| 54 StreamController controller; | 54 StreamController controller; |
| 55 Timer pendingSend; | 55 Timer pendingSend; |
| 56 | 56 |
| 57 DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) { | 57 DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) { |
| 58 controller = new StreamController(onPauseStateChange: onPauseStateChange); | 58 controller = new StreamController(onPauseStateChange: onPauseStateChange); |
| 59 Timer.run(send); | 59 Timer.run(send); |
| 60 } | 60 } |
| 61 | 61 |
| 62 Stream get stream => controller.stream; | 62 Stream get stream => controller.stream; |
| 63 |
| 63 send() { | 64 send() { |
| 64 if (pendingSend != null) { | 65 if (pendingSend != null) { |
| 65 pendingSend.cancel(); | 66 pendingSend.cancel(); |
| 66 pendingSend = null; | 67 pendingSend = null; |
| 67 } | 68 } |
| 68 if (controller.isPaused) return; | 69 if (controller.isPaused) return; |
| 69 if (sentCount == targetCount) { | 70 if (sentCount == targetCount) { |
| 70 controller.close(); | 71 controller.close(); |
| 71 return; | 72 return; |
| 72 } | 73 } |
| (...skipping 17 matching lines...) Expand all Loading... |
| 90 send(); | 91 send(); |
| 91 } | 92 } |
| 92 } | 93 } |
| 93 | 94 |
| 94 main() { | 95 main() { |
| 95 var port = new ReceivePort(); | 96 var port = new ReceivePort(); |
| 96 // The data provider can deliver 800MB/s of data. It sends 100MB of data to | 97 // The data provider can deliver 800MB/s of data. It sends 100MB of data to |
| 97 // the slower consumer who can only read 200MB/s. The data is sent in 1MB | 98 // the slower consumer who can only read 200MB/s. The data is sent in 1MB |
| 98 // chunks. | 99 // chunks. |
| 99 // | 100 // |
| 100 // This test is limited to 32MB of heap-space (see VMOptions on top of the | 101 // This test is limited to 64MB of heap-space (see VMOptions on top of the |
| 101 // file). If the consumer doesn't pause the data-provider it will run out of | 102 // file). If the consumer doesn't pause the data-provider it will run out of |
| 102 // heap-space. | 103 // heap-space. |
| 103 | 104 |
| 104 new DataProvider(800 * MB, 100 * MB, 1 * MB).stream | 105 new DataProvider(800 * MB, 100 * MB, 1 * MB).stream |
| 105 .pipe(new SlowConsumer(200 * MB)) | 106 .pipe(new SlowConsumer(200 * MB)) |
| 106 .then((count) { | 107 .then((count) { |
| 107 port.close(); | 108 port.close(); |
| 108 Expect.equals(100 * MB, count); | 109 Expect.equals(100 * MB, count); |
| 109 }); | 110 }); |
| 110 } | 111 } |
| OLD | NEW |