You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/09/07 09:28:05 UTC
[GitHub] [flink] HuangXingBo opened a new pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
HuangXingBo opened a new pull request #17180:
URL: https://github.com/apache/flink/pull/17180
## What is the purpose of the change
*This pull request will optimize the python operator instances of the same job to share one python environment resources in the same jvm*
## Brief change log
- *Optimize the python operator instances of the same job to share one python environment resources in the same jvm*
## Verifying this change
This change added tests and can be verified as follows:
- *Original tests*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
}, {
"hash" : "32b10796fbf98d90fe2cd1282aff793179078939",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842",
"triggerID" : "32b10796fbf98d90fe2cd1282aff793179078939",
"triggerType" : "PUSH"
}, {
"hash" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678)
* 32b10796fbf98d90fe2cd1282aff793179078939 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842)
* 1824c6258348afd31d899eee5edb7c80cf09d7f3 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914162215
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 (Tue Sep 07 09:49:10 UTC 2021)
✅no warnings
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] HuangXingBo closed pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
HuangXingBo closed pull request #17180:
URL: https://github.com/apache/flink/pull/17180
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu commented on a change in pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #17180:
URL: https://github.com/apache/flink/pull/17180#discussion_r703969699
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -98,29 +104,48 @@
private transient Thread shutdownHook;
+ /** Flag whether is responsible for preparing python environment. */
+ private transient boolean isResponsibleForEnvironment;
+
@NotNull private final PythonDependencyInfo dependencyInfo;
@NotNull private final Map<String, String> systemEnv;
@NotNull private final String[] tmpDirectories;
+ @NotNull private final PythonEnvResources pythonEnvResource;
public ProcessPythonEnvironmentManager(
@NotNull PythonDependencyInfo dependencyInfo,
@NotNull String[] tmpDirectories,
- @NotNull Map<String, String> systemEnv) {
+ @NotNull Map<String, String> systemEnv,
+ @NotNull JobID jobID) {
this.dependencyInfo = Objects.requireNonNull(dependencyInfo);
this.tmpDirectories = Objects.requireNonNull(tmpDirectories);
this.systemEnv = Objects.requireNonNull(systemEnv);
+ this.pythonEnvResource =
Review comment:
Move it to open and declare pythonEnvResource as transient
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -98,29 +104,48 @@
private transient Thread shutdownHook;
+ /** Flag whether is responsible for preparing python environment. */
+ private transient boolean isResponsibleForEnvironment;
+
@NotNull private final PythonDependencyInfo dependencyInfo;
@NotNull private final Map<String, String> systemEnv;
@NotNull private final String[] tmpDirectories;
+ @NotNull private final PythonEnvResources pythonEnvResource;
public ProcessPythonEnvironmentManager(
@NotNull PythonDependencyInfo dependencyInfo,
@NotNull String[] tmpDirectories,
- @NotNull Map<String, String> systemEnv) {
+ @NotNull Map<String, String> systemEnv,
+ @NotNull JobID jobID) {
this.dependencyInfo = Objects.requireNonNull(dependencyInfo);
this.tmpDirectories = Objects.requireNonNull(tmpDirectories);
this.systemEnv = Objects.requireNonNull(systemEnv);
+ this.pythonEnvResource =
Review comment:
Move it to open and declare pythonEnvResource as transient
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +479,41 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static class PythonEnvResources {
+ /** The base directory of the Python Environment. */
+ private String baseDirectory;
+
+ /** Keep track of the number of the threads sharing this Python environment resources. */
+ private long refCount = 0;
+
+ /** The flag whether the Python environment is ready. */
+ private AtomicBoolean isReady = new AtomicBoolean(false);
Review comment:
```suggestion
private final AtomicBoolean isReady = new AtomicBoolean(false);
```
##########
File path: flink-python/pyflink/datastream/stream_execution_environment.py
##########
@@ -874,20 +874,11 @@ def _generate_stream_graph(self, clear_transformations: bool = False, job_name:
j_configuration = get_j_env_configuration(self._j_stream_execution_environment)
if not self._remote_mode and is_local_deployment(j_configuration):
jvm = gateway.jvm
- env_config = JPythonConfigUtil.getEnvironmentConfig(
- self._j_stream_execution_environment)
- parallelism = self.get_parallelism()
- if parallelism > 1 and env_config.containsKey(jvm.PythonOptions.PYTHON_ARCHIVES.key()):
- import logging
- logging.warning("Lookback mode is disabled as python archives are used and the "
- "parallelism of the job is greater than 1. The Python user-defined "
- "functions will be executed in an independent Python process.")
- else:
- from pyflink.fn_execution.beam.beam_worker_pool_service import \
- BeamFnLoopbackWorkerPoolServicer
- j_env = jvm.System.getenv()
- get_field_value(j_env, "m").put(
- 'PYFLINK_LOOPBACK_SERVER_ADDRESS', BeamFnLoopbackWorkerPoolServicer().start())
+ from pyflink.fn_execution.beam.beam_worker_pool_service import \
+ BeamFnLoopbackWorkerPoolServicer
+ j_env = jvm.System.getenv()
+ get_field_value(j_env, "m").put(
+ 'PYFLINK_LOOPBACK_SERVER_ADDRESS', BeamFnLoopbackWorkerPoolServicer().start())
Review comment:
Should we also revert the changes made in integration_test.sh? That's the following:
```
pytest "$FLINK_PYTHON_DIR/pyflink/datastream/tests/test_stream_execution_environment.py"
```
##########
File path: flink-python/pyflink/table/tests/test_dependency.py
##########
@@ -159,11 +159,9 @@ def add_from_file(i):
with open("data/data.txt", 'r') as f:
return i + int(f.read())
- self.t_env.get_config().get_configuration().set_string("parallelism.default", "1")
self.t_env.create_temporary_system_function("add_from_file",
udf(add_from_file, DataTypes.BIGINT(),
DataTypes.BIGINT()))
- self.t_env._remote_mode = True
Review comment:
The variable `_remote_mode` could also be removed from TableEnvironment
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -163,18 +196,34 @@ public void close() throws Exception {
}
@Override
- public PythonEnvironment createEnvironment() throws IOException {
- Map<String, String> env = constructEnvironmentVariables();
-
- if (dependencyInfo.getRequirementsFilePath().isPresent()) {
- LOG.info("Trying to pip install the Python requirements...");
- PythonEnvironmentManagerUtils.pipInstallRequirements(
- dependencyInfo.getRequirementsFilePath().get(),
- dependencyInfo.getRequirementsCacheDir().orElse(null),
- requirementsDirectory,
- dependencyInfo.getPythonExec(),
- env);
+ public PythonEnvironment createEnvironment() throws Exception {
+ Map<String, String> env;
+ if (isResponsibleForEnvironment) {
+ env = constructEnvironmentVariables();
+
+ if (dependencyInfo.getRequirementsFilePath().isPresent()) {
+ LOG.info("Trying to pip install the Python requirements...");
+ PythonEnvironmentManagerUtils.pipInstallRequirements(
+ dependencyInfo.getRequirementsFilePath().get(),
+ dependencyInfo.getRequirementsCacheDir().orElse(null),
+ requirementsDirectory,
+ dependencyInfo.getPythonExec(),
+ env);
+ }
+ pythonEnvResource.setEnvironmentVariables(env);
+ synchronized (pythonEnvResource) {
+ pythonEnvResource.finish();
+ pythonEnvResource.notifyAll();
+ }
+ } else {
+ synchronized (pythonEnvResource) {
+ if (!pythonEnvResource.isReady()) {
+ pythonEnvResource.wait();
+ }
+ env = new HashMap<>(pythonEnvResource.env);
Review comment:
`env = pythonEnvResource.env` is enough
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -129,27 +154,35 @@ public void open() throws Exception {
@Override
public void close() throws Exception {
+ long refCount;
+ // The last thread to exit is responsible for cleaning up resources.
+ synchronized (pythonEnvResource) {
Review comment:
It's a good practice to set pythonEnvResource as null to avoid possibilities decRef is called multiple times in certain cases.
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +479,41 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static class PythonEnvResources {
+ /** The base directory of the Python Environment. */
+ private String baseDirectory;
+
+ /** Keep track of the number of the threads sharing this Python environment resources. */
+ private long refCount = 0;
+
+ /** The flag whether the Python environment is ready. */
+ private AtomicBoolean isReady = new AtomicBoolean(false);
+
+ private Map<String, String> env;
+
+ void setBaseDirectory(String baseDirectory) {
Review comment:
this method could be removed as baseDirectory could be accessed directly.
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -98,29 +104,48 @@
private transient Thread shutdownHook;
+ /** Flag whether is responsible for preparing python environment. */
+ private transient boolean isResponsibleForEnvironment;
+
@NotNull private final PythonDependencyInfo dependencyInfo;
@NotNull private final Map<String, String> systemEnv;
@NotNull private final String[] tmpDirectories;
+ @NotNull private final PythonEnvResources pythonEnvResource;
public ProcessPythonEnvironmentManager(
@NotNull PythonDependencyInfo dependencyInfo,
@NotNull String[] tmpDirectories,
- @NotNull Map<String, String> systemEnv) {
+ @NotNull Map<String, String> systemEnv,
+ @NotNull JobID jobID) {
this.dependencyInfo = Objects.requireNonNull(dependencyInfo);
this.tmpDirectories = Objects.requireNonNull(tmpDirectories);
this.systemEnv = Objects.requireNonNull(systemEnv);
+ this.pythonEnvResource =
+ sharedPythonEnvResources.computeIfAbsent(
+ Objects.requireNonNull(jobID), k -> new PythonEnvResources());
}
@Override
public void open() throws Exception {
- baseDirectory = createBaseDirectory(tmpDirectories);
- archivesDirectory = String.join(File.separator, baseDirectory, PYTHON_ARCHIVES_DIR);
- requirementsDirectory = String.join(File.separator, baseDirectory, PYTHON_REQUIREMENTS_DIR);
- filesDirectory = String.join(File.separator, baseDirectory, PYTHON_FILES_DIR);
-
- File baseDirectoryFile = new File(baseDirectory);
- if (!baseDirectoryFile.exists() && !baseDirectoryFile.mkdir()) {
- throw new IOException("Could not create the base directory: " + baseDirectory);
+ synchronized (pythonEnvResource) {
+ if (pythonEnvResource.baseDirectory == null) {
+ baseDirectory = createBaseDirectory(tmpDirectories);
+
+ File baseDirectoryFile = new File(baseDirectory);
+ if (!baseDirectoryFile.exists() && !baseDirectoryFile.mkdir()) {
+ throw new IOException("Could not create the base directory: " + baseDirectory);
+ }
+ pythonEnvResource.setBaseDirectory(baseDirectory);
+ isResponsibleForEnvironment = true;
+ } else {
+ baseDirectory = pythonEnvResource.baseDirectory;
+ }
+ archivesDirectory = String.join(File.separator, baseDirectory, PYTHON_ARCHIVES_DIR);
+ requirementsDirectory =
+ String.join(File.separator, baseDirectory, PYTHON_REQUIREMENTS_DIR);
+ filesDirectory = String.join(File.separator, baseDirectory, PYTHON_FILES_DIR);
+
+ pythonEnvResource.incRef();
Review comment:
`pythonEnvResource.incRef();` should be wrapped in try/finally statements. Otherwise, if exceptions were thrown before `pythonEnvResource.incRef();`, it will not increase the reference counter, however, it will still decrease the reference counter in `close`.
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +479,41 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static class PythonEnvResources {
+ /** The base directory of the Python Environment. */
+ private String baseDirectory;
+
+ /** Keep track of the number of the threads sharing this Python environment resources. */
Review comment:
```suggestion
/** Keep track of the number of threads sharing this Python environment resources. */
```
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +479,41 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static class PythonEnvResources {
Review comment:
```suggestion
private static final class PythonEnvResources {
```
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -163,18 +196,34 @@ public void close() throws Exception {
}
@Override
- public PythonEnvironment createEnvironment() throws IOException {
- Map<String, String> env = constructEnvironmentVariables();
-
- if (dependencyInfo.getRequirementsFilePath().isPresent()) {
- LOG.info("Trying to pip install the Python requirements...");
- PythonEnvironmentManagerUtils.pipInstallRequirements(
- dependencyInfo.getRequirementsFilePath().get(),
- dependencyInfo.getRequirementsCacheDir().orElse(null),
- requirementsDirectory,
- dependencyInfo.getPythonExec(),
- env);
+ public PythonEnvironment createEnvironment() throws Exception {
+ Map<String, String> env;
+ if (isResponsibleForEnvironment) {
+ env = constructEnvironmentVariables();
+
+ if (dependencyInfo.getRequirementsFilePath().isPresent()) {
+ LOG.info("Trying to pip install the Python requirements...");
+ PythonEnvironmentManagerUtils.pipInstallRequirements(
+ dependencyInfo.getRequirementsFilePath().get(),
+ dependencyInfo.getRequirementsCacheDir().orElse(null),
+ requirementsDirectory,
+ dependencyInfo.getPythonExec(),
+ env);
+ }
+ pythonEnvResource.setEnvironmentVariables(env);
+ synchronized (pythonEnvResource) {
+ pythonEnvResource.finish();
Review comment:
Currently, there are multiple places handling the concurrent access of PythonEnvResource using wait/notify. This is error prone. Could we refactor them a bit to avoid this? SharedResources is a good example on how to achieve this.
Some thoughts in my mind: All concurrent access of PythonEnvResource should be handled in one place. Maybe we could created class SharedPythonEnvResources and handle all concurrent access in SharedPythonEnvResources.
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +479,41 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static class PythonEnvResources {
+ /** The base directory of the Python Environment. */
+ private String baseDirectory;
+
+ /** Keep track of the number of the threads sharing this Python environment resources. */
+ private long refCount = 0;
+
+ /** The flag whether the Python environment is ready. */
Review comment:
```suggestion
/** The flag indicating whether the Python environment is ready. */
```
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -98,29 +104,48 @@
private transient Thread shutdownHook;
+ /** Flag whether is responsible for preparing python environment. */
+ private transient boolean isResponsibleForEnvironment;
+
@NotNull private final PythonDependencyInfo dependencyInfo;
@NotNull private final Map<String, String> systemEnv;
@NotNull private final String[] tmpDirectories;
+ @NotNull private final PythonEnvResources pythonEnvResource;
public ProcessPythonEnvironmentManager(
@NotNull PythonDependencyInfo dependencyInfo,
@NotNull String[] tmpDirectories,
- @NotNull Map<String, String> systemEnv) {
+ @NotNull Map<String, String> systemEnv,
+ @NotNull JobID jobID) {
this.dependencyInfo = Objects.requireNonNull(dependencyInfo);
this.tmpDirectories = Objects.requireNonNull(tmpDirectories);
this.systemEnv = Objects.requireNonNull(systemEnv);
+ this.pythonEnvResource =
Review comment:
Move it to open and declare pythonEnvResource as transient
##########
File path: flink-python/pyflink/fn_execution/beam/beam_worker_pool_service.py
##########
@@ -155,7 +159,12 @@ def _start_sdk_worker_main(self, start_worker_request: beam_fn_api_pb2.StartWork
_LOGGER.exception('Python sdk harness failed: ')
raise
finally:
- if base_dir:
- os.chdir(base_dir)
+ self._parse_param_lock.acquire()
+ self._ref_cnt -= 1
+ # The last thread to exit is responsible for changing working directory.
+ if self._ref_cnt == 0 and self._old_working_dir is not None:
Review comment:
What's the motivation for this? I'm asking because I'm wondering that if we need to change back the worker directory, should we also remove the changed PYTHONPATH?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu commented on a change in pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #17180:
URL: https://github.com/apache/flink/pull/17180#discussion_r705265024
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +442,117 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static final class PythonEnvResources {
+
+ private static final ReentrantLock lock = new ReentrantLock();
+
+ @GuardedBy("lock")
+ private static final Map<JobID, PythonLeasedResource> reservedResources = new HashMap<>();
+
+ static PythonLeasedResource getOrAllocateSharedResource(
+ JobID jobID,
+ FunctionWithException<JobID, Tuple2<String, Map<String, String>>, Exception>
+ initializer)
+ throws Exception {
+ try {
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted which preparing python environment.");
+ }
+
+ try {
+ PythonLeasedResource resource = reservedResources.get(jobID);
+ if (resource == null) {
+ resource = createResource(initializer, jobID);
+ reservedResources.put(jobID, resource);
+ }
+ resource.incRef();
+ return resource;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public static void release(JobID jobID) throws Exception {
+ lock.lock();
+ try {
+ final PythonLeasedResource resource = reservedResources.get(jobID);
+ if (resource == null) {
+ return;
+ }
+ resource.decRef();
+ if (resource.refCount == 0) {
+ reservedResources.remove(jobID);
+ resource.close();
+ }
+
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private static PythonLeasedResource createResource(
+ FunctionWithException<JobID, Tuple2<String, Map<String, String>>, Exception>
+ initializer,
+ JobID jobID)
+ throws Exception {
+ Tuple2<String, Map<String, String>> resource = initializer.apply(jobID);
+ String baseDirectory = resource.f0;
+ Map<String, String> env = resource.f1;
+ return new PythonLeasedResource(baseDirectory, env);
+ }
+
+ private static final class PythonLeasedResource implements AutoCloseable {
+ private Map<String, String> env;
Review comment:
```suggestion
private final Map<String, String> env;
```
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -347,10 +350,13 @@ private void constructArchivesDirectory(Map<String, String> env) throws IOExcept
}
}
- private void constructRequirementsDirectory(Map<String, String> env) throws IOException {
+ private void constructRequirementsDirectory(Map<String, String> env, String baseDirectory)
Review comment:
baseDirectory could be got from `getBaseDirectory` directly and so this change could be reverted
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +442,117 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static final class PythonEnvResources {
+
+ private static final ReentrantLock lock = new ReentrantLock();
+
+ @GuardedBy("lock")
+ private static final Map<JobID, PythonLeasedResource> reservedResources = new HashMap<>();
Review comment:
```suggestion
private static final Map<Object, PythonLeasedResource> reservedResources = new HashMap<>();
```
It doesn't necessarily to be JobID and we just happen to use JobID. So it should be enough to declare it as Object type.
##########
File path: flink-python/src/main/java/org/apache/flink/python/env/beam/ProcessPythonEnvironmentManager.java
##########
@@ -430,4 +442,117 @@ private static String createBaseDirectory(String[] tmpDirectories) throws IOExce
+ Arrays.toString(tmpDirectories)
+ "' for storing the generated files of python dependency.");
}
+
+ private static final class PythonEnvResources {
+
+ private static final ReentrantLock lock = new ReentrantLock();
+
+ @GuardedBy("lock")
+ private static final Map<JobID, PythonLeasedResource> reservedResources = new HashMap<>();
+
+ static PythonLeasedResource getOrAllocateSharedResource(
+ JobID jobID,
+ FunctionWithException<JobID, Tuple2<String, Map<String, String>>, Exception>
+ initializer)
+ throws Exception {
+ try {
+ lock.lockInterruptibly();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted which preparing python environment.");
+ }
+
+ try {
+ PythonLeasedResource resource = reservedResources.get(jobID);
+ if (resource == null) {
+ resource = createResource(initializer, jobID);
+ reservedResources.put(jobID, resource);
+ }
+ resource.incRef();
+ return resource;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public static void release(JobID jobID) throws Exception {
+ lock.lock();
+ try {
+ final PythonLeasedResource resource = reservedResources.get(jobID);
+ if (resource == null) {
+ return;
+ }
+ resource.decRef();
+ if (resource.refCount == 0) {
+ reservedResources.remove(jobID);
+ resource.close();
+ }
+
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private static PythonLeasedResource createResource(
+ FunctionWithException<JobID, Tuple2<String, Map<String, String>>, Exception>
+ initializer,
+ JobID jobID)
+ throws Exception {
+ Tuple2<String, Map<String, String>> resource = initializer.apply(jobID);
+ String baseDirectory = resource.f0;
+ Map<String, String> env = resource.f1;
+ return new PythonLeasedResource(baseDirectory, env);
+ }
+
+ private static final class PythonLeasedResource implements AutoCloseable {
+ private Map<String, String> env;
+
+ /** The base directory of the Python Environment. */
+ private String baseDirectory;
Review comment:
```suggestion
private final String baseDirectory;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
}, {
"hash" : "32b10796fbf98d90fe2cd1282aff793179078939",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842",
"triggerID" : "32b10796fbf98d90fe2cd1282aff793179078939",
"triggerType" : "PUSH"
}, {
"hash" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23856",
"triggerID" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1824c6258348afd31d899eee5edb7c80cf09d7f3 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23856)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
}, {
"hash" : "32b10796fbf98d90fe2cd1282aff793179078939",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842",
"triggerID" : "32b10796fbf98d90fe2cd1282aff793179078939",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678)
* 32b10796fbf98d90fe2cd1282aff793179078939 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
}, {
"hash" : "32b10796fbf98d90fe2cd1282aff793179078939",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842",
"triggerID" : "32b10796fbf98d90fe2cd1282aff793179078939",
"triggerType" : "PUSH"
}, {
"hash" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23856",
"triggerID" : "1824c6258348afd31d899eee5edb7c80cf09d7f3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 32b10796fbf98d90fe2cd1282aff793179078939 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23842)
* 1824c6258348afd31d899eee5edb7c80cf09d7f3 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23856)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
}, {
"hash" : "32b10796fbf98d90fe2cd1282aff793179078939",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "32b10796fbf98d90fe2cd1282aff793179078939",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678)
* 32b10796fbf98d90fe2cd1282aff793179078939 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dianfu commented on a change in pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
dianfu commented on a change in pull request #17180:
URL: https://github.com/apache/flink/pull/17180#discussion_r703992024
##########
File path: flink-python/pyflink/fn_execution/beam/beam_worker_pool_service.py
##########
@@ -155,7 +159,12 @@ def _start_sdk_worker_main(self, start_worker_request: beam_fn_api_pb2.StartWork
_LOGGER.exception('Python sdk harness failed: ')
raise
finally:
- if base_dir:
- os.chdir(base_dir)
+ self._parse_param_lock.acquire()
+ self._ref_cnt -= 1
+ # The last thread to exit is responsible for changing working directory.
+ if self._ref_cnt == 0 and self._old_working_dir is not None:
Review comment:
What's the motivation for this? I'm asking because I'm wondering that if we need to change back the worker directory, should we also revert back the changed PYTHONPATH?
##########
File path: flink-python/pyflink/fn_execution/beam/beam_worker_pool_service.py
##########
@@ -155,7 +159,12 @@ def _start_sdk_worker_main(self, start_worker_request: beam_fn_api_pb2.StartWork
_LOGGER.exception('Python sdk harness failed: ')
raise
finally:
- if base_dir:
- os.chdir(base_dir)
+ self._parse_param_lock.acquire()
+ self._ref_cnt -= 1
+ # The last thread to exit is responsible for changing working directory.
+ if self._ref_cnt == 0 and self._old_working_dir is not None:
Review comment:
What's the motivation for this? I'm asking because I'm wondering that if we need to change back the worker directory, should we also revert the changed PYTHONPATH?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17180: [FLINK-24123][python] Optimize the python operator instances of the same job to share one python environment resources in the same jvm
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17180:
URL: https://github.com/apache/flink/pull/17180#issuecomment-914169654
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678",
"triggerID" : "b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b2eb0fe23bcfe8680fb28dc1e8a72f64405bd0f1 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23678)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org