You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ju...@apache.org on 2011/01/18 16:21:58 UTC
svn commit: r1060413 - in
/tika/trunk/tika-core/src/main/java/org/apache/tika/fork:
ClassLoaderProxy.java ForkClient.java ForkSerializer.java ForkServer.java
MemoryURLConnection.java MemoryURLStreamHandler.java
MemoryURLStreamHandlerFactory.java
Author: jukka
Date: Tue Jan 18 15:21:58 2011
New Revision: 1060413
URL: http://svn.apache.org/viewvc?rev=1060413&view=rev
Log:
TIKA-416: Out-of-process text extraction
Various optimizations. Add a watchdog timer thread to automatically terminate unused server processes.
Added:
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLConnection.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandler.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandlerFactory.java
Modified:
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderProxy.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkSerializer.java
tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderProxy.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderProxy.java?rev=1060413&r1=1060412&r2=1060413&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderProxy.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ClassLoaderProxy.java Tue Jan 18 15:21:58 2011
@@ -19,24 +19,28 @@ package org.apache.tika.fork;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
-import java.io.File;
-import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.OutputStream;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
class ClassLoaderProxy extends ClassLoader implements ForkProxy {
/** Serial version UID */
- private static final long serialVersionUID = 4350939227765568438L;
+ private static final long serialVersionUID = -7303109260448540420L;
- private final int resource;
+ /**
+ * Names of resources that could not be found. Used to avoid repeated
+ * lookup of commonly accessed, but often not present, resources like
+ * <code>META-INF/services/javax.xml.parsers.SAXParserFactory</code>.
+ */
+ private final Set<String> notFound = new HashSet<String>();
- private int count = 0;
+ private final int resource;
private transient DataInputStream input;
@@ -53,6 +57,9 @@ class ClassLoaderProxy extends ClassLoad
@Override
protected synchronized URL findResource(String name) {
+ if (notFound.contains(name)) {
+ return null;
+ }
try {
// Send a request to load the resource data
output.write(ForkServer.RESOURCE);
@@ -63,8 +70,9 @@ class ClassLoaderProxy extends ClassLoad
// Receive the response
if (input.readBoolean()) {
- return readStreamToFile().toURI().toURL();
+ return MemoryURLStreamHandler.createURL(readStream());
} else {
+ notFound.add(name);
return null;
}
} catch (IOException e) {
@@ -85,7 +93,7 @@ class ClassLoaderProxy extends ClassLoad
// Receive the response
List<URL> resources = new ArrayList<URL>();
while (input.readBoolean()) {
- resources.add(readStreamToFile().toURI().toURL());
+ resources.add(MemoryURLStreamHandler.createURL(readStream()));
}
return Collections.enumeration(resources);
}
@@ -103,18 +111,17 @@ class ClassLoaderProxy extends ClassLoad
// Receive the response
if (input.readBoolean()) {
- byte[] data = readStreamToMemory();
+ byte[] data = readStream();
return defineClass(name, data, 0, data.length);
} else {
- return null;
+ throw new ClassNotFoundException("Unable to find class " + name);
}
} catch (IOException e) {
- e.printStackTrace();
- throw new ClassNotFoundException("Unable load class " + name, e);
+ throw new ClassNotFoundException("Unable to load class " + name, e);
}
}
- private byte[] readStreamToMemory() throws IOException {
+ private byte[] readStream() throws IOException {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
byte[] buffer = new byte[0xffff];
int n;
@@ -125,22 +132,4 @@ class ClassLoaderProxy extends ClassLoad
return stream.toByteArray();
}
- private File readStreamToFile() throws IOException {
- File file = new File("resource-" + count++ + ".bin");
-
- OutputStream stream = new FileOutputStream(file);
- try {
- byte[] buffer = new byte[0xffff];
- int n;
- while ((n = input.readUnsignedShort()) > 0) {
- input.readFully(buffer, 0, n);
- stream.write(buffer, 0, n);
- }
- } finally {
- stream.close();
- }
-
- return file;
- }
-
}
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java?rev=1060413&r1=1060412&r2=1060413&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkClient.java Tue Jan 18 15:21:58 2011
@@ -24,9 +24,11 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectOutputStream;
-import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
+import java.util.jar.JarEntry;
+import java.util.jar.JarOutputStream;
+import java.util.zip.ZipEntry;
import org.apache.tika.io.IOUtils;
import org.xml.sax.ContentHandler;
@@ -37,7 +39,7 @@ class ForkClient {
private final List<ForkResource> resources = new ArrayList<ForkResource>();
- private final File directory;
+ private final File jar;
private final Process process;
@@ -48,21 +50,14 @@ class ForkClient {
private final InputStream error;
public ForkClient(ClassLoader loader, Object object) throws IOException {
- this.directory = File.createTempFile("apache-tika-", "-fork");
- directory.delete();
- directory.mkdir();
-
boolean ok = false;
try {
- copyClassToDirectory(ForkServer.class);
- copyClassToDirectory(ForkSerializer.class);
- copyClassToDirectory(ForkProxy.class);
- copyClassToDirectory(ClassLoaderProxy.class);
+ this.jar = createBootstrapJar();
ProcessBuilder builder = new ProcessBuilder();
- builder.directory(directory);
- builder.command(java, ForkServer.class.getName());
+ builder.command(java, "-jar", jar.getPath());
this.process = builder.start();
+
this.output = new DataOutputStream(process.getOutputStream());
this.input = new DataInputStream(process.getInputStream());
this.error = process.getErrorStream();
@@ -73,35 +68,8 @@ class ForkClient {
ok = true;
} finally {
if (!ok) {
- delete(directory);
- }
- }
- }
-
- /**
- * Copies the <code>.class</code> file of the given class to the
- * directory from where the forked server process can load it
- * during startup before setting up the stdin/out communication
- * channel with the parent process.
- *
- * @param klass the class to be copied
- * @throws IOException if the class could not be copied
- */
- private void copyClassToDirectory(Class<?> klass) throws IOException {
- String path = klass.getName().replace('.', '/') + ".class";
- ClassLoader loader = klass.getClassLoader();
- InputStream input = loader.getResourceAsStream(path);
- try {
- File file = new File(directory, path);
- file.getParentFile().mkdirs();
- OutputStream output = new FileOutputStream(file);
- try {
- IOUtils.copy(input, output);
- } finally {
- output.close();
+ close();
}
- } finally {
- input.close();
}
}
@@ -151,13 +119,23 @@ class ForkClient {
public synchronized void close() {
try {
- output.close();
- input.close();
- error.close();
+ if (output != null) {
+ output.close();
+ }
+ if (input != null) {
+ input.close();
+ }
+ if (error != null) {
+ error.close();
+ }
} catch (IOException ignore) {
}
- process.destroy();
- delete(directory);
+ if (process != null) {
+ process.destroy();
+ }
+ if (jar != null) {
+ // jar.delete();
+ }
}
private byte waitForResponse(List<ForkResource> resources)
@@ -200,18 +178,62 @@ class ForkClient {
}
/**
- * Recursively deletes the given file or directory.
+ * Creates a temporary jar file that can be used to bootstrap the forked
+ * server process. Remember to remove the file when no longer used.
+ *
+ * @return the created jar file
+ * @throws IOException if the bootstrap archive could not be created
+ */
+ private static File createBootstrapJar() throws IOException {
+ File file = File.createTempFile("apache-tika-fork-", ".jar");
+ boolean ok = false;
+ try {
+ fillBootstrapJar(file);
+ ok = true;
+ } finally {
+ if (!ok) {
+ file.delete();
+ }
+ }
+ return file;
+ }
+
+ /**
+ * Fills in the jar file used to bootstrap the forked server process.
+ * All the required <code>.class</code> files and a manifest with a
+ * <code>Main-Class</code> entry are written into the archive.
*
- * @param file file or directory
+ * @param file file to hold the bootstrap archive
+ * @throws IOException if the bootstrap archive could not be created
*/
- private void delete(File file) {
- File[] children = file.listFiles();
- if (children != null) {
- for (File child : children) {
- delete(child);
+ private static void fillBootstrapJar(File file) throws IOException {
+ JarOutputStream jar = new JarOutputStream(new FileOutputStream(file));
+ try {
+ String manifest =
+ "Main-Class: " + ForkServer.class.getName() + "\n";
+ jar.putNextEntry(new ZipEntry("META-INF/MANIFEST.MF"));
+ jar.write(manifest.getBytes("UTF-8"));
+
+ Class<?>[] bootstrap = {
+ ForkServer.class, ForkSerializer.class,
+ ForkProxy.class, ClassLoaderProxy.class,
+ MemoryURLConnection.class, MemoryURLStreamHandler.class,
+ MemoryURLStreamHandlerFactory.class
+ };
+ ClassLoader loader = ForkServer.class.getClassLoader();
+ for (Class<?> klass : bootstrap) {
+ String path = klass.getName().replace('.', '/') + ".class";
+ InputStream input = loader.getResourceAsStream(path);
+ try {
+ jar.putNextEntry(new JarEntry(path));
+ IOUtils.copy(input, jar);
+ } finally {
+ input.close();
+ }
}
+ } finally {
+ jar.close();
}
- file.delete();
}
}
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkSerializer.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkSerializer.java?rev=1060413&r1=1060412&r2=1060413&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkSerializer.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkSerializer.java Tue Jan 18 15:21:58 2011
@@ -16,14 +16,9 @@
*/
package org.apache.tika.fork;
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.io.ObjectStreamClass;
/**
Modified: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java?rev=1060413&r1=1060412&r2=1060413&view=diff
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java (original)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/ForkServer.java Tue Jan 18 15:21:58 2011
@@ -24,16 +24,20 @@ import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.OutputStream;
import java.lang.reflect.Method;
+import java.net.URL;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
+import java.util.zip.Checksum;
-class ForkServer implements Runnable {
+class ForkServer implements Runnable, Checksum {
- public static final byte ERROR = -1;
-
- public static final byte REPLY = 0;
+ public static final byte DONE = 0;
public static final byte CALL = 1;
- public static final byte RESOURCE = 2;
+ public static final byte PING = 2;
+
+ public static final byte RESOURCE = 3;
/**
* Starts a forked server process using the standard input and output
@@ -45,10 +49,17 @@ class ForkServer implements Runnable {
* @throws Exception if the server could not be started
*/
public static void main(String[] args) throws Exception {
+ URL.setURLStreamHandlerFactory(new MemoryURLStreamHandlerFactory());
+
ForkServer server = new ForkServer(System.in, System.out);
System.setIn(new ByteArrayInputStream(new byte[0]));
System.setOut(System.err);
- server.run();
+
+ Thread watchdog = new Thread(server, "Tika Watchdog");
+ watchdog.setDaemon(true);
+ watchdog.start();
+
+ server.processRequests();
}
/** Input stream for reading from the parent process */
@@ -57,6 +68,8 @@ class ForkServer implements Runnable {
/** Output stream for writing to the parent process */
private final DataOutputStream output;
+ private volatile boolean active = true;
+
/**
* Sets up a forked server instance using the given stdin/out
* communication channel.
@@ -67,26 +80,43 @@ class ForkServer implements Runnable {
*/
public ForkServer(InputStream input, OutputStream output)
throws IOException {
- this.input = new DataInputStream(input);
- this.output = new DataOutputStream(output);
+ this.input =
+ new DataInputStream(new CheckedInputStream(input, this));
+ this.output =
+ new DataOutputStream(new CheckedOutputStream(output, this));
}
public void run() {
try {
+ while (active) {
+ active = false;
+ Thread.sleep(5000);
+ }
+ System.exit(0);
+ } catch (InterruptedException e) {
+ }
+ }
+
+ public void processRequests() {
+ try {
ClassLoader loader = (ClassLoader) readObject(
ForkServer.class.getClassLoader());
Thread.currentThread().setContextClassLoader(loader);
Object object = readObject(loader);
while (true) {
- Method method = getMethod(object, input.readUTF());
- Object[] args = new Object[method.getParameterTypes().length];
- for (int i = 0; i < args.length; i++) {
- args[i] = readObject(loader);
+ int request = input.read();
+ if (request == CALL) {
+ Method method = getMethod(object, input.readUTF());
+ Object[] args = new Object[method.getParameterTypes().length];
+ for (int i = 0; i < args.length; i++) {
+ args[i] = readObject(loader);
+ }
+ method.invoke(object, args);
+ } else if (request != PING) {
+
}
- method.invoke(object, args);
-
- output.write(REPLY);
+ output.write(DONE);
output.flush();
}
} catch (Throwable t) {
@@ -128,10 +158,27 @@ class ForkServer implements Runnable {
}
// Tell the parent process that we successfully received this object
- output.writeByte(ForkServer.REPLY);
+ output.writeByte(ForkServer.DONE);
output.flush();
return object;
}
+ //-------------------------------------------------------------< Checsum >
+
+ public void update(int b) {
+ active = true;
+ }
+
+ public void update(byte[] b, int off, int len) {
+ active = true;
+ }
+
+ public long getValue() {
+ return 0;
+ }
+
+ public void reset() {
+ }
+
}
Added: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLConnection.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLConnection.java?rev=1060413&view=auto
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLConnection.java (added)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLConnection.java Tue Jan 18 15:21:58 2011
@@ -0,0 +1,42 @@
+/*
+ * 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.tika.fork;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.net.URL;
+import java.net.URLConnection;
+
+class MemoryURLConnection extends URLConnection {
+
+ private final byte[] data;
+
+ MemoryURLConnection(URL url, byte[] data) {
+ super(url);
+ this.data = data;
+ }
+
+ @Override
+ public void connect() {
+ }
+
+ @Override
+ public InputStream getInputStream() {
+ return new ByteArrayInputStream(data);
+ }
+
+}
\ No newline at end of file
Added: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandler.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandler.java?rev=1060413&view=auto
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandler.java (added)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandler.java Tue Jan 18 15:21:58 2011
@@ -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.tika.fork;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLConnection;
+import java.net.URLStreamHandler;
+import java.util.Map;
+import java.util.WeakHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+class MemoryURLStreamHandler extends URLStreamHandler {
+
+ private static final AtomicInteger counter = new AtomicInteger();
+
+ private static final Map<URL, byte[]> URLs = new WeakHashMap<URL, byte[]>();
+
+ public static URL createURL(byte[] data) {
+ try {
+ int i = counter.incrementAndGet();
+ URL url = new URL("tika-in-memory", "localhost", "/" + i);
+ URLs.put(url, data);
+ return url;
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected URLConnection openConnection(URL u) throws IOException {
+ byte[] data = URLs.get(u);
+ if (data != null) {
+ return new MemoryURLConnection(u, URLs.get(u));
+ } else {
+ throw new IOException("Unknown URL: " + u);
+ }
+ }
+
+}
Added: tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandlerFactory.java
URL: http://svn.apache.org/viewvc/tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandlerFactory.java?rev=1060413&view=auto
==============================================================================
--- tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandlerFactory.java (added)
+++ tika/trunk/tika-core/src/main/java/org/apache/tika/fork/MemoryURLStreamHandlerFactory.java Tue Jan 18 15:21:58 2011
@@ -0,0 +1,32 @@
+/*
+ * 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.tika.fork;
+
+import java.net.URLStreamHandler;
+import java.net.URLStreamHandlerFactory;
+
+class MemoryURLStreamHandlerFactory implements URLStreamHandlerFactory {
+
+ public URLStreamHandler createURLStreamHandler(String protocol) {
+ if ("tika-in-memory".equals(protocol)) {
+ return new MemoryURLStreamHandler();
+ } else {
+ return null;
+ }
+ }
+
+}
\ No newline at end of file