[feature] Use of ContentFilePool

Replaces direct construction of in memory RPC results with pooled
instances.

Signed-off-by: Patrick Reinhart <[email protected]>

Author: reinhapa

exist-core/src/main/java/org/exist/util/io/ContentFilePool.java

@@ -26,6 +26,8 @@
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.exist.util.Configuration;
 
+import javax.annotation.Nullable;
+
 /**
  * Generic pool for {@link ContentFile} instances used to represent RPC server data up to a defined size in memory first
  * before storing the data in the file system using the {@link TemporaryFileManager}.
@@ -50,23 +52,25 @@ public final class ContentFilePool extends GenericObjectPool<ContentFile> {
      * Creates a new pool using the givem temporary file manager, configuration and maximum idle time.
      *
      * @param tempFileManager the temporary file manager used when need to swap out data bigger than max in memory size
+     * @param brokerId the optional broker id
      * @param config the configuration used to configure the main pool properties
      */
-    public ContentFilePool(final TemporaryFileManager tempFileManager, final Configuration config) {
-        super(new ContentFilePoolObjectFactory(tempFileManager, toInMemorySize(config)), toPoolConfig(config));
+    public ContentFilePool(final TemporaryFileManager tempFileManager, @Nullable final String brokerId, final Configuration config) {
+        super(new ContentFilePoolObjectFactory(tempFileManager, toInMemorySize(config)), toPoolConfig(brokerId, config));
     }
 
     private static int toInMemorySize(final Configuration config) {
         return config.getInteger(PROPERTY_IN_MEMORY_SIZE, VirtualTempPath.DEFAULT_IN_MEMORY_SIZE);
     }
 
-    private static GenericObjectPoolConfig<ContentFile> toPoolConfig(final Configuration config) {
+    private static GenericObjectPoolConfig<ContentFile> toPoolConfig(@Nullable final String brokerId, final Configuration config) {
         final GenericObjectPoolConfig<ContentFile> poolConfig = new GenericObjectPoolConfig<>();
         poolConfig.setBlockWhenExhausted(false);
         poolConfig.setLifo(true);
         poolConfig.setMaxIdle(config.getInteger(PROPERTY_POOL_MAX_IDLE));
         poolConfig.setMaxTotal(config.getInteger(PROPERTY_POOL_SIZE));
-        poolConfig.setJmxNameBase("org.exist.management.exist:type=ContentFilePool");
+        final String poolName = brokerId == null ? "" : "pool." + brokerId;
+        poolConfig.setJmxNameBase("org.exist.management.exist:type=ContentFilePool,name=" + poolName);
         return poolConfig;
     }
 

exist-core/src/main/java/org/exist/xmlrpc/CachedContentFile.java

@@ -23,15 +23,19 @@
 
 import org.exist.util.io.ContentFile;
 
+import java.util.function.Consumer;
+
 /**
  * @author <a href="mailto:[email protected]">Patrick Reinhart</a>
  */
-public final class CachedContentFile extends AbstractCachedResult {
+final class CachedContentFile extends AbstractCachedResult {
     private final ContentFile result;
+    private final Consumer<ContentFile> poolConsumer;
 
-    public CachedContentFile(final ContentFile result) {
+    CachedContentFile(final ContentFile result, final Consumer<ContentFile> poolConsumer) {
         super(0);
         this.result = result;
+        this.poolConsumer = poolConsumer;
     }
 
     @Override
@@ -42,6 +46,7 @@ public ContentFile getResult() {
     @Override
     protected void doClose() {
         if (result != null) {
+            poolConsumer.accept(result);
             result.close();
         }
     }

exist-core/src/main/java/org/exist/xmlrpc/RpcConnection.java

@@ -73,8 +73,8 @@
 import org.exist.util.crypto.digest.DigestType;
 import org.exist.util.crypto.digest.MessageDigest;
 import org.exist.util.io.ContentFile;
+import org.exist.util.io.ContentFilePool;
 import org.exist.util.io.TemporaryFileManager;
-import org.exist.util.io.VirtualTempPath;
 import org.exist.util.serializer.SAXSerializer;
 import org.exist.util.serializer.SerializerPool;
 import org.exist.validation.ValidationReport;
@@ -137,12 +137,14 @@ public class RpcConnection implements RpcAPI {
     private final static Charset DEFAULT_ENCODING = StandardCharsets.UTF_8;
 
     private final XmldbRequestProcessorFactory factory;
+    private final ContentFilePool filePool;
     private final Subject user;
     private final Random random = new Random();
 
-    public RpcConnection(final XmldbRequestProcessorFactory factory, final Subject user) {
+    public RpcConnection(final XmldbRequestProcessorFactory factory, final ContentFilePool filePool, final Subject user) {
         super();
         this.factory = factory;
+        this.filePool = filePool;
         this.user = user;
     }
 
@@ -169,10 +171,6 @@ private TemporaryFileManager temporaryFileManager() {
         return TemporaryFileManager.getInstance();
     }
 
-    private VirtualTempPath createVirtualTempPath() {
-        return new VirtualTempPath(temporaryFileManager());
-    }
-
     private boolean createCollection(final XmldbURI collUri, final Date created) throws PermissionDeniedException, EXistException {
         withDb((broker, transaction) -> {
             Collection current = broker.getCollection(collUri);
@@ -704,7 +702,7 @@ public Map<String, Object> getDocumentData(final String docName, final Map<Strin
 
             // A tweak for very large resources, VirtualTempFile
             final Map<String, Object> result = new HashMap<>();
-            final VirtualTempPath tempFile = createVirtualTempPath();
+            final ContentFile tempFile = filePool.borrowObject();
 
             if (document.getResourceType() == DocumentImpl.XML_FILE) {
                 try (final OutputStream out = tempFile.newOutputStream();
@@ -724,11 +722,11 @@ public Map<String, Object> getDocumentData(final String docName, final Map<Strin
             if (tempFile.size() > MAX_DOWNLOAD_CHUNK_SIZE) {
                 offset = firstChunk.length;
 
-                final int handle = factory.resultSets.add(new CachedContentFile(tempFile));
+                final int handle = factory.resultSets.add(new CachedContentFile(tempFile, filePool::returnObject));
                 result.put("handle", Integer.toString(handle));
                 result.put("supports-long-offset", Boolean.TRUE);
             } else {
-                tempFile.close();
+                filePool.returnObject(tempFile);
             }
             result.put("offset", offset);
 
@@ -2180,7 +2178,7 @@ public Map<String, Object> retrieveFirstChunk(final String docName, final String
             final NodeProxy node = new NodeProxy(null, document, nodeId);
 
             final Map<String, Object> result = new HashMap<>();
-            final VirtualTempPath tempFile = createVirtualTempPath();
+            final ContentFile tempFile = filePool.borrowObject();
 
             if (compression && LOG.isDebugEnabled()) {
                 LOG.debug("retrieveFirstChunk with compression");
@@ -2199,11 +2197,11 @@ public Map<String, Object> retrieveFirstChunk(final String docName, final String
             if (tempFile.size() > MAX_DOWNLOAD_CHUNK_SIZE) {
                 offset = firstChunk.length;
 
-                final int handle = factory.resultSets.add(new CachedContentFile(tempFile));
+                final int handle = factory.resultSets.add(new CachedContentFile(tempFile, filePool::returnObject));
                 result.put("handle", Integer.toString(handle));
                 result.put("supports-long-offset", Boolean.TRUE);
             } else {
-                tempFile.close();
+                filePool.returnObject(tempFile);
             }
             result.put("offset", offset);
             return result;
@@ -2283,7 +2281,7 @@ public Map<String, Object> retrieveFirstChunk(final int resultId, final int num,
             }
 
             final Map<String, Object> result = new HashMap<>();
-            final VirtualTempPath tempFile = createVirtualTempPath();
+            final ContentFile tempFile = filePool.borrowObject();
 
             if (compression && LOG.isDebugEnabled()) {
                 LOG.debug("retrieveFirstChunk with compression");
@@ -2312,11 +2310,11 @@ public Map<String, Object> retrieveFirstChunk(final int resultId, final int num,
             if (tempFile.size() > MAX_DOWNLOAD_CHUNK_SIZE) {
                 offset = firstChunk.length;
 
-                final int handle = factory.resultSets.add(new CachedContentFile(tempFile));
+                final int handle = factory.resultSets.add(new CachedContentFile(tempFile, filePool::returnObject));
                 result.put("handle", Integer.toString(handle));
                 result.put("supports-long-offset", Boolean.TRUE);
             } else {
-                tempFile.close();
+                filePool.returnObject(tempFile);
             }
             result.put("offset", offset);
             return result;
@@ -2400,7 +2398,7 @@ public Map<String, Object> retrieveAllFirstChunk(final int resultId, final Map<S
             try {
 
                 final Map<String, Object> result = new HashMap<>();
-                final VirtualTempPath tempFile = createVirtualTempPath();
+                final ContentFile tempFile = filePool.borrowObject();
 
                 if (compression && LOG.isDebugEnabled()) {
                     LOG.debug("retrieveAllFirstChunk with compression");
@@ -2454,11 +2452,11 @@ public Map<String, Object> retrieveAllFirstChunk(final int resultId, final Map<S
                 if (tempFile.size() > MAX_DOWNLOAD_CHUNK_SIZE) {
                     offset = firstChunk.length;
 
-                    final int handle = factory.resultSets.add(new CachedContentFile(tempFile));
+                    final int handle = factory.resultSets.add(new CachedContentFile(tempFile, filePool::returnObject));
                     result.put("handle", Integer.toString(handle));
                     result.put("supports-long-offset", Boolean.TRUE);
                 } else {
-                    tempFile.close();
+                    filePool.returnObject(tempFile);
                 }
                 result.put("offset", offset);
                 return result;

exist-core/src/main/java/org/exist/xmlrpc/XmldbRequestProcessorFactory.java

@@ -36,6 +36,8 @@
 import org.exist.security.Subject;
 import org.exist.storage.BrokerPool;
 import org.exist.util.NamedThreadFactory;
+import org.exist.util.io.ContentFilePool;
+import org.exist.util.io.TemporaryFileManager;
 
 import java.util.Map;
 import java.util.UUID;
@@ -56,6 +58,7 @@ public class XmldbRequestProcessorFactory implements RequestProcessorFactoryFact
 
     private final boolean useDefaultUser;
     private final BrokerPool brokerPool;
+    private final ContentFilePool contentFilePool;
     protected final QueryResultCache resultSets = new QueryResultCache();
 
     protected final AtomicLazyVal<ExecutorService> restoreExecutorService;
@@ -72,14 +75,15 @@ public XmldbRequestProcessorFactory(final String databaseId, final boolean useDe
             this.databaseId = databaseId;
         }
         this.brokerPool = BrokerPool.getInstance(this.databaseId);
+        this.contentFilePool = new ContentFilePool(TemporaryFileManager.getInstance(), brokerPool.getId(),  brokerPool.getConfiguration());
         this.restoreExecutorService = new AtomicLazyVal<>(() -> Executors.newCachedThreadPool(new NamedThreadFactory(brokerPool, "rpc-db-restore")));
     }
 
     @Override
     public Object getRequestProcessor(final XmlRpcRequest pRequest) throws XmlRpcException {
         final XmlRpcHttpRequestConfig config = (XmlRpcHttpRequestConfig) pRequest.getConfig();
         final Subject user = authenticate(config.getBasicUserName(), config.getBasicPassword());
-        return new RpcConnection(this, user);
+        return new RpcConnection(this, contentFilePool, user);
     }
 
     protected Subject authenticate(String username, String password) throws XmlRpcException {

exist-core/src/test/java/org/exist/util/io/ContentFilePoolTest.java

@@ -55,7 +55,7 @@ void prepare() throws DatabaseConfigurationException {
         configuration = new Configuration();
         configuration.setProperty(PROPERTY_POOL_SIZE, 1);
         configuration.setProperty(PROPERTY_IN_MEMORY_SIZE, 10);
-        pool = new ContentFilePool(temporaryFileManager, configuration);
+        pool = new ContentFilePool(temporaryFileManager, null, configuration);
         replay(contentFile, temporaryFileManager);
     }
 

exist-core/src/test/java/org/exist/xmlrpc/CachedContentFileTest.java

@@ -29,6 +29,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import java.util.function.Consumer;
+
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
 import static org.assertj.core.api.Assertions.*;
@@ -40,31 +42,35 @@
 class CachedContentFileTest {
     @Mock
     ContentFile contentFile;
+    @Mock
+    Consumer<ContentFile> contentFileConsumer;
+
     CachedContentFile cachedContentFile;
     CachedContentFile cachedContentFileNoContent;
 
     @BeforeEach
     void prepare() {
-        cachedContentFile = new CachedContentFile(contentFile);
-        cachedContentFileNoContent= new CachedContentFile(null);
+        cachedContentFile = new CachedContentFile(contentFile, contentFileConsumer);
+        cachedContentFileNoContent= new CachedContentFile(null, contentFileConsumer);
     }
 
     @AfterEach
     void verifyMocks() {
-        verify(contentFile);
+        verify(contentFile, contentFileConsumer);
     }
 
     @Test
     void testGetResult() {
-        replay(contentFile);
+        replay(contentFile, contentFileConsumer);
         assertThat(cachedContentFile.getResult()).isEqualTo(contentFile);
         assertThat(cachedContentFileNoContent.getResult()).isNull();
     }
 
     @Test
     void testDoClose() {
         contentFile.close();
-        replay(contentFile);
+        contentFileConsumer.accept(contentFile);
+        replay(contentFile, contentFileConsumer);
         assertThatNoException().isThrownBy(cachedContentFile::doClose);
         assertThatNoException().isThrownBy(cachedContentFileNoContent::doClose);
     }