diff --git a/build/tasks/src/org/jetbrains/intellij/build/io/zip.kt b/build/tasks/src/org/jetbrains/intellij/build/io/zip.kt index 3261aa280f6a..2a039e08b9a0 100644 --- a/build/tasks/src/org/jetbrains/intellij/build/io/zip.kt +++ b/build/tasks/src/org/jetbrains/intellij/build/io/zip.kt @@ -14,15 +14,19 @@ enum class AddDirEntriesMode { ALL } +// `createFileParentDirs = false` can be used for performance reasons if you do create a lot of zip files fun zipWithCompression( targetFile: Path, dirs: Map, compressionLevel: Int = Deflater.DEFAULT_COMPRESSION, addDirEntriesMode: AddDirEntriesMode = AddDirEntriesMode.NONE, overwrite: Boolean = false, + createFileParentDirs: Boolean = true, fileFilter: ((name: String) -> Boolean)? = null, ) { - Files.createDirectories(targetFile.parent) + if (createFileParentDirs) { + Files.createDirectories(targetFile.parent) + } ZipFileWriter( channel = FileChannel.open(targetFile, if (overwrite) W_OVERWRITE else W_CREATE_NEW), deflater = if (compressionLevel == Deflater.NO_COMPRESSION) null else Deflater(compressionLevel, true), diff --git a/build/tasks/src/org/jetbrains/intellij/build/logging/BuildMessageLoggerBase.kt b/build/tasks/src/org/jetbrains/intellij/build/logging/BuildMessageLoggerBase.kt index b5452a626006..783d863b1f50 100644 --- a/build/tasks/src/org/jetbrains/intellij/build/logging/BuildMessageLoggerBase.kt +++ b/build/tasks/src/org/jetbrains/intellij/build/logging/BuildMessageLoggerBase.kt @@ -22,8 +22,7 @@ abstract class BuildMessageLoggerBase : BuildMessageLogger() { Span.current().addEvent("artifact built: ${message.text}") } LogMessage.Kind.COMPILATION_ERRORS -> { - val errorsString = (message as CompilationErrorsLogMessage).errorMessages.joinToString(separator = "\n") - Span.current().addEvent("compilation errors (${message.compilerName}):\n$errorsString") + // reported as span event } LogMessage.Kind.BUILD_CANCEL, LogMessage.Kind.BUILD_PROBLEM -> { throw BuildScriptsLoggedError(message.text) diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/BuildOptions.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/BuildOptions.kt index b7d94df60c76..e8700d674c38 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/BuildOptions.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/BuildOptions.kt @@ -7,6 +7,7 @@ import kotlinx.collections.immutable.persistentListOf import kotlinx.collections.immutable.toPersistentMap import org.jetbrains.annotations.ApiStatus import org.jetbrains.annotations.TestOnly +import org.jetbrains.intellij.build.BuildOptions.Companion.BUILD_STEPS_TO_SKIP_PROPERTY import org.jetbrains.intellij.build.BuildPaths.Companion.COMMUNITY_ROOT import org.jetbrains.intellij.build.dependencies.DependenciesProperties import org.jetbrains.intellij.build.dependencies.TeamCityHelper @@ -15,6 +16,8 @@ import java.nio.file.Path import java.util.* import java.util.concurrent.TimeUnit import kotlin.random.Random +import kotlin.time.Duration +import kotlin.time.Duration.Companion.minutes data class BuildOptions( @ApiStatus.Internal @JvmField val jarCacheDir: Path? = null, @@ -99,6 +102,11 @@ data class BuildOptions( * If `true`, the project modules will be compiled incrementally. */ var incrementalCompilation: Boolean = getBooleanProperty(INTELLIJ_BUILD_INCREMENTAL_COMPILATION), + + /** + * Full rebuild will be triggered if this timeout is exceeded for incremental compilation. + */ + val incrementalCompilationTimeout: Duration? = System.getProperty("intellij.build.incremental.compilation.timeoutMin")?.toLong()?.minutes, ) { companion object { /** @@ -385,11 +393,6 @@ data class BuildOptions( */ var incrementalCompilationFallbackRebuild: Boolean = getBooleanProperty(INCREMENTAL_COMPILATION_FALLBACK_REBUILD_PROPERTY, true) - /** - * Full rebuild will be triggered if this timeout is exceeded for incremental compilation. - */ - val incrementalCompilationTimeout: Long = System.getProperty("intellij.build.incremental.compilation.timeoutMin")?.toLong() ?: Long.MAX_VALUE - /** * Use [BuildContext.buildNumber] to get the actual build number in build scripts. * @see BuildContext.reportDistributionBuildNumber diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/classpath.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/classpath.kt index 9c540c262aa0..e8d5e836d92a 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/classpath.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/classpath.kt @@ -150,7 +150,7 @@ fun reorderJar(jarFile: Path, orderedNames: List) { packageIndexBuilder.addFile(entry.name) zipCreator.uncompressedData( nameString = entry.name, - data = entry.getByteBuffer(sourceZip), + data = entry.getByteBuffer(sourceZip, null), indexWriter = packageIndexBuilder.indexWriter, ) } diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnection.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnection.kt index 2889bbc22ee5..81ecbfee586e 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnection.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnection.kt @@ -3,20 +3,16 @@ package org.jetbrains.intellij.build.http2Client -import io.netty.buffer.ByteBufUtil import io.netty.channel.ChannelHandlerContext import io.netty.handler.codec.http.HttpHeaderNames -import io.netty.handler.codec.http.HttpHeaderValues import io.netty.handler.codec.http.HttpMethod import io.netty.handler.codec.http.HttpResponseStatus import io.netty.handler.codec.http2.Http2Headers import io.netty.handler.codec.http2.Http2HeadersFrame import io.netty.handler.codec.http2.ReadOnlyHttp2Headers import io.netty.util.AsciiString -import kotlinx.serialization.DeserializationStrategy -import kotlinx.serialization.serializer -// https://cabulous.medium.com/http-2-and-how-it-works-9f645458e4b2 +// user-facing API on top of implementation API Http2ConnectionProvider internal class Http2ClientConnection internal constructor( private val scheme: AsciiString, private val authority: AsciiString, @@ -33,30 +29,13 @@ internal class Http2ClientConnection internal constructor( connection.close() } - internal suspend inline fun post(path: String, data: String, contentType: AsciiString): T { - return post(path = AsciiString.of(path), data = data, contentType = contentType, deserializer = serializer()) - } - - private suspend fun post(path: AsciiString, data: String, contentType: AsciiString, deserializer: DeserializationStrategy): T { - return connection.stream { stream, result -> - val bufferAllocator = stream.alloc() - stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = bufferAllocator, deserializer = deserializer)) - - stream.writeHeaders( - headers = ReadOnlyHttp2Headers.clientHeaders( - true, - HttpMethod.POST.asciiName(), - path, - scheme, - authority, - HttpHeaderNames.CONTENT_TYPE, contentType, - HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP, - *commonHeaders - ), - endStream = false, - ) - stream.writeData(ByteBufUtil.writeUtf8(bufferAllocator, data), endStream = true) - } + fun withAuth(authHeader: CharSequence): Http2ClientConnection { + return Http2ClientConnection( + scheme = scheme, + authority = scheme, + commonHeaders = commonHeaders + arrayOf(HttpHeaderNames.AUTHORIZATION, AsciiString.of(authHeader)), + connection = connection, + ) } suspend fun head(path: CharSequence): HttpResponseStatus { @@ -92,8 +71,11 @@ internal class Http2ClientConnection internal constructor( } } - - internal fun createHeaders(method: HttpMethod, path: AsciiString): Http2Headers { - return ReadOnlyHttp2Headers.clientHeaders(true, method.asciiName(), path, scheme, authority, *commonHeaders) + internal fun createHeaders(method: HttpMethod, path: AsciiString, vararg extraHeaders: AsciiString): Http2Headers { + var otherHeaders = commonHeaders + if (extraHeaders.isNotEmpty()) { + otherHeaders += extraHeaders + } + return ReadOnlyHttp2Headers.clientHeaders(true, method.asciiName(), path, scheme, authority, *otherHeaders) } } \ No newline at end of file diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnectionFactory.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnectionFactory.kt index 842adcb5380f..1e595f258eda 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnectionFactory.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ClientConnectionFactory.kt @@ -62,7 +62,7 @@ internal suspend fun withHttp2ClientConnectionFactory( useSsl: Boolean = true, trustAll: Boolean = false, block: suspend (Http2ClientConnectionFactory) -> T, -) { +): T { var connection: Http2ClientConnectionFactory? = null try { return supervisorScope { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ConnectionProvider.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ConnectionProvider.kt index 1076ec9e1089..1f8bd3333c32 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ConnectionProvider.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2ConnectionProvider.kt @@ -10,6 +10,8 @@ import io.netty.channel.Channel import io.netty.channel.ChannelHandlerContext import io.netty.channel.ChannelInitializer import io.netty.channel.SimpleChannelInboundHandler +import io.netty.handler.codec.http.HttpResponseStatus +import io.netty.handler.codec.http.HttpStatusClass import io.netty.handler.codec.http2.* import io.netty.handler.ssl.SslContext import io.opentelemetry.api.common.AttributeKey @@ -29,6 +31,8 @@ import kotlin.random.asJavaRandom import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Duration.Companion.seconds +internal class UnexpectedHttpStatus(@JvmField val status: HttpResponseStatus) : RuntimeException("Unexpected HTTP response status: $status") + private class ConnectionState( @JvmField val bootstrap: Http2StreamChannelBootstrap, @JvmField val channel: Channel, @@ -37,6 +41,7 @@ private class ConnectionState( private const val MAX_ATTEMPTS = 2 +// https://cabulous.medium.com/http-2-and-how-it-works-9f645458e4b2 // https://stackoverflow.com/questions/55087292/how-to-handle-http-2-goaway-with-java-net-httpclient // Server can send GOAWAY frame after X streams, that's why we need manager for channel - open a new one in case of such error internal class Http2ConnectionProvider( @@ -152,6 +157,12 @@ internal class Http2ConnectionProvider( continue } } + catch (e: UnexpectedHttpStatus) { + // retry only for sever errors + if (e.status.codeClass() != HttpStatusClass.SERVER_ERROR) { + throw e + } + } catch (e: Throwable) { if (attemptIndex >= MAX_ATTEMPTS) { if (suppressedExceptions != null) { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2StreamJsonInboundHandler.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2StreamJsonInboundHandler.kt index a6bfa899ef4c..92b38f590748 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2StreamJsonInboundHandler.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/Http2StreamJsonInboundHandler.kt @@ -3,27 +3,101 @@ package org.jetbrains.intellij.build.http2Client import io.netty.buffer.ByteBufAllocator import io.netty.buffer.ByteBufInputStream +import io.netty.buffer.ByteBufUtil import io.netty.channel.ChannelHandlerContext import io.netty.handler.codec.http.HttpHeaderNames import io.netty.handler.codec.http.HttpHeaderValues +import io.netty.handler.codec.http.HttpMethod import io.netty.handler.codec.http.HttpResponseStatus import io.netty.handler.codec.http2.Http2DataFrame import io.netty.handler.codec.http2.Http2HeadersFrame import io.netty.handler.codec.http2.Http2StreamFrame +import io.netty.util.AsciiString import kotlinx.coroutines.CompletableDeferred import kotlinx.serialization.DeserializationStrategy +import kotlinx.serialization.KSerializer import kotlinx.serialization.json.Json import kotlinx.serialization.json.decodeFromStream +import kotlinx.serialization.serializer import java.util.zip.GZIPInputStream private val json = Json { ignoreUnknownKeys = true } -internal class Http2StreamJsonInboundHandler( +internal suspend fun Http2ClientConnection.getString(path: CharSequence): String { + return connection.stream { stream, result -> + stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = stream.alloc(), deserializer = null)) + + stream.writeHeaders( + headers = createHeaders( + HttpMethod.GET, + AsciiString.of(path), + HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP, + ), + endStream = true, + ) + } +} + +internal suspend inline fun Http2ClientConnection.getJsonOrDefaultIfNotFound( + path: CharSequence, + defaultIfNotFound: T, +): T { + return doGetJsonOrDefaultIfNotFound(path = path, deserializer = serializer(), defaultIfNotFound = defaultIfNotFound) +} + +internal suspend inline fun Http2ClientConnection.post(path: CharSequence, data: CharSequence, contentType: AsciiString): T { + return post(path = AsciiString.of(path), data = data, contentType = contentType, deserializer = serializer()) +} + +private suspend fun Http2ClientConnection.doGetJsonOrDefaultIfNotFound( + path: CharSequence, + deserializer: KSerializer, + defaultIfNotFound: T, +): T { + return connection.stream { stream, result -> + stream.pipeline().addLast(Http2StreamJsonInboundHandler( + result = result, + bufferAllocator = stream.alloc(), + deserializer = deserializer, + defaultIfNotFound = defaultIfNotFound, + )) + + stream.writeHeaders( + headers = createHeaders( + HttpMethod.GET, + AsciiString.of(path), + HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP, + ), + endStream = true, + ) + } +} + +private suspend fun Http2ClientConnection.post(path: AsciiString, data: CharSequence, contentType: AsciiString, deserializer: DeserializationStrategy): T { + return connection.stream { stream, result -> + val bufferAllocator = stream.alloc() + stream.pipeline().addLast(Http2StreamJsonInboundHandler(result = result, bufferAllocator = bufferAllocator, deserializer = deserializer)) + + stream.writeHeaders( + headers = createHeaders( + HttpMethod.POST, + path, + HttpHeaderNames.CONTENT_TYPE, contentType, + HttpHeaderNames.ACCEPT_ENCODING, HttpHeaderValues.GZIP, + ), + endStream = false, + ) + stream.writeData(ByteBufUtil.writeUtf8(bufferAllocator, data), endStream = true) + } +} + +private class Http2StreamJsonInboundHandler( private val bufferAllocator: ByteBufAllocator, private val result: CompletableDeferred, - private val deserializer: DeserializationStrategy, + private val deserializer: DeserializationStrategy?, + private val defaultIfNotFound: T? = null, ) : InboundHandlerResultTracker(result) { private var isGzip = false @@ -52,18 +126,36 @@ internal class Http2StreamJsonInboundHandler( val response = if (isGzip) { GZIPInputStream(ByteBufInputStream(data)).use { - json.decodeFromStream(deserializer, it) + if (deserializer == null) { + @Suppress("UNCHECKED_CAST") + it.reader().readText() as T + } + else { + json.decodeFromStream(deserializer, it) + } } } else { - json.decodeFromString(deserializer, data.toString(Charsets.UTF_8)) + val string = data.toString(Charsets.UTF_8) + if (deserializer == null) { + @Suppress("UNCHECKED_CAST") + string as T + } + else { + json.decodeFromString(deserializer, string) + } } result.complete(response) } else if (frame is Http2HeadersFrame) { val status = HttpResponseStatus.parseLine(frame.headers().status()) if (status != HttpResponseStatus.OK) { - result.completeExceptionally(IllegalStateException("Failed with $status")) + if (status == HttpResponseStatus.NOT_FOUND && defaultIfNotFound != null) { + result.complete(defaultIfNotFound) + } + else { + result.completeExceptionally(UnexpectedHttpStatus(status)) + } return } diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/downloadFile.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/downloadFile.kt index 12a2eae65e0b..9d49b6eca9f9 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/downloadFile.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/downloadFile.kt @@ -24,6 +24,16 @@ private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardO internal data class DownloadResult(@JvmField var size: Long, @JvmField val digest: MessageDigest) +internal suspend fun Http2ClientConnection.download(path: String, file: Path): Long { + Files.createDirectories(file.parent) + + return connection.stream { stream, result -> + stream.pipeline().addLast(FileDownloadHandler(result = result, file = file)) + + stream.writeHeaders(createHeaders(HttpMethod.GET, AsciiString.of(path)), endStream = true) + } +} + internal suspend fun Http2ClientConnection.download( path: String, file: Path, @@ -34,7 +44,7 @@ internal suspend fun Http2ClientConnection.download( return connection.stream { stream, result -> stream.pipeline().addLast( - DownloadHandler( + ZstdDecompressingFileDownloadHandler( result = result, downloadResult = DownloadResult(size = 0, digest = digestFactory()), file = file, @@ -43,12 +53,10 @@ internal suspend fun Http2ClientConnection.download( ) stream.writeHeaders(createHeaders(HttpMethod.GET, AsciiString.of(path)), endStream = true) - - result.await() } } -private class DownloadHandler( +private class ZstdDecompressingFileDownloadHandler( private val result: CompletableDeferred, private val downloadResult: DownloadResult, private val file: Path, @@ -82,7 +90,7 @@ private class DownloadHandler( if (frame is Http2HeadersFrame) { val status = HttpResponseStatus.parseLine(frame.headers().status()) if (status != HttpResponseStatus.OK) { - result.completeExceptionally(IllegalStateException("Unexpected response status: $status")) + result.completeExceptionally(UnexpectedHttpStatus(status)) } } else if (frame is Http2DataFrame) { @@ -125,3 +133,50 @@ private class DownloadHandler( while (sourceBuffer.hasRemaining()) } } + +private class FileDownloadHandler( + private val result: CompletableDeferred, + private val file: Path, +) : InboundHandlerResultTracker(result) { + private var offset = 0L + private var fileChannel: FileChannel? = null + + override fun acceptInboundMessage(message: Any): Boolean = message is Http2DataFrame || message is Http2HeadersFrame + + override fun handlerAdded(ctx: ChannelHandlerContext?) { + fileChannel = FileChannel.open(file, OVERWRITE_OPERATION) + } + + override fun handlerRemoved(context: ChannelHandlerContext) { + fileChannel?.close() + fileChannel = null + } + + override fun channelRead0(context: ChannelHandlerContext, frame: Http2StreamFrame) { + if (frame is Http2HeadersFrame) { + val status = HttpResponseStatus.parseLine(frame.headers().status()) + if (status != HttpResponseStatus.OK) { + if (status == HttpResponseStatus.NOT_FOUND) { + result.complete(-1) + } + else { + result.completeExceptionally(UnexpectedHttpStatus(status)) + } + } + } + else if (frame is Http2DataFrame) { + val content = frame.content() + val fileChannel = fileChannel!! + while (true) { + val readableBytes = content.readableBytes() + if (readableBytes == 0) { + break + } + offset += content.readBytes(fileChannel, offset, readableBytes) + } + if (frame.isEndStream) { + result.complete(offset) + } + } + } +} diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/uploadFile.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/uploadFile.kt index 0b3622c0a9f9..59546097260b 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/uploadFile.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/http2Client/uploadFile.kt @@ -1,9 +1,13 @@ // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +@file:Suppress("DuplicatedCode") + package org.jetbrains.intellij.build.http2Client import com.github.luben.zstd.Zstd import com.github.luben.zstd.ZstdCompressCtx import io.netty.channel.ChannelHandlerContext +import io.netty.handler.codec.http.HttpHeaderNames +import io.netty.handler.codec.http.HttpHeaderValues import io.netty.handler.codec.http.HttpMethod import io.netty.handler.codec.http.HttpResponseStatus import io.netty.handler.codec.http2.DefaultHttp2DataFrame @@ -20,39 +24,54 @@ import java.util.* import kotlin.math.min internal val READ_OPERATION = EnumSet.of(StandardOpenOption.READ) + internal data class UploadResult(@JvmField var uploadedSize: Long, @JvmField var fileSize: Long) +internal suspend fun Http2ClientConnection.upload(path: CharSequence, file: Path): UploadResult { + return upload(path = path, file = file, sourceBlockSize = 1024 * 1024, zstdCompressContextPool = null) +} + internal suspend fun Http2ClientConnection.upload( - path: AsciiString, + path: CharSequence, file: Path, sourceBlockSize: Int, - zstdCompressContextPool: ZstdCompressContextPool, + zstdCompressContextPool: ZstdCompressContextPool?, ): UploadResult { return connection.stream { stream, result -> val handler = WebDavPutStatusChecker(result) stream.pipeline().addLast(handler) - stream.writeHeaders(createHeaders(HttpMethod.PUT, path), endStream = false) + stream.writeHeaders(createHeaders(HttpMethod.PUT, AsciiString.of(path), HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_OCTET_STREAM), endStream = false) - val fileBuffer = FileChannel.open(file, READ_OPERATION).use { channel -> - channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size()) - } - try { - val fileSize = fileBuffer.remaining().toLong() - zstdCompressContextPool.withZstd(fileSize) { zstd -> - compressAndUpload( - fileBuffer = fileBuffer, - sourceBlockSize = sourceBlockSize, - zstd = zstd, - stream = stream, - fileSize = fileSize, - ) { + if (zstdCompressContextPool == null) { + FileChannel.open(file, READ_OPERATION).use { channel -> + uploadUncompressed(fileChannel = channel, sourceBlockSize = sourceBlockSize, stream = stream, fileSize = channel.size()) { handler.uploadedResult = it } } } - finally { - unmapBuffer(fileBuffer) + else { + val fileBuffer = FileChannel.open(file, READ_OPERATION).use { channel -> + channel.map(FileChannel.MapMode.READ_ONLY, 0, channel.size()) + } + try { + val fileSize = fileBuffer.remaining().toLong() + + zstdCompressContextPool.withZstd(fileSize) { zstd -> + compressAndUpload( + fileBuffer = fileBuffer, + sourceBlockSize = sourceBlockSize, + zstd = zstd, + stream = stream, + fileSize = fileSize, + ) { + handler.uploadedResult = it + } + } + } + finally { + unmapBuffer(fileBuffer) + } } // 1. writer must send the last data frame with endStream=true @@ -76,7 +95,7 @@ private class WebDavPutStatusChecker(private val result: CompletableDeferred Unit, +) { + var position = 0L + var uploadedSize = 0L + while (true) { + val chunkSize = min(fileSize - position, sourceBlockSize.toLong()).toInt() + val targetNettyBuffer = stream.alloc().directBuffer(chunkSize) + + var written = 0 + while (written != chunkSize) { + val n = targetNettyBuffer.writeBytes(fileChannel, position, chunkSize) + if (n == -1) { + throw RuntimeException("unexpected EOF") + } + written += n + } + + position += chunkSize + uploadedSize += chunkSize + + val endStream = position >= fileSize + val writeFuture = stream.writeAndFlush(DefaultHttp2DataFrame(targetNettyBuffer, endStream)) + if (endStream) { + uploadResultConsumer(UploadResult(uploadedSize = uploadedSize, fileSize = fileSize)) + } + writeFuture.joinCancellable() + if (endStream) { return } diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/JpsCompilationRunner.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/JpsCompilationRunner.kt index b99b4c08af42..218328d193a1 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/JpsCompilationRunner.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/JpsCompilationRunner.kt @@ -8,8 +8,8 @@ import io.opentelemetry.api.trace.Span import kotlinx.coroutines.Dispatchers import org.jetbrains.intellij.build.CompilationContext import org.jetbrains.intellij.build.impl.logging.jps.withJpsLogging +import org.jetbrains.intellij.build.telemetry.TraceManager import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder -import org.jetbrains.intellij.build.telemetry.block import org.jetbrains.intellij.build.telemetry.use import org.jetbrains.jps.api.CanceledStatus import org.jetbrains.jps.api.CmdlineRemoteProto.Message.ControllerMessage.ParametersMessage.TargetTypeBuildScope @@ -279,7 +279,8 @@ internal class JpsCompilationRunner(private val context: CompilationContext) { TargetTypeBuildScope.newBuilder() .setTypeId(RuntimeModuleRepositoryBuildConstants.TARGET_TYPE_ID) .setForceBuild(false) - .setAllTargets(true).build() + .setAllTargets(true) + .build() ) } @@ -299,7 +300,7 @@ internal class JpsCompilationRunner(private val context: CompilationContext) { .setAttribute("modules", moduleSet.joinToString(separator = ", ")) .setAttribute("incremental", context.options.incrementalCompilation) .setAttribute("cacheDir", compilationData.dataStorageRoot.toString()) - .block(Dispatchers.IO) { span -> + .use(Dispatchers.IO) { span -> withJpsLogging(context, span) { messageHandler -> Standalone.runBuild( { context.projectModel }, @@ -311,18 +312,27 @@ internal class JpsCompilationRunner(private val context: CompilationContext) { canceledStatus, ) - if (!messageHandler.errorMessagesByCompiler.isEmpty) { - for ((key, value) in messageHandler.errorMessagesByCompiler.entrySet()) { - @Suppress("UNCHECKED_CAST") - context.messages.compilationErrors(key, value as List) + if (!messageHandler.errorMessagesByCompiler.isEmpty()) { + for ((key, value) in messageHandler.errorMessagesByCompiler) { + span.addEvent( + "compilation error", + Attributes.of( + AttributeKey.stringKey("compiler"), key, + AttributeKey.stringArrayKey("errors"), value, + ) + ) + + context.messages.compilationErrors(key, value) } + span.recordException(RuntimeException("Compilation failed")) + TraceManager.scheduleExportPendingSpans() throw RuntimeException("Compilation failed") } else if (!compilationData.statisticsReported && context.options.compilationLogEnabled) { messageHandler.printPerModuleCompilationStatistics(compilationStart) context.messages.reportStatisticValue( "Compilation time, ms", - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - compilationStart).toString() + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - compilationStart).toString(), ) compilationData.statisticsReported = true } diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/TestingTasksImpl.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/TestingTasksImpl.kt index 2d88f75de9e5..a8a862bb9e6c 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/TestingTasksImpl.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/TestingTasksImpl.kt @@ -37,9 +37,10 @@ import java.util.regex.Pattern import kotlin.io.path.absolutePathString import kotlin.io.path.readLines +private const val NO_TESTS_ERROR = 42 + internal class TestingTasksImpl(context: CompilationContext, private val options: TestingOptions) : TestingTasks { private val context: CompilationContext = if (options.useArchivedCompiledClasses) ArchivedCompilationContext(context) else context - private val NO_TESTS_ERROR = 42 private fun loadRunConfigurations(name: String): List { return try { @@ -259,13 +260,13 @@ internal class TestingTasksImpl(context: CompilationContext, private val options val testDiscovery = "intellij-test-discovery" val library = context.projectModel.project.libraryCollection.findLibrary(testDiscovery) - ?: throw RuntimeException("Can't find the ${testDiscovery} library, but test discovery capturing enabled.") + ?: throw RuntimeException("Can't find the $testDiscovery library, but test discovery capturing enabled.") val agentJar = library.getPaths(JpsOrderRootType.COMPILED) .firstOrNull { val name = it.fileName.toString() name.startsWith("intellij-test-discovery") && name.endsWith(".jar") - } ?: throw RuntimeException("Can't find the agent in ${testDiscovery} library, but test discovery capturing enabled.") + } ?: throw RuntimeException("Can't find the agent in $testDiscovery library, but test discovery capturing enabled.") additionalJvmOptions += "-javaagent:${agentJar}" val excludeRoots = context.projectModel.global.libraryCollection.getLibraries(JpsJavaSdkType.INSTANCE) @@ -292,7 +293,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options val testConfigurationType = System.getProperty("teamcity.remote-debug.type") if (testConfigurationType != "junit") { context.messages.error( - "Remote debugging is supported for junit run configurations only, but 'teamcity.remote-debug.type' is ${testConfigurationType}") + "Remote debugging is supported for junit run configurations only, but 'teamcity.remote-debug.type' is $testConfigurationType") } val testObject = System.getProperty("teamcity.remote-debug.junit.type") val junitClass = System.getProperty("teamcity.remote-debug.junit.class") @@ -300,7 +301,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options val message = "Remote debugging supports debugging all test methods in a class for now, debugging isn't supported for '${testObject}'" if (testObject == "method") { context.messages.warning(message) - context.messages.warning("Launching all test methods in the class ${junitClass}") + context.messages.warning("Launching all test methods in the class $junitClass") } else { context.messages.error(message) @@ -386,7 +387,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options prepareEnvForTestRun(allJvmArgs, systemProperties, bootstrapClasspath, remoteDebugging) val messages = context.messages if (isRunningInBatchMode) { - messages.info("Running tests from ${mainModule} matched by '${options.batchTestIncludes}' pattern.") + messages.info("Running tests from $mainModule matched by '${options.batchTestIncludes}' pattern.") } else { messages.info("Starting tests from groups '${testGroups}' from classpath of module '${mainModule}'") @@ -397,20 +398,20 @@ internal class TestingTasksImpl(context: CompilationContext, private val options spanBuilder("test classpath and runtime info").use { withContext(Dispatchers.IO) { val runtime = getRuntimeExecutablePath().toString() - messages.info("Runtime: ${runtime}") + messages.info("Runtime: $runtime") runProcess(args = listOf(runtime, "-version"), inheritOut = true, inheritErrToOut = true) } - messages.info("Runtime options: ${allJvmArgs}") - messages.info("System properties: ${systemProperties}") - messages.info("Bootstrap classpath: ${bootstrapClasspath}") - messages.info("Tests classpath: ${testClasspath}") + messages.info("Runtime options: $allJvmArgs") + messages.info("System properties: $systemProperties") + messages.info("Bootstrap classpath: $bootstrapClasspath") + messages.info("Tests classpath: $testClasspath") modulePath?.let { mp -> @Suppress("SpellCheckingInspection") messages.info("Tests modulepath: $mp") } if (!envVariables.isEmpty()) { - messages.info("Environment variables: ${envVariables}") + messages.info("Environment variables: $envVariables") } } runJUnit5Engine( @@ -448,7 +449,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options if (options.customRuntimePath != null) { runtimeDir = Path.of(checkNotNull(options.customRuntimePath)) check(Files.isDirectory(runtimeDir)) { - "Custom Jre path from system property '${TestingOptions.TEST_JRE_PROPERTY}' is missing: ${runtimeDir}" + "Custom Jre path from system property '${TestingOptions.TEST_JRE_PROPERTY}' is missing: $runtimeDir" } } else { @@ -459,7 +460,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options if (SystemInfoRt.isMac && Files.notExists(java)) { java = runtimeDir.resolve("Contents/Home/bin/java") } - check(Files.exists(java)) { "java executable is missing: ${java}" } + check(Files.exists(java)) { "java executable is missing: $java" } return java } @@ -666,7 +667,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options } ?: listOf() if (testClasses.isEmpty()) { - throw RuntimeException("No tests were found in ${root} with ${filteringPattern}") + throw RuntimeException("No tests were found in $root with $filteringPattern") } return testClasses @@ -740,12 +741,12 @@ internal class TestingTasksImpl(context: CompilationContext, private val options noTestsInAllClasses = noTestsInAllClasses && noTests } catch (e: Throwable) { - throw RuntimeException("Failed to process ${qName}", e) + throw RuntimeException("Failed to process $qName", e) } } if (noTestsInAllClasses) { - throw RuntimeException("No tests were found in ${mainModule} with ${pattern}") + throw RuntimeException("No tests were found in $mainModule with $pattern") } } @@ -1041,7 +1042,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options val argFile = CommandLineWrapperUtil.createArgumentFile(args, Charset.defaultCharset()) val runtime = runBlocking(Dispatchers.IO) { getRuntimeExecutablePath().toString() } - context.messages.info("Starting tests on runtime ${runtime}") + context.messages.info("Starting tests on runtime $runtime") val builder = ProcessBuilder(runtime, "@" + argFile.absolutePath) builder.environment().putAll(envVariables) if(TeamCityHelper.isUnderTeamCity) { @@ -1054,7 +1055,7 @@ internal class TestingTasksImpl(context: CompilationContext, private val options } val exitCode = builder.start().waitFor() if (exitCode != 0 && exitCode != NO_TESTS_ERROR) { - context.messages.error("Tests failed with exit code ${exitCode}") + context.messages.error("Tests failed with exit code $exitCode") } return exitCode } @@ -1114,7 +1115,7 @@ private fun removeStandardJvmOptions(vmOptions: List): List { private suspend fun publishTestDiscovery(messages: BuildMessages, file: String?) { val serverUrl = System.getProperty("intellij.test.discovery.url") val token = System.getProperty("intellij.test.discovery.token") - messages.info("Trying to upload ${file} into ${serverUrl}.") + messages.info("Trying to upload $file into ${serverUrl}.") val path = file?.let { Path.of(it) } if (path != null && Files.exists(path)) { if (serverUrl == null) { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompilationPartsUtil.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompilationPartsUtil.kt index 96becb4987e3..8dd99630aa3a 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompilationPartsUtil.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompilationPartsUtil.kt @@ -9,8 +9,8 @@ import io.opentelemetry.api.common.Attributes import io.opentelemetry.api.trace.Span import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable import kotlinx.serialization.encodeToString @@ -19,21 +19,16 @@ import org.jetbrains.annotations.VisibleForTesting import org.jetbrains.intellij.build.BuildMessages import org.jetbrains.intellij.build.CompilationContext import org.jetbrains.intellij.build.forEachConcurrent -import org.jetbrains.intellij.build.http2Client.READ_OPERATION import org.jetbrains.intellij.build.http2Client.withHttp2ClientConnectionFactory import org.jetbrains.intellij.build.io.AddDirEntriesMode import org.jetbrains.intellij.build.io.zip import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.use -import java.math.BigInteger import java.net.InetSocketAddress import java.net.URI -import java.nio.ByteBuffer -import java.nio.channels.FileChannel import java.nio.file.* import java.nio.file.attribute.BasicFileAttributes import java.nio.file.attribute.FileTime -import java.security.MessageDigest import java.util.* import java.util.concurrent.CancellationException import java.util.concurrent.ConcurrentHashMap @@ -310,7 +305,7 @@ suspend fun fetchAndUnpackCompiledClasses( } true } - .forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtMost(4)) { item -> + .forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtLeast(4)) { item -> val file = item.file when { Files.notExists(file) -> toDownload.add(item) @@ -331,40 +326,9 @@ suspend fun fetchAndUnpackCompiledClasses( } spanBuilder("cleanup outdated compiled class archives").use { - val start = System.nanoTime() - var count = 0 - var bytes = 0L - try { - val preserve = items.mapTo(HashSet(items.size)) { it.file } - val epoch = FileTime.fromMillis(0) - val daysAgo = FileTime.fromMillis(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4)) - Files.createDirectories(tempDownloadStorage) - // we need to traverse with depth 3 since the first level is [production, test], second level is module name, third is a file - Files.walk(tempDownloadStorage, 3, FileVisitOption.FOLLOW_LINKS).use { stream -> - stream - .filter { !preserve.contains(it) } - .forEach { file -> - val attr = Files.readAttributes(file, BasicFileAttributes::class.java) - if (attr.isRegularFile) { - val lastAccessTime = attr.lastAccessTime() - if (lastAccessTime > epoch && lastAccessTime < daysAgo) { - count++ - bytes += attr.size() - Files.deleteIfExists(file) - } - } - } - } - } - catch (e: Throwable) { - Span.current().addEvent("failed to cleanup outdated archives", Attributes.of(AttributeKey.stringKey("error"), e.message ?: "")) - } - - reportStatisticValue("compile-parts:verify:time", TimeUnit.NANOSECONDS.toMillis(verifyTime).toString()) - reportStatisticValue("compile-parts:cleanup:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString()) - reportStatisticValue("compile-parts:removed:bytes", bytes.toString()) - reportStatisticValue("compile-parts:removed:count", count.toString()) + cleanupOutdatedCompiledClassArchives(items = items, tempDownloadStorage = tempDownloadStorage, reportStatisticValue = reportStatisticValue) } + reportStatisticValue("compile-parts:verify:time", TimeUnit.NANOSECONDS.toMillis(verifyTime).toString()) spanBuilder("fetch compiled classes archives").use { val start = System.nanoTime() @@ -389,12 +353,10 @@ suspend fun fetchAndUnpackCompiledClasses( } val start = System.nanoTime() - spanBuilder("unpack compiled classes archives").use { - for (item in toUnpack) { - launch { - spanBuilder("unpack").setAttribute("name", item.name).use { - unpackArchive(item, saveHash) - } + spanBuilder("unpack compiled classes archives").use(Dispatchers.IO) { + toUnpack.forEachConcurrent(Runtime.getRuntime().availableProcessors().coerceAtLeast(4)) { item -> + spanBuilder("unpack").setAttribute("name", item.name).use { + unpackCompilationPartArchive(item, saveHash) } } } @@ -403,6 +365,45 @@ suspend fun fetchAndUnpackCompiledClasses( reportStatisticValue("compile-parts:unpack:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString()) } +private fun cleanupOutdatedCompiledClassArchives( + items: List, + tempDownloadStorage: Path, + reportStatisticValue: (key: String, value: String) -> Unit, +) { + val start = System.nanoTime() + var count = 0 + var bytes = 0L + try { + val preserve = items.mapTo(HashSet(items.size)) { it.file } + val epoch = FileTime.fromMillis(0) + val daysAgo = FileTime.fromMillis(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4)) + Files.createDirectories(tempDownloadStorage) + // we need to traverse with depth 3 since the first level is [production, test], second level is module name, third is a file + Files.walk(tempDownloadStorage, 3, FileVisitOption.FOLLOW_LINKS).use { stream -> + stream + .filter { !preserve.contains(it) } + .forEach { file -> + val attr = Files.readAttributes(file, BasicFileAttributes::class.java) + if (attr.isRegularFile) { + val lastAccessTime = attr.lastAccessTime() + if (lastAccessTime > epoch && lastAccessTime < daysAgo) { + count++ + bytes += attr.size() + Files.deleteIfExists(file) + } + } + } + } + } + catch (e: Throwable) { + Span.current().addEvent("failed to cleanup outdated archives", Attributes.of(AttributeKey.stringKey("error"), e.message ?: "")) + } + + reportStatisticValue("compile-parts:cleanup:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString()) + reportStatisticValue("compile-parts:removed:bytes", bytes.toString()) + reportStatisticValue("compile-parts:removed:count", count.toString()) +} + private suspend fun checkPreviouslyUnpackedDirectories( items: List, span: Span, @@ -415,54 +416,52 @@ private suspend fun checkPreviouslyUnpackedDirectories( } val start = System.nanoTime() - coroutineScope { - launch(Dispatchers.IO) { + withContext(Dispatchers.IO) { + launch { spanBuilder("remove stalled directories not present in metadata").setAttribute(AttributeKey.stringArrayKey("keys"), java.util.List.copyOf(metadata.files.keys)).use { removeStalledDirs(metadata, classOutput) } } - for (item in items) { - launch { - val out = item.output - if (Files.notExists(out)) { - span.addEvent("output directory doesn't exist", Attributes.of(AttributeKey.stringKey("name"), item.name, AttributeKey.stringKey("outDir"), out.toString())) - return@launch - } + items.forEachConcurrent { item -> + val out = item.output + if (Files.notExists(out)) { + span.addEvent("output directory doesn't exist", Attributes.of(AttributeKey.stringKey("name"), item.name, AttributeKey.stringKey("outDir"), out.toString())) + return@forEachConcurrent + } - val hashFile = out.resolve(".hash") - if (!Files.isRegularFile(hashFile)) { - span.addEvent("no .hash file in output directory", Attributes.of(AttributeKey.stringKey("name"), item.name)) - out.deleteRecursively() - return@launch - } + val hashFile = out.resolve(".hash") + if (!Files.isRegularFile(hashFile)) { + span.addEvent("no .hash file in output directory", Attributes.of(AttributeKey.stringKey("name"), item.name)) + out.deleteRecursively() + return@forEachConcurrent + } - try { - val actual = Files.readString(hashFile) - if (actual == item.hash) { - upToDate.add(item.name) - } - else { - span.addEvent( - "output directory hash mismatch", - Attributes.of( - AttributeKey.stringKey("name"), item.name, - AttributeKey.stringKey("expected"), item.hash, - AttributeKey.stringKey("actual"), actual, - ) + try { + val actual = Files.readString(hashFile) + if (actual == item.hash) { + upToDate.add(item.name) + } + else { + span.addEvent( + "output directory hash mismatch", + Attributes.of( + AttributeKey.stringKey("name"), item.name, + AttributeKey.stringKey("expected"), item.hash, + AttributeKey.stringKey("actual"), actual, ) - out.deleteRecursively() - } - } - catch (e: CancellationException) { - throw e - } - catch (e: Throwable) { - span.addEvent("output directory hash calculation failed", Attributes.of(AttributeKey.stringKey("name"), item.name)) - span.recordException(e, Attributes.of(AttributeKey.stringKey("name"), item.name)) + ) out.deleteRecursively() } } + catch (e: CancellationException) { + throw e + } + catch (e: Throwable) { + span.addEvent("output directory hash calculation failed", Attributes.of(AttributeKey.stringKey("name"), item.name)) + span.recordException(e, Attributes.of(AttributeKey.stringKey("name"), item.name)) + out.deleteRecursively() + } } } return System.nanoTime() - start @@ -505,34 +504,6 @@ private fun CoroutineScope.removeStalledDirs( } } -private val sharedDigest = MessageDigest.getInstance("SHA-256", java.security.Security.getProvider("SUN")) -internal fun sha256() = sharedDigest.clone() as MessageDigest - -internal fun computeHash(file: Path): String { - val messageDigest = sha256() - FileChannel.open(file, READ_OPERATION).use { channel -> - val fileSize = channel.size() - // java message digest doesn't support native buffer (copies to a heap byte array in any case) - val bufferSize = 256 * 1024 - val buffer = ByteBuffer.allocate(bufferSize) - var offset = 0L - var readBytes: Int - while (offset < fileSize) { - buffer.clear() - readBytes = channel.read(buffer, offset) - if (readBytes <= 0) { - break - } - - messageDigest.update(buffer.array(), 0, readBytes) - offset += readBytes - } - } - return digestToString(messageDigest) -} - -internal fun digestToString(digest: MessageDigest): String = BigInteger(1, digest.digest()).toString(36) + "z" - internal data class PackAndUploadItem( @JvmField val output: Path, @JvmField val name: String, diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompiledClasses.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompiledClasses.kt index 91144df22c3e..8c30bfa8725f 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompiledClasses.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/CompiledClasses.kt @@ -9,7 +9,6 @@ import io.opentelemetry.api.common.Attributes import io.opentelemetry.api.trace.Span import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.isActive -import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeout import org.jetbrains.intellij.build.BuildOptions import org.jetbrains.intellij.build.CompilationContext @@ -22,8 +21,6 @@ import org.jetbrains.intellij.build.telemetry.use import org.jetbrains.jps.api.CanceledStatus import org.jetbrains.jps.incremental.storage.ProjectStamps import java.nio.file.Path -import kotlin.time.Duration -import kotlin.time.Duration.Companion.minutes internal fun checkCompilationOptions(context: CompilationContext) { val options = context.options @@ -189,18 +186,20 @@ internal suspend fun doCompile( check(isCompilationRequired(context.options)) { "Unexpected compilation request, unable to proceed" } - context.messages.progress("Compiling project") context.compilationData.statisticsReported = false val runner = JpsCompilationRunner(context) try { val (status, isIncrementalCompilation) = when { context.options.forceRebuild -> "forced rebuild" to false availableCommitDepth >= 0 -> portableJpsCacheUsageStatus(availableCommitDepth) to true - isIncrementalCompilationDataAvailable(context) -> "compiled using local cache" to true + isIncrementalCompilationDataAvailable(context) -> "compile using local cache" to true else -> "clean build" to false } context.options.incrementalCompilation = isIncrementalCompilation - if (!isIncrementalCompilation) { + if (isIncrementalCompilation) { + Span.current().addEvent("status: $status") + } + else { Span.current().addEvent( "no compiled classes can be reused", Attributes.of( @@ -209,20 +208,22 @@ internal suspend fun doCompile( ) ) } - block(status) { - if (isIncrementalCompilation) { - // workaround for KT-55695 - compileWithTimeout( - compilationRunner = runner, + + val incrementalCompilationTimeout = context.options.incrementalCompilationTimeout + if (isIncrementalCompilation && incrementalCompilationTimeout != null) { + // workaround for KT-55695 + withTimeout(incrementalCompilationTimeout) { + compile( + jpsCompilationRunner = runner, context = context, moduleNames = moduleNames, includingTestsInModules = includingTestsInModules, - timeout = context.options.incrementalCompilationTimeout.minutes, + canceledStatus = CanceledStatus { !isActive }, ) } - else { - compile(jpsCompilationRunner = runner, context = context, moduleNames = moduleNames, includingTestsInModules = includingTestsInModules) - } + } + else { + compile(jpsCompilationRunner = runner, context = context, moduleNames = moduleNames, includingTestsInModules = includingTestsInModules) } context.messages.buildStatus(status) } @@ -246,26 +247,6 @@ private suspend fun unpackCompiledClasses(classOutput: Path, context: Compilatio } } -private suspend fun compileWithTimeout( - compilationRunner: JpsCompilationRunner, - context: CompilationContext, - moduleNames: Collection?, - includingTestsInModules: List?, - timeout: Duration, -) { - withTimeout(timeout) { - launch { - compile( - jpsCompilationRunner = compilationRunner, - context = context, - moduleNames = moduleNames, - includingTestsInModules = includingTestsInModules, - canceledStatus = CanceledStatus { !isActive }, - ) - } - } -} - private suspend fun retryCompilation( context: CompilationContext, runner: JpsCompilationRunner, diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCache.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCache.kt index bb194bb5722a..0a1989fee532 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCache.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCache.kt @@ -1,6 +1,7 @@ // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package org.jetbrains.intellij.build.impl.compilation +import io.netty.util.AsciiString import io.opentelemetry.api.trace.Span import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.runBlocking @@ -14,12 +15,13 @@ import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.block import org.jetbrains.intellij.build.telemetry.use import org.jetbrains.jps.incremental.storage.ProjectStamps +import java.net.URI import java.nio.file.Path import java.util.* import java.util.concurrent.CancellationException internal val IS_PORTABLE_COMPILATION_CACHE_ENABLED: Boolean - get() = ProjectStamps.PORTABLE_CACHES && IS_CONFIGURED + get() = ProjectStamps.PORTABLE_CACHES && IS_JPS_CACHE_URL_CONFIGURED private var isAlreadyUpdated = false @@ -29,7 +31,7 @@ internal object TestJpsCompilationCacheDownload { System.setProperty("jps.cache.test", "true") System.setProperty("org.jetbrains.jps.portable.caches", "true") if (System.getProperty(URL_PROPERTY) == null) { - System.setProperty(URL_PROPERTY, "https://127.0.0.1:1900") + System.setProperty(URL_PROPERTY, "https://127.0.0.1:1900/cache/jps") } val projectHome = ULTIMATE_HOME @@ -42,34 +44,55 @@ internal object TestJpsCompilationCacheDownload { useCompiledClassesFromProjectOutput = true, ), ) - downloadCacheAndCompileProject(forceDownload = false, context = context) + downloadCacheAndCompileProject(forceDownload = false, gitUrl = computeRemoteGitUrl(), context = context) } } internal suspend fun downloadJpsCacheAndCompileProject(context: CompilationContext) { - downloadCacheAndCompileProject(forceDownload = System.getProperty(FORCE_DOWNLOAD_PROPERTY).toBoolean(), context = context) + downloadCacheAndCompileProject(forceDownload = System.getProperty(FORCE_DOWNLOAD_PROPERTY).toBoolean(), gitUrl = computeRemoteGitUrl(), context = context) } /** - * Upload local [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig] + * Upload local [PortableCompilationCache]. */ suspend fun uploadPortableCompilationCache(context: CompilationContext) { - createPortableCompilationCacheUploader(context).upload(context.messages) + uploadJpsCache( + forcedUpload = context.options.forceRebuild, + commitHash = getCommitCache(), + s3Dir = getS3Dir(), + authHeader = getAuthHeader(), + uploadUrl = getJpsCacheUploadUrl(), + context = context, + ) } /** - * Publish already uploaded [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig] + * Publish already uploaded [PortableCompilationCache]. */ -suspend fun publishPortableCompilationCache(context: CompilationContext) { - createPortableCompilationCacheUploader(context).updateCommitHistory() +suspend fun publishPortableCompilationCache(context: CompilationContext, overrideCommits: Set? = null) { + updateJpsCacheCommitHistory( + overrideCommits = overrideCommits, + remoteGitUrl = computeRemoteGitUrl(), + commitHash = getCommitCache(), + uploadUrl = getJpsCacheUploadUrl(), + authHeader = getAuthHeader(), + context = context, + s3Dir = getS3Dir(), + ) +} + +private fun getCommitCache() = require(COMMIT_HASH_PROPERTY, "Repository commit") + +private fun getS3Dir(): Path? { + return System.getProperty(AWS_SYNC_FOLDER_PROPERTY)?.let { Path.of(it) } } /** - * Publish already uploaded [PortableCompilationCache] to [PortableJpsCacheRemoteCacheConfig] overriding existing [CommitsHistory]. + * Publish already uploaded [PortableCompilationCache] overriding existing [CommitsHistory]. * Used in force rebuild and cleanup. */ suspend fun publishUploadedJpsCacheWithCommitHistoryOverride(forceRebuiltCommits: Set, context: CompilationContext) { - createPortableCompilationCacheUploader(context).updateCommitHistory(overrideCommits = forceRebuiltCommits) + publishPortableCompilationCache(context = context, overrideCommits = forceRebuiltCommits) } /** @@ -84,7 +107,7 @@ suspend fun publishUploadedJpsCacheWithCommitHistoryOverride(forceRebuiltCommits * For more details see [org.jetbrains.jps.backwardRefs.JavaBackwardReferenceIndexWriter.initialize] */ @Suppress("KDocUnresolvedReference") -private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, context: CompilationContext) { +private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, gitUrl: String, context: CompilationContext) { val forceRebuild = context.options.forceRebuild spanBuilder("download JPS cache and compile") .setAttribute("forceRebuild", forceRebuild) @@ -103,16 +126,11 @@ private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, conte cleanOutput(context = context, keepCompilationState = false) } - val downloader = PortableCompilationCacheDownloader( - context = context, - git = Git(context.paths.projectHome), - remoteCache = PortableJpsCacheRemoteCacheConfig(), - gitUrl = computeRemoteGitUrl(), - ) + val downloader = PortableCompilationCacheDownloader(context = context, git = Git(context.paths.projectHome)) val portableCompilationCache = PortableCompilationCache(forceDownload = forceDownload) val availableCommitDepth = if (!forceRebuild && (forceDownload || !isIncrementalCompilationDataAvailable(context))) { - portableCompilationCache.downloadCache(downloader, context) + portableCompilationCache.downloadCache(downloader = downloader, gitUrl = computeRemoteGitUrl(), context = context) } else { -1 @@ -121,18 +139,21 @@ private suspend fun downloadCacheAndCompileProject(forceDownload: Boolean, conte context.options.incrementalCompilation = !forceRebuild // compilation is executed unconditionally here even if the exact commit cache is downloaded // to have an additional validation step and not to ignore a local changes, for example, in TeamCity Remote Run - doCompile( - availableCommitDepth = availableCommitDepth, - context = context, - handleCompilationFailureBeforeRetry = { successMessage -> - portableCompilationCache.handleCompilationFailureBeforeRetry( - successMessage = successMessage, - portableCompilationCacheDownloader = downloader, - forceDownload = portableCompilationCache.forceDownload, - context = context, - ) - }, - ) + if (!context.options.useCompiledClassesFromProjectOutput) { + doCompile( + availableCommitDepth = availableCommitDepth, + context = context, + handleCompilationFailureBeforeRetry = { successMessage -> + portableCompilationCache.handleCompilationFailureBeforeRetry( + successMessage = successMessage, + portableCompilationCacheDownloader = downloader, + forceDownload = portableCompilationCache.forceDownload, + gitUrl = gitUrl, + context = context, + ) + }, + ) + } isAlreadyUpdated = true context.options.incrementalCompilation = true } @@ -150,6 +171,7 @@ private class PortableCompilationCache(forceDownload: Boolean) { portableCompilationCacheDownloader: PortableCompilationCacheDownloader, context: CompilationContext, forceDownload: Boolean, + gitUrl: String, ): String { when { forceDownload -> { @@ -162,7 +184,7 @@ private class PortableCompilationCache(forceDownload: Boolean) { // If download isn't forced, then locally available cache will be used which may suffer from those issues. // Hence, compilation failure. Replacing local cache with remote one may help. Span.current().addEvent("Incremental compilation using locally available caches failed. Re-trying using Remote Cache.") - val availableCommitDepth = downloadCache(portableCompilationCacheDownloader, context) + val availableCommitDepth = downloadCache(portableCompilationCacheDownloader, gitUrl = gitUrl, context) if (availableCommitDepth >= 0) { return portableJpsCacheUsageStatus(availableCommitDepth) } @@ -171,10 +193,10 @@ private class PortableCompilationCache(forceDownload: Boolean) { return successMessage } - suspend fun downloadCache(downloader: PortableCompilationCacheDownloader, context: CompilationContext): Int { + suspend fun downloadCache(downloader: PortableCompilationCacheDownloader, gitUrl: String, context: CompilationContext): Int { return spanBuilder("downloading Portable Compilation Cache").use { span -> try { - downloader.download() + downloader.download(cacheUrl = require(URL_PROPERTY, "Remote Cache url"), gitUrl = gitUrl, authHeader = getAuthHeader()) } catch (e: CancellationException) { throw e @@ -211,10 +233,7 @@ private const val UPLOAD_URL_PROPERTY = "intellij.jps.remote.cache.upload.url" */ private const val URL_PROPERTY = "intellij.jps.remote.cache.url" -/** - * If true then [PortableJpsCacheRemoteCacheConfig] is configured to be used - */ -private val IS_CONFIGURED = !System.getProperty(URL_PROPERTY).isNullOrBlank() +private val IS_JPS_CACHE_URL_CONFIGURED = !System.getProperty(URL_PROPERTY).isNullOrBlank() /** * IntelliJ repository git remote url @@ -256,32 +275,16 @@ internal class CompilationOutput( @JvmField val path: Path, ) -/** - * Server which stores [PortableCompilationCache] - */ -internal class PortableJpsCacheRemoteCacheConfig { - val url by lazy { require(URL_PROPERTY, "Remote Cache url") } - val uploadUrl by lazy { require(UPLOAD_URL_PROPERTY, "Remote Cache upload url") } - val authHeader by lazy { - val username = System.getProperty("jps.auth.spaceUsername") - val password = System.getProperty("jps.auth.spacePassword") - when { - password == null -> "" - username == null -> "Bearer $password" - else -> "Basic " + Base64.getEncoder().encodeToString("$username:$password".toByteArray()) - } - } -} +private fun getJpsCacheUploadUrl(): URI = URI(require(UPLOAD_URL_PROPERTY, "Remote Cache upload url")) -internal fun createPortableCompilationCacheUploader(context: CompilationContext): PortableCompilationCacheUploader { - return PortableCompilationCacheUploader( - context = context, - remoteCache = PortableJpsCacheRemoteCacheConfig(), - remoteGitUrl = computeRemoteGitUrl(), - commitHash = require(COMMIT_HASH_PROPERTY, "Repository commit"), - s3Folder = Path.of(require(AWS_SYNC_FOLDER_PROPERTY, "AWS S3 sync folder")), - forcedUpload = context.options.forceRebuild, - ) +private fun getAuthHeader(): CharSequence { + val username = System.getProperty("jps.auth.spaceUsername") + val password = System.getProperty("jps.auth.spacePassword") + return when { + password == null -> AsciiString.EMPTY_STRING + username == null -> AsciiString.of("Bearer $password") + else -> AsciiString.of("Basic " + Base64.getEncoder().encodeToString("$username:$password".toByteArray())) + } } private fun computeRemoteGitUrl(): String { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheDownloader.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheDownloader.kt index 7c3bd85bece8..d1cc31e8a127 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheDownloader.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheDownloader.kt @@ -3,22 +3,23 @@ package org.jetbrains.intellij.build.impl.compilation -import com.intellij.util.io.Decompressor +import com.google.gson.stream.JsonReader import io.opentelemetry.api.common.AttributeKey import io.opentelemetry.api.common.Attributes import io.opentelemetry.api.trace.Span -import kotlinx.coroutines.* -import okhttp3.Request -import okio.sink +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.launch import org.jetbrains.intellij.build.CompilationContext import org.jetbrains.intellij.build.forEachConcurrent +import org.jetbrains.intellij.build.http2Client.* import org.jetbrains.intellij.build.impl.compilation.cache.CommitsHistory import org.jetbrains.intellij.build.impl.compilation.cache.getAllCompilationOutputs -import org.jetbrains.intellij.build.impl.compilation.cache.parseSourcesStateFile import org.jetbrains.intellij.build.retryWithExponentialBackOff import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.use -import java.io.IOException +import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState +import java.net.URI import java.nio.file.Files import java.nio.file.NoSuchFileException import java.nio.file.Path @@ -28,83 +29,53 @@ import java.util.concurrent.atomic.LongAdder private const val COMMITS_COUNT = 1_000 -internal class PortableCompilationCacheDownloader( - private val context: CompilationContext, - private val git: Git, - private val remoteCache: PortableJpsCacheRemoteCacheConfig, - private val gitUrl: String, -) { - private val remoteCacheUrl = remoteCache.url.trimEnd('/') - +internal class PortableCompilationCacheDownloader(private val context: CompilationContext, private val git: Git) { private val lastCommits by lazy { git.log(COMMITS_COUNT) } - private suspend fun downloadString(url: String): String = retryWithExponentialBackOff { - if (url.isS3()) { - awsS3Cli("cp", url, "-") - } - else { - httpClient.get(url, remoteCache.authHeader) { it.body.string() } - } - } - - private suspend fun downloadToFile(url: String, file: Path, spanName: String, notFound: LongAdder) { - spanBuilder(spanName).setAttribute("url", url).setAttribute("path", file.toString()).use { span -> - Files.createDirectories(file.parent) - retryWithExponentialBackOff { - if (url.isS3()) { - awsS3Cli("cp", url, file.toString()) + private suspend fun downloadToFile(urlPath: String, file: Path, spanName: String, notFound: LongAdder, connection: Http2ClientConnection?): Long { + return spanBuilder(spanName).setAttribute("urlPath", urlPath).setAttribute("path", file.toString()).use { span -> + if (connection == null) { + Files.createDirectories(file.parent) + require(urlPath.isS3()) + retryWithExponentialBackOff { + awsS3Cli("cp", urlPath, file.toString()) } - else { - httpClient.newCall(Request.Builder().url(url).header("Authorization", remoteCache.authHeader).build()) - .executeAsync() - .use { response -> - when { - response.isSuccessful -> { - file.sink().use { - response.body.source().readAll(it) - } - } - response.code == 404 -> { - span.addEvent("resource not found") - notFound.increment() - } - else -> throw IOException("Unexpected code $response") - } - } + return@use try { + Files.size(file) + } + catch (e: NoSuchFileException) { + 0 } } - } - } - @OptIn(DelicateCoroutinesApi::class) - private val availableCachesKeysLazyTask = GlobalScope.async(Dispatchers.Unconfined, start = CoroutineStart.LAZY) { - val commitHistory = "$remoteCacheUrl/${CommitsHistory.JSON_FILE}" - if (isExist(commitHistory)) { - val json = downloadString(commitHistory) - CommitsHistory(json).commitsForRemote(gitUrl) - } - else { - emptyList() - } - } - - private suspend fun isExist(path: String): Boolean { - return spanBuilder("head").setAttribute("url", remoteCacheUrl).use { - retryWithExponentialBackOff { - httpClient.head(path, remoteCache.authHeader) == 200 + val sizeOnDisk = connection.download(path = urlPath, file = file) + if (sizeOnDisk == -1L) { + span.addEvent("resource not found") + notFound.increment() } + sizeOnDisk } } - suspend fun download(): Int { - val availableCachesKeys = availableCachesKeysLazyTask.await() + private suspend fun getAvailableCachesKeysLazyTask(urlPathPrefix: String, gitUrl: String, connection: Http2ClientConnection?): Collection { + val commitHistoryUrl = "$urlPathPrefix/${CommitsHistory.JSON_FILE}" + require(!commitHistoryUrl.isS3() && connection != null) + val json: Map> = connection.getJsonOrDefaultIfNotFound(path = commitHistoryUrl, defaultIfNotFound = emptyMap()) + if (json.isEmpty()) { + return emptyList() + } + return CommitsHistory(json).commitsForRemote(gitUrl) + } + + private suspend fun prepareDownload(urlPathPrefix: String, gitUrl: String, connection: Http2ClientConnection?): Pair? { + val availableCachesKeys = getAvailableCachesKeysLazyTask(urlPathPrefix = urlPathPrefix, gitUrl = gitUrl, connection = connection) val availableCommitDepth = lastCommits.indexOfFirst { availableCachesKeys.contains(it) } if (availableCommitDepth !in 0 until lastCommits.count()) { - Span.current().addEvent("Unable to find cache for any of last ${lastCommits.count()} commits.") - return -1 + Span.current().addEvent("unable to find cache for any of last ${lastCommits.count()} commits.") + return null } val lastCachedCommit = lastCommits.get(availableCommitDepth) @@ -112,30 +83,47 @@ internal class PortableCompilationCacheDownloader( "using cache for commit $lastCachedCommit ($availableCommitDepth behind last commit)", Attributes.of( AttributeKey.longKey("behind last commit"), availableCommitDepth.toLong(), - AttributeKey.stringArrayKey("available cache keys"), availableCachesKeysLazyTask.await().toList(), + AttributeKey.stringArrayKey("available cache keys"), availableCachesKeys.toList(), ) ) + return lastCachedCommit to availableCommitDepth + } + + suspend fun download(cacheUrl: String, authHeader: CharSequence, gitUrl: String): Int { val start = System.nanoTime() val totalDownloadedBytes = LongAdder() val notFound = LongAdder() - var total = -1 - withContext(Dispatchers.IO) { - launch { - spanBuilder("get and unpack jps cache").setAttribute("commit", lastCachedCommit).use { - downloadAndUnpackJpsCache(commitHash = lastCachedCommit, notFound = notFound, totalBytes = totalDownloadedBytes) + var availableCommitDepth = -1 + val total = if (cacheUrl.isS3()) { + val info = prepareDownload(urlPathPrefix = cacheUrl, gitUrl = gitUrl, connection = null) ?: return -1 + availableCommitDepth = info.second + doDownload( + urlPathPrefix = cacheUrl, + lastCachedCommit = info.first, + notFound = notFound, + totalDownloadedBytes = totalDownloadedBytes, + connection = null, + ) + } + else { + val serverUri = URI(cacheUrl) + withHttp2ClientConnectionFactory(trustAll = serverUri.host == "127.0.0.1") { client -> + client.connect(serverUri.host, serverUri.port).withAuth(authHeader).use { connection -> + val info = prepareDownload(urlPathPrefix = serverUri.path, gitUrl = gitUrl, connection = connection) ?: return@use -1 + availableCommitDepth = info.second + doDownload( + urlPathPrefix = serverUri.path, + lastCachedCommit = info.first, + notFound = notFound, + totalDownloadedBytes = totalDownloadedBytes, + connection = connection, + ) } } + } - val sourcesState = parseSourcesStateFile(downloadString("$remoteCacheUrl/metadata/$lastCachedCommit")) - val outputs = getAllCompilationOutputs(sourcesState, context.classesOutputDirectory) - total = outputs.size - spanBuilder("download compilation output parts").setAttribute(AttributeKey.longKey("count"), outputs.size.toLong()).use { - outputs.forEachConcurrent(downloadParallelism) { output -> - spanBuilder("get and unpack output").setAttribute("part", output.remotePath).use { - downloadAndUnpackCompilationOutput(compilationOutput = output, notFound = notFound, totalDownloadedBytes = totalDownloadedBytes) - } - } - } + if (availableCommitDepth == -1) { + return -1 } reportStatisticValue("jps-cache:download:time", TimeUnit.NANOSECONDS.toMillis((System.nanoTime() - start)).toString()) @@ -146,19 +134,75 @@ internal class PortableCompilationCacheDownloader( return availableCommitDepth } - private suspend fun downloadAndUnpackJpsCache(commitHash: String, notFound: LongAdder, totalBytes: LongAdder) { + private suspend fun PortableCompilationCacheDownloader.doDownload( + urlPathPrefix: String, + lastCachedCommit: String, + notFound: LongAdder, + totalDownloadedBytes: LongAdder, + connection: Http2ClientConnection?, + ): Int { + return coroutineScope { + launch { + spanBuilder("get and unpack jps cache").setAttribute("commit", lastCachedCommit).use { + downloadAndUnpackJpsCache( + urlPathPrefix = urlPathPrefix, + commitHash = lastCachedCommit, + notFound = notFound, + totalBytes = totalDownloadedBytes, + connection = connection, + ) + } + } + + val metadataUrlPath = "$urlPathPrefix/metadata/$lastCachedCommit" + val json = if (connection == null) { + require(metadataUrlPath.isS3()) + retryWithExponentialBackOff { + awsS3Cli("cp", metadataUrlPath, "-") + } + } + else { + connection.getString(path = metadataUrlPath) + } + + val outputs = getAllCompilationOutputs(sourceState = BuildTargetSourcesState.readJson(JsonReader(json.reader())), classOutDir = context.classesOutputDirectory) + spanBuilder("download compilation output parts").setAttribute(AttributeKey.longKey("count"), outputs.size.toLong()).use { + outputs.forEachConcurrent(downloadParallelism) { output -> + spanBuilder("get and unpack output").setAttribute("part", output.remotePath).use { + downloadAndUnpackCompilationOutput( + urlPathPrefix = urlPathPrefix, + compilationOutput = output, + notFound = notFound, + totalDownloadedBytes = totalDownloadedBytes, + connection = connection, + ) + } + } + } + outputs.size + } + } + + private suspend fun downloadAndUnpackJpsCache(urlPathPrefix: String, commitHash: String, notFound: LongAdder, totalBytes: LongAdder, connection: Http2ClientConnection?) { val cacheArchive = Files.createTempFile("cache", ".zip") try { - downloadToFile(url = "$remoteCacheUrl/caches/$commitHash", file = cacheArchive, spanName = "download jps cache", notFound = notFound) + val sizeOnDisk = downloadToFile( + urlPath = "$urlPathPrefix/caches/$commitHash", + file = cacheArchive, + spanName = "download jps cache", + notFound = notFound, + connection = connection, + ) - totalBytes.add(Files.size(cacheArchive)) + require(sizeOnDisk > 0) + totalBytes.add(sizeOnDisk) val cacheDestination = context.compilationData.dataStorageRoot spanBuilder("unpack jps cache") - .setAttribute("archive", "$cacheArchive") - .setAttribute("destination", "$cacheDestination") - .use { - Decompressor.Zip(cacheArchive).overwrite(true).extract(cacheDestination) + .setAttribute("archive", cacheArchive.toString()) + .setAttribute("destination", cacheDestination.toString()) + .use(Dispatchers.IO) { + unpackArchiveUsingNettyByteBufferPool(archiveFile = cacheArchive, outDir = cacheDestination, isCompressed = true) } } finally { @@ -166,32 +210,45 @@ internal class PortableCompilationCacheDownloader( } } - private suspend fun downloadAndUnpackCompilationOutput(compilationOutput: CompilationOutput, notFound: LongAdder, totalDownloadedBytes: LongAdder) { + private suspend fun downloadAndUnpackCompilationOutput( + urlPathPrefix: String, + compilationOutput: CompilationOutput, + notFound: LongAdder, + totalDownloadedBytes: LongAdder, + connection: Http2ClientConnection?, + ) { val tempFile = compilationOutput.path.resolve("tmp-output.zip") - downloadToFile(url = "$remoteCacheUrl/${compilationOutput.remotePath}", file = tempFile, spanName = "download output", notFound = notFound) - try { - totalDownloadedBytes.add(Files.size(tempFile)) - } - catch (e: NoSuchFileException) { - // We assume that the error is logged by downloadToFile. - // In the future, we will implement stricter behavior. - // For now, the JPS cache reports non-existent compilation outputs, and we have to use such a workaround. + val urlPath = "$urlPathPrefix/${compilationOutput.remotePath.trimStart('/')}" + val sizeOnDisk = downloadToFile( + urlPath = urlPath, + file = tempFile, + spanName = "download output", + notFound = notFound, + connection = connection, + ) + + if (sizeOnDisk <= 0) { return } + // We assume that the error is logged by downloadToFile. + // In the future, we will implement stricter behavior. + // For now, the JPS cache reports non-existent compilation outputs, and we have to use such a workaround. + totalDownloadedBytes.add(sizeOnDisk) + try { spanBuilder("unpack output") .setAttribute("archive", tempFile.toString()) .setAttribute("destination", compilationOutput.path.toString()) - .use { - Decompressor.Zip(tempFile).overwrite(true).extract(compilationOutput.path) + .use(Dispatchers.IO) { + unpackArchiveUsingNettyByteBufferPool(archiveFile = tempFile, outDir = compilationOutput.path, isCompressed = true) } } catch (e: CancellationException) { throw e } catch (e: Exception) { - throw Exception("Unable to unpack $remoteCacheUrl/${compilationOutput.remotePath} to ${compilationOutput.path}", e) + throw Exception("Unable to unpack $urlPath to ${compilationOutput.path}", e) } finally { Files.deleteIfExists(tempFile) diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheUploader.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheUploader.kt index 92e48d3f1d94..fe46e77a1d34 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheUploader.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/PortableCompilationCacheUploader.kt @@ -3,270 +3,279 @@ package org.jetbrains.intellij.build.impl.compilation import com.google.gson.stream.JsonReader -import com.intellij.util.io.Compressor +import io.netty.handler.codec.http.HttpResponseStatus import io.opentelemetry.api.common.AttributeKey import io.opentelemetry.api.common.Attributes import io.opentelemetry.api.trace.Span import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch import kotlinx.coroutines.withContext -import okhttp3.MediaType.Companion.toMediaType -import okhttp3.Request -import okhttp3.RequestBody -import okio.BufferedSink -import okio.source -import org.jetbrains.intellij.build.BuildMessages import org.jetbrains.intellij.build.CompilationContext import org.jetbrains.intellij.build.forEachConcurrent +import org.jetbrains.intellij.build.http2Client.Http2ClientConnection +import org.jetbrains.intellij.build.http2Client.getJsonOrDefaultIfNotFound +import org.jetbrains.intellij.build.http2Client.upload +import org.jetbrains.intellij.build.http2Client.withHttp2ClientConnectionFactory import org.jetbrains.intellij.build.impl.compilation.cache.CommitsHistory import org.jetbrains.intellij.build.impl.compilation.cache.getAllCompilationOutputs import org.jetbrains.intellij.build.io.copyFile import org.jetbrains.intellij.build.io.moveFile import org.jetbrains.intellij.build.io.zipWithCompression -import org.jetbrains.intellij.build.retryWithExponentialBackOff import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.use import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState import org.jetbrains.jps.incremental.storage.ProjectStamps +import java.net.URI import java.nio.file.Files import java.nio.file.NoSuchFileException import java.nio.file.Path import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicInteger -import java.util.concurrent.atomic.AtomicLong +import java.util.concurrent.atomic.LongAdder import kotlin.io.path.ExperimentalPathApi import kotlin.io.path.deleteRecursively +import kotlin.random.Random private const val SOURCES_STATE_FILE_NAME = "target_sources_state.json" -private val MEDIA_TYPE_BINARY = "application/octet-stream".toMediaType() -internal class PortableCompilationCacheUploader( - private val context: CompilationContext, - internal val remoteCache: PortableJpsCacheRemoteCacheConfig, - @JvmField val remoteGitUrl: String, - private val commitHash: String, - private val s3Folder: Path, - private val forcedUpload: Boolean, +internal suspend fun uploadJpsCache( + forcedUpload: Boolean, + commitHash: String, + authHeader: CharSequence, + s3Dir: Path?, + uploadUrl: URI, + context: CompilationContext, ) { - init { - s3Folder.deleteRecursively() - Files.createDirectories(s3Folder) + if (s3Dir != null) { + s3Dir.deleteRecursively() + Files.createDirectories(s3Dir) } - suspend fun upload(messages: BuildMessages) { - val sourceStateFile = context.compilationData.dataStorageRoot.resolve(SOURCES_STATE_FILE_NAME) - - val sourceState = try { - Files.newBufferedReader(sourceStateFile).use { - BuildTargetSourcesState.readJson(JsonReader(it)) - } - } - catch (e: NoSuchFileException) { - throw IllegalStateException( - "Compilation outputs doesn't contain source state file, " + - "please enable '${ProjectStamps.PORTABLE_CACHES_PROPERTY}' flag", e - ) + val sourceStateFile = context.compilationData.dataStorageRoot.resolve(SOURCES_STATE_FILE_NAME) + val sourceState = try { + Files.newBufferedReader(sourceStateFile).use { + BuildTargetSourcesState.readJson(JsonReader(it)) } + } + catch (e: NoSuchFileException) { + throw IllegalStateException("Compilation outputs doesn't contain source state file, please enable '${ProjectStamps.PORTABLE_CACHES_PROPERTY}' flag", e) + } - val start = System.nanoTime() - val totalUploadedBytes = AtomicLong() - val uploadedOutputCount = AtomicInteger() - val uploader = Uploader(remoteCache.uploadUrl, remoteCache.authHeader) - withContext(Dispatchers.IO) { - // Jps Caches upload is started first because of significant size - launch { - spanBuilder("upload jps cache").use { - uploadJpsCaches(uploader) + val start = System.nanoTime() + val totalUploadedBytes = LongAdder() + val uploadedOutputCount = LongAdder() + val messages = context.messages + withHttp2ClientConnectionFactory(trustAll = uploadUrl.host == "127.0.0.1") { client -> + client.connect(uploadUrl.host, uploadUrl.port).withAuth(authHeader).use { connection -> + val urlPathPrefix = uploadUrl.path + withContext(Dispatchers.IO) { + launch { + spanBuilder("upload jps cache").use { + uploadJpsCaches(urlPathPrefix = urlPathPrefix, connection = connection, s3Dir = s3Dir, forcedUpload = forcedUpload, commitHash = commitHash, context = context) + } + } + + spanBuilder("upload compilation outputs").use { + val allCompilationOutputs = getAllCompilationOutputs(sourceState = sourceState, classOutDir = context.classesOutputDirectory) + val tempDir = Files.createTempDirectory(context.paths.tempDir, "jps-cache-bytecode-") + try { + uploadCompilationOutputs( + uploadedOutputCount = uploadedOutputCount, + totalUploadedBytes = totalUploadedBytes, + allCompilationOutputs = allCompilationOutputs, + urlPathPrefix = urlPathPrefix, + connection = connection, + s3Dir = s3Dir, + forcedUpload = forcedUpload, + tempDir = tempDir, + ) + } + finally { + tempDir.deleteRecursively() + } + messages.reportStatisticValue("Total outputs", allCompilationOutputs.size.toString()) } } - spanBuilder("upload compilation outputs").use { - val allCompilationOutputs = getAllCompilationOutputs(sourceState = sourceState, classOutDir = context.classesOutputDirectory) - uploadCompilationOutputs( - uploader = uploader, - uploadedOutputCount = uploadedOutputCount, - allCompilationOutputs = allCompilationOutputs, - ) - messages.reportStatisticValue("Total outputs", allCompilationOutputs.size.toString()) - } - } - - messages.reportStatisticValue("jps-cache:upload:time", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start).toString()) - messages.reportStatisticValue("jps-cache:uploaded:count", (uploadedOutputCount.get() + 1).toString()) - messages.reportStatisticValue("jps-cache:uploaded:bytes", "$totalUploadedBytes") - - messages.reportStatisticValue("Uploaded outputs", uploadedOutputCount.get().toString()) - - uploadMetadata(sourceStateFile = sourceStateFile, uploader = uploader) - uploadToS3() - } - - private suspend fun uploadToS3() { - spanBuilder("aws s3 sync").use { - awsS3Cli("cp", "--no-progress", "--include", "*", "--recursive", s3Folder.toString(), "s3://intellij-jps-cache", returnStdOut = false) - } - } - - private suspend fun uploadJpsCaches(uploader: Uploader) { - val dataStorageRoot = context.compilationData.dataStorageRoot - val zipFile = dataStorageRoot.parent.resolve(commitHash) - Compressor.Zip(zipFile).use { zip -> - zip.addDirectory(dataStorageRoot) - } - val cachePath = "caches/$commitHash" - if (forcedUpload || !uploader.isExist(cachePath, true)) { - uploader.upload(cachePath, zipFile) - } - moveFile(zipFile, s3Folder.resolve(cachePath)) - } - - private suspend fun uploadMetadata(sourceStateFile: Path, uploader: Uploader) { - val metadataPath = "metadata/$commitHash" - spanBuilder("upload metadata").setAttribute("path", metadataPath).use { - uploader.upload(metadataPath, sourceStateFile) - copyFile(sourceStateFile, s3Folder.resolve(metadataPath)) - } - } - - private suspend fun uploadCompilationOutputs( - uploader: Uploader, - uploadedOutputCount: AtomicInteger, - allCompilationOutputs: List, - ) { - allCompilationOutputs.forEachConcurrent(uploadParallelism) { compilationOutput -> - val outDir = compilationOutput.path - spanBuilder("upload output part") - .setAttribute("part", compilationOutput.remotePath) - .setAttribute("path", outDir.toString()) - .use { span -> - val sourcePath = compilationOutput.remotePath - if (Files.notExists(outDir)) { - span.addEvent("doesn't exist, was a respective module removed?") - return@use - } - - val zipFile = context.paths.tempDir.resolve("compilation-output-zips").resolve(sourcePath) - zipWithCompression(zipFile, mapOf(outDir to "")) - if (forcedUpload || !uploader.isExist(sourcePath)) { - uploader.upload(sourcePath, zipFile) - uploadedOutputCount.incrementAndGet() - } - moveFile(zipFile, s3Folder.resolve(sourcePath)) + val metadataPath = "metadata/$commitHash" + val metadataUrl = "$urlPathPrefix/$metadataPath" + spanBuilder("upload metadata").setAttribute("path", metadataUrl).use { + connection.upload(metadataUrl, sourceStateFile) + if (s3Dir != null) { + copyFile(file = sourceStateFile, target = s3Dir.resolve(metadataPath)) } - } - } - - /** - * Upload and publish a file with commits history - */ - internal suspend fun updateCommitHistory(overrideCommits: Set? = null) { - val overrideRemoteHistory = overrideCommits != null - val commitHistory = CommitsHistory(mapOf(remoteGitUrl to (overrideCommits ?: setOf(commitHash)))) - val uploader = Uploader(serverUrl = remoteCache.uploadUrl, authHeader = remoteCache.authHeader) - for (commitHash in commitHistory.commitsForRemote(remoteGitUrl)) { - val cacheUploaded = uploader.isExist("caches/$commitHash") - val metadataUploaded = uploader.isExist("metadata/$commitHash") - if (!cacheUploaded && !metadataUploaded) { - val msg = "Unable to publish $commitHash due to missing caches/$commitHash and metadata/$commitHash. " + - "Probably caused by previous cleanup build." - if (overrideRemoteHistory) context.messages.error(msg) else context.messages.warning(msg) - return - } - check(cacheUploaded == metadataUploaded) { - "JPS Caches are uploaded: $cacheUploaded, metadata is uploaded: $metadataUploaded" } } - val newHistory = if (overrideRemoteHistory) commitHistory else commitHistory + remoteCommitHistory(uploader) - uploader.upload(path = CommitsHistory.JSON_FILE, file = writeCommitHistory(newHistory)) - val expected = newHistory.commitsForRemote(remoteGitUrl).toSet() - val actual = remoteCommitHistory(uploader).commitsForRemote(remoteGitUrl).toSet() - val missing = expected - actual - val unexpected = actual - expected - check(missing.none() && unexpected.none()) { - """ - Missing: $missing - Unexpected: $unexpected - """.trimIndent() - } } - private suspend fun remoteCommitHistory(uploader: Uploader): CommitsHistory { - return if (uploader.isExist(CommitsHistory.JSON_FILE)) { - CommitsHistory(uploader.getAsString(CommitsHistory.JSON_FILE, remoteCache.authHeader)) - } - else { - CommitsHistory(emptyMap()) - } - } + messages.reportStatisticValue("jps-cache:upload:time", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start).toString()) + messages.reportStatisticValue("jps-cache:uploaded:count", (uploadedOutputCount.sum() + 1).toString()) + messages.reportStatisticValue("jps-cache:uploaded:bytes", totalUploadedBytes.sum().toString()) - private fun writeCommitHistory(commitHistory: CommitsHistory): Path { - val commitHistoryFile = s3Folder.resolve(CommitsHistory.JSON_FILE) - Files.createDirectories(commitHistoryFile.parent) - val json = commitHistory.toJson() - Files.writeString(commitHistoryFile, json) - Span.current().addEvent("write commit history", Attributes.of(AttributeKey.stringKey("data"), json)) - return commitHistoryFile + messages.reportStatisticValue("Uploaded outputs", uploadedOutputCount.sum().toString()) + + if (s3Dir != null) { + uploadToS3(s3Dir) } } -private class Uploader(serverUrl: String, @JvmField val authHeader: String) { - private val serverUrl = serverUrl.trimEnd('/') +private suspend fun uploadToS3(s3Dir: Path) { + spanBuilder("aws s3 sync").use { + awsS3Cli("cp", "--no-progress", "--include", "*", "--recursive", s3Dir.toString(), "s3://intellij-jps-cache", returnStdOut = false) + } +} - suspend fun upload(path: String, file: Path) { - val url = pathToUrl(path) - spanBuilder("upload").setAttribute("url", url).setAttribute("path", path).use { - val fileSize = Files.size(file) - retryWithExponentialBackOff { - httpClient.newCall(Request.Builder().url(url) - .header("Authorization", authHeader) - .put(object : RequestBody() { - override fun contentType() = MEDIA_TYPE_BINARY +private suspend fun uploadJpsCaches( + urlPathPrefix: String, + connection: Http2ClientConnection, + commitHash: String, + forcedUpload: Boolean, + s3Dir: Path?, + context: CompilationContext, +) { + val dataStorageRoot = context.compilationData.dataStorageRoot + val zipFile = context.paths.tempDir.resolve("$commitHash-${java.lang.Long.toUnsignedString(Random.nextLong(), Character.MAX_RADIX)}.zip") + try { + val compressed by lazy { + zipWithCompression(zipFile, mapOf(dataStorageRoot to "")) + zipFile + } - override fun contentLength(): Long = fileSize + val cachePath = "caches/$commitHash" + val urlPath = "$urlPathPrefix/$cachePath" + if (forcedUpload || !checkExists(connection, urlPath, logIfExists = true)) { + connection.upload(path = urlPath, file = compressed) + } - override fun writeTo(sink: BufferedSink) { - file.source().use { - sink.writeAll(it) - } - } - }).build()).executeAsync().useSuccessful {} - } + if (s3Dir != null) { + moveFile(compressed, s3Dir.resolve(cachePath)) } } + finally { + Files.deleteIfExists(zipFile) + } +} - suspend fun isExist(path: String, logIfExists: Boolean = false): Boolean { - val url = pathToUrl(path) - return spanBuilder("head").setAttribute("url", url).use { span -> - val code = retryWithExponentialBackOff { - httpClient.head(url, authHeader) - } +private suspend fun uploadCompilationOutputs( + uploadedOutputCount: LongAdder, + allCompilationOutputs: List, + urlPathPrefix: String, + connection: Http2ClientConnection, + totalUploadedBytes: LongAdder, + forcedUpload: Boolean, + tempDir: Path, + s3Dir: Path?, +) { + allCompilationOutputs.forEachConcurrent(uploadParallelism) { compilationOutput -> + val outDir = compilationOutput.path + spanBuilder("upload output part") + .setAttribute("part", compilationOutput.remotePath) + .setAttribute("path", outDir.toString()) + .use { span -> + if (Files.notExists(outDir)) { + span.addEvent("doesn't exist, was a respective module removed?") + return@use + } - if (code != 200) { - return@use false - } + val sourcePath = compilationOutput.remotePath + val zipFile = tempDir.resolve("${sourcePath.replace('/', '_')}.zip") + zipWithCompression(targetFile = zipFile, dirs = mapOf(outDir to ""), createFileParentDirs = false) - try { - /** - * FIXME dirty workaround for unreliable [serverUrl] - */ - httpClient.get(url, authHeader) { - it.peekBody(byteCount = 1) + val urlPath = "$urlPathPrefix/$sourcePath" + if (forcedUpload || !checkExists(connection, urlPath)) { + spanBuilder("upload").setAttribute("urlPath", urlPath).setAttribute("path", sourcePath).use { + val result = connection.upload(path = urlPath, file = zipFile) + totalUploadedBytes.add(result.uploadedSize) + } + uploadedOutputCount.increment() + } + + if (s3Dir != null) { + moveFile(zipFile, s3Dir.resolve(sourcePath)) } } - catch (_: Exception) { - return@use false + } +} + +/** + * Upload and publish a file with commits history + */ +internal suspend fun updateJpsCacheCommitHistory( + overrideCommits: Set?, + remoteGitUrl: String, + commitHash: String, + uploadUrl: URI, + authHeader: CharSequence, + s3Dir: Path?, + context: CompilationContext, +) { + val overrideRemoteHistory = overrideCommits != null + val commitHistory = CommitsHistory(mapOf(remoteGitUrl to (overrideCommits ?: setOf(commitHash)))) + withHttp2ClientConnectionFactory(trustAll = uploadUrl.host == "127.0.0.1") { client -> + val urlPathPrefix = uploadUrl.path + client.connect(uploadUrl.host, uploadUrl.port).withAuth(authHeader).use { connection -> + for (commitHashForRemote in commitHistory.commitsForRemote(remoteGitUrl)) { + val cacheUploaded = checkExists(connection, "$urlPathPrefix/caches/$commitHashForRemote") + val metadataUploaded = checkExists(connection, "$urlPathPrefix/metadata/$commitHashForRemote") + if (!cacheUploaded && !metadataUploaded) { + val msg = "Unable to publish $commitHashForRemote due to missing caches/$commitHashForRemote and metadata/$commitHashForRemote." + + " Probably caused by previous cleanup build." + if (overrideRemoteHistory) { + context.messages.error(msg) + } + else { + context.messages.warning(msg) + } + return@use false + } + check(cacheUploaded == metadataUploaded) { + "JPS Caches are uploaded: $cacheUploaded, metadata is uploaded: $metadataUploaded" + } } - if (logIfExists) { - span.addEvent("File '$path' already exists on server, nothing to upload") + val newHistory = if (overrideRemoteHistory) commitHistory else commitHistory + remoteCommitHistory(connection, urlPathPrefix) + connection.upload(path = "$urlPathPrefix/${CommitsHistory.JSON_FILE}", file = writeCommitHistory(commitHistory = newHistory, context = context, s3Dir = s3Dir)) + val expected = newHistory.commitsForRemote(remoteGitUrl).toSet() + val actual = remoteCommitHistory(connection, urlPathPrefix).commitsForRemote(remoteGitUrl).toSet() + val missing = expected - actual + val unexpected = actual - expected + check(missing.none() && unexpected.none()) { + """ + Missing: $missing + Unexpected: $unexpected + """.trimIndent() } - true } } +} - suspend fun getAsString(path: String, authHeader: String) = retryWithExponentialBackOff { - httpClient.get(pathToUrl(path), authHeader) { it.body.string() } +private suspend fun remoteCommitHistory(connection: Http2ClientConnection, urlPathPrefix: String): CommitsHistory { + return CommitsHistory(connection.getJsonOrDefaultIfNotFound(path = "$urlPathPrefix/${CommitsHistory.JSON_FILE}", defaultIfNotFound = emptyMap())) +} + +private fun writeCommitHistory(commitHistory: CommitsHistory, context: CompilationContext, s3Dir: Path?): Path { + val commitHistoryFile = (s3Dir ?: context.paths.tempDir).resolve(CommitsHistory.JSON_FILE) + Files.createDirectories(commitHistoryFile.parent) + val json = commitHistory.toJson() + Files.writeString(commitHistoryFile, json) + Span.current().addEvent("write commit history", Attributes.of(AttributeKey.stringKey("data"), json)) + return commitHistoryFile +} + +private suspend fun checkExists( + connection: Http2ClientConnection, + urlPath: String, + logIfExists: Boolean = false, +): Boolean { + val status = connection.head(urlPath) + if (status == HttpResponseStatus.OK) { + // already exists + if (logIfExists) { + Span.current().addEvent("File $urlPath already exists on server, nothing to upload") + } + return true } - - private fun pathToUrl(path: String) = "$serverUrl/${path.trimStart('/')}" + else if (status != HttpResponseStatus.NOT_FOUND) { + Span.current().addEvent("unexpected response status for HEAD request", Attributes.of(AttributeKey.stringKey("status"), status.toString())) + } + return false } \ No newline at end of file diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/CommitsHistory.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/CommitsHistory.kt index a4bce0ca0efc..c738ce5ec85c 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/CommitsHistory.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/CommitsHistory.kt @@ -1,19 +1,18 @@ -// Copyright 2000-2022 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package org.jetbrains.intellij.build.impl.compilation.cache -import com.google.gson.Gson -import com.google.gson.reflect.TypeToken +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json class CommitsHistory(private val commitsPerRemote: Map>) { companion object { - const val JSON_FILE = "commit_history.json" - private val JSON_TYPE = object : TypeToken>>() {}.type + internal const val JSON_FILE = "commit_history.json" } - constructor(json: String) : this(Gson().fromJson(json, JSON_TYPE) as Map>) + constructor(json: String) : this(Json.decodeFromString>>(json)) fun toJson(): String { - return Gson().toJson(commitsPerRemote) + return Json.encodeToString(commitsPerRemote) } fun commitsForRemote(remote: String): Collection { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/SourcesStateProcessor.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/SourcesStateProcessor.kt index 334f16a67921..48b5b1077718 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/SourcesStateProcessor.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/cache/SourcesStateProcessor.kt @@ -1,12 +1,10 @@ // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package org.jetbrains.intellij.build.impl.compilation.cache -import com.google.gson.stream.JsonReader import org.jetbrains.intellij.build.impl.compilation.CompilationOutput import org.jetbrains.jps.builders.java.JavaModuleBuildTargetType import org.jetbrains.jps.builders.java.ResourcesTargetType import org.jetbrains.jps.cache.model.BuildTargetState -import org.jetbrains.jps.incremental.storage.BuildTargetSourcesState import java.nio.file.Path private const val IDENTIFIER = "\$BUILD_DIR\$" @@ -23,10 +21,6 @@ internal fun getAllCompilationOutputs(sourceState: Map> { - return BuildTargetSourcesState.readJson(JsonReader(json.reader())) -} - private fun getProductionCompilationOutputs(currentSourcesState: Map>, classOutDir: Path): List { return getCompilationOutputs( // production modules with both classes and resources diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/digest.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/digest.kt new file mode 100644 index 000000000000..813a1d20d614 --- /dev/null +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/digest.kt @@ -0,0 +1,37 @@ +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +package org.jetbrains.intellij.build.impl.compilation + +import org.jetbrains.intellij.build.http2Client.READ_OPERATION +import java.math.BigInteger +import java.nio.ByteBuffer +import java.nio.channels.FileChannel +import java.nio.file.Path +import java.security.MessageDigest + +private val sharedDigest = MessageDigest.getInstance("SHA-256", java.security.Security.getProvider("SUN")) +internal fun sha256() = sharedDigest.clone() as MessageDigest + +internal fun computeHash(file: Path): String { + val messageDigest = sha256() + FileChannel.open(file, READ_OPERATION).use { channel -> + val fileSize = channel.size() + // java message digest doesn't support native buffer (copies to a heap byte array in any case) + val bufferSize = 256 * 1024 + val buffer = ByteBuffer.allocate(bufferSize) + var offset = 0L + var readBytes: Int + while (offset < fileSize) { + buffer.clear() + readBytes = channel.read(buffer, offset) + if (readBytes <= 0) { + break + } + + messageDigest.update(buffer.array(), 0, readBytes) + offset += readBytes + } + } + return digestToString(messageDigest) +} + +internal fun digestToString(digest: MessageDigest): String = BigInteger(1, digest.digest()).toString(36) + "z" \ No newline at end of file diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/download.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/download.kt index 95a3bb4fed9c..0157206ccd5b 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/download.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/download.kt @@ -1,7 +1,6 @@ // Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package org.jetbrains.intellij.build.impl.compilation -import com.intellij.util.lang.HashMapZipFile import io.opentelemetry.api.common.AttributeKey import io.opentelemetry.api.common.Attributes import kotlinx.coroutines.isActive @@ -11,21 +10,14 @@ import org.jetbrains.intellij.build.http2Client.Http2ClientConnection import org.jetbrains.intellij.build.http2Client.Http2ClientConnectionFactory import org.jetbrains.intellij.build.http2Client.ZstdDecompressContextPool import org.jetbrains.intellij.build.http2Client.download -import org.jetbrains.intellij.build.io.INDEX_FILENAME import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.use import java.net.URI -import java.nio.channels.FileChannel import java.nio.file.Files -import java.nio.file.Path -import java.nio.file.StandardOpenOption -import java.util.* import java.util.concurrent.CancellationException import java.util.concurrent.atomic.AtomicLong import kotlin.io.path.name -private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING) - internal suspend fun downloadCompilationCache( serverUrl: String, client: Http2ClientConnectionFactory, @@ -35,38 +27,10 @@ internal suspend fun downloadCompilationCache( skipUnpack: Boolean, saveHash: Boolean, ) { - var urlPathWithPrefix = "/$prefix/" - // first let's check for initial redirect (mirror selection) - val initialServerUri = URI(serverUrl) - var effectiveServerUri = initialServerUri - var connection: Http2ClientConnection? = client.connect(effectiveServerUri.host, effectiveServerUri.port) - try { - spanBuilder("mirror selection").use { span -> - val newLocation = connection!!.getRedirectLocation(urlPathWithPrefix) - if (newLocation == null) { - span.addEvent("origin server will be used", Attributes.of(AttributeKey.stringKey("url"), urlPathWithPrefix)) - } - else { - effectiveServerUri = URI(newLocation.toString()) - urlPathWithPrefix = effectiveServerUri.path - span.addEvent("redirected to mirror", Attributes.of(AttributeKey.stringKey("url"), urlPathWithPrefix)) - } - } - } - finally { - if (initialServerUri != effectiveServerUri) { - connection?.close() - connection = null - } - } - - if (connection == null) { - connection = client.connect(effectiveServerUri.host, effectiveServerUri.port) - } - try { + checkMirrorAndConnect(serverUrl = serverUrl, client = client, prefix = prefix) { connection, urlPathPrefix -> val zstdDecompressContextPool = ZstdDecompressContextPool() toDownload.forEachConcurrent(downloadParallelism) { item -> - val urlPath = "$urlPathWithPrefix${item.name}/${item.file.fileName}" + val urlPath = "$urlPathPrefix/${item.name}/${item.file.fileName}" spanBuilder("download").setAttribute("name", item.name).setAttribute("urlPath", urlPath).use { span -> try { downloadedBytes.getAndAdd( @@ -93,8 +57,46 @@ internal suspend fun downloadCompilationCache( } } } +} + +internal suspend fun checkMirrorAndConnect( + serverUrl: String, + client: Http2ClientConnectionFactory, + prefix: String, + block: suspend (connection: Http2ClientConnection, urlPathPrefix: String) -> Unit, +) { + var urlPathWithSlash = "/$prefix/" + // first let's check for initial redirect (mirror selection) + val initialServerUri = URI(serverUrl) + var effectiveServerUri = initialServerUri + var connection: Http2ClientConnection? = client.connect(effectiveServerUri.host, effectiveServerUri.port) + var connectionToClose = connection + try { + spanBuilder("mirror selection").use { span -> + val newLocation = connection!!.getRedirectLocation(urlPathWithSlash) + if (newLocation == null) { + span.addEvent("origin server will be used", Attributes.of(AttributeKey.stringKey("url"), urlPathWithSlash)) + connectionToClose = null + } + else { + effectiveServerUri = URI(newLocation.toString()) + urlPathWithSlash = effectiveServerUri.path + span.addEvent("redirected to mirror", Attributes.of(AttributeKey.stringKey("url"), urlPathWithSlash)) + connection = null + } + } + } finally { - connection.close() + connectionToClose?.close() + connectionToClose = null + } + + val effectiveConnection = connection ?: client.connect(effectiveServerUri.host, effectiveServerUri.port) + try { + block(effectiveConnection, urlPathWithSlash.trimEnd('/')) + } + finally { + effectiveConnection.close() } } @@ -123,7 +125,7 @@ private suspend fun download( if (!skipUnpack) { spanBuilder("unpack").setAttribute("name", item.name).use { - unpackArchive(item, saveHash) + unpackCompilationPartArchive(item, saveHash) } } return downloaded @@ -135,29 +137,8 @@ internal class CompilePartDownloadFailedError(@JvmField val item: FetchAndUnpack internal class HashMismatchException(message: String) : IOException(message) -internal fun unpackArchive(item: FetchAndUnpackItem, saveHash: Boolean) { - HashMapZipFile.load(item.file).use { zipFile -> - val root = item.output - Files.createDirectories(root) - val createdDirs = HashSet() - createdDirs.add(root) - for (entry in zipFile.entries) { - if (entry.isDirectory || entry.name == INDEX_FILENAME) { - continue - } - - val file = root.resolve(entry.name) - val parent = file.parent - if (createdDirs.add(parent)) { - Files.createDirectories(parent) - } - - FileChannel.open(file, OVERWRITE_OPERATION).use { channel -> - channel.write(entry.getByteBuffer(zipFile), 0) - } - } - } - +internal fun unpackCompilationPartArchive(item: FetchAndUnpackItem, saveHash: Boolean) { + unpackArchiveUsingNettyByteBufferPool(archiveFile = item.file, outDir = item.output, isCompressed = false) if (saveHash) { // save actual hash Files.writeString(item.output.resolve(".hash"), item.hash) diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/http.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/http.kt index 1007f0e72c08..63d87aeb38a4 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/http.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/http.kt @@ -5,34 +5,10 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.suspendCancellableCoroutine import okhttp3.* import okhttp3.internal.closeQuietly -import org.jetbrains.intellij.build.NoMoreRetriesException import java.io.IOException import java.util.concurrent.TimeUnit import kotlin.coroutines.resumeWithException -internal suspend fun OkHttpClient.head(url: String, authHeader: String): Int { - return newCall(Request.Builder().url(url).head().header("Authorization", authHeader).build()).executeAsync().use { response -> - if (response.code != 200 && response.code != 404) { - throw IOException("Unexpected code $response") - } - response.code - } -} - -internal suspend fun OkHttpClient.get(url: String, authHeader: String, task: (Response) -> T): T { - return newCall(Request.Builder().url(url).header("Authorization", authHeader).build()).executeAsync().useSuccessful(task) -} - -internal inline fun Response.useSuccessful(task: (Response) -> T): T { - return use { response -> - when { - response.isSuccessful -> task(response) - response.code == 404 -> throw NoMoreRetriesException("Unexpected code $response") - else -> throw IOException("Unexpected code $response") - } - } -} - internal val httpClient: OkHttpClient by lazy { val timeout = 1L val unit = TimeUnit.MINUTES @@ -79,7 +55,7 @@ internal val httpClient: OkHttpClient by lazy { .build() } -@ExperimentalCoroutinesApi // resume with a resource cleanup. +@ExperimentalCoroutinesApi internal suspend fun Call.executeAsync(): Response { return suspendCancellableCoroutine { continuation -> continuation.invokeOnCancellation { diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/upload.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/upload.kt index ea5b3410e945..03863d1a74d7 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/upload.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/upload.kt @@ -13,6 +13,7 @@ import kotlinx.serialization.Serializable import org.jetbrains.intellij.build.forEachConcurrent import org.jetbrains.intellij.build.http2Client.Http2ClientConnection import org.jetbrains.intellij.build.http2Client.ZstdCompressContextPool +import org.jetbrains.intellij.build.http2Client.post import org.jetbrains.intellij.build.http2Client.upload import org.jetbrains.intellij.build.telemetry.TraceManager.spanBuilder import org.jetbrains.intellij.build.telemetry.use diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/zip.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/zip.kt new file mode 100644 index 000000000000..43ac70c17c18 --- /dev/null +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/compilation/zip.kt @@ -0,0 +1,57 @@ +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +package org.jetbrains.intellij.build.impl.compilation + +import com.intellij.util.lang.HashMapZipFile +import io.netty.buffer.ByteBuf +import io.netty.buffer.PooledByteBufAllocator +import org.jetbrains.intellij.build.io.INDEX_FILENAME +import java.nio.channels.FileChannel +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption +import java.util.* +import java.util.function.IntFunction + +private val OVERWRITE_OPERATION = EnumSet.of(StandardOpenOption.WRITE, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING) + +internal fun unpackArchiveUsingNettyByteBufferPool(archiveFile: Path, outDir: Path, isCompressed: Boolean) { + Files.createDirectories(outDir) + HashMapZipFile.load(archiveFile).use { zipFile -> + val createdDirs = HashSet() + createdDirs.add(outDir) + for (entry in zipFile.entries) { + if (entry.isDirectory || entry.name == INDEX_FILENAME) { + continue + } + + val file = outDir.resolve(entry.name) + val parent = file.parent + if (createdDirs.add(parent)) { + Files.createDirectories(parent) + } + + FileChannel.open(file, OVERWRITE_OPERATION).use { channel -> + if (isCompressed) { + // we use netty buffer as a part of compilation cache/parts tasks - reuse it + var nettyBuffer: ByteBuf? = null + try { + val byteBuffer = entry.getByteBuffer( + zipFile, + IntFunction { size -> + val buffer = PooledByteBufAllocator.DEFAULT.directBuffer(size) + nettyBuffer = buffer + buffer.nioBuffer(0, size) + }) + channel.write(byteBuffer, 0) + } + finally { + nettyBuffer?.release() + } + } + else { + channel.write(entry.getByteBuffer(zipFile, null)) + } + } + } + } +} \ No newline at end of file diff --git a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/logging/jps/JpsLoggerFactory.kt b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/logging/jps/JpsLoggerFactory.kt index c400451ef119..c68b220f7220 100644 --- a/platform/build-scripts/src/org/jetbrains/intellij/build/impl/logging/jps/JpsLoggerFactory.kt +++ b/platform/build-scripts/src/org/jetbrains/intellij/build/impl/logging/jps/JpsLoggerFactory.kt @@ -1,11 +1,10 @@ -// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. @file:Suppress("ReplacePutWithAssignment", "ReplaceGetOrSet", "HardCodedStringLiteral") package org.jetbrains.intellij.build.impl.logging.jps import com.intellij.openapi.diagnostic.DefaultLogger import com.intellij.openapi.diagnostic.Logger -import com.intellij.util.containers.MultiMap import com.jetbrains.plugin.structure.base.utils.createParentDirs import io.opentelemetry.api.common.AttributeKey import io.opentelemetry.api.common.Attributes @@ -23,11 +22,12 @@ import java.nio.file.Files import java.util.* import java.util.concurrent.CancellationException import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.CopyOnWriteArrayList import java.util.concurrent.TimeUnit import java.util.function.BiConsumer @Internal -suspend fun withJpsLogging(context: CompilationContext, span: Span, action: suspend (JpsMessageHandler) -> Unit) { +internal suspend fun withJpsLogging(context: CompilationContext, span: Span, action: suspend (JpsMessageHandler) -> Unit) { val messageHandler = JpsMessageHandler(context, span) JpsLoggerFactory.messageHandler = messageHandler if (context.options.compilationLogEnabled) { @@ -69,7 +69,9 @@ suspend fun withJpsLogging(context: CompilationContext, span: Span, action: susp @VisibleForTesting class JpsLoggerFactory : Logger.Factory { companion object { - var messageHandler: JpsMessageHandler? = null + @JvmField + internal var messageHandler: JpsMessageHandler? = null + @JvmField var fileLoggerFactory: JpsFileLoggerFactory? = null } @@ -78,12 +80,12 @@ class JpsLoggerFactory : Logger.Factory { } } -@Internal -class JpsMessageHandler(private val context: CompilationContext, private val span: Span) : MessageHandler { - val errorMessagesByCompiler = MultiMap.createConcurrent() +internal class JpsMessageHandler(private val context: CompilationContext, private val span: Span) : MessageHandler { + @JvmField + val errorMessagesByCompiler = ConcurrentHashMap>() private val compilationStartTimeForTarget = ConcurrentHashMap() private val compilationFinishTimeForTarget = ConcurrentHashMap() - private var progress = (-1.0).toFloat() + private var progress = -1.0f override fun processMessage(message: BuildMessage): Unit = TeamCityBuildMessageLogger.withFlow(span) { val text = message.messageText @@ -97,8 +99,10 @@ class JpsMessageHandler(private val context: CompilationContext, private val spa messageText = buildString { if (sourcePath != null) { append(sourcePath) - if (message.line != -1L) append(":" + message.line) - appendLine(":") + if (message.line != -1L) { + append(':').append(message.line) + } + appendLine(':') } append(text) val moduleNames = message.moduleNames @@ -111,7 +115,7 @@ class JpsMessageHandler(private val context: CompilationContext, private val spa compilerName = "" messageText = text } - errorMessagesByCompiler.putValue(compilerName, messageText) + errorMessagesByCompiler.computeIfAbsent(compilerName) { CopyOnWriteArrayList() }.add(messageText) } BuildMessage.Kind.WARNING -> context.messages.warning(text) BuildMessage.Kind.INFO, BuildMessage.Kind.JPS_INFO -> if (message is BuilderStatisticsMessage) { diff --git a/platform/util/zip/src/com/intellij/util/lang/HashMapZipFile.java b/platform/util/zip/src/com/intellij/util/lang/HashMapZipFile.java index 388442748dfe..5badb5d5df47 100644 --- a/platform/util/zip/src/com/intellij/util/lang/HashMapZipFile.java +++ b/platform/util/zip/src/com/intellij/util/lang/HashMapZipFile.java @@ -1,4 +1,4 @@ -// Copyright 2000-2023 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package com.intellij.util.lang; import org.jetbrains.annotations.ApiStatus; @@ -96,7 +96,7 @@ public final class HashMapZipFile implements ZipFile { @Override public ByteBuffer getByteBuffer(@NotNull String path) throws IOException { ImmutableZipEntry entry = getRawEntry(path.charAt(0) == '/' ? path.substring(1) : path); - return entry == null ? null : entry.getByteBuffer(this); + return entry == null ? null : entry.getByteBuffer(this, null); } /** @@ -122,7 +122,7 @@ public final class HashMapZipFile implements ZipFile { @Override public @NotNull ByteBuffer getByteBuffer() throws IOException { - return entry.getByteBuffer(HashMapZipFile.this); + return entry.getByteBuffer(HashMapZipFile.this, null); } @Override diff --git a/platform/util/zip/src/com/intellij/util/lang/ImmutableZipEntry.java b/platform/util/zip/src/com/intellij/util/lang/ImmutableZipEntry.java index 2c9084e1e3a0..cedb28d72066 100644 --- a/platform/util/zip/src/com/intellij/util/lang/ImmutableZipEntry.java +++ b/platform/util/zip/src/com/intellij/util/lang/ImmutableZipEntry.java @@ -1,13 +1,15 @@ -// Copyright 2000-2021 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license that can be found in the LICENSE file. +// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. package com.intellij.util.lang; import org.jetbrains.annotations.ApiStatus; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import java.io.EOFException; import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.util.function.IntFunction; import java.util.zip.DataFormatException; import java.util.zip.Inflater; import java.util.zip.ZipException; @@ -111,14 +113,14 @@ public final class ImmutableZipEntry { @ApiStatus.Internal public InputStream getInputStream(@NotNull HashMapZipFile file) throws IOException { - return new DirectByteBufferBackedInputStream(getByteBuffer(file), method == DEFLATED); + return new DirectByteBufferBackedInputStream(getByteBuffer(file, null), method == DEFLATED); } /** * Release returned buffer using {@link ZipFile#releaseBuffer} after use. */ @ApiStatus.Internal - public ByteBuffer getByteBuffer(@NotNull HashMapZipFile file) throws IOException { + public ByteBuffer getByteBuffer(@NotNull HashMapZipFile file, @Nullable IntFunction allocator) throws IOException { if (uncompressedSize < 0) { throw new IOException("no data"); } @@ -135,7 +137,9 @@ public final class ImmutableZipEntry { Inflater inflater = new Inflater(true); inflater.setInput(inputBuffer); try { - ByteBuffer result = DirectByteBufferPool.DEFAULT_POOL.allocate(uncompressedSize); + ByteBuffer result = allocator == null + ? DirectByteBufferPool.DEFAULT_POOL.allocate(uncompressedSize) + : allocator.apply(uncompressedSize); while (result.hasRemaining()) { if (inflater.inflate(result) == 0) { throw new IllegalStateException("Inflater wants input, but input was already set");