Supporting concurrent execution of CachedCompiler.loadFromJava

Author: glukos

src/main/java/net/openhft/compiler/CachedCompiler.java

@@ -32,6 +32,8 @@
 import java.io.IOException;
 import java.io.PrintWriter;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import static net.openhft.compiler.CompilerUtils.*;
 
@@ -48,8 +50,7 @@ public class CachedCompiler implements Closeable {
     @Nullable
     private final File classDir;
 
-    private final Map<String, JavaFileObject> javaFileObjects =
-            new HashMap<String, JavaFileObject>();
+    private final ConcurrentMap<String, JavaFileObject> javaFileObjects = new ConcurrentHashMap<>();
 
     public CachedCompiler(@Nullable File sourceDir, @Nullable File classDir) {
         this.sourceDir = sourceDir;
@@ -97,7 +98,7 @@ Map<String, byte[]> compileFromJava(@NotNull String className,
 
         } else {
             javaFileObjects.put(className, new JavaSourceFromString(className, javaCode));
-            compilationUnits = javaFileObjects.values();
+            compilationUnits = new ArrayList<>(javaFileObjects.values()); // To prevent CME from compiler code
         }
         // reuse the same file manager to allow caching of jar files
         List<String> options = Arrays.asList("-g", "-nowarn");
@@ -109,7 +110,7 @@ public void report(Diagnostic<? extends JavaFileObject> diagnostic) {
                 }
             }
         }, options, null, compilationUnits).call();
-        Map<String, byte[]> result = fileManager.getAllBuffers();
+
         if (!ok) {
             // compilation error, so we want to exclude this file from future compilation passes
             if (sourceDir == null)
@@ -118,7 +119,11 @@ public void report(Diagnostic<? extends JavaFileObject> diagnostic) {
             // nothing to return due to compiler error
             return Collections.emptyMap();
         }
-        return result;
+        else {
+            Map<String, byte[]> result = fileManager.getAllBuffers();
+
+            return result;
+        }
     }
 
     public Class loadFromJava(@NotNull ClassLoader classLoader,
@@ -143,7 +148,8 @@ public Class loadFromJava(@NotNull ClassLoader classLoader,
             StandardJavaFileManager standardJavaFileManager = s_compiler.getStandardFileManager(null, null, null);
             fileManagerMap.put(classLoader, fileManager = new MyJavaFileManager(standardJavaFileManager));
         }
-        for (Map.Entry<String, byte[]> entry : compileFromJava(className, javaCode, printWriter, fileManager).entrySet()) {
+        final Map<String, byte[]> compiled = compileFromJava(className, javaCode, printWriter, fileManager);
+        for (Map.Entry<String, byte[]> entry : compiled.entrySet()) {
             String className2 = entry.getKey();
             synchronized (loadedClassesMap) {
                 if (loadedClasses.containsKey(className2))
@@ -157,9 +163,17 @@ public Class loadFromJava(@NotNull ClassLoader classLoader,
                     LOG.info("Updated {} in {}", className2, classDir);
                 }
             }
-            Class clazz2 = CompilerUtils.defineClass(classLoader, className2, bytes);
-            synchronized (loadedClassesMap) {
-                loadedClasses.put(className2, clazz2);
+
+            synchronized (className2.intern()) { // To prevent duplicate class definition error
+                synchronized (loadedClassesMap) {
+                    if (loadedClasses.containsKey(className2))
+                        continue;
+                }
+
+                Class<?> clazz2 = CompilerUtils.defineClass(classLoader, className2, bytes);
+                synchronized (loadedClassesMap) {
+                    loadedClasses.put(className2, clazz2);
+                }
             }
         }
         synchronized (loadedClassesMap) {

src/main/java/net/openhft/compiler/CloseableByteArrayOutputStream.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2014 Higher Frequency Trading
+ *
+ * http://chronicle.software
+ *
+ * Licensed 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 net.openhft.compiler;
+
+import java.io.ByteArrayOutputStream;
+import java.util.concurrent.CompletableFuture;
+
+public class CloseableByteArrayOutputStream extends ByteArrayOutputStream {
+    private final CompletableFuture<?> closeFuture = new CompletableFuture<>();
+
+    @Override
+    public void close() {
+        closeFuture.complete(null);
+    }
+
+    public CompletableFuture<?> closeFuture() {
+        return closeFuture;
+    }
+}

src/main/java/net/openhft/compiler/MyJavaFileManager.java

@@ -19,6 +19,8 @@
 package net.openhft.compiler;
 
 import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import sun.misc.Unsafe;
 
 import javax.tools.*;
@@ -30,8 +32,12 @@
 import java.lang.reflect.Method;
 import java.net.URI;
 import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 class MyJavaFileManager implements JavaFileManager {
+    private static final Logger LOG = LoggerFactory.getLogger(MyJavaFileManager.class);
     private final static Unsafe unsafe;
     private static final long OVERRIDE_OFFSET;
 
@@ -50,7 +56,7 @@ class MyJavaFileManager implements JavaFileManager {
     private final StandardJavaFileManager fileManager;
 
     // synchronizing due to ConcurrentModificationException
-    private final Map<String, ByteArrayOutputStream> buffers = Collections.synchronizedMap(new LinkedHashMap<>());
+    private final Map<String, CloseableByteArrayOutputStream> buffers = Collections.synchronizedMap(new LinkedHashMap<>());
 
     MyJavaFileManager(StandardJavaFileManager fileManager) {
         this.fileManager = fileManager;
@@ -115,8 +121,13 @@ public JavaFileObject getJavaFileForOutput(Location location, final String class
         return new SimpleJavaFileObject(URI.create(className), kind) {
             @NotNull
             public OutputStream openOutputStream() {
-                ByteArrayOutputStream baos = new ByteArrayOutputStream();
-                buffers.put(className, baos);
+                // CloseableByteArrayOutputStream.closed is used to filter partial results from getAllBuffers()
+                CloseableByteArrayOutputStream baos = new CloseableByteArrayOutputStream();
+
+                // Reads from getAllBuffers() should be repeatable:
+                // let's ignore compile result in case compilation of this class was triggered before
+                buffers.putIfAbsent(className, baos);
+
                 return baos;
             }
         };
@@ -148,13 +159,35 @@ public void clearBuffers() {
 
     @NotNull
     public Map<String, byte[]> getAllBuffers() {
+        Map<String, byte[]> ret = new LinkedHashMap<>(buffers.size() * 2);
+        Map<String, CloseableByteArrayOutputStream> compiledClasses = new LinkedHashMap<>(ret.size());
+
         synchronized (buffers) {
-            Map<String, byte[]> ret = new LinkedHashMap<>(buffers.size() * 2);
-            for (Map.Entry<String, ByteArrayOutputStream> entry : buffers.entrySet()) {
-                ret.put(entry.getKey(), entry.getValue().toByteArray());
+            compiledClasses.putAll(buffers);
+        }
+
+        for (Map.Entry<String, CloseableByteArrayOutputStream> e : compiledClasses.entrySet()) {
+            try {
+                // Await for compilation in case class is still being compiled in previous compiler run.
+                e.getValue().closeFuture().get(30, TimeUnit.SECONDS);
+            } catch (InterruptedException t) {
+                Thread.currentThread().interrupt();
+
+                LOG.warn("Interrupted while waiting for compilation result [class=" + e.getKey() + "]");
+
+                break;
+            } catch (ExecutionException | TimeoutException t) {
+                LOG.warn("Failed to wait for compilation result [class=" + e.getKey() + "]", t);
+
+                continue;
             }
-            return ret;
+
+            final byte[] value = e.getValue().toByteArray();
+
+            ret.put(e.getKey(), value);
         }
+
+        return ret;
     }
 
     @SuppressWarnings("unchecked")