You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2014/10/06 16:31:43 UTC
[3/4] Integration of new BLOB service.
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
new file mode 100644
index 0000000..942a0ef
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
@@ -0,0 +1,279 @@
+/*
+ * 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.librarycache;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.blob.BlobService;
+import org.apache.flink.runtime.jobgraph.JobID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * For each job graph that is submitted to the system the library cache manager maintains
+ * a set of libraries (typically JAR files) which the job requires to run. The library cache manager
+ * caches library files in order to avoid unnecessary retransmission of data. It is based on a singleton
+ * programming pattern, so there exists at most on library manager at a time.
+ * <p>
+ * This class is thread-safe.
+ */
+public final class BlobLibraryCacheManager extends TimerTask implements LibraryCacheManager {
+
+ private static Logger LOG = LoggerFactory.getLogger(BlobLibraryCacheManager.class);
+
+ /**
+ * Dummy object used in the lock map.
+ */
+ private final Object lockObject = new Object();
+
+ /**
+ * Map to translate a job ID to the responsible class loaders.
+ */
+ private final ConcurrentMap<JobID, URLClassLoader> classLoaders = new
+ ConcurrentHashMap<JobID, URLClassLoader>();
+
+ /**
+ * Map to store the number of references to a specific library manager entry.
+ */
+ private final Map<JobID, Integer> libraryReferenceCounter = new HashMap<JobID,
+ Integer>();
+
+ /**
+ * Map to store the blob keys referenced by a specific job
+ */
+ private final Map<JobID, Collection<BlobKey>> requiredJars = new
+ HashMap<JobID, Collection<BlobKey>>();
+
+ /**
+ * Map to store the number of reference to a specific file
+ */
+ private final Map<BlobKey, Integer> blobKeyReferenceCounter = new
+ HashMap<BlobKey, Integer>();
+
+
+ /**
+ * All registered blobs
+ */
+ private final Set<BlobKey> registeredBlobs = new HashSet<BlobKey>();
+
+ private final BlobService blobService;
+
+ public BlobLibraryCacheManager(BlobService blobService, Configuration configuration){
+ this.blobService = blobService;
+
+ // Initializing the clean up task
+ Timer timer = new Timer();
+ long cleanupInterval = configuration.getLong(
+ ConfigConstants.LIBRARY_CACHE_MANAGER_CLEANUP_INTERVAL,
+ ConfigConstants.DEFAULT_LIBRARY_CACHE_MANAGER_CLEANUP_INTERVAL)*1000;
+ timer.schedule(this, cleanupInterval);
+ }
+
+ /**
+ * Increments the reference counter of the corrsponding map
+ *
+ * @param key
+ * the key identifying the counter to increment
+ * @return the increased reference counter
+ */
+ private <K> int incrementReferenceCounter(final K key, final Map<K,
+ Integer> map) {
+
+ if(!map.containsKey(key)){
+ map.put(key, 1);
+
+ return 1;
+ }else{
+ int counter = map.get(key) + 1;
+ map.put(key, counter);
+
+ return counter;
+ }
+ }
+
+ /**
+ * Decrements the reference counter associated with the key
+ *
+ * @param key
+ * the key identifying the counter to decrement
+ * @return the decremented reference counter
+ */
+ private <K> int decrementReferenceCounter(final K key, final Map<K,
+ Integer> map) {
+
+ if (!map.containsKey(key)) {
+ throw new IllegalStateException("Cannot find reference counter entry for key " + key);
+ }else{
+ int counter = map.get(key) -1;
+
+ if(counter == 0){
+ map.remove(key);
+ }else{
+ map.put(key, counter);
+ }
+
+ return counter;
+ }
+ }
+
+ /**
+ * Registers a job ID with a set of library paths that are required to run the job. For every registered
+ * job the library cache manager creates a class loader that is used to instantiate the vertex's environment later
+ * on.
+ *
+ * @param id
+ * the ID of the job to be registered.
+ * @param requiredJarFiles
+ * the client path's of the required libraries
+ * @throws IOException
+ * thrown if one of the requested libraries is not in the cache
+ */
+ @Override
+ public void register(final JobID id, final Collection<BlobKey> requiredJarFiles) throws
+ IOException {
+
+ synchronized (lockObject) {
+ if (incrementReferenceCounter(id, libraryReferenceCounter) > 1) {
+ return;
+ }
+
+ // Check if library manager entry for this id already exists
+ if (this.classLoaders.containsKey(id)) {
+ throw new IllegalStateException("Library cache manager already contains " +
+ "class loader entry for job ID " + id);
+ }
+
+ if (requiredJars.containsKey(id)) {
+ throw new IllegalStateException("Library cache manager already contains blob keys" +
+ " entry for job ID " + id);
+ }
+
+ requiredJars.put(id, requiredJarFiles);
+
+ URL[] urls = new URL[requiredJarFiles.size()];
+ int count = 0;
+
+ for (BlobKey blobKey : requiredJarFiles) {
+ urls[count++] = registerBlobKeyAndGetURL(blobKey);
+ }
+
+ final URLClassLoader classLoader = new URLClassLoader(urls);
+ this.classLoaders.put(id, classLoader);
+ }
+ }
+
+ private URL registerBlobKeyAndGetURL(BlobKey key) throws IOException{
+ if(incrementReferenceCounter(key, blobKeyReferenceCounter) == 1){
+ // registration might happen even if the file is already stored locally
+ registeredBlobs.add(key);
+ }
+
+ return blobService.getURL(key);
+ }
+
+ /**
+ * Unregisters a job ID and releases the resources associated with it.
+ *
+ * @param id
+ * the job ID to unregister
+ */
+ @Override
+ public void unregister(final JobID id) {
+ synchronized (lockObject) {
+ if (decrementReferenceCounter(id, libraryReferenceCounter) == 0) {
+ this.classLoaders.remove(id);
+
+ Collection<BlobKey> keys = requiredJars.get(id);
+
+ for (BlobKey key : keys) {
+ decrementReferenceCounter(key, blobKeyReferenceCounter);
+ }
+
+ requiredJars.remove(id);
+ }
+ }
+
+ }
+
+ /**
+ * Returns the class loader to the specified vertex.
+ *
+ * @param id
+ * the ID of the job to return the class loader for
+ * @return the class loader of requested vertex or <code>null</code> if no class loader has been registered with the
+ * given ID.
+ */
+ @Override
+ public ClassLoader getClassLoader(final JobID id) {
+ return this.classLoaders.get(id);
+ }
+
+ @Override
+ public File getFile(BlobKey blobKey) throws IOException {
+ return new File(blobService.getURL(blobKey).getFile());
+ }
+
+ public int getBlobServerPort() {
+ return blobService.getPort();
+ }
+
+ @Override
+ public void shutdown() throws IOException{
+ blobService.shutdown();
+ }
+
+ /**
+ * Cleans up blobs which are not referenced anymore
+ */
+ @Override
+ public void run() {
+ synchronized (lockObject) {
+ Iterator<BlobKey> it = registeredBlobs.iterator();
+
+ while (it.hasNext()) {
+ BlobKey key = it.next();
+
+ try {
+ if (!blobKeyReferenceCounter.containsKey(key)) {
+ blobService.delete(key);
+ it.remove();
+ }
+ } catch (IOException ioe) {
+ LOG.warn("Could not delete file with blob key" + key, ioe);
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FallbackLibraryCacheManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FallbackLibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FallbackLibraryCacheManager.java
new file mode 100644
index 0000000..5c162a0
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/FallbackLibraryCacheManager.java
@@ -0,0 +1,55 @@
+/*
+ * 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.librarycache;
+
+import org.apache.flink.runtime.blob.BlobKey;
+import org.apache.flink.runtime.jobgraph.JobID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+
+public class FallbackLibraryCacheManager implements LibraryCacheManager {
+ private static Logger LOG = LoggerFactory.getLogger(FallbackLibraryCacheManager.class);
+
+ @Override
+ public ClassLoader getClassLoader(JobID id) {
+ return getClass().getClassLoader();
+ }
+
+ @Override
+ public File getFile(BlobKey blobKey) throws IOException {
+ throw new IOException("There is no file associated to the blob key " + blobKey);
+ }
+
+ @Override
+ public void register(JobID id, Collection<BlobKey> requiredJarFiles) throws IOException {
+ LOG.warn("FallbackLibraryCacheManager cannot download files associated with blob keys.");
+ }
+
+ @Override
+ public void unregister(JobID id) {
+ LOG.warn("FallbackLibraryCacheManager does not book keeping of job IDs.");
+ }
+
+ @Override
+ public void shutdown() {}
+}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
index 0f19121..a0a95ab 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheManager.java
@@ -16,817 +16,53 @@
* limitations under the License.
*/
-
package org.apache.flink.runtime.execution.librarycache;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.File;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FSDataOutputStream;
-import org.apache.flink.core.fs.FileStatus;
-import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
-import org.apache.flink.core.io.StringRecord;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.jobgraph.JobID;
-import org.apache.flink.util.StringUtils;
-
-/**
- * For each job graph that is submitted to the system the library cache manager maintains
- * a set of libraries (typically JAR files) which the job requires to run. The library cache manager
- * caches library files in order to avoid unnecessary retransmission of data. It is based on a singleton
- * programming pattern, so there exists at most on library manager at a time.
- * <p>
- * This class is thread-safe.
- *
- */
-public final class LibraryCacheManager {
-
- /**
- * The instance of the library cache manager accessible through a singleton pattern.
- */
- private static LibraryCacheManager libraryManager = null;
-
- /**
- * Map to translate client paths of libraries to the file name used by the cache manager.
- */
- private final ConcurrentMap<LibraryTranslationKey, String> clientPathToCacheName = new ConcurrentHashMap<LibraryTranslationKey, String>();
-
- /**
- * Name of the directory to put cached libraries in.
- */
- private static final String LIBRARYCACHENAME = "libraryCache";
-
- /**
- * Algorithm to be used for calculating the checksum of the libraries.
- */
- private static final String HASHING_ALGORITHM = "SHA-1";
-
- /**
- * Dummy object used in the lock map.
- */
- private static final Object LOCK_OBJECT = new Object();
-
- /**
- * File system object used to access the local file system.
- */
- private final FileSystem fs;
-
- /**
- * The message digest object used to calculate the checksums of the libraries.
- */
- private final MessageDigest md;
-
- /**
- * Path pointing to the library cache directory.
- */
- private final Path libraryCachePath;
-
- /**
- * Map to translate a job ID to the responsible library cache manager entry.
- */
- private final ConcurrentMap<JobID, LibraryManagerEntry> libraryManagerEntries = new ConcurrentHashMap<JobID, LibraryManagerEntry>();
-
- /**
- * Map to store the number of references to a specific library manager entry.
- */
- private final ConcurrentMap<JobID, AtomicInteger> libraryReferenceCounter = new ConcurrentHashMap<JobID, AtomicInteger>();
-
- /**
- * Map to guarantee atomicity of of register/unregister operations.
- */
- private final ConcurrentMap<JobID, Object> lockMap = new ConcurrentHashMap<JobID, Object>();
-
- /**
- * Returns the singleton instance of the library cache manager.
- *
- * @return the singleton instance of the library cache manager.
- * @throws IOException
- * thrown if access to the file system can not be obtained or the requested hashing algorithm does not exist
- */
- private static synchronized LibraryCacheManager get() throws IOException {
-
- // Lazy initialization
- if (libraryManager == null) {
- libraryManager = new LibraryCacheManager();
- }
-
- return libraryManager;
- }
-
- /**
- * Constructs a new instance of the library cache manager.
- *
- * @throws IOException
- * thrown if access to the file system can not be obtained or the requested hashing algorithm does not exist
- */
- private LibraryCacheManager() throws IOException {
-
- // Check if the library cache directory exists, otherwise create it
- final String tmp = System.getProperty("java.io.tmpdir");
- if (tmp == null) {
- throw new IOException("Cannot find directory for temporary files");
- }
-
- this.fs = FileSystem.getLocalFileSystem();
-
- // Add user name of library cache path to avoid conflicts with regard to file permissions
- String userName = System.getProperty("user.name");
- if (userName == null) {
- userName = "default";
- }
-
- // Use the File object to the convert the path to a proper URI
- final File path = new File(tmp + File.separator + LIBRARYCACHENAME + "-" + userName);
- final URI uri = path.toURI();
-
- this.libraryCachePath = new Path(uri);
-
- this.fs.mkdirs(this.libraryCachePath);
-
- // Create an MD5 message digest object we can use
- try {
- this.md = MessageDigest.getInstance(HASHING_ALGORITHM);
- } catch (NoSuchAlgorithmException e) {
- throw new IOException("Cannot find algorithm " + HASHING_ALGORITHM + ": "
- + StringUtils.stringifyException(e));
- }
- }
-
- /**
- * Increments the reference counter for the library manager entry with the given job ID.
- *
- * @param jobID
- * the job ID identifying the library manager entry
- * @return the increased reference counter
- */
- private int incrementReferenceCounter(final JobID jobID) {
-
- while (true) {
-
- AtomicInteger ai = this.libraryReferenceCounter.get(jobID);
- if (ai == null) {
-
- ai = new AtomicInteger(1);
- if (this.libraryReferenceCounter.putIfAbsent(jobID, ai) == null) {
- return 1;
- }
-
- // We had a race, try again
- } else {
- return ai.incrementAndGet();
- }
- }
- }
-
- /**
- * Decrements the reference counter for the library manager entry with the given job ID.
- *
- * @param jobID
- * the job ID identifying the library manager entry
- * @return the decremented reference counter
- */
- private int decrementReferenceCounter(final JobID jobID) {
-
- final AtomicInteger ai = this.libraryReferenceCounter.get(jobID);
-
- if (ai == null) {
- throw new IllegalStateException("Cannot find reference counter entry for job " + jobID);
- }
-
- int retVal = ai.decrementAndGet();
-
- if (retVal == 0) {
- this.libraryReferenceCounter.remove(jobID);
- }
-
- return retVal;
- }
-
- /**
- * Registers a job ID with a set of library paths that are required to run the job. The library paths are given in
- * terms
- * of client paths, so the method first translates the client paths into the corresponding internal cache names. For
- * every registered
- * job the library cache manager creates a class loader that is used to instantiate the job's environment later on.
- *
- * @param id
- * the ID of the job to be registered
- * @param clientPaths
- * the client path's of the required libraries
- * @throws IOException
- * thrown if the library cache manager could not be instantiated, no mapping between the job ID and a job ID
- * exists or the requested library is not in the cache.
- */
- public static void register(final JobID id, final Path[] clientPaths) throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.registerInternal(id, clientPaths);
- }
-
- /**
- * Registers a job ID with a set of library paths that are required to run the job. The library paths are given in
- * terms
- * of client paths, so the method first translates the client paths into the corresponding internal cache names. For
- * every registered
- * job the library cache manager creates a class loader that is used to instantiate the job's environment later on.
- *
- * @param id
- * the ID of the job to be registered.
- * @param clientPaths
- * the client path's of the required libraries
- * @throws IOException
- * thrown if no mapping between the job ID and a job ID exists or the requested library is not in the cache.
- */
- private void registerInternal(final JobID id, final Path[] clientPaths) throws IOException {
-
- final String[] cacheNames = new String[clientPaths.length];
-
- for (int i = 0; i < clientPaths.length; ++i) {
- final LibraryTranslationKey key = new LibraryTranslationKey(id, clientPaths[i]);
- cacheNames[i] = this.clientPathToCacheName.get(key);
- if (cacheNames[i] == null) {
- throw new IOException("Cannot map" + clientPaths[i].toString() + " to cache name");
- }
- }
-
- // Register as regular
- registerInternal(id, cacheNames);
- }
-
- /**
- * Registers a job ID with a set of library paths that are required to run the job. For every registered
- * job the library cache manager creates a class loader that is used to instantiate the job's environment later on.
- *
- * @param id
- * the ID of the job to be registered.
- * @param requiredJarFiles
- * the client path's of the required libraries
- * @throws IOException
- * thrown if the library cache manager could not be instantiated or one of the requested libraries is not in
- * the cache
- */
- public static void register(final JobID id, final String[] requiredJarFiles) throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.registerInternal(id, requiredJarFiles);
- }
-
- /**
- * Registers a job ID with a set of library paths that are required to run the job. For every registered
- * job the library cache manager creates a class loader that is used to instantiate the vertex's environment later
- * on.
- *
- * @param id
- * the ID of the job to be registered.
- * @param requiredJarFiles
- * the client path's of the required libraries
- * @throws IOException
- * thrown if one of the requested libraries is not in the cache
- */
- private void registerInternal(final JobID id, final String[] requiredJarFiles) throws IOException {
-
- // Use spin lock here
- while (this.lockMap.putIfAbsent(id, LOCK_OBJECT) != null);
-
- try {
- if (incrementReferenceCounter(id) > 1) {
- return;
- }
-
- // Check if library manager entry for this id already exists
- if (this.libraryManagerEntries.containsKey(id)) {
- throw new IllegalStateException("Library cache manager already contains entry for job ID " + id);
- }
-
- // Check if all the required jar files exist in the cache
- URL[] urls = null;
- if (requiredJarFiles != null) {
-
- urls = new URL[requiredJarFiles.length];
-
- for (int i = 0; i < requiredJarFiles.length; i++) {
- final Path p = contains(requiredJarFiles[i]);
- if (p == null) {
- throw new IOException(requiredJarFiles[i] + " does not exist in the library cache");
- }
-
- // Add file to the URL array
- try {
- urls[i] = p.toUri().toURL();
- } catch (MalformedURLException e) {
- throw new IOException(StringUtils.stringifyException(e));
- }
- }
- }
-
- final LibraryManagerEntry entry = new LibraryManagerEntry(id, requiredJarFiles, urls);
-
- this.libraryManagerEntries.put(id, entry);
-
- } finally {
- this.lockMap.remove(id);
- }
- }
-
- /**
- * Unregisters a job ID and releases the resources associated with it.
- *
- * @param id
- * the job ID to unregister
- * @throws IOException
- * thrown if the library cache manager could not be instantiated
- */
- public static void unregister(final JobID id) throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.unregisterInternal(id);
- }
- /**
- * Unregisters a job ID and releases the resources associated with it.
- *
- * @param id
- * the job ID to unregister
- */
- private void unregisterInternal(final JobID id) {
-
- // Use spin lock here
- while (this.lockMap.putIfAbsent(id, LOCK_OBJECT) != null);
- try {
- if (decrementReferenceCounter(id) == 0) {
- this.libraryManagerEntries.remove(id);
- }
- }
- finally {
- this.lockMap.remove(id);
- }
- }
-
- /**
- * Checks if the given library is in the local cache.
- *
- * @param cacheName
- * The name of the library to be checked for.
- * @return the path object of the library if it is cached, <code>null</code> otherwise
- * @throws IOException
- * thrown if the library cache manager could not be instantiated or no access to the file system could be
- * obtained
- */
- public static Path contains(final String cacheName) throws IOException {
-
- final LibraryCacheManager lib = get();
- return lib.containsInternal(cacheName);
- }
-
- /**
- * Checks if the given library is in the local cache.
- *
- * @param cacheName
- * The name of the library to be checked for.
- * @return the path object of the library if it is cached, <code>null</code> otherwise
- * @throws IOException
- * thrown if no access to the file system could be obtained
- */
- private Path containsInternal(final String cacheName) throws IOException {
-
- // Create a path object from the external name string
- final Path p = new Path(this.libraryCachePath + File.separator + cacheName);
-
- synchronized (this.fs) {
- if (fs.exists(p)) {
- return p;
- }
- }
-
- return null;
- }
-
- /**
- * Returns the class loader to the specified vertex.
- *
- * @param id
- * the ID of the job to return the class loader for
- * @return the class loader of requested vertex or <code>null</code> if no class loader has been registered with the
- * given ID.
- * @throws IOException
- * thrown if the library cache manager could not be instantiated
- */
- public static ClassLoader getClassLoader(final JobID id) throws IOException {
-
- if (id == null) {
- return null;
- }
-
- final LibraryCacheManager lib = get();
- return lib.getClassLoaderInternal(id);
- }
-
- /**
- * Returns the class loader to the specified vertex.
- *
- * @param id
- * the ID of the job to return the class loader for
- * @return the class loader of requested vertex or <code>null</code> if no class loader has been registered with the
- * given ID.
- */
- private ClassLoader getClassLoaderInternal(final JobID id) {
-
- final LibraryManagerEntry entry = this.libraryManagerEntries.get(id);
-
- if (entry == null) {
- return null;
- }
-
- return entry.getClassLoader();
- }
-
- /**
- * Returns the names of the required libraries of the specified job.
- *
- * @param id
- * the ID of the job to return the names of required libraries for.
- * @return the names of the required libraries or <code>null</code> if the specified job ID is unknown
- * @throws IOException
- * thrown if the library cache manager could not be instantiated
- */
- public static String[] getRequiredJarFiles(final JobID id) throws IOException {
-
- if (id == null) {
- return new String[0];
- }
-
- final LibraryCacheManager lib = get();
-
- return lib.getRequiredJarFilesInternal(id);
- }
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+public interface LibraryCacheManager {
/**
- * Returns the names of the required libraries of the specified job.
- *
- * @param id
- * the ID of the job to return the names of required libraries for.
- * @return the names of the required libraries or <code>null</code> if the specified job ID is unknown
+ * Returns the user code class loader associated with id.
+ *
+ * @param id identifying the job
+ * @return ClassLoader which can load the user code
*/
- private String[] getRequiredJarFilesInternal(final JobID id) {
-
- final LibraryManagerEntry entry = this.libraryManagerEntries.get(id);
-
- if (entry == null) {
- return null;
- }
-
- return entry.getRequiredJarFiles();
- }
+ ClassLoader getClassLoader(final JobID id);
/**
- * Writes data from the library with the given file name to the specified stream.
- *
- * @param libraryFileName
- * the name of the library
- * @param out
- * the stream to write the data to
+ * Returns a file handle to the file identified by the blob key.
+ *
+ * @param blobKey identifying the requested file
+ * @return File handle
* @throws IOException
- * thrown if an error occurs while writing the data
*/
- public static void writeLibraryToStream(final String libraryFileName, final DataOutput out) throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.writeLibraryToStreamInternal(libraryFileName, out);
-
- }
+ File getFile(final BlobKey blobKey) throws IOException;
/**
- * Writes data from the library with the given file name to the specified stream.
- *
- * @param libraryFileName
- * the name of the library
- * @param out
- * the stream to write the data to
+ * Registers a job with its required jar files. The jar files are identified by their blob keys.
+ *
+ * @param id job ID
+ * @param requiredJarFiles collection of blob keys identifying the required jar files
* @throws IOException
- * thrown if an error occurs while writing the data
*/
- private void writeLibraryToStreamInternal(final String libraryFileName, final DataOutput out) throws IOException {
-
- if (libraryFileName == null) {
- throw new IOException("libraryName is null!");
- }
-
- final Path storePath = new Path(this.libraryCachePath + "/" + libraryFileName);
-
- synchronized (this.fs) {
-
- if (!fs.exists(storePath)) {
- throw new IOException(storePath + " does not exist!");
- }
-
- final FileStatus status = fs.getFileStatus(storePath);
-
- StringRecord.writeString(out, libraryFileName);
- out.writeLong(status.getLen());
-
- final FSDataInputStream inStream = fs.open(storePath);
- final byte[] buf = new byte[8192]; // 8K Buffer*/
- int read = inStream.read(buf, 0, buf.length);
- while (read > 0) {
- out.write(buf, 0, read);
- read = inStream.read(buf, 0, buf.length);
- }
-
- inStream.close();
- }
- }
+ void register(final JobID id, final Collection<BlobKey> requiredJarFiles) throws IOException;
/**
- * Reads library data from the given stream.
- *
- * @param in
- * the stream to read the library data from
- * @throws IOException
- * throws if an error occurs while reading from the stream
+ * Unregisters a job from the library cache manager.
+ *
+ * @param id job ID
*/
- public static void readLibraryFromStream(final DataInput in) throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.readLibraryFromStreamInternal(in);
-
- }
+ void unregister(final JobID id);
/**
- * Reads library data from the given stream.
- *
- * @param in
- * the stream to read the library data from
+ * Shutdown method
+ *
* @throws IOException
- * throws if an error occurs while reading from the stream
*/
- private void readLibraryFromStreamInternal(final DataInput in) throws IOException {
-
- final String libraryFileName = StringRecord.readString(in);
-
- if (libraryFileName == null) {
- throw new IOException("libraryFileName is null!");
- }
-
- final long length = in.readLong();
-
- if (length > (long) Integer.MAX_VALUE) {
- throw new IOException("Submitted jar file " + libraryFileName + " is too large");
- }
-
- final byte[] buf = new byte[(int) length];
- in.readFully(buf);
-
- final Path storePath = new Path(this.libraryCachePath + "/" + libraryFileName);
-
- synchronized (this.fs) {
-
- // Check if file already exists in our library cache, if not write it to the cache directory
- if (!fs.exists(storePath)) {
- final FSDataOutputStream fos = fs.create(storePath, false);
- fos.write(buf, 0, buf.length);
- fos.close();
- }
- }
- }
-
- /**
- * Reads a library from the given input stream and adds it to the local library cache. The cache name of
- * the library is determined by the checksum of the received data and cannot be specified manually.
- *
- * @param jobID
- * the ID of the job the library data belongs to
- * @param name
- * the name of the library at the clients host
- * @param size
- * the size of the library to be read from the input stream
- * @param in
- * the data input stream
- * @throws IOException
- * thrown if the library cache manager could not be instantiated or an error occurred while reading the
- * library data from the input stream
- */
- public static void addLibrary(final JobID jobID, final Path name, final long size, final DataInput in)
- throws IOException {
-
- final LibraryCacheManager lib = get();
- lib.addLibraryInternal(jobID, name, size, in);
- }
-
- /**
- * Reads a library from the given input stream and adds it to the local library cache. The cache name of
- * the library is determined by the checksum of the received data and cannot be specified manually.
- *
- * @param jobID
- * the ID of the job the library data belongs to
- * @param name
- * the name of the library at the clients host
- * @param size
- * the size of the library to be read from the input stream
- * @param in
- * the data input stream
- * @throws IOException
- * thrown if an error occurred while reading the library data from the input stream
- */
- private void addLibraryInternal(final JobID jobID, final Path name, final long size, final DataInput in)
- throws IOException {
-
- if (size > (long) Integer.MAX_VALUE) {
- throw new IOException("Submitted jar file " + name + " is too large");
- }
-
- // Map the entire jar file to memory
- final byte[] buf = new byte[(int) size];
- in.readFully(buf);
-
- // Reset and calculate message digest from jar file
- this.md.reset();
- this.md.update(buf);
-
- // Construct internal jar name from digest
- final String cacheName = StringUtils.byteToHexString(md.digest()) + ".jar";
- final Path storePath = new Path(this.libraryCachePath + "/" + cacheName);
-
- synchronized (this.fs) {
-
- // Check if file already exists in our library cache, if not write it to the cache directory
- if (!this.fs.exists(storePath)) {
- final FSDataOutputStream fos = this.fs.create(storePath, false);
- fos.write(buf, 0, buf.length);
- fos.close();
- }
- }
-
- // Create mapping for client path and cache name
- final LibraryTranslationKey key = new LibraryTranslationKey(jobID, name);
- this.clientPathToCacheName.putIfAbsent(key, cacheName);
- }
-
- /**
- * Auxiliary class that stores the class loader object as well as the names of the required
- * libraries for a job vertex.
- *
- */
- private static class LibraryManagerEntry {
-
- /**
- * The class loader object for the Nephele job this object belongs to.
- */
- private final ClassLoader classLoader;
-
- /**
- * A list containing the names of the JAR files required by the Nephele job this object belongs to.
- */
- private final String[] requiredJarFiles;
-
- /**
- * Constructs a <code>LibraryManagerEntry</code> object from the given job ID and array of required library
- * files.
- *
- * @param id
- * the ID of the job to create a <code>LibraryManagerEntry</code> for.
- * @param requiredJarFiles
- * an array with the names of required libraries by the corresponding job (plain names)
- * @param urls
- * an array with the names of required libraries by the corresponding job (URL objects required by the
- * class loader)
- */
- public LibraryManagerEntry(final JobID id, final String[] requiredJarFiles, URL[] urls) {
-
- String[] temp = requiredJarFiles;
- if (temp == null) {
- temp = new String[0];
- }
-
- this.requiredJarFiles = temp;
-
- if (urls == null) {
- urls = new URL[0];
- }
-
- this.classLoader = new URLClassLoader(urls, ClassLoader.getSystemClassLoader());
- }
-
- /**
- * Returns the class loader associated with this library manager entry.
- *
- * @return the class loader associated with this library manager entry
- */
- public ClassLoader getClassLoader() {
- return this.classLoader;
- }
-
- /**
- * Returns a (possibly empty) array of library names required by the associated job vertex to run.
- *
- * @return a (possibly empty) array of library names required by the associated job vertex to run
- */
- public String[] getRequiredJarFiles() {
- return this.requiredJarFiles;
- }
- }
-
- /**
- * Auxiliary class that acts as a key for the translation of the names a client uses to refer to required libraries
- * for a vertex
- * and the internal names used by the library cache manager.
- *
- */
- private static class LibraryTranslationKey {
-
- /**
- * The ID of the job this object belongs to.
- */
- private final JobID jobID;
-
- /**
- * The path at which the library has been stored at the client.
- */
- private final Path clientPath;
-
- /**
- * Construct a <code>LibraryTranslationKey</code> object from a fiven job ID and a client path that specifies
- * the name of
- * required library at the job client.
- *
- * @param jobID
- * the job ID
- * @param clientPath
- * the client path
- */
- public LibraryTranslationKey(final JobID jobID, final Path clientPath) {
-
- this.jobID = jobID;
- this.clientPath = clientPath;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public int hashCode() {
-
- final long temp = (this.jobID.hashCode() + this.clientPath.hashCode()) % Integer.MAX_VALUE;
-
- return (int) temp;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public boolean equals(final Object obj) {
-
- if (obj == null) {
- return false;
- }
-
- if (this.jobID == null) {
- return false;
- }
-
- if (this.clientPath == null) {
- return false;
- }
-
- if (obj instanceof LibraryTranslationKey) {
-
- final LibraryTranslationKey key = (LibraryTranslationKey) obj;
- if (this.jobID.equals(key.getJobID()) && this.clientPath.equals(key.getClientPath())) {
- return true;
- }
- }
-
- return false;
- }
-
- /**
- * Returns the client path associated with this object.
- *
- * @return the client path associated with this object
- */
- public Path getClientPath() {
- return this.clientPath;
- }
-
- /**
- * Returns the job ID associated with this object.
- *
- * @return the job ID associated with this object
- */
- public JobID getJobID() {
- return this.jobID;
- }
- }
+ void shutdown() throws IOException;
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileRequest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileRequest.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileRequest.java
deleted file mode 100644
index 2f9b181..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileRequest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * 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.librarycache;
-
-import java.io.IOException;
-
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.io.StringRecord;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-/**
- * A library cache profile request includes a set of library names and issues a task manager to report which of these
- * libraries
- * are currently available in its local cache.
- *
- */
-public class LibraryCacheProfileRequest implements IOReadableWritable {
-
- /**
- * List of the required libraries' names.
- */
- private String[] requiredLibraries;
-
- /**
- * Returns the names of libraries whose cache status is to be retrieved.
- *
- * @return the names of libraries whose cache status is to be retrieved
- */
- public String[] getRequiredLibraries() {
- return requiredLibraries;
- }
-
- /**
- * Sets the names of libraries whose cache status is to be retrieved.
- *
- * @param requiredLibraries
- * the names of libraries whose cache status is to be retrieved
- */
- public void setRequiredLibraries(final String[] requiredLibraries) {
- this.requiredLibraries = requiredLibraries;
- }
-
-
- @Override
- public void read(final DataInputView in) throws IOException {
-
- // Read required jar files
- this.requiredLibraries = new String[in.readInt()];
-
- for (int i = 0; i < this.requiredLibraries.length; i++) {
- this.requiredLibraries[i] = StringRecord.readString(in);
- }
- }
-
-
- @Override
- public void write(final DataOutputView out) throws IOException {
-
- if (this.requiredLibraries == null) {
- throw new IOException("requiredLibraries is null");
- }
-
- // Number of required jar files
- out.writeInt(this.requiredLibraries.length);
-
- for (int i = 0; i < this.requiredLibraries.length; i++) {
- StringRecord.writeString(out, this.requiredLibraries[i]);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileResponse.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileResponse.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileResponse.java
deleted file mode 100644
index 8bbf7f0..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheProfileResponse.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * 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.librarycache;
-
-import java.io.IOException;
-
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.io.StringRecord;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-/**
- * A library cache profile response is the response to a library cache profile request. It contains the set of
- * library names originally included in the request message and additionally a bit vector stating which of them
- * are available in the respective task manager's local cache.
- *
- */
-public class LibraryCacheProfileResponse implements IOReadableWritable {
-
- /**
- * List of the requires libraries' names.
- */
- private String[] requiredLibraries = null;
-
- /**
- * Lists if library at given index is in cache (<code>true</code>) or not (<code>false</code>).
- */
- private boolean[] cached = null;
-
- /**
- * Construct a library cache profile response from a given library cache profile
- * request and initially sets the cache status for all included library names to <code>false</code>.
- *
- * @param request
- * the library cache profile request the response belongs to
- */
- public LibraryCacheProfileResponse(final LibraryCacheProfileRequest request) {
-
- this.requiredLibraries = request.getRequiredLibraries();
- this.cached = new boolean[this.requiredLibraries.length];
- }
-
- /**
- * Constructs an empty library cache profile response. This constructor is required
- * for deserialization and should not be called in a regular application.
- */
- public LibraryCacheProfileResponse() {
- }
-
- /**
- * Sets the cache status of the library at the given position.
- *
- * @param pos
- * the position of the library name whose cache status is to be set
- * @param cached
- * <code>true</code> if the library at the given position is in the local cache, <code>false</code> otherwise
- */
- public void setCached(final int pos, final boolean cached) {
-
- if (pos < this.cached.length) {
- this.cached[pos] = cached;
- }
- }
-
- /**
- * Returns the cache status of the library at the given position.
- *
- * @param pos
- * the position of the library name whose cache status is to be retrieved
- * @return <code>true</code> if the library at the given position is in the local cache, <code>false</code>
- * otherwise
- */
- public boolean isCached(final int pos) {
-
- if (pos < this.cached.length) {
- return this.cached[pos];
- }
-
- return false;
- }
-
-
- @Override
- public void read(final DataInputView in) throws IOException {
-
- // Read the names of the required jar files
- this.requiredLibraries = new String[in.readInt()];
- for (int i = 0; i < this.requiredLibraries.length; i++) {
- this.requiredLibraries[i] = StringRecord.readString(in);
- }
-
- // Read cache status
- this.cached = new boolean[in.readInt()];
- for (int i = 0; i < this.cached.length; i++) {
- this.cached[i] = in.readBoolean();
- }
-
- }
-
-
- @Override
- public void write(final DataOutputView out) throws IOException {
-
- if (this.requiredLibraries == null) {
- throw new IOException("requiredLibraries is null");
- }
-
- if (this.cached == null) {
- throw new IOException("cached is null");
- }
-
- // Write the names of the required jar files
- out.writeInt(this.requiredLibraries.length);
- for (int i = 0; i < this.requiredLibraries.length; i++) {
- StringRecord.writeString(out, this.requiredLibraries[i]);
- }
-
- // Write out cache status
- out.writeInt(this.cached.length);
- for (int i = 0; i < this.cached.length; i++) {
- out.writeBoolean(this.cached[i]);
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheUpdate.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheUpdate.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheUpdate.java
deleted file mode 100644
index c376eea..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/LibraryCacheUpdate.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.librarycache;
-
-import java.io.IOException;
-
-import org.apache.flink.core.io.IOReadableWritable;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-
-/**
- * This class is used to encapsulate the transmission of a library file in a RPC call.
- */
-public class LibraryCacheUpdate implements IOReadableWritable {
-
- /** The name of the library file that is transmitted with this object. */
- private String libraryFileName;
-
- /**
- * Constructs a new library cache update object.
- *
- * @param libraryFileName
- * the name of the library that should be transported within this object.
- */
- public LibraryCacheUpdate(String libraryFileName) {
- if (libraryFileName == null) {
- throw new IllegalArgumentException("libraryFileName must not be null");
- }
-
- this.libraryFileName = libraryFileName;
- }
-
- /**
- * Constructor used to reconstruct the object at the receiver of the RPC call.
- */
- public LibraryCacheUpdate() {}
-
-
- @Override
- public void read(DataInputView in) throws IOException {
- LibraryCacheManager.readLibraryFromStream(in);
- }
-
-
- @Override
- public void write(DataOutputView out) throws IOException {
- if (this.libraryFileName == null) {
- throw new IOException("libraryFileName is null");
- }
-
- LibraryCacheManager.writeLibraryToStream(this.libraryFileName, out);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 1c050fe..2f881d7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -268,8 +268,7 @@ public class Execution {
public void run() {
try {
Instance instance = slot.getInstance();
- instance.checkLibraryAvailability(vertex.getJobId());
-
+
TaskOperationResult result = instance.getTaskManagerProxy().submitTask(deployment);
if (result == null) {
markFailed(new Exception("Failed to deploy the task to slot " + slot + ": TaskOperationResult was null"));
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 07e2455..1954d70 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.execution.ExecutionListener;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.instance.Instance;
@@ -77,12 +78,13 @@ public class ExecutionGraph {
/** All vertices, in the order in which they were created **/
private final List<ExecutionJobVertex> verticesInCreationOrder;
-
+
/** All intermediate results that are part of this graph */
private final ConcurrentHashMap<IntermediateDataSetID, IntermediateResult> intermediateResults;
/** The currently executed tasks, for callbacks */
private final ConcurrentHashMap<ExecutionAttemptID, Execution> currentExecutions;
+
private final Map<ChannelID, ExecutionEdge> edges = new HashMap<ChannelID, ExecutionEdge>();
@@ -90,9 +92,8 @@ public class ExecutionGraph {
/** An executor that can run long actions (involving remote calls) */
private final ExecutorService executor;
-
-
- private final List<String> userCodeJarFiles;
+
+ private final List<BlobKey> requiredJarFiles;
private final List<JobStatusListener> jobStatusListeners;
@@ -116,10 +117,11 @@ public class ExecutionGraph {
public ExecutionGraph(JobID jobId, String jobName, Configuration jobConfig) {
- this(jobId, jobName, jobConfig, null);
+ this(jobId, jobName, jobConfig, new ArrayList<BlobKey>(), null);
}
- public ExecutionGraph(JobID jobId, String jobName, Configuration jobConfig, ExecutorService executor) {
+ public ExecutionGraph(JobID jobId, String jobName, Configuration jobConfig,
+ List<BlobKey> requiredJarFiles,ExecutorService executor) {
if (jobId == null || jobName == null || jobConfig == null) {
throw new NullPointerException();
}
@@ -134,12 +136,13 @@ public class ExecutionGraph {
this.verticesInCreationOrder = new ArrayList<ExecutionJobVertex>();
this.currentExecutions = new ConcurrentHashMap<ExecutionAttemptID, Execution>();
- this.userCodeJarFiles = new ArrayList<String>();
this.jobStatusListeners = new CopyOnWriteArrayList<JobStatusListener>();
this.executionListeners = new CopyOnWriteArrayList<ExecutionListener>();
this.stateTimestamps = new long[JobStatus.values().length];
this.stateTimestamps[JobStatus.CREATED.ordinal()] = System.currentTimeMillis();
+
+ this.requiredJarFiles = requiredJarFiles;
}
// --------------------------------------------------------------------------------------------
@@ -175,15 +178,15 @@ public class ExecutionGraph {
this.verticesInCreationOrder.add(ejv);
}
}
-
- public void addUserCodeJarFile(String jarFile) {
- this.userCodeJarFiles.add(jarFile);
- }
-
- public String[] getUserCodeJarFiles() {
- return (String[]) this.userCodeJarFiles.toArray(new String[this.userCodeJarFiles.size()]);
+
+ /**
+ * Returns a list of BLOB keys referring to the JAR files required to run this job
+ * @return list of BLOB keys referring to the JAR files required to run this job
+ */
+ public List<BlobKey> getRequiredJarFiles() {
+ return this.requiredJarFiles;
}
-
+
// --------------------------------------------------------------------------------------------
public JobID getJobID() {
@@ -624,7 +627,7 @@ public class ExecutionGraph {
* @param error
*/
void notifyExecutionChange(JobVertexID vertexId, int subtask, ExecutionAttemptID executionId, ExecutionState newExecutionState, Throwable error) {
-
+
// we must be very careful here with exceptions
if (this.executionListeners.size() > 0) {
@@ -644,7 +647,7 @@ public class ExecutionGraph {
fail(error);
}
}
-
+
// --------------------------------------------------------------------------------------------
// Miscellaneous
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index b6c7d78..3ea1afc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -27,6 +27,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
+import org.apache.flink.runtime.blob.BlobKey;
import org.slf4j.Logger;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.deployment.GateDeploymentDescriptor;
@@ -384,7 +385,7 @@ public class ExecutionVertex {
}
}
- String[] jarFiles = getExecutionGraph().getUserCodeJarFiles();
+ List<BlobKey> jarFiles = getExecutionGraph().getRequiredJarFiles();
return new TaskDeploymentDescriptor(getJobId(), getJobvertexId(), executionId, getTaskName(),
subTaskIndex, getTotalNumberOfParallelSubtasks(),
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
index 095b022..88450c2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/instance/Instance.java
@@ -27,10 +27,6 @@ import java.util.List;
import java.util.Queue;
import java.util.Set;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheProfileRequest;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheProfileResponse;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheUpdate;
import org.apache.flink.runtime.ipc.RPC;
import org.apache.flink.runtime.jobgraph.JobID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotAvailablilityListener;
@@ -190,30 +186,8 @@ public class Instance {
}
}
- public void checkLibraryAvailability(JobID jobID) throws IOException {
- String[] requiredLibraries = LibraryCacheManager.getRequiredJarFiles(jobID);
- if (requiredLibraries == null) {
- throw new IOException("No entry of required libraries for job " + jobID);
- }
-// if (requiredLibraries.length > 0) {
- LibraryCacheProfileRequest request = new LibraryCacheProfileRequest();
- request.setRequiredLibraries(requiredLibraries);
-
- // Send the request
- LibraryCacheProfileResponse response = getTaskManagerProxy().getLibraryCacheProfile(request);
-
- // Check response and transfer libraries if necessary
- for (int k = 0; k < requiredLibraries.length; k++) {
- if (!response.isCached(k)) {
- LibraryCacheUpdate update = new LibraryCacheUpdate(requiredLibraries[k]);
- getTaskManagerProxy().updateLibraryCache(update);
- }
- }
-// }
- }
-
// --------------------------------------------------------------------------------------------
// Heartbeats
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
index d72b1d5..d400908 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
@@ -159,7 +159,7 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
public RuntimeUDFContext createRuntimeContext(String taskName) {
Environment env = getEnvironment();
return new IterativeRuntimeUdfContext(taskName, env.getCurrentNumberOfSubtasks(),
- env.getIndexInSubtaskGroup(), userCodeClassLoader);
+ env.getIndexInSubtaskGroup(), getUserCodeClassLoader());
}
// --------------------------------------------------------------------------------------------
@@ -337,7 +337,8 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
private TypeSerializer<OT> getOutputSerializer() {
TypeSerializerFactory<OT> serializerFactory;
- if ((serializerFactory = getLastTasksConfig().getOutputSerializer(userCodeClassLoader)) == null) {
+ if ((serializerFactory = getLastTasksConfig().getOutputSerializer(getUserCodeClassLoader())) ==
+ null) {
throw new RuntimeException("Missing output serializer for workset update.");
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
index 0c04538..ea8efb6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
@@ -116,8 +116,9 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
// add the outputs for the final solution
this.finalOutputWriters = new ArrayList<BufferWriter>();
final TaskConfig finalOutConfig = this.config.getIterationHeadFinalOutputConfig();
+ final ClassLoader userCodeClassLoader = getUserCodeClassLoader();
this.finalOutputCollector = RegularPactTask.getOutputCollector(this, finalOutConfig,
- this.userCodeClassLoader, this.finalOutputWriters, finalOutConfig.getNumOutputs());
+ userCodeClassLoader, this.finalOutputWriters, finalOutConfig.getNumOutputs());
// sanity check the setup
final int writersIntoStepFunction = this.eventualOutputs.size();
@@ -159,6 +160,7 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
private <BT> CompactingHashTable<BT> initCompactingHashTable() throws Exception {
// get some memory
double hashjoinMemorySize = config.getRelativeSolutionSetMemory();
+ final ClassLoader userCodeClassLoader = getUserCodeClassLoader();
TypeSerializerFactory<BT> solutionTypeSerializerFactory = config.getSolutionSetSerializer(userCodeClassLoader);
TypeComparatorFactory<BT> solutionTypeComparatorFactory = config.getSolutionSetComparator(userCodeClassLoader);
@@ -196,8 +198,10 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
}
private <BT> JoinHashMap<BT> initJoinHashMap() {
- TypeSerializerFactory<BT> solutionTypeSerializerFactory = config.getSolutionSetSerializer(userCodeClassLoader);
- TypeComparatorFactory<BT> solutionTypeComparatorFactory = config.getSolutionSetComparator(userCodeClassLoader);
+ TypeSerializerFactory<BT> solutionTypeSerializerFactory = config.getSolutionSetSerializer
+ (getUserCodeClassLoader());
+ TypeComparatorFactory<BT> solutionTypeComparatorFactory = config.getSolutionSetComparator
+ (getUserCodeClassLoader());
TypeSerializer<BT> solutionTypeSerializer = solutionTypeSerializerFactory.getSerializer();
TypeComparator<BT> solutionTypeComparator = solutionTypeComparatorFactory.createComparator();
@@ -221,7 +225,7 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
}
private SuperstepBarrier initSuperstepBarrier() {
- SuperstepBarrier barrier = new SuperstepBarrier(userCodeClassLoader);
+ SuperstepBarrier barrier = new SuperstepBarrier(getUserCodeClassLoader());
this.toSync.subscribeToEvent(barrier, AllWorkersDoneEvent.class);
this.toSync.subscribeToEvent(barrier, TerminationEvent.class);
return barrier;
@@ -258,7 +262,8 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
if (isWorksetIteration) {
initialSolutionSetInput = config.getIterationHeadSolutionSetInputIndex();
- TypeSerializerFactory<X> solutionTypeSerializerFactory = config.getSolutionSetSerializer(userCodeClassLoader);
+ TypeSerializerFactory<X> solutionTypeSerializerFactory = config
+ .getSolutionSetSerializer(getUserCodeClassLoader());
solutionTypeSerializer = solutionTypeSerializerFactory;
// setup the index for the solution set
@@ -296,7 +301,8 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
}
// instantiate all aggregators and register them at the iteration global registry
- aggregatorRegistry = new RuntimeAggregatorRegistry(config.getIterationAggregators(userCodeClassLoader));
+ aggregatorRegistry = new RuntimeAggregatorRegistry(config.getIterationAggregators
+ (getUserCodeClassLoader()));
IterationAggregatorBroker.instance().handIn(brokerKey, aggregatorRegistry);
DataInputView superstepResult = null;
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
index f75ba61..91ae1e9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationSynchronizationSinkTask.java
@@ -30,7 +30,6 @@ import org.apache.flink.api.common.aggregators.Aggregator;
import org.apache.flink.api.common.aggregators.AggregatorWithName;
import org.apache.flink.api.common.aggregators.ConvergenceCriterion;
import org.apache.flink.runtime.event.task.AbstractTaskEvent;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
import org.apache.flink.runtime.io.network.api.MutableRecordReader;
import org.apache.flink.runtime.iterative.event.AllWorkersDoneEvent;
import org.apache.flink.runtime.iterative.event.TerminationEvent;
@@ -55,8 +54,6 @@ public class IterationSynchronizationSinkTask extends AbstractInvokable implemen
private MutableRecordReader<IntegerRecord> headEventReader;
- private ClassLoader userCodeClassLoader;
-
private SyncEventHandler eventHandler;
private ConvergenceCriterion<Value> convergenceCriterion;
@@ -81,18 +78,17 @@ public class IterationSynchronizationSinkTask extends AbstractInvokable implemen
@Override
public void invoke() throws Exception {
- userCodeClassLoader = LibraryCacheManager.getClassLoader(getEnvironment().getJobID());
TaskConfig taskConfig = new TaskConfig(getTaskConfiguration());
// store all aggregators
this.aggregators = new HashMap<String, Aggregator<?>>();
- for (AggregatorWithName<?> aggWithName : taskConfig.getIterationAggregators(userCodeClassLoader)) {
+ for (AggregatorWithName<?> aggWithName : taskConfig.getIterationAggregators(getUserCodeClassLoader())) {
aggregators.put(aggWithName.getName(), aggWithName.getAggregator());
}
// store the aggregator convergence criterion
if (taskConfig.usesConvergenceCriterion()) {
- convergenceCriterion = taskConfig.getConvergenceCriterion(userCodeClassLoader);
+ convergenceCriterion = taskConfig.getConvergenceCriterion(getUserCodeClassLoader());
convergenceAggregatorName = taskConfig.getConvergenceCriterionAggregatorName();
Preconditions.checkNotNull(convergenceAggregatorName);
}
@@ -101,7 +97,8 @@ public class IterationSynchronizationSinkTask extends AbstractInvokable implemen
// set up the event handler
int numEventsTillEndOfSuperstep = taskConfig.getNumberOfEventsUntilInterruptInIterativeGate(0);
- eventHandler = new SyncEventHandler(numEventsTillEndOfSuperstep, aggregators, userCodeClassLoader);
+ eventHandler = new SyncEventHandler(numEventsTillEndOfSuperstep, aggregators,
+ getEnvironment().getUserClassLoader());
headEventReader.subscribeToEvent(eventHandler, WorkerDoneEvent.class);
IntegerRecord dummy = new IntegerRecord();
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
index 7df8967..c42bf92 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.jobgraph;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
@@ -32,7 +33,6 @@ import java.util.Map;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FSDataInputStream;
-import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.io.IOReadableWritable;
@@ -40,7 +40,8 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStream;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.core.memory.DataOutputViewStream;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
+import org.apache.flink.runtime.blob.BlobClient;
+import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.types.StringValue;
/**
@@ -48,22 +49,21 @@ import org.apache.flink.types.StringValue;
*/
public class JobGraph implements IOReadableWritable {
- /** Size of the buffer to be allocated for transferring attached files. */
- private static final int BUFFERSIZE = 8192;
-
-
// --------------------------------------------------------------------------------------------
// Members that define the structure / topology of the graph
// --------------------------------------------------------------------------------------------
- /** List of JAR files required to run this job. */
- private final ArrayList<Path> userJars = new ArrayList<Path>();
-
/** List of task vertices included in this job graph. */
private final Map<JobVertexID, AbstractJobVertex> taskVertices = new LinkedHashMap<JobVertexID, AbstractJobVertex>();
/** The job configuration attached to this job. */
private final Configuration jobConfiguration = new Configuration();
+
+ /** Set of JAR files required to run this job. */
+ private final transient List<Path> userJars = new ArrayList<Path>();
+
+ /** Set of blob keys identifying the JAR files required to run this job. */
+ private final List<BlobKey> userJarBlobKeys = new ArrayList<BlobKey>();
/** ID of this job. */
private final JobID jobID;
@@ -336,7 +336,7 @@ public class JobGraph implements IOReadableWritable {
ois.close();
// Read required jar files
- readRequiredJarFiles(in);
+ readJarBlobKeys(in);
}
@@ -359,9 +359,47 @@ public class JobGraph implements IOReadableWritable {
oos.close();
// Write out all required jar files
- writeRequiredJarFiles(out);
+ writeJarBlobKeys(out);
+ }
+
+ /**
+ * Writes the BLOB keys of the jar files required to run this job to the given {@link org.apache.flink.core.memory.DataOutputView}.
+ *
+ * @param out
+ * the data output to write the BLOB keys to
+ * @throws IOException
+ * thrown if an error occurs while writing to the data output
+ */
+ private void writeJarBlobKeys(final DataOutputView out) throws IOException {
+
+ out.writeInt(this.userJarBlobKeys.size());
+
+ for (final Iterator<BlobKey> it = this.userJarBlobKeys.iterator(); it.hasNext();) {
+ it.next().write(out);
+ }
}
+ /**
+ * Reads the BLOB keys for the JAR files required to run this job and registers them.
+ *
+ * @param in
+ * the data stream to read the BLOB keys from
+ * @throws IOException
+ * thrown if an error occurs while reading the stream
+ */
+ private void readJarBlobKeys(final DataInputView in) throws IOException {
+
+ // Do jar files follow;
+ final int numberOfBlobKeys = in.readInt();
+
+ for (int i = 0; i < numberOfBlobKeys; ++i) {
+ final BlobKey key = new BlobKey();
+ key.read(in);
+ this.userJarBlobKeys.add(key);
+ }
+ }
+
+
// --------------------------------------------------------------------------------------------
// Handling of attached JAR files
// --------------------------------------------------------------------------------------------
@@ -383,89 +421,53 @@ public class JobGraph implements IOReadableWritable {
}
/**
- * Returns a (possibly empty) array of paths to JAR files which are required to run the job on a task manager.
- *
- * @return a (possibly empty) array of paths to JAR files which are required to run the job on a task manager
+ * Returns a set of BLOB keys referring to the JAR files required to run this job.
+ *
+ * @return set of BLOB keys referring to the JAR files required to run this job
*/
- public Path[] getJars() {
- return userJars.toArray(new Path[userJars.size()]);
- }
-
- /**
- * Reads required JAR files from an input stream and adds them to the
- * library cache manager.
- *
- * @param in
- * the data stream to read the JAR files from
- * @throws IOException
- * thrown if an error occurs while reading the stream
- */
- private void readRequiredJarFiles(final DataInputView in) throws IOException {
+ public List<BlobKey> getUserJarBlobKeys() {
- // Do jar files follow;
- final int numJars = in.readInt();
-
- if (numJars > 0) {
-
- for (int i = 0; i < numJars; i++) {
-
- final Path p = new Path();
- p.read(in);
- this.userJars.add(p);
-
- // Read the size of the jar file
- final long sizeOfJar = in.readLong();
-
- // Add the jar to the library manager
- LibraryCacheManager.addLibrary(this.jobID, p, sizeOfJar, in);
- }
-
- }
-
- // Register this job with the library cache manager
- LibraryCacheManager.register(this.jobID, this.userJars.toArray(new Path[0]));
+ return this.userJarBlobKeys;
}
-
+
/**
- * Writes the JAR files of all vertices in array <code>jobVertices</code> to the specified output stream.
- *
- * @param out
- * the output stream to write the JAR files to
+ * Uploads the previously added user jar file to the job manager through the job manager's BLOB server.
+ *
+ * @param serverAddress
+ * the network address of the BLOB server
* @throws IOException
- * thrown if an error occurs while writing to the stream
+ * thrown if an I/O error occurs during the upload
*/
- private void writeRequiredJarFiles(DataOutputView out) throws IOException {
+ public void uploadRequiredJarFiles(final InetSocketAddress serverAddress) throws IOException {
- // Now check if all the collected jar files really exist
- final FileSystem fs = FileSystem.getLocalFileSystem();
-
- for (int i = 0; i < this.userJars.size(); i++) {
- if (!fs.exists(this.userJars.get(i))) {
- throw new IOException("Cannot find jar file " + this.userJars.get(i));
- }
+ if (this.userJars.isEmpty()) {
+ return;
}
- // How many jar files follow?
- out.writeInt(this.userJars.size());
-
- for (int i = 0; i < this.userJars.size(); i++) {
+ BlobClient bc = null;
+ try {
- final Path jar = this.userJars.get(i);
+ bc = new BlobClient(serverAddress);
- // Write out the actual path
- jar.write(out);
+ for (final Iterator<Path> it = this.userJars.iterator(); it.hasNext();) {
- // Write out the length of the file
- final FileStatus file = fs.getFileStatus(jar);
- out.writeLong(file.getLen());
+ final Path jar = it.next();
+ final FileSystem fs = jar.getFileSystem();
+ FSDataInputStream is = null;
+ try {
+ is = fs.open(jar);
+ final BlobKey key = bc.put(is);
+ this.userJarBlobKeys.add(key);
+ } finally {
+ if (is != null) {
+ is.close();
+ }
+ }
+ }
- // Now write the jar file
- final FSDataInputStream inStream = fs.open(this.userJars.get(i));
- final byte[] buf = new byte[BUFFERSIZE];
- int read = inStream.read(buf, 0, buf.length);
- while (read > 0) {
- out.write(buf, 0, read);
- read = inStream.read(buf, 0, buf.length);
+ } finally {
+ if (bc != null) {
+ bc.close();
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/ca4380c3/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
index f0280ce..1666788 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.Environment;
/**
- * Abstract base class for every task class in Nephele.
+ * Abstract base class for every task class in Flink.
*/
public abstract class AbstractInvokable {
@@ -64,6 +64,15 @@ public abstract class AbstractInvokable {
return this.environment;
}
+ /**
+ * Returns the user code class loader of this invokable.
+ *
+ * @return user code class loader of this invokable.
+ */
+ public ClassLoader getUserCodeClassLoader() {
+ return getEnvironment().getUserClassLoader();
+ }
+
/**
* Returns the current number of subtasks the respective task is split into.