use dynamic buffer

apache · Aug 8, 2023 · bd550f8 · bd550f8
1 parent 6efdfc5
commit bd550f8
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 8 deletions.
diff --git a/src/main/java/org/apache/sysds/runtime/util/CommonThreadPool.java b/src/main/java/org/apache/sysds/runtime/util/CommonThreadPool.java
@@ -156,8 +156,8 @@ public static <T> void invokeAndShutdown(ExecutorService pool, Collection<? exte
 	 * 
 	 * @return A dynamic thread pool.
 	 */
-	public static ExecutorService getDynamicPool() {
-		if(asyncPool != null)
+	public synchronized static ExecutorService getDynamicPool() {
+		if(asyncPool != null && (asyncPool.isShutdown() || asyncPool.isTerminated()) )
 			return asyncPool;
 		else {
 			asyncPool = Executors.newCachedThreadPool();

diff --git a/src/main/java/org/apache/sysds/runtime/util/DoubleBufferingOutputStream.java b/src/main/java/org/apache/sysds/runtime/util/DoubleBufferingOutputStream.java
@@ -23,14 +23,11 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
 import org.apache.commons.lang3.concurrent.ConcurrentUtils;
 
-public class DoubleBufferingOutputStream extends FilterOutputStream
-{
-	protected ExecutorService _pool = CommonThreadPool.getDynamicPool();
+public class DoubleBufferingOutputStream extends FilterOutputStream {
 	protected Future<?>[] _locks;
 	protected byte[][] _buff;
 	private int _pos;
@@ -69,7 +66,7 @@ public void write(byte[] b, int off, int len)
 				System.arraycopy(b, off, _buff[_pos], 0, len);
 
 				//submit write request 
-				_locks[_pos] = _pool.submit(new WriteTask(_buff[_pos], len));
+				_locks[_pos] = CommonThreadPool.getDynamicPool().submit(new WriteTask(_buff[_pos], len));
 				_pos = (_pos+1) % _buff.length;
 			}
 		}
@@ -103,7 +100,6 @@ public void flush() throws IOException {
 
 	@Override
 	public void close() throws IOException {
-		_pool.shutdown();
 		super.close();
 	}