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/08 03:15:31 UTC
[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
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