<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.runtime.execution;

import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.deployment.GateDeploymentDescriptor;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.Buffer;
import org.apache.flink.runtime.io.network.bufferprovider.BufferAvailabilityListener;
import org.apache.flink.runtime.io.network.bufferprovider.BufferProvider;
import org.apache.flink.runtime.io.network.bufferprovider.GlobalBufferPool;
import org.apache.flink.runtime.io.network.bufferprovider.LocalBufferPool;
import org.apache.flink.runtime.io.network.bufferprovider.LocalBufferPoolOwner;
import org.apache.flink.runtime.io.network.channels.ChannelID;
import org.apache.flink.runtime.io.network.channels.OutputChannel;
import org.apache.flink.runtime.io.network.gates.GateID;
import org.apache.flink.runtime.io.network.gates.InputGate;
import org.apache.flink.runtime.io.network.gates.OutputGate;
import org.apache.flink.runtime.jobgraph.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.protocols.AccumulatorProtocol;
import org.apache.flink.runtime.taskmanager.Task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;


public class RuntimeEnvironment implements Environment, BufferProvider, LocalBufferPoolOwner, Runnable {

	/** The log object used for debugging. */
	private static final Logger LOG = LoggerFactory.getLogger(RuntimeEnvironment.class);

	private static final ThreadGroup TASK_THREADS = new ThreadGroup("Task Threads");
	
	/** The interval to sleep in case a communication channel is not yet entirely set up (in milliseconds). */
	private static final int SLEEPINTERVAL = 100;
	
	// --------------------------------------------------------------------------------------------

	/** The task that owns this environment */
	private final Task owner;
	
	
	/** The job configuration encapsulated in the environment object. */
	private final Configuration jobConfiguration;

	/** The task configuration encapsulated in the environment object. */
	private final Configuration taskConfiguration;
	
	
	/** ClassLoader for all user code classes */
	private final ClassLoader userCodeClassLoader;
	
	/** Class of the task to run in this environment. */
	private final Class<? extends AbstractInvokable> invokableClass;

	/** Instance of the class to be run in this environment. */
	private final AbstractInvokable invokable;
	
	
	/** List of output gates created by the task. */
	private final ArrayList<OutputGate> outputGates = new ArrayList<OutputGate>();

	/** List of input gates created by the task. */
	private final ArrayList<InputGate<? extends IOReadableWritable>> inputGates = new ArrayList<InputGate<? extends IOReadableWritable>>();

	/** Unbound input gate IDs which are required for deserializing an environment in the course of an RPC call. */
	private final Queue<GateID> unboundInputGateIDs = new ArrayDeque<GateID>();

	/** The memory manager of the current environment (currently the one associated with the executing TaskManager). */
	private final MemoryManager memoryManager;

	/** The I/O manager of the current environment (currently the one associated with the executing TaskManager). */
	private final IOManager ioManager;

	/** The input split provider that can be queried for new input splits.  */
	private final InputSplitProvider inputSplitProvider;

	
	/** The thread executing the task in the environment. */
	private Thread executingThread;

	/**
	 * The RPC proxy to report accumulators to JobManager
	 */
	private final AccumulatorProtocol accumulatorProtocolProxy;

	private final Map<String,FutureTask<Path>> cacheCopyTasks = new HashMap<String, FutureTask<Path>>();
	
	private final BroadcastVariableManager bcVarManager;
	
	private LocalBufferPool outputBufferPool;
	
	private AtomicBoolean canceled = new AtomicBoolean();


	public RuntimeEnvironment(Task owner, TaskDeploymentDescriptor tdd,
							ClassLoader userCodeClassLoader,
							MemoryManager memoryManager, IOManager ioManager,
							InputSplitProvider inputSplitProvider,
							AccumulatorProtocol accumulatorProtocolProxy,
							BroadcastVariableManager bcVarManager)
		throws Exception
	{
		Preconditions.checkNotNull(owner);
		Preconditions.checkNotNull(memoryManager);
		Preconditions.checkNotNull(ioManager);
		Preconditions.checkNotNull(inputSplitProvider);
		Preconditions.checkNotNull(accumulatorProtocolProxy);
		Preconditions.checkNotNull(userCodeClassLoader);
		Preconditions.checkNotNull(bcVarManager);
		
		this.owner = owner;

		this.memoryManager = memoryManager;
		this.ioManager = ioManager;
		this.inputSplitProvider = inputSplitProvider;
		this.accumulatorProtocolProxy = accumulatorProtocolProxy;
		this.bcVarManager = bcVarManager;

		// load and instantiate the invokable class
		this.userCodeClassLoader = userCodeClassLoader;
		try {
			final String className = tdd.getInvokableClassName();
			this.invokableClass = Class.forName(className, true, userCodeClassLoader).asSubclass(AbstractInvokable.class);
		}
		catch (Throwable t) {
			throw new Exception("Could not load invokable class.", t);
		}
		
		try {
			this.invokable = this.invokableClass.newInstance();
		}
		catch (Throwable t) {
			throw new Exception("Could not instantiate the invokable class.", t);
		}
		
		this.jobConfiguration = tdd.getJobConfiguration();
		this.taskConfiguration = tdd.getTaskConfiguration();
		
		this.invokable.setEnvironment(this);
		this.invokable.registerInputOutput();

		List<GateDeploymentDescriptor> inGates = tdd.getInputGates();
		List<GateDeploymentDescriptor> outGates = tdd.getOutputGates();
		
		
		if (this.inputGates.size() != inGates.size()) {
			throw new Exception("The number of readers created in 'registerInputOutput()' "
					+ "is different than the number of connected incoming edges in the job graph.");
		}
		if (this.outputGates.size() != outGates.size()) {
			throw new Exception("The number of writers created in 'registerInputOutput()' "<commit_msg>[FLINK-1265] Fix user classloader bug for registerInputOutput() method<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.runtime.execution;

import java.io.IOException;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.deployment.GateDeploymentDescriptor;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.Buffer;
import org.apache.flink.runtime.io.network.bufferprovider.BufferAvailabilityListener;
import org.apache.flink.runtime.io.network.bufferprovider.BufferProvider;
import org.apache.flink.runtime.io.network.bufferprovider.GlobalBufferPool;
import org.apache.flink.runtime.io.network.bufferprovider.LocalBufferPool;
import org.apache.flink.runtime.io.network.bufferprovider.LocalBufferPoolOwner;
import org.apache.flink.runtime.io.network.channels.ChannelID;
import org.apache.flink.runtime.io.network.channels.OutputChannel;
import org.apache.flink.runtime.io.network.gates.GateID;
import org.apache.flink.runtime.io.network.gates.InputGate;
import org.apache.flink.runtime.io.network.gates.OutputGate;
import org.apache.flink.runtime.jobgraph.JobID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.memorymanager.MemoryManager;
import org.apache.flink.runtime.protocols.AccumulatorProtocol;
import org.apache.flink.runtime.taskmanager.Task;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;


public class RuntimeEnvironment implements Environment, BufferProvider, LocalBufferPoolOwner, Runnable {

	/** The log object used for debugging. */
	private static final Logger LOG = LoggerFactory.getLogger(RuntimeEnvironment.class);

	private static final ThreadGroup TASK_THREADS = new ThreadGroup("Task Threads");
	
	/** The interval to sleep in case a communication channel is not yet entirely set up (in milliseconds). */
	private static final int SLEEPINTERVAL = 100;
	
	// --------------------------------------------------------------------------------------------

	/** The task that owns this environment */
	private final Task owner;
	
	
	/** The job configuration encapsulated in the environment object. */
	private final Configuration jobConfiguration;

	/** The task configuration encapsulated in the environment object. */
	private final Configuration taskConfiguration;
	
	
	/** ClassLoader for all user code classes */
	private final ClassLoader userCodeClassLoader;
	
	/** Class of the task to run in this environment. */
	private final Class<? extends AbstractInvokable> invokableClass;

	/** Instance of the class to be run in this environment. */
	private final AbstractInvokable invokable;
	
	
	/** List of output gates created by the task. */
	private final ArrayList<OutputGate> outputGates = new ArrayList<OutputGate>();

	/** List of input gates created by the task. */
	private final ArrayList<InputGate<? extends IOReadableWritable>> inputGates = new ArrayList<InputGate<? extends IOReadableWritable>>();

	/** Unbound input gate IDs which are required for deserializing an environment in the course of an RPC call. */
	private final Queue<GateID> unboundInputGateIDs = new ArrayDeque<GateID>();

	/** The memory manager of the current environment (currently the one associated with the executing TaskManager). */
	private final MemoryManager memoryManager;

	/** The I/O manager of the current environment (currently the one associated with the executing TaskManager). */
	private final IOManager ioManager;

	/** The input split provider that can be queried for new input splits.  */
	private final InputSplitProvider inputSplitProvider;

	
	/** The thread executing the task in the environment. */
	private Thread executingThread;

	/**
	 * The RPC proxy to report accumulators to JobManager
	 */
	private final AccumulatorProtocol accumulatorProtocolProxy;

	private final Map<String,FutureTask<Path>> cacheCopyTasks = new HashMap<String, FutureTask<Path>>();
	
	private final BroadcastVariableManager bcVarManager;
	
	private LocalBufferPool outputBufferPool;
	
	private AtomicBoolean canceled = new AtomicBoolean();


	public RuntimeEnvironment(Task owner, TaskDeploymentDescriptor tdd,
							ClassLoader userCodeClassLoader,
							MemoryManager memoryManager, IOManager ioManager,
							InputSplitProvider inputSplitProvider,
							AccumulatorProtocol accumulatorProtocolProxy,
							BroadcastVariableManager bcVarManager)
		throws Exception
	{
		Preconditions.checkNotNull(owner);
		Preconditions.checkNotNull(memoryManager);
		Preconditions.checkNotNull(ioManager);
		Preconditions.checkNotNull(inputSplitProvider);
		Preconditions.checkNotNull(accumulatorProtocolProxy);
		Preconditions.checkNotNull(userCodeClassLoader);
		Preconditions.checkNotNull(bcVarManager);
		
		this.owner = owner;

		this.memoryManager = memoryManager;
		this.ioManager = ioManager;
		this.inputSplitProvider = inputSplitProvider;
		this.accumulatorProtocolProxy = accumulatorProtocolProxy;
		this.bcVarManager = bcVarManager;

		// load and instantiate the invokable class
		this.userCodeClassLoader = userCodeClassLoader;
		try {
			final String className = tdd.getInvokableClassName();
			this.invokableClass = Class.forName(className, true, userCodeClassLoader).asSubclass(AbstractInvokable.class);
		}
		catch (Throwable t) {
			throw new Exception("Could not load invokable class.", t);
		}
		
		try {
			this.invokable = this.invokableClass.newInstance();
		}
		catch (Throwable t) {
			throw new Exception("Could not instantiate the invokable class.", t);
		}
		
		this.jobConfiguration = tdd.getJobConfiguration();
		this.taskConfiguration = tdd.getTaskConfiguration();
		
		this.invokable.setEnvironment(this);
		
		// make sure that user classloader is available, because registerInputOutput might call usercode
		{
			Thread currentThread = Thread.currentThread();
			ClassLoader context = currentThread.getContextClassLoader();
			currentThread.setContextClassLoader(userCodeClassLoader);
			this.invokable.registerInputOutput();

			currentThread.setContextClassLoader(context);
		}

		List<GateDeploymentDescriptor> inGates = tdd.getInputGates();
		List<GateDeploymentDescriptor> outGates = tdd.getOutputGates();
		
		
		if (this.inputGates.size() != inGates.size()) {
			throw new Exception("The number of readers created in 'registerInputOutput()' "
					+ "is different than the number of connected incoming edges in the job graph.");
		}
		if (this.outputGates.size() != outGates.size()) {
			throw new Exception("The number of writers created in 'registerInputOutput()' "
{'fraction_non_alphanumeric': 0.09681706005188888, 'fraction_numerical': 0.0013921407327722585, 'mean_word_length': 4.72816237767307, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6991579', 'n_tokens_mistral': 4496, 'n_tokens_neox': 4146, 'n_words': 1233}