diff --git a/python/python-exec-service/resources/intellij.python.community.execService.xml b/python/python-exec-service/resources/intellij.python.community.execService.xml index 180149211d2c..188e07c95195 100644 --- a/python/python-exec-service/resources/intellij.python.community.execService.xml +++ b/python/python-exec-service/resources/intellij.python.community.execService.xml @@ -18,4 +18,18 @@ /> + + + + + + + + \ No newline at end of file diff --git a/python/python-exec-service/src/com/intellij/python/community/execService/api.kt b/python/python-exec-service/src/com/intellij/python/community/execService/api.kt index abc502c764ef..de1eb7763bf1 100644 --- a/python/python-exec-service/src/com/intellij/python/community/execService/api.kt +++ b/python/python-exec-service/src/com/intellij/python/community/execService/api.kt @@ -6,6 +6,8 @@ import com.intellij.execution.process.ProcessOutputTypes import com.intellij.execution.target.FullPathOnTarget import com.intellij.execution.target.TargetEnvironmentConfiguration import com.intellij.execution.target.TargetedCommandLineBuilder +import com.intellij.openapi.application.ApplicationManager +import com.intellij.openapi.components.service import com.intellij.openapi.util.NlsSafe import com.intellij.platform.eel.EelApi import com.intellij.platform.eel.getShell @@ -33,7 +35,7 @@ import kotlin.time.Duration.Companion.minutes /** * Default service implementation */ -fun ExecService(): ExecService = ExecServiceImpl +fun ExecService(): ExecService = ApplicationManager.getApplication().service() /** @@ -219,18 +221,29 @@ open class ZeroCodeStdoutParserTransformer(val stdoutParser: (String) -> Resu } } +/** + * Each process launched with [ExecOptions] belongs to one of these categories. The lighter proces is, the more processes system can run. + * Limits are set via Registry. + */ +enum class ConcurrentProcessWeight { + LIGHT, + MEDIUM, + HEAVY +} /** * @property[env] Environment variables to be applied with the process run * @property[timeout] Process gets killed after this timeout * @property[processDescription] optional description to be displayed to user * @property[tty] Much like [com.intellij.platform.eel.EelExecApi.Pty] + * @property[weight] use it to limit the number of concurrent processes not to exhaust user resources, see [ConcurrentProcessWeight] */ data class ExecOptions( override val env: Map = emptyMap(), override val processDescription: @Nls String? = null, val timeout: Duration = 5.minutes, override val tty: TtySize? = null, + val weight: ConcurrentProcessWeight = ConcurrentProcessWeight.LIGHT, ) : ExecOptionsBase diff --git a/python/python-exec-service/src/com/intellij/python/community/execService/impl/ConcurrencyLimiter.kt b/python/python-exec-service/src/com/intellij/python/community/execService/impl/ConcurrencyLimiter.kt new file mode 100644 index 000000000000..7655b9818f37 --- /dev/null +++ b/python/python-exec-service/src/com/intellij/python/community/execService/impl/ConcurrencyLimiter.kt @@ -0,0 +1,31 @@ +// Copyright 2000-2025 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +package com.intellij.python.community.execService.impl + +import com.intellij.openapi.diagnostic.fileLogger +import com.intellij.python.community.execService.ConcurrentProcessWeight +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.sync.withPermit +import java.util.* + +internal class ConcurrencyLimiter(private val getBucketSize: (ConcurrentProcessWeight) -> Int) { + companion object { + private val logger = fileLogger() + } + + // Registry might not be accessible at the early stage, but once this object is constructed, the app is started, hence there is a registry. + private val semaphores = EnumMap(ConcurrentProcessWeight::class.java) + private val accessSemaMutex = Mutex() + + + suspend fun exec(weight: ConcurrentProcessWeight, code: suspend () -> T): T { + val sema = accessSemaMutex.withLock { semaphores.getOrPut(weight) { Semaphore(getBucketSize(weight)) } } + return sema.withPermit { + if (logger.isTraceEnabled) { + logger.trace("For $weight perms. left: ${sema.availablePermits}") + } + code() + } + } +} \ No newline at end of file diff --git a/python/python-exec-service/src/com/intellij/python/community/execService/impl/ExecServiceImpl.kt b/python/python-exec-service/src/com/intellij/python/community/execService/impl/ExecServiceImpl.kt index 24d9e130c50e..9961a4424312 100644 --- a/python/python-exec-service/src/com/intellij/python/community/execService/impl/ExecServiceImpl.kt +++ b/python/python-exec-service/src/com/intellij/python/community/execService/impl/ExecServiceImpl.kt @@ -5,6 +5,7 @@ import com.intellij.openapi.application.ApplicationManager import com.intellij.openapi.components.Service import com.intellij.openapi.components.service import com.intellij.openapi.diagnostic.fileLogger +import com.intellij.openapi.util.registry.Registry import com.intellij.python.community.execService.* import com.intellij.python.community.execService.impl.processLaunchers.LaunchRequest import com.intellij.python.community.execService.impl.processLaunchers.ProcessLauncher @@ -19,7 +20,15 @@ import kotlinx.coroutines.withTimeout import org.jetbrains.annotations.Nls -internal object ExecServiceImpl : ExecService { +@Service(Service.Level.APP) +internal class ExecServiceImpl private constructor() : ExecService { + private val execLimiter = ConcurrencyLimiter { + when (it) { + ConcurrentProcessWeight.LIGHT -> Registry.intValue("python.execService.limit.light") + ConcurrentProcessWeight.MEDIUM -> Registry.intValue("python.execService.limit.medium") + ConcurrentProcessWeight.HEAVY -> Registry.intValue("python.execService.limit.heavy") + } + } override suspend fun executeGetProcess(binary: BinaryToExec, args: Args, scopeToBind: CoroutineScope?, options: ExecGetProcessOptions): Result> { val launcher = create(binary, args, options, scopeToBind).getOr { return it } @@ -51,46 +60,57 @@ internal object ExecServiceImpl : ExecService { return@coroutineScope it.asPyError() } - val description = options.processDescription - ?: PyExecBundle.message("py.exec.defaultName.process", (listOf(processLauncher.exeForError.toString()) + processLauncher.args).joinToString(" ")) - val process = processLauncher.start().getOr { - val message = PyExecBundle.message("py.exec.start.error", description, it.error.cantExecProcessError, it.error.errNo - ?: "unknown") - return@coroutineScope processLauncher.createExecError( - messageToUser = message, - errorReason = it.error - ) + return@coroutineScope execLimiter.exec(options.weight) { + executeProcessGetResultImpl(binary, options, processLauncher, processInteractiveHandler) } - - val result = try { - withTimeout(options.timeout) { - val interactiveResult = processInteractiveHandler.getResultFromProcess(binary, processLauncher.args, process) - - val successResult = interactiveResult.getOr { failure -> - val (output, customErrorMessage) = failure.error - val additionalMessage = customErrorMessage ?: run { - PyExecBundle.message("py.exec.exitCode.error", description, output.exitCode) - } - return@withTimeout processLauncher.createExecError( - messageToUser = additionalMessage, - errorReason = ExecErrorReason.UnexpectedProcessTermination(output), - loggedProcessId = process.loggedProcess.id, - ) - } - Result.success(successResult) - } - } - catch (_: TimeoutCancellationException) { - processLauncher.killAndJoin() - processLauncher.createExecError( - messageToUser = PyExecBundle.message("py.exec.timeout.error", description, options.timeout), - errorReason = ExecErrorReason.Timeout, - loggedProcessId = process.loggedProcess.id, - ) - } - return@coroutineScope result } } + + private suspend fun executeProcessGetResultImpl( + binary: BinaryToExec, + options: ExecOptions, + processLauncher: ProcessLauncher, + processInteractiveHandler: ProcessInteractiveHandler, + ): Result { + val description = options.processDescription + ?: PyExecBundle.message("py.exec.defaultName.process", (listOf(processLauncher.exeForError.toString()) + processLauncher.args).joinToString(" ")) + val process = processLauncher.start().getOr { + val message = PyExecBundle.message("py.exec.start.error", description, it.error.cantExecProcessError, it.error.errNo + ?: "unknown") + return processLauncher.createExecError( + messageToUser = message, + errorReason = it.error + ) + } + + val result = try { + withTimeout(options.timeout) { + val interactiveResult = processInteractiveHandler.getResultFromProcess(binary, processLauncher.args, process) + + val successResult = interactiveResult.getOr { failure -> + val (output, customErrorMessage) = failure.error + val additionalMessage = customErrorMessage ?: run { + PyExecBundle.message("py.exec.exitCode.error", description, output.exitCode) + } + return@withTimeout processLauncher.createExecError( + messageToUser = additionalMessage, + errorReason = ExecErrorReason.UnexpectedProcessTermination(output), + loggedProcessId = process.loggedProcess.id, + ) + } + Result.success(successResult) + } + } + catch (_: TimeoutCancellationException) { + processLauncher.killAndJoin() + processLauncher.createExecError( + messageToUser = PyExecBundle.message("py.exec.timeout.error", description, options.timeout), + errorReason = ExecErrorReason.Timeout, + loggedProcessId = process.loggedProcess.id, + ) + } + return result + } } private fun ProcessLauncher.createExecError( diff --git a/python/python-exec-service/tests/com/intellij/python/junit5Tests/unit/ExecLimiterTest.kt b/python/python-exec-service/tests/com/intellij/python/junit5Tests/unit/ExecLimiterTest.kt new file mode 100644 index 000000000000..69332204fed9 --- /dev/null +++ b/python/python-exec-service/tests/com/intellij/python/junit5Tests/unit/ExecLimiterTest.kt @@ -0,0 +1,60 @@ +// Copyright 2000-2025 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license. +package com.intellij.python.junit5Tests.unit + +import com.intellij.execution.Platform +import com.intellij.openapi.util.registry.Registry +import com.intellij.platform.eel.getShell +import com.intellij.platform.eel.provider.asNioPath +import com.intellij.platform.eel.provider.localEel +import com.intellij.platform.eel.provider.systemOs +import com.intellij.python.community.execService.* +import com.intellij.testFramework.common.timeoutRunBlocking +import com.intellij.testFramework.junit5.TestApplication +import kotlinx.coroutines.* +import kotlinx.coroutines.sync.Semaphore +import org.junit.jupiter.api.Assertions +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.minutes + +@TestApplication +class ExecLimiterTest { + @Test + fun testLimit(@TempDir dir: Path): Unit = timeoutRunBlocking(1.minutes) { + + + val (shell, arg) = localEel.exec.getShell() + val sleepCmd = when (localEel.systemOs().platform) { + Platform.WINDOWS -> "pause" + // sleep(1) can't be used as it creates a separate process which can't be killed as ijent doesn't kill the process group + // cat, from the other hand, stops as soon as its stdin gets closed, so killing sh is enough + Platform.UNIX -> "cat" + } + val execService = ExecService() + + val weight = ConcurrentProcessWeight.HEAVY + val limit = Registry.intValue("python.execService.limit.heavy") + + val workersCount = limit * 10 + val lock = Semaphore(permits = workersCount, acquiredPermits = workersCount) + repeat(workersCount) { n -> + launch(Dispatchers.Unconfined, start = CoroutineStart.UNDISPATCHED) { + lock.release() + val file = dir.resolve("$n.txt") + execService.execGetStdout( + binary = shell.asNioPath(), + args = Args(arg, "echo 1 > $file && $sleepCmd"), + options = ExecOptions(weight = weight)).orThrow() + } + } + repeat(workersCount) { + lock.acquire() + } + delay(500.milliseconds) + Assertions.assertEquals(limit, dir.listDirectoryEntries().size, "No more than $limit processes must run at the same time") + coroutineContext.job.cancelChildren() + } +} \ No newline at end of file diff --git a/python/python-sdk-configurator/backend/resources/intellij.python.sdkConfigurator.backend.xml b/python/python-sdk-configurator/backend/resources/intellij.python.sdkConfigurator.backend.xml index a5d1745f6f32..02c3050e282a 100644 --- a/python/python-sdk-configurator/backend/resources/intellij.python.sdkConfigurator.backend.xml +++ b/python/python-sdk-configurator/backend/resources/intellij.python.sdkConfigurator.backend.xml @@ -17,10 +17,6 @@ - - - diff --git a/python/python-sdk-configurator/backend/src/impl/ModulesSdkConfigurator.kt b/python/python-sdk-configurator/backend/src/impl/ModulesSdkConfigurator.kt index ebe0613c3d9b..0dd0dda43140 100644 --- a/python/python-sdk-configurator/backend/src/impl/ModulesSdkConfigurator.kt +++ b/python/python-sdk-configurator/backend/src/impl/ModulesSdkConfigurator.kt @@ -6,10 +6,8 @@ import com.intellij.openapi.module.Module import com.intellij.openapi.project.Project import com.intellij.openapi.project.guessModuleDir import com.intellij.openapi.project.modules -import com.intellij.openapi.roots.ModuleRootManager import com.intellij.openapi.roots.ModuleRootModificationUtil import com.intellij.openapi.util.Key -import com.intellij.openapi.util.registry.Registry import com.intellij.openapi.util.removeUserData import com.intellij.platform.ide.progress.withBackgroundProgress import com.intellij.python.common.tools.ToolId @@ -32,8 +30,6 @@ import kotlinx.collections.immutable.toPersistentList import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll -import kotlinx.coroutines.sync.Semaphore -import kotlinx.coroutines.sync.withPermit import kotlinx.coroutines.withContext /** @@ -107,14 +103,11 @@ internal class ModulesSdkConfigurator private constructor( private suspend fun getModulesWithoutSDKCreateInfo(project: Project): Map = withBackgroundProgress(project, PySdkConfiguratorBundle.message("intellij.python.sdk.looking")) { val tools = PyProjectSdkConfigurationExtension.createMap() - val limit = Semaphore(permits = Registry.intValue("intellij.python.sdkConfigurator.backend.sdk.parallel")) val now = System.currentTimeMillis() val resultDef = project.modules.filter { PythonSdkUtil.findPythonSdk(it) == null }.map { module -> - limit.withPermit { - async { - val moduleInfo = getModuleInfo(module, tools) ?: return@async null - Pair(module, moduleInfo) - } + async { + val moduleInfo = getModuleInfo(module, tools) ?: return@async null + Pair(module, moduleInfo) } } val result = resultDef.awaitAll().filterNotNull() diff --git a/python/src/com/jetbrains/python/sdk/PySdkCommandRunner.kt b/python/src/com/jetbrains/python/sdk/PySdkCommandRunner.kt index 1c1fcbce0971..52c6fd56986e 100644 --- a/python/src/com/jetbrains/python/sdk/PySdkCommandRunner.kt +++ b/python/src/com/jetbrains/python/sdk/PySdkCommandRunner.kt @@ -26,8 +26,9 @@ suspend fun runExecutableWithProgress( env: Map = emptyMap(), vararg args: String, transformer: ProcessOutputTransformer, + processWeight: ConcurrentProcessWeight = ConcurrentProcessWeight.LIGHT ): PyResult { - val execOptions = ExecOptions(timeout = timeout, env = env) + val execOptions = ExecOptions(timeout = timeout, env = env, weight = processWeight) val errorHandlerTransformer: ProcessOutputTransformer = { output -> when { diff --git a/python/src/com/jetbrains/python/sdk/conda/execution/CondaExecutor.kt b/python/src/com/jetbrains/python/sdk/conda/execution/CondaExecutor.kt index 1156ad06de1a..047923225227 100644 --- a/python/src/com/jetbrains/python/sdk/conda/execution/CondaExecutor.kt +++ b/python/src/com/jetbrains/python/sdk/conda/execution/CondaExecutor.kt @@ -133,7 +133,7 @@ object CondaExecutor { ): PyResult { val envs = getFixedEnvs(binaryToExec).getOr { return it } val runArgs = prepareCondaRunArgs(args, emptyList(), condaEnvIdentity).toTypedArray() - return runExecutableWithProgress(binaryToExec, timeout, env = envs, *runArgs, transformer = transformer) + return runExecutableWithProgress(binaryToExec, timeout, env = envs, *runArgs, transformer = transformer, processWeight = ConcurrentProcessWeight.HEAVY) } private fun getFixedEnvs(binaryToExec: BinaryToExec): PyResult> {