Skip to content

Commit a07e731

Browse files
lecousinlecousin
committed
add test + fix BufferedIO
1 parent 1d59bb8 commit a07e731

File tree

3 files changed

+81
-17
lines changed

3 files changed

+81
-17
lines changed

‎net.lecousin.core/src/main/java/net/lecousin/framework/io/buffering/BufferedIO.java

Lines changed: 28 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -700,13 +700,13 @@ protected int read() throws IOException {
700700
int bufferIndex = getBufferIndex(pos);
701701
Buffer buffer = useBufferSync(bufferIndex);
702702
buffer.lastRead = System.currentTimeMillis();
703-
buffer.inUse--;
704703
byte b = buffer.buffer[getBufferOffset(pos)];
705704
pos++;
706705
if (pos < size && (bufferIndex != 0 || startReadSecondBufferWhenFirstBufferFullyRead)) {
707706
int nextIndex = getBufferIndex(pos);
708707
if (nextIndex != bufferIndex) loadBuffer(nextIndex);
709708
}
709+
buffer.inUse--;
710710
return b & 0xFF;
711711
}
712712

@@ -751,11 +751,14 @@ protected int readAsync() throws IOException {
751751
int bufferIndex = getBufferIndex(pos);
752752
Buffer buffer = useBufferAsync(bufferIndex);
753753
SynchronizationPoint<NoException> sp = buffer.loading;
754-
if (sp != null && !sp.isUnblocked()) return -2;
754+
if (sp != null && !sp.isUnblocked()) {
755+
buffer.lastRead = System.currentTimeMillis();
756+
buffer.inUse--;
757+
return -2;
758+
}
755759
if (buffer.error != null) throw buffer.error;
756-
buffer.lastRead = System.currentTimeMillis();
757-
buffer.inUse--;
758760
byte b = buffer.buffer[getBufferOffset(pos)];
761+
buffer.inUse--;
759762
pos++;
760763
if (pos < size && (bufferIndex != 0 || startReadSecondBufferWhenFirstBufferFullyRead)) {
761764
int nextIndex = getBufferIndex(pos);
@@ -786,6 +789,7 @@ public Integer run() throws IOException {
786789
throw buffer.error;
787790
int start = getBufferOffset(pos);
788791
if (start >= buffer.len) {
792+
buffer.inUse--;
789793
if (ondone != null) ondone.run(new Pair<>(Integer.valueOf(0), null));
790794
return Integer.valueOf(0);
791795
}
@@ -1034,14 +1038,15 @@ public void run() {
10341038
buffers.add(new Buffer(this));
10351039
while (firstIndex <= lastIndex) {
10361040
Buffer b;
1037-
if (firstIndex >= buffers.size()) {
1041+
if (firstIndex == buffers.size()) {
10381042
b = new Buffer(this);
1039-
newBuffers.add(b);
10401043
buffers.add(b);
10411044
} else
10421045
b = buffers.get(firstIndex);
1043-
if (b.buffer == null)
1046+
if (b.buffer == null) {
10441047
b.buffer = new byte[firstIndex > 0 ? bufferSize : firstBufferSize];
1048+
newBuffers.add(b);
1049+
}
10451050
b.len = firstIndex < lastIndex ? b.buffer.length : (int)(newSize - getBufferStart(firstIndex));
10461051
b.lastRead = System.currentTimeMillis();
10471052
firstIndex++;
@@ -1151,22 +1156,26 @@ protected AsyncWork<Integer, IOException> writeAsync(
11511156
List<Buffer> buffers = this.buffers;
11521157
if (closing || buffers == null)
11531158
return new AsyncWork<>(null, null, new CancelException("IO closed"));
1159+
boolean isNew = false;
11541160
synchronized (buffers) {
11551161
if (bufferIndex == buffers.size()) {
11561162
buffer = new Buffer(this);
11571163
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
11581164
buffer.len = 0;
11591165
buffer.inUse = 1;
11601166
buffers.add(buffer);
1167+
isNew = true;
11611168
} else {
11621169
buffer = buffers.get(bufferIndex);
11631170
if (buffer.buffer == null) {
11641171
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
11651172
buffer.len = 0;
1173+
isNew = true;
11661174
}
11671175
buffer.inUse++;
11681176
}
11691177
}
1178+
if (isNew) manager.newBuffer(buffer);
11701179
}
11711180
SynchronizationPoint<NoException> sp = buffer.loading;
11721181
AsyncWork<Integer,IOException> done = new AsyncWork<>();
@@ -1230,22 +1239,26 @@ protected int writeSync(long pos, ByteBuffer buf, int alreadyDone) throws IOExce
12301239
if (pos < size || (size > 0 && getBufferIndex(size - 1) == bufferIndex))
12311240
buffer = useBufferAsync(bufferIndex);
12321241
else {
1242+
boolean isNew = false;
12331243
synchronized (buffers) {
12341244
if (bufferIndex == buffers.size()) {
12351245
buffer = new Buffer(this);
12361246
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
12371247
buffer.len = 0;
12381248
buffer.inUse = 1;
12391249
buffers.add(buffer);
1250+
isNew = true;
12401251
} else {
12411252
buffer = buffers.get(bufferIndex);
12421253
if (buffer.buffer == null) {
12431254
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
12441255
buffer.len = 0;
1256+
isNew = true;
12451257
}
12461258
buffer.inUse++;
12471259
}
12481260
}
1261+
if (isNew) manager.newBuffer(buffer);
12491262
}
12501263
SynchronizationPoint<NoException> sp = buffer.loading;
12511264
if (sp != null)
@@ -1278,22 +1291,26 @@ protected void write(byte[] buf, int offset, int length) throws IOException {
12781291
if (pos < size || (size > 0 && getBufferIndex(size - 1) == bufferIndex))
12791292
buffer = useBufferAsync(bufferIndex);
12801293
else {
1294+
boolean isNew = false;
12811295
synchronized (buffers) {
12821296
if (bufferIndex == buffers.size()) {
12831297
buffer = new Buffer(this);
12841298
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
12851299
buffer.len = 0;
12861300
buffer.inUse = 1;
12871301
buffers.add(buffer);
1302+
isNew = true;
12881303
} else {
12891304
buffer = buffers.get(bufferIndex);
12901305
if (buffer.buffer == null) {
12911306
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
12921307
buffer.len = 0;
1308+
isNew = true;
12931309
}
12941310
buffer.inUse++;
12951311
}
12961312
}
1313+
if (isNew) manager.newBuffer(buffer);
12971314
}
12981315
SynchronizationPoint<NoException> sp = buffer.loading;
12991316
if (sp != null)
@@ -1325,22 +1342,26 @@ protected void write(byte b) throws IOException {
13251342
if (pos < size || (size > 0 && getBufferIndex(size - 1) == bufferIndex))
13261343
buffer = useBufferAsync(bufferIndex);
13271344
else {
1345+
boolean isNew = false;
13281346
synchronized (buffers) {
13291347
if (bufferIndex == buffers.size()) {
13301348
buffer = new Buffer(this);
13311349
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
13321350
buffer.len = 0;
13331351
buffer.inUse = 1;
13341352
buffers.add(buffer);
1353+
isNew = true;
13351354
} else {
13361355
buffer = buffers.get(bufferIndex);
13371356
if (buffer.buffer == null) {
13381357
buffer.buffer = new byte[bufferIndex == 0 ? firstBufferSize : bufferSize];
13391358
buffer.len = 0;
1359+
isNew = true;
13401360
}
13411361
buffer.inUse++;
13421362
}
13431363
}
1364+
if (isNew) manager.newBuffer(buffer);
13441365
}
13451366
SynchronizationPoint<NoException> sp = buffer.loading;
13461367
if (sp != null)

‎net.lecousin.core/src/main/java/net/lecousin/framework/io/buffering/BufferingManager.java

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,13 @@
2727
*/
2828
public class BufferingManager implements Closeable, IMemoryManageable {
2929

30+
/** By default set to 55MB. */
31+
public static final int DEFAULT_MEMORY_THRESHOLD = 55 * 1024 * 1024;
32+
/** By default set to 64MB. */
33+
public static final int DEFAULT_MAX_MEMORY = 64 * 1024 * 1024;
34+
/** By default set to 12MB. */
35+
public static final int DEFAULT_TO_BE_WRITTEN_THRESHOLD = 12 * 1024 * 1024;
36+
3037
private Background background;
3138

3239
private BufferingManager() {
@@ -58,11 +65,11 @@ static class Buffer {
5865

5966
private LinkedArrayList<Buffer> buffers = new LinkedArrayList<>(25);
6067
private long totalMemory = 0;
61-
private long maxMemory = 64 * 1024 * 1024;
62-
private long memoryThreshold = 55 * 1024 * 1024;
68+
private long maxMemory = DEFAULT_MAX_MEMORY;
69+
private long memoryThreshold = DEFAULT_MEMORY_THRESHOLD;
6370
private long lastFree = 0;
6471
private long toBeWritten = 0;
65-
private long toBeWrittenThreshold = 12 * 1024 * 1024;
72+
private long toBeWrittenThreshold = DEFAULT_TO_BE_WRITTEN_THRESHOLD;
6673

6774
public long getMemoryThreshold() { return memoryThreshold; }
6875

‎net.lecousin.core/src/test/java/net/lecousin/framework/core/tests/io/buffered/TestBufferedIOReadWrite.java

Lines changed: 43 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,64 @@
11
package net.lecousin.framework.core.tests.io.buffered;
22

33
import java.io.File;
4+
import java.util.ArrayList;
45
import java.util.Collection;
5-
6-
import org.junit.runner.RunWith;
7-
import org.junit.runners.Parameterized;
8-
import org.junit.runners.Parameterized.Parameters;
6+
import java.util.List;
97

108
import net.lecousin.framework.concurrent.Task;
119
import net.lecousin.framework.core.test.io.TestIO;
1210
import net.lecousin.framework.core.test.io.TestReadWrite;
1311
import net.lecousin.framework.io.FileIO;
1412
import net.lecousin.framework.io.buffering.BufferedIO;
13+
import net.lecousin.framework.io.buffering.BufferingManager;
14+
15+
import org.junit.After;
16+
import org.junit.Before;
17+
import org.junit.runner.RunWith;
18+
import org.junit.runners.Parameterized;
19+
import org.junit.runners.Parameterized.Parameters;
1520

1621
@RunWith(Parameterized.class)
1722
public class TestBufferedIOReadWrite extends TestReadWrite {
1823

19-
@Parameters(name = "nbBuf = {1}")
24+
@Parameters(name = "nbBuf = {1}, small memory = {2}")
2025
public static Collection<Object[]> parameters() {
21-
return TestIO.UsingTestData.generateTestCases(false);
26+
List<Object[]> tests = TestIO.UsingTestData.generateTestCases(false);
27+
List<Object[]> params = new ArrayList<>(tests.size() * 2);
28+
for (Object[] o : tests) {
29+
Object[] p;
30+
p = new Object[3];
31+
p[0] = o[0];
32+
p[1] = o[1];
33+
p[2] = Boolean.FALSE;
34+
params.add(p);
35+
p = new Object[3];
36+
p[0] = o[0];
37+
p[1] = o[1];
38+
p[2] = Boolean.TRUE;
39+
params.add(p);
40+
}
41+
return params;
2242
}
2343

24-
public TestBufferedIOReadWrite(byte[] testBuf, int nbBuf) {
44+
public TestBufferedIOReadWrite(byte[] testBuf, int nbBuf, boolean smallMemory) {
2545
super(testBuf, nbBuf);
46+
this.smallMemory = smallMemory;
47+
}
48+
49+
private boolean smallMemory;
50+
51+
@Before
52+
public void initMemory() {
53+
if (smallMemory)
54+
BufferingManager.get().setMemoryLimits(48 * 1024, 64 * 1024, 4 * 1024);
55+
else
56+
BufferingManager.get().setMemoryLimits(BufferingManager.DEFAULT_MEMORY_THRESHOLD, BufferingManager.DEFAULT_MAX_MEMORY, BufferingManager.DEFAULT_TO_BE_WRITTEN_THRESHOLD);
57+
}
58+
59+
@After
60+
public void resetMemory() {
61+
BufferingManager.get().setMemoryLimits(BufferingManager.DEFAULT_MEMORY_THRESHOLD, BufferingManager.DEFAULT_MAX_MEMORY, BufferingManager.DEFAULT_TO_BE_WRITTEN_THRESHOLD);
2662
}
2763

2864
@SuppressWarnings({ "unchecked", "resource" })

0 commit comments

Comments
 (0)