Skip to content

Commit 8853980

Browse files
oleg-cherednikoleg-cherednik
committed
#20 Extract entries async
1 parent 41d6959 commit 8853980

File tree

9 files changed

+263
-77
lines changed

9 files changed

+263
-77
lines changed

‎src/main/java/ru/olegcherednik/zip4jvm/engine/unzip/DataInputThreadLocal.java

+61
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

+17-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,19 @@ 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+
System.out.println(unzipExtractEngine.getClass().getSimpleName());
53+
}
54+
55+
private static UnzipExtractEngine createUnzipExtractEngine(UnzipSettings settings, ZipModel zipModel) {
56+
PasswordProvider passwordProvider = settings.getPasswordProvider();
57+
58+
if (settings.getAsyncThreads() == UnzipSettings.ASYNC_THREADS_OFF)
59+
return new UnzipExtractEngine(passwordProvider, zipModel);
60+
61+
int totalThreads = settings.getAsyncThreads();
62+
return new UnzipExtractAsyncEngine(passwordProvider, zipModel, totalThreads);
5163
}
5264

5365
// ---------- ZipFile.Reader ----------
@@ -105,9 +117,9 @@ public ZipFile.Entry next() {
105117
};
106118
}
107119

108-
public static RandomAccessDataInput createRandomAccessDataInput(SrcZip srcZip) throws IOException {
109-
return srcZip.isSolid() ? new SolidRandomAccessDataInput(srcZip)
110-
: new SplitRandomAccessDataInput(srcZip);
120+
public static RandomAccessDataInput createRandomAccessDataInput(SrcZip srcZip) {
121+
return Quietly.doRuntime(() -> srcZip.isSolid() ? new SolidRandomAccessDataInput(srcZip)
122+
: new SplitRandomAccessDataInput(srcZip));
111123
}
112124

113125
}

‎src/main/java/ru/olegcherednik/zip4jvm/engine/unzip/UnzipExtractAsyncEngine.java

+108
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
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 java.nio.file.Path;
29+
import java.util.Iterator;
30+
import java.util.LinkedList;
31+
import java.util.List;
32+
import java.util.Map;
33+
import java.util.Optional;
34+
import java.util.concurrent.CompletableFuture;
35+
import java.util.concurrent.Executor;
36+
import java.util.concurrent.ExecutorService;
37+
import java.util.concurrent.ForkJoinPool;
38+
import java.util.concurrent.ForkJoinWorkerThread;
39+
import java.util.concurrent.atomic.AtomicInteger;
40+
41+
/**
42+
* @author Oleg Cherednik
43+
* @since 28.12.2024
44+
*/
45+
public class UnzipExtractAsyncEngine extends UnzipExtractEngine {
46+
47+
protected final int totalThreads;
48+
49+
public UnzipExtractAsyncEngine(PasswordProvider passwordProvider, ZipModel zipModel, int totalThreads) {
50+
super(passwordProvider, zipModel);
51+
this.totalThreads = totalThreads;
52+
}
53+
54+
// ---------- UnzipExtractEngine ----------
55+
56+
@Override
57+
protected void extractEntry(Path dstDir, Map<String, String> map) {
58+
List<CompletableFuture<Void>> tasks = new LinkedList<>();
59+
Iterator<ZipEntry> it = zipModel.absOffsAscIterator();
60+
61+
DataInputThreadLocal<ConsecutiveAccessDataInput> threadLocalDataInput =
62+
new DataInputThreadLocal<>(this::createConsecutiveDataInput);
63+
ExecutorService executor = createExecutor();
64+
65+
try {
66+
while (it.hasNext()) {
67+
ZipEntry zipEntry = it.next();
68+
69+
if (map != null && !map.containsKey(zipEntry.getFileName()))
70+
continue;
71+
72+
String fileName = Optional.ofNullable(map)
73+
.map(m -> m.get(zipEntry.getFileName()))
74+
.orElse(zipEntry.getFileName());
75+
Path file = dstDir.resolve(fileName);
76+
77+
CompletableFuture<Void> task = createCompletableFuture(
78+
() -> extractEntry(file, zipEntry, threadLocalDataInput.get()), executor);
79+
80+
tasks.add(task);
81+
}
82+
83+
tasks.forEach(CompletableFuture::join);
84+
} finally {
85+
threadLocalDataInput.release();
86+
executor.shutdown();
87+
}
88+
}
89+
90+
protected ExecutorService createExecutor() {
91+
AtomicInteger counter = new AtomicInteger();
92+
int parallelism = totalThreads <= 0 ? Runtime.getRuntime().availableProcessors() : totalThreads;
93+
String format = String.format("zip4jvm-extract-%%0%dd", String.valueOf(parallelism).length());
94+
95+
ForkJoinPool.ForkJoinWorkerThreadFactory factory = pool -> {
96+
ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
97+
thread.setName(String.format(format, counter.incrementAndGet()));
98+
return thread;
99+
};
100+
101+
return new ForkJoinPool(parallelism, factory, null, false);
102+
}
103+
104+
protected CompletableFuture<Void> createCompletableFuture(RunnableWithException task, Executor executor) {
105+
return CompletableFuture.runAsync(() -> Quietly.doRuntime(task), executor);
106+
}
107+
108+
}

‎src/main/java/ru/olegcherednik/zip4jvm/engine/unzip/UnzipExtractEngine.java

+25-25
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import ru.olegcherednik.zip4jvm.model.password.PasswordProvider;
3232
import ru.olegcherednik.zip4jvm.model.src.SrcZip;
3333
import ru.olegcherednik.zip4jvm.utils.ZipUtils;
34+
import ru.olegcherednik.zip4jvm.utils.quitely.Quietly;
3435
import ru.olegcherednik.zip4jvm.utils.time.DosTimestampConverterUtils;
3536

3637
import lombok.RequiredArgsConstructor;
@@ -105,31 +106,31 @@ protected List<ZipEntry> getEntriesByPrefix(String prefix) {
105106
.collect(Collectors.toList());
106107
}
107108

108-
// ----------
109-
110109
protected void extractEntry(Path dstDir, Map<String, String> map) {
111-
try (ConsecutiveAccessDataInput in = createConsecutiveDataInput(zipModel.getSrcZip())) {
110+
try (ConsecutiveAccessDataInput in = createConsecutiveDataInput()) {
112111
Iterator<ZipEntry> it = zipModel.absOffsAscIterator();
113112

114113
while (it.hasNext()) {
115114
ZipEntry zipEntry = it.next();
116115

117-
if (map == null || map.containsKey(zipEntry.getFileName())) {
118-
in.seekForward(zipEntry.getLocalFileHeaderAbsOffs());
116+
if (map != null && !map.containsKey(zipEntry.getFileName()))
117+
continue;
118+
119+
String fileName = Optional.ofNullable(map)
120+
.map(m -> m.get(zipEntry.getFileName()))
121+
.orElse(zipEntry.getFileName());
122+
Path file = dstDir.resolve(fileName);
119123

120-
String fileName = Optional.ofNullable(map)
121-
.map(m -> m.get(zipEntry.getFileName()))
122-
.orElse(zipEntry.getFileName());
123-
Path file = dstDir.resolve(fileName);
124-
extractEntry(file, zipEntry, in);
125-
}
124+
extractEntry(file, zipEntry, in);
126125
}
127126
} catch (IOException e) {
128127
throw new Zip4jvmException(e);
129128
}
130129
}
131130

132-
protected void extractEntry(Path file, ZipEntry zipEntry, DataInput in) throws IOException {
131+
protected void extractEntry(Path file, ZipEntry zipEntry, ConsecutiveAccessDataInput in) throws IOException {
132+
in.seekForward(zipEntry.getLocalFileHeaderAbsOffs());
133+
133134
if (zipEntry.isSymlink())
134135
extractSymlink(file, zipEntry, in);
135136
else if (zipEntry.isDirectory())
@@ -142,7 +143,7 @@ else if (zipEntry.isDirectory())
142143
setFileLastModifiedTime(file, zipEntry);
143144
}
144145

145-
protected static void extractSymlink(Path symlink, ZipEntry zipEntry, DataInput in) throws IOException {
146+
protected void extractSymlink(Path symlink, ZipEntry zipEntry, DataInput in) throws IOException {
146147
String target = IOUtils.toString(zipEntry.createInputStream(in), Charsets.UTF_8);
147148

148149
if (target.startsWith("/"))
@@ -154,7 +155,7 @@ else if (target.contains(":"))
154155
ZipSymlinkEngine.createRelativeSymlink(symlink, symlink.getParent().resolve(target));
155156
}
156157

157-
protected static void extractEmptyDirectory(Path dir) throws IOException {
158+
protected void extractEmptyDirectory(Path dir) throws IOException {
158159
Files.createDirectories(dir);
159160
}
160161

@@ -164,17 +165,24 @@ protected void extractRegularFile(Path file, ZipEntry zipEntry, DataInput in) th
164165
ZipUtils.copyLarge(zipEntry.createInputStream(in), getOutputStream(file));
165166
}
166167

167-
protected static void setFileAttributes(Path path, ZipEntry zipEntry) throws IOException {
168+
public ConsecutiveAccessDataInput createConsecutiveDataInput() {
169+
SrcZip srcZip = zipModel.getSrcZip();
170+
171+
return Quietly.doRuntime(() -> srcZip.isSolid() ? new SolidConsecutiveAccessDataInput(srcZip)
172+
: new SplitConsecutiveAccessDataInput(srcZip));
173+
}
174+
175+
protected void setFileAttributes(Path path, ZipEntry zipEntry) throws IOException {
168176
if (zipEntry.getExternalFileAttributes() != null)
169177
zipEntry.getExternalFileAttributes().apply(path);
170178
}
171179

172-
private static void setFileLastModifiedTime(Path path, ZipEntry zipEntry) throws IOException {
180+
protected void setFileLastModifiedTime(Path path, ZipEntry zipEntry) throws IOException {
173181
long lastModifiedTime = DosTimestampConverterUtils.dosToJavaTime(zipEntry.getLastModifiedTime());
174182
Files.setLastModifiedTime(path, FileTime.fromMillis(lastModifiedTime));
175183
}
176184

177-
protected static OutputStream getOutputStream(Path file) throws IOException {
185+
protected OutputStream getOutputStream(Path file) throws IOException {
178186
Path parent = file.getParent();
179187

180188
if (!Files.exists(parent))
@@ -184,12 +192,4 @@ protected static OutputStream getOutputStream(Path file) throws IOException {
184192
return Files.newOutputStream(file);
185193
}
186194

187-
// ---------- static ----------
188-
189-
public static ConsecutiveAccessDataInput createConsecutiveDataInput(SrcZip srcZip) throws IOException {
190-
return srcZip.isSolid() ? new SolidConsecutiveAccessDataInput(srcZip)
191-
: new SplitConsecutiveAccessDataInput(srcZip);
192-
193-
}
194-
195195
}

‎src/main/java/ru/olegcherednik/zip4jvm/io/in/file/consecutive/SolidConsecutiveAccessDataInput.java

+2
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@ public class SolidConsecutiveAccessDataInput extends BaseConsecutiveAccessDataIn
4242
private final InputStream in;
4343

4444
public SolidConsecutiveAccessDataInput(SrcZip srcZip) throws IOException {
45+
System.out.println(Thread.currentThread().getName());
46+
4547
byteOrder = srcZip.getByteOrder();
4648
in = new BufferedInputStream(Files.newInputStream(srcZip.getDiskByNo(0).getPath()));
4749
}

0 commit comments

Comments
 (0)