Skip to content

Commit 6f806bc

Browse files
committed
fix: resolve race condition in ShardingIndexedCodec during parallel encoding and enable parallel decoding
1 parent 4612451 commit 6f806bc

1 file changed

Lines changed: 18 additions & 18 deletions

File tree

src/main/java/dev/zarr/zarrjava/v3/codec/core/ShardingIndexedCodec.java

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,7 @@ public ByteBuffer encode(final Array shardArray) throws ZarrException {
113113
extendArrayBy1(chunksPerShard, 2));
114114
final List<ByteBuffer> chunkBytesList = new ArrayList<>(chunkCount);
115115

116-
Arrays.stream(
117-
IndexingUtils.computeChunkCoords(shardMetadata.shape, shardMetadata.chunkShape))
116+
Arrays.stream(IndexingUtils.computeChunkCoords(shardMetadata.shape, shardMetadata.chunkShape))
118117
.parallel()
119118
.forEach(
120119
chunkCoords -> {
@@ -128,8 +127,10 @@ public ByteBuffer encode(final Array shardArray) throws ZarrException {
128127
null
129128
);
130129
if (MultiArrayUtils.allValuesEqual(chunkArray, shardMetadata.parsedFillValue)) {
131-
setValueFromShardIndexArray(shardIndexArray, chunkCoords, 0, -1);
132-
setValueFromShardIndexArray(shardIndexArray, chunkCoords, 1, -1);
130+
synchronized (chunkBytesList) {
131+
setValueFromShardIndexArray(shardIndexArray, chunkCoords, 0, -1);
132+
setValueFromShardIndexArray(shardIndexArray, chunkCoords, 1, -1);
133+
}
133134
} else {
134135
final ByteBuffer chunkBytes = codecPipeline.encode(chunkArray);
135136
synchronized (chunkBytesList) {
@@ -207,33 +208,32 @@ private Array decodeInternal(
207208
Utils.toLongArray(shape));
208209

209210
Arrays.stream(allChunkCoords)
210-
// .parallel()
211+
.parallel()
211212
.forEach(
212213
chunkCoords -> {
213214
try {
214215
final long chunkByteOffset = getValueFromShardIndexArray(shardIndexArray,
215216
chunkCoords, 0);
216217
final long chunkByteLength = getValueFromShardIndexArray(shardIndexArray,
217218
chunkCoords, 1);
218-
Array chunkArray = null;
219+
if (chunkByteOffset == -1 || chunkByteLength == -1) {
220+
return;
221+
}
219222
final IndexingUtils.ChunkProjection chunkProjection =
220223
IndexingUtils.computeProjection(chunkCoords, shardMetadata.shape,
221224
shardMetadata.chunkShape, offset, Utils.toLongArray(shape)
222225
);
223-
if (chunkByteOffset != -1 && chunkByteLength != -1) {
224-
final ByteBuffer chunkBytes = dataProvider.read(chunkByteOffset, chunkByteLength);
225-
if (chunkBytes == null) {
226-
throw new ZarrException(String.format("Could not load byte data for chunk %s",
227-
Arrays.toString(chunkCoords)));
228-
}
229-
chunkArray = codecPipeline.decode(chunkBytes);
226+
final ByteBuffer chunkBytes = dataProvider.read(chunkByteOffset, chunkByteLength);
227+
if (chunkBytes == null) {
228+
throw new ZarrException(String.format("Could not load byte data for chunk %s",
229+
Arrays.toString(chunkCoords)));
230230
}
231-
if (chunkArray == null) {
232-
chunkArray = shardMetadata.allocateFillValueChunk();
231+
Array chunkArray = codecPipeline.decode(chunkBytes);
232+
synchronized (outputArray) {
233+
MultiArrayUtils.copyRegion(chunkArray, chunkProjection.chunkOffset, outputArray,
234+
chunkProjection.outOffset, chunkProjection.shape
235+
);
233236
}
234-
MultiArrayUtils.copyRegion(chunkArray, chunkProjection.chunkOffset, outputArray,
235-
chunkProjection.outOffset, chunkProjection.shape
236-
);
237237
} catch (ZarrException e) {
238238
throw new RuntimeException(e);
239239
}

0 commit comments

Comments
 (0)