Skip to content

Commit 4980ba1

Browse files
#20 Extract entries async
1 parent 41d6959 commit 4980ba1

File tree

14 files changed

+333
-146
lines changed

14 files changed

+333
-146
lines changed

src/main/java/ru/olegcherednik/zip4jvm/decompose/LocalFileHeaderDecompose.java

+6-5
Original file line numberDiff line numberDiff line change
@@ -128,26 +128,27 @@ private void copyPayload(Path dir, ZipEntry zipEntry, ZipEntryBlock.LocalFileHea
128128

129129
Block content = diagLocalFileHeader.getContent();
130130
long size = zipEntry.getCompressedSize();
131-
long offs = content.getDiskOffs() + content.getSize();
131+
// TODO here we should use SrcZip methods
132+
long absOffs = content.getDiskOffs() + content.getSize();
132133

133134
EncryptionMethod encryptionMethod = zipEntry.getEncryptionMethod();
134135

135136
if (encryptionMethod.isAes()) {
136137
AesEncryptionHeaderBlock block = (AesEncryptionHeaderBlock) encryptionHeaderBlock;
137138

138-
offs += block.getSalt().getSize();
139-
offs += block.getPasswordChecksum().getSize();
139+
absOffs += block.getSalt().getSize();
140+
absOffs += block.getPasswordChecksum().getSize();
140141

141142
size -= block.getSalt().getSize();
142143
size -= block.getPasswordChecksum().getSize();
143144
size -= block.getMac().getSize();
144145
} else if (encryptionMethod == EncryptionMethod.PKWARE) {
145146
PkwareEncryptionHeaderBlock block = (PkwareEncryptionHeaderBlock) encryptionHeaderBlock;
146-
offs += block.getSize();
147+
absOffs += block.getSize();
147148
size -= block.getSize();
148149
}
149150

150-
Utils.copyLarge(blockModel.getZipModel(), dir.resolve("payload" + EXT_DATA), offs, size);
151+
Utils.copyLarge(blockModel.getZipModel(), dir.resolve("payload" + EXT_DATA), absOffs, absOffs, size);
151152
}
152153

153154
private EncryptionHeaderDecompose encryptionHeader(EncryptionMethod encryptionMethod,

src/main/java/ru/olegcherednik/zip4jvm/decompose/Utils.java

+6-14
Original file line numberDiff line numberDiff line change
@@ -50,30 +50,22 @@ public static void print(Path file, Consumer<PrintStream> consumer) throws FileN
5050
}
5151

5252
public static void copyLarge(ZipModel zipModel, Path out, Block block) throws IOException {
53-
copyLarge(zipModel, out, block.getDiskOffs(), block.getSize());
53+
copyLarge(zipModel, out, block.getDiskOffs(), block.getAbsOffs(), block.getSize());
5454
}
5555

56-
public static void copyLarge(ZipModel zipModel, Path out, long offs, long size) throws IOException {
57-
Path file = zipModel.getSrcZip().getDiskByAbsOffs(offs).getPath();
56+
public static void copyLarge(ZipModel zipModel, Path out, long diskOffs, long absOffs, long size)
57+
throws IOException {
58+
Path file = zipModel.getSrcZip().getDiskByAbsOffs(absOffs).getPath();
5859

5960
try (InputStream fis = Files.newInputStream(file);
6061
OutputStream fos = Files.newOutputStream(out)) {
61-
long skipBytes = fis.skip(offs);
62-
assert skipBytes == offs;
62+
long skipBytes = fis.skip(diskOffs);
63+
assert skipBytes == diskOffs;
6364

6465
IOUtils.copyLarge(fis, fos, 0, size);
6566
}
6667
}
6768

68-
public static void copyByteArray(Path out, byte[] buf, Block block) throws IOException {
69-
ValidationUtils.requireLessOrEqual(block.getAbsOffs(), Integer.MAX_VALUE, "block.absoluteOffs");
70-
ValidationUtils.requireLessOrEqual(block.getSize(), Integer.MAX_VALUE, "block.size");
71-
72-
try (OutputStream fos = Files.newOutputStream(out)) {
73-
fos.write(buf, (int) block.getAbsOffs(), (int) block.getSize());
74-
}
75-
}
76-
7769
public static void copyByteArray(Path out, byte[] buf) throws IOException {
7870
Files.write(out, buf);
7971
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package ru.olegcherednik.zip4jvm.engine.unzip;
20+
21+
import ru.olegcherednik.zip4jvm.io.in.DataInput;
22+
23+
import lombok.RequiredArgsConstructor;
24+
import org.apache.commons.io.IOUtils;
25+
26+
import java.util.List;
27+
import java.util.concurrent.CopyOnWriteArrayList;
28+
import java.util.function.Supplier;
29+
30+
/**
31+
* @param <T> {@link DataInput} definition
32+
* @author Oleg Cherednik
33+
* @since 28.12.2024
34+
*/
35+
@RequiredArgsConstructor
36+
public class DataInputThreadLocal<T extends DataInput> extends ThreadLocal<T> {
37+
38+
private final Supplier<T> dataInputSup;
39+
private final List<T> dataInputs = new CopyOnWriteArrayList<>();
40+
41+
public void release() {
42+
dataInputs.forEach(IOUtils::closeQuietly);
43+
dataInputs.clear();
44+
}
45+
46+
// ---------- ThreadLocal ----------
47+
48+
@Override
49+
public T get() {
50+
T in = super.get();
51+
52+
if (in == null) {
53+
in = dataInputSup.get();
54+
set(in);
55+
dataInputs.add(in);
56+
}
57+
58+
return in;
59+
}
60+
61+
}

src/main/java/ru/olegcherednik/zip4jvm/engine/unzip/UnzipEngine.java

+15-5
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,8 @@
2828
import ru.olegcherednik.zip4jvm.model.password.PasswordProvider;
2929
import ru.olegcherednik.zip4jvm.model.settings.UnzipSettings;
3030
import ru.olegcherednik.zip4jvm.model.src.SrcZip;
31+
import ru.olegcherednik.zip4jvm.utils.quitely.Quietly;
3132

32-
import java.io.IOException;
3333
import java.nio.file.Path;
3434
import java.util.Collection;
3535
import java.util.Collections;
@@ -47,7 +47,17 @@ public final class UnzipEngine implements ZipFile.Reader {
4747
public UnzipEngine(SrcZip srcZip, UnzipSettings settings) {
4848
PasswordProvider passwordProvider = settings.getPasswordProvider();
4949
zipModel = ZipModelBuilder.read(srcZip, settings.getCharsetCustomizer(), passwordProvider);
50-
unzipExtractEngine = new UnzipExtractEngine(passwordProvider, zipModel);
50+
unzipExtractEngine = createUnzipExtractEngine(settings, zipModel);
51+
}
52+
53+
private static UnzipExtractEngine createUnzipExtractEngine(UnzipSettings settings, ZipModel zipModel) {
54+
PasswordProvider passwordProvider = settings.getPasswordProvider();
55+
56+
if (settings.getAsyncThreads() == UnzipSettings.ASYNC_THREADS_OFF)
57+
return new UnzipExtractEngine(passwordProvider, zipModel);
58+
59+
int totalThreads = settings.getAsyncThreads();
60+
return new UnzipExtractAsyncEngine(passwordProvider, zipModel, totalThreads);
5161
}
5262

5363
// ---------- ZipFile.Reader ----------
@@ -105,9 +115,9 @@ public ZipFile.Entry next() {
105115
};
106116
}
107117

108-
public static RandomAccessDataInput createRandomAccessDataInput(SrcZip srcZip) throws IOException {
109-
return srcZip.isSolid() ? new SolidRandomAccessDataInput(srcZip)
110-
: new SplitRandomAccessDataInput(srcZip);
118+
public static RandomAccessDataInput createRandomAccessDataInput(SrcZip srcZip) {
119+
return Quietly.doRuntime(() -> srcZip.isSolid() ? new SolidRandomAccessDataInput(srcZip)
120+
: new SplitRandomAccessDataInput(srcZip));
111121
}
112122

113123
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package ru.olegcherednik.zip4jvm.engine.unzip;
20+
21+
import ru.olegcherednik.zip4jvm.io.in.file.consecutive.ConsecutiveAccessDataInput;
22+
import ru.olegcherednik.zip4jvm.model.ZipModel;
23+
import ru.olegcherednik.zip4jvm.model.entry.ZipEntry;
24+
import ru.olegcherednik.zip4jvm.model.password.PasswordProvider;
25+
import ru.olegcherednik.zip4jvm.utils.quitely.Quietly;
26+
import ru.olegcherednik.zip4jvm.utils.quitely.functions.RunnableWithException;
27+
28+
import org.apache.commons.collections4.CollectionUtils;
29+
30+
import java.nio.file.Path;
31+
import java.util.Iterator;
32+
import java.util.LinkedList;
33+
import java.util.List;
34+
import java.util.Set;
35+
import java.util.concurrent.CompletableFuture;
36+
import java.util.concurrent.Executor;
37+
import java.util.concurrent.ExecutorService;
38+
import java.util.concurrent.ForkJoinPool;
39+
import java.util.concurrent.ForkJoinWorkerThread;
40+
import java.util.concurrent.atomic.AtomicInteger;
41+
42+
/**
43+
* @author Oleg Cherednik
44+
* @since 28.12.2024
45+
*/
46+
public class UnzipExtractAsyncEngine extends UnzipExtractEngine {
47+
48+
protected final int totalThreads;
49+
50+
public UnzipExtractAsyncEngine(PasswordProvider passwordProvider, ZipModel zipModel, int totalThreads) {
51+
super(passwordProvider, zipModel);
52+
this.totalThreads = totalThreads <= 0 ? Runtime.getRuntime().availableProcessors() : totalThreads;
53+
}
54+
55+
// ---------- UnzipExtractEngine ----------
56+
57+
@Override
58+
protected void extractAllEntries(Path dstDir) {
59+
List<CompletableFuture<Void>> tasks = new LinkedList<>();
60+
Iterator<ZipEntry> it = zipModel.absOffsAscIterator();
61+
62+
DataInputThreadLocal<ConsecutiveAccessDataInput> threadLocalDataInput =
63+
new DataInputThreadLocal<>(this::createConsecutiveDataInput);
64+
ExecutorService executor = createExecutor();
65+
66+
try {
67+
while (it.hasNext()) {
68+
ZipEntry zipEntry = it.next();
69+
Path file = dstDir.resolve(zipEntry.getFileName());
70+
71+
CompletableFuture<Void> task = createCompletableFuture(
72+
() -> extractEntry(file, zipEntry, threadLocalDataInput.get()), executor);
73+
74+
tasks.add(task);
75+
}
76+
77+
tasks.forEach(CompletableFuture::join);
78+
} finally {
79+
threadLocalDataInput.release();
80+
executor.shutdown();
81+
}
82+
}
83+
84+
@Override
85+
protected void extractEntryByPrefix(Path dstDir, Set<String> prefixes) {
86+
assert CollectionUtils.isNotEmpty(prefixes);
87+
88+
List<CompletableFuture<Void>> tasks = new LinkedList<>();
89+
Iterator<ZipEntry> it = zipModel.absOffsAscIterator();
90+
91+
DataInputThreadLocal<ConsecutiveAccessDataInput> threadLocalDataInput =
92+
new DataInputThreadLocal<>(this::createConsecutiveDataInput);
93+
ExecutorService executor = createExecutor();
94+
95+
try {
96+
while (it.hasNext()) {
97+
ZipEntry zipEntry = it.next();
98+
String fileName = getFileName(zipEntry, prefixes);
99+
100+
if (fileName != null) {
101+
Path file = dstDir.resolve(fileName);
102+
CompletableFuture<Void> task = createCompletableFuture(
103+
() -> extractEntry(file, zipEntry, threadLocalDataInput.get()), executor);
104+
105+
tasks.add(task);
106+
}
107+
}
108+
109+
tasks.forEach(CompletableFuture::join);
110+
} finally {
111+
threadLocalDataInput.release();
112+
executor.shutdown();
113+
}
114+
}
115+
116+
// ----------
117+
118+
protected ExecutorService createExecutor() {
119+
AtomicInteger counter = new AtomicInteger();
120+
String format = String.format("zip4jvm-extract-%%0%dd", String.valueOf(totalThreads).length());
121+
122+
ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
123+
ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
124+
thread.setName(String.format(format, counter.incrementAndGet()));
125+
return thread;
126+
};
127+
128+
return new ForkJoinPool(totalThreads, factory, null, false);
129+
}
130+
131+
protected CompletableFuture<Void> createCompletableFuture(RunnableWithException task, Executor executor) {
132+
return CompletableFuture.runAsync(() -> Quietly.doRuntime(task), executor);
133+
}
134+
135+
}

0 commit comments

Comments
 (0)