Skip to content

Commit ac08669

Browse files
itzgitzg
authored
Delegate file writing via queue and option to disable HTTP/2 (#621)
1 parent e861fdd commit ac08669

File tree

7 files changed

+140
-42
lines changed

7 files changed

+140
-42
lines changed

‎.run/modrinth-modpack (slug).run.xml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,16 @@
1212
<option name="ENABLED" value="true" />
1313
</pattern>
1414
</extension>
15+
<extension name="net.ashald.envfile">
16+
<option name="IS_ENABLED" value="false" />
17+
<option name="IS_SUBST" value="false" />
18+
<option name="IS_PATH_MACRO_SUPPORTED" value="false" />
19+
<option name="IS_IGNORE_MISSING_FILES" value="false" />
20+
<option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
21+
<ENTRIES>
22+
<ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
23+
</ENTRIES>
24+
</extension>
1525
<method v="2">
1626
<option name="Make" enabled="true" />
1727
</method>

‎src/main/java/me/itzg/helpers/curseforge/CurseForgeInstaller.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,8 @@ else if (Manifests.allFilesPresent(outputDir, context.prevInstallManifest, ignor
384384
}
385385
}
386386
else {
387+
log.info("Downloading modpack zip for {}", modFile.getDisplayName());
388+
387389
modpackZip = context.cfApi.downloadTemp(modFile, ".zip",
388390
(status, uri, file) ->
389391
log.debug("Modpack file retrieval: status={} uri={} file={}", status, uri, file)

‎src/main/java/me/itzg/helpers/fabric/FabricMetaClient.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -208,7 +208,7 @@ private static void debugDownloadedContent(Path path) {
208208
log.debug("Downloaded launcher jar content starts with: {}",
209209
Hex.encodeHexString(ByteBuffer.wrap(buf, 0, amount))
210210
);
211-
} catch (IOException e) {
211+
} catch (IOException|IndexOutOfBoundsException e) {
212212
log.warn("Failed to debug content of launcher jar", e);
213213
}
214214
}

‎src/main/java/me/itzg/helpers/files/ByteBufQueue.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
package me.itzg.helpers.files;
2+
3+
import io.netty.buffer.ByteBuf;
4+
import java.util.LinkedList;
5+
import java.util.concurrent.locks.Condition;
6+
import java.util.concurrent.locks.Lock;
7+
import java.util.concurrent.locks.ReentrantLock;
8+
9+
class ByteBufQueue {
10+
11+
final Lock lock = new ReentrantLock();
12+
final Condition readyOrFinished = lock.newCondition();
13+
final LinkedList<ByteBuf> buffers = new LinkedList<>();
14+
boolean finished = false;
15+
16+
public void add(ByteBuf buf) {
17+
lock.lock();
18+
try {
19+
buffers.add(buf);
20+
} finally {
21+
readyOrFinished.signal();
22+
lock.unlock();
23+
}
24+
}
25+
26+
public ByteBuf take() {
27+
while (true) {
28+
lock.lock();
29+
30+
try {
31+
if (!buffers.isEmpty()) {
32+
return buffers.removeFirst();
33+
}
34+
else if (finished) {
35+
return null;
36+
}
37+
readyOrFinished.awaitUninterruptibly();
38+
} finally {
39+
lock.unlock();
40+
}
41+
}
42+
}
43+
44+
public void finish() {
45+
lock.lock();
46+
try {
47+
finished = true;
48+
readyOrFinished.signal();
49+
} finally {
50+
lock.unlock();
51+
}
52+
}
53+
}

‎src/main/java/me/itzg/helpers/files/ReactiveFileUtils.java‎

Lines changed: 39 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,17 @@
11
package me.itzg.helpers.files;
22

3-
import java.io.IOException;
3+
import io.netty.buffer.ByteBuf;
44
import java.nio.channels.FileChannel;
55
import java.nio.file.Files;
66
import java.nio.file.Path;
77
import java.nio.file.StandardOpenOption;
88
import java.time.Instant;
99
import java.util.stream.Collectors;
1010
import lombok.extern.slf4j.Slf4j;
11-
import reactor.core.Exceptions;
1211
import reactor.core.publisher.Mono;
1312
import reactor.core.scheduler.Schedulers;
1413
import reactor.netty.ByteBufFlux;
14+
import reactor.util.function.Tuple2;
1515

1616
@Slf4j
1717
public class ReactiveFileUtils {
@@ -42,36 +42,46 @@ public static Mono<Path> createDirectories(Path dir) {
4242
}
4343

4444
public static Mono<Long> writeByteBufFluxToFile(ByteBufFlux byteBufFlux, Path file) {
45-
return Mono.fromCallable(() -> {
46-
log.trace("Opening {} for writing", file);
47-
return FileChannel.open(file,
48-
StandardOpenOption.WRITE,
49-
StandardOpenOption.CREATE,
50-
StandardOpenOption.TRUNCATE_EXISTING
51-
);
52-
}
53-
)
54-
.subscribeOn(Schedulers.boundedElastic())
55-
.flatMap(outChannel ->
56-
byteBufFlux
57-
.asByteBuffer()
58-
.subscribeOn(Schedulers.boundedElastic())
59-
.<Integer>handle((byteBuffer, sink) -> {
60-
try {
61-
sink.next(outChannel.write(byteBuffer));
62-
} catch (IOException e) {
63-
sink.error(Exceptions.propagate(e));
45+
final ByteBufQueue byteBufQueue = new ByteBufQueue();
46+
47+
// Separate this into a pair of concurrent mono's
48+
return Mono.zip(
49+
// ...file writer
50+
Mono.fromCallable(() -> {
51+
try (FileChannel channel = FileChannel.open(file,
52+
StandardOpenOption.WRITE,
53+
StandardOpenOption.CREATE,
54+
StandardOpenOption.TRUNCATE_EXISTING
55+
)) {
56+
ByteBuf byteBuf;
57+
while ((byteBuf = byteBufQueue.take()) != null) {
58+
try {
59+
//noinspection ResultOfMethodCallIgnored
60+
channel.write(byteBuf.nioBuffer());
61+
} finally {
62+
byteBuf.release();
63+
}
64+
}
65+
66+
return file;
6467
}
6568
})
66-
.doOnTerminate(() -> {
67-
try {
68-
outChannel.close();
69-
log.trace("Closed {}", file);
70-
} catch (IOException e) {
71-
log.warn("Failed to close {}", file, e);
72-
}
69+
// ...which runs in a separate thread
70+
.subscribeOn(Schedulers.boundedElastic()),
71+
// ...and the network consumer flux
72+
byteBufFlux
73+
// Mark the bytebufs as retained so they can be released after
74+
// they are written by the mono above
75+
.retain()
76+
.map(byteBuf -> {
77+
final int amount = byteBuf.readableBytes();
78+
byteBufQueue.add(byteBuf);
79+
return amount;
7380
})
81+
.doOnTerminate(byteBufQueue::finish)
7482
.collect(Collectors.<Integer>summingLong(value -> value))
75-
);
83+
)
84+
// Just expose the total bytes read from network
85+
.map(Tuple2::getT2);
7686
}
7787
}

‎src/main/java/me/itzg/helpers/http/SharedFetch.java‎

Lines changed: 28 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import lombok.extern.slf4j.Slf4j;
1313
import me.itzg.helpers.McImageHelper;
1414
import me.itzg.helpers.errors.GenericException;
15+
import reactor.netty.http.Http11SslContextSpec;
1516
import reactor.netty.http.Http2SslContextSpec;
1617
import reactor.netty.http.HttpProtocol;
1718
import reactor.netty.http.client.HttpClient;
@@ -60,8 +61,6 @@ public SharedFetch(String forCommand, Options options) {
6061

6162
reactiveClient = HttpClient.create(connectionProvider)
6263
.proxyWithSystemProperties()
63-
// https://projectreactor.io/docs/netty/release/reference/http-client.html#HTTP2
64-
.protocol(HttpProtocol.HTTP11, HttpProtocol.H2)
6564
.headers(headers -> {
6665
headers
6766
.set(HttpHeaderNames.USER_AGENT.toString(), userAgent)
@@ -72,15 +71,29 @@ public SharedFetch(String forCommand, Options options) {
7271
}
7372
)
7473
// Reference https://projectreactor.io/docs/netty/release/reference/index.html#response-timeout
75-
.responseTimeout(options.getResponseTimeout())
76-
.secure(spec ->
77-
// Http2 SSL supports both HTTP/2 and HTTP/1.1
78-
spec.sslContext((GenericSslContextSpec<?>) Http2SslContextSpec.forClient())
79-
// Reference https://projectreactor.io/docs/netty/release/reference/index.html#ssl-tls-timeout
80-
.handshakeTimeout(options.getTlsHandshakeTimeout())
81-
)
82-
83-
;
74+
.responseTimeout(options.getResponseTimeout());
75+
76+
if (options.isUseHttp2()) {
77+
log.debug("Using HTTP/2");
78+
reactiveClient
79+
// https://projectreactor.io/docs/netty/release/reference/http-client.html#HTTP2
80+
.protocol(HttpProtocol.HTTP11, HttpProtocol.H2)
81+
.secure(spec ->
82+
// Http2 SSL supports both HTTP/2 and HTTP/1.1
83+
spec.sslContext((GenericSslContextSpec<?>) Http2SslContextSpec.forClient())
84+
// Reference https://projectreactor.io/docs/netty/release/reference/index.html#ssl-tls-timeout
85+
.handshakeTimeout(options.getTlsHandshakeTimeout())
86+
);
87+
}
88+
else {
89+
log.debug("Using HTTP/1.1");
90+
reactiveClient
91+
.secure(spec ->
92+
spec.sslContext((GenericSslContextSpec<?>) Http11SslContextSpec.forClient())
93+
// Reference https://projectreactor.io/docs/netty/release/reference/index.html#ssl-tls-timeout
94+
.handshakeTimeout(options.getTlsHandshakeTimeout())
95+
);
96+
}
8497

8598
headers.put("x-fetch-session", fetchSessionId);
8699

@@ -135,14 +148,17 @@ public static class Options {
135148
*/
136149
private final URI filesViaUrl;
137150

151+
@Default
152+
private final boolean useHttp2 = true;
153+
138154
public Options withHeader(String key, String value) {
139155
final Map<String, String> newHeaders = extraHeaders != null ?
140156
new HashMap<>(extraHeaders) : new HashMap<>();
141157
newHeaders.put(key, value);
142158

143159
return new Options(
144160
responseTimeout, tlsHandshakeTimeout, maxIdleTimeout, pendingAcquireTimeout,
145-
newHeaders, filesViaUrl
161+
newHeaders, filesViaUrl, useHttp2
146162
);
147163
}
148164
}

‎src/main/java/me/itzg/helpers/http/SharedFetchArgs.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,13 @@ public void setPendingAcquireTimeout(Duration timeout) {
4848
optionsBuilder.pendingAcquireTimeout(timeout);
4949
}
5050

51+
@Option(names = "--use-http2", defaultValue = "${env:FETCH_USE_HTTP2:-true}",
52+
description = "Whether to use HTTP/2. Default: ${DEFAULT-VALUE}"
53+
)
54+
public void setUseHttp2(boolean useHttp2) {
55+
optionsBuilder.useHttp2(useHttp2);
56+
}
57+
5158
public Options options() {
5259
return optionsBuilder.build();
5360
}

0 commit comments

Comments
 (0)