Add java client TableServiceAsync

Util/src/main/java/io/deephaven/util/reference/CleanupReferenceProcessor.java

@@ -37,7 +37,6 @@ public interface ExceptionHandler {
      * The reference queue from the most recent initialization.
      */
     private volatile ReferenceQueue<?> referenceQueue;
-    private Thread cleanerThread;
 
     /**
      * The cleaner thread from the most recent initialization, guarded by the lock on {@code this}.

java-client/session-dagger/src/test/java/io/deephaven/client/SessionTableServicesTest.java

@@ -6,16 +6,14 @@
 import io.deephaven.client.impl.TableHandle;
 import io.deephaven.client.impl.TableHandle.TableHandleException;
 import io.deephaven.client.impl.TableHandleManager;
-import io.deephaven.client.impl.TableServiceAsync;
-import io.deephaven.client.impl.TableServiceAsync.TableHandleFuture;
-import io.deephaven.client.impl.TableServices;
+import io.deephaven.client.impl.TableService;
+import io.deephaven.client.impl.TableService.TableHandleFuture;
 import io.deephaven.qst.TableCreator;
 import io.deephaven.qst.table.TableSpec;
 import org.junit.Ignore;
 import org.junit.Test;
 
 import java.time.Duration;
-import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
 
@@ -43,13 +41,13 @@ public void distinctBatchManagerIsNotStateful() throws TableHandleException, Int
 
     @Test
     public void distinctTableServicesIsNotStateful() throws TableHandleException, InterruptedException {
-        checkState(session.tableServices(), session.tableServices(), NOT_STATEFUL);
+        checkState(session.newStatefulTableService(), session.newStatefulTableService(), NOT_STATEFUL);
     }
 
     @Test
     public void distinctTableServiceAsyncsIsNotStateful()
             throws InterruptedException, ExecutionException, TimeoutException {
-        checkAsyncState(session.tableServices(), session.tableServices(), NOT_STATEFUL);
+        checkAsyncState(session.newStatefulTableService(), session.newStatefulTableService(), NOT_STATEFUL);
     }
 
     // this is currently broken; serial clients *can't* reliably execute the same non-trivial TableSpec DAG
@@ -61,21 +59,36 @@ public void singleSerialManagerIsStateful() throws TableHandleException, Interru
     }
 
     @Test
-    public void singleBatchManagerIsStateful() throws TableHandleException, InterruptedException {
+    public void singleBatchManagerNotIsStateful() throws TableHandleException, InterruptedException {
         final TableHandleManager manager = session.batch();
-        checkState(manager, manager, STATEFUL);
+        checkState(manager, manager, NOT_STATEFUL);
     }
 
     @Test
-    public void singleTableServiceIsStateful() throws TableHandleException, InterruptedException {
-        final TableServices tableServices = session.tableServices();
-        checkState(tableServices, tableServices, STATEFUL);
+    public void newStatefulTableServiceIsStateful() throws TableHandleException, InterruptedException {
+        final TableService ts = session.newStatefulTableService();
+        checkState(ts, ts, STATEFUL);
     }
 
     @Test
-    public void singleTableServiceAsyncIsStateful() throws InterruptedException, ExecutionException, TimeoutException {
-        final TableServices tableServices = session.tableServices();
-        checkAsyncState(tableServices, tableServices, STATEFUL);
+    public void newStatefulTableServiceAsyncIsStateful()
+            throws InterruptedException, ExecutionException, TimeoutException {
+        final TableService ts = session.newStatefulTableService();
+        checkAsyncState(ts, ts, STATEFUL);
+    }
+
+    @Test
+    public void newStatefulTableServiceBatchIsStateful() throws TableHandleException, InterruptedException {
+        final TableService ts = session.newStatefulTableService();
+        checkState(ts.batch(), ts.batch(), STATEFUL);
+    }
+
+    // this is currently broken; serial clients *can't* reliably execute the same non-trivial TableSpec DAG
+    @Ignore
+    @Test
+    public void newStatefulTableServiceSerialIsStateful() throws TableHandleException, InterruptedException {
+        final TableService ts = session.newStatefulTableService();
+        checkState(ts.serial(), ts.serial(), STATEFUL);
     }
 
     static void checkState(TableHandleManager m1, TableHandleManager m2, boolean expectEquals)
@@ -84,7 +97,7 @@ static void checkState(TableHandleManager m1, TableHandleManager m2, boolean exp
         checkExecuteState(m1, m2, expectEquals);
     }
 
-    static void checkAsyncState(TableServiceAsync a1, TableServiceAsync a2, boolean expectEquals)
+    static void checkAsyncState(TableService a1, TableService a2, boolean expectEquals)
             throws InterruptedException, ExecutionException, TimeoutException {
         checkExecuteAsyncState(a1, a2, expectEquals);
     }
@@ -117,12 +130,35 @@ private static void checkExecuteState(TableHandleManager m1, TableHandleManager
         }
     }
 
-    private static void checkExecuteAsyncState(TableServiceAsync a1, TableServiceAsync a2, boolean expectEquals)
+    private static void checkExecuteAsyncState(TableService a1, TableService a2, boolean expectEquals)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        checkExecuteAsyncStateOneAtTime(a1, a2, expectEquals);
+        checkExecuteAsyncStateAtSameTime(a1, a2, expectEquals);
+    }
+
+    private static void checkExecuteAsyncStateOneAtTime(TableService a1, TableService a2, boolean expectEquals)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        final TableSpec q = tableSpec();
+        try (
+                final TableHandle h1 = a1.executeAsync(q).getOrCancel(Duration.ofSeconds(5));
+                final TableHandle h2 = a2.executeAsync(q).getOrCancel(Duration.ofSeconds(5))) {
+            assertThat(h1.exportId().toString().equals(h2.exportId().toString())).isEqualTo(expectEquals);
+            try (
+                    final TableHandle h3 = h1.updateView("K=ii");
+                    final TableHandle h4 = h2.updateView("K=ii")) {
+                assertThat(h3.exportId().toString().equals(h4.exportId().toString())).isEqualTo(expectEquals);
+            }
+        }
+    }
+
+    private static void checkExecuteAsyncStateAtSameTime(TableService a1, TableService a2, boolean expectEquals)
             throws InterruptedException, ExecutionException, TimeoutException {
         final TableSpec q = tableSpec();
+        final TableHandleFuture f1 = a1.executeAsync(q);
+        final TableHandleFuture f2 = a2.executeAsync(q);
         try (
-                final TableHandle h1 = TableHandleFuture.get(a1.executeAsync(q), Duration.ofSeconds(5));
-                final TableHandle h2 = TableHandleFuture.get(a2.executeAsync(q), Duration.ofSeconds(5))) {
+                final TableHandle h1 = f1.getOrCancel(Duration.ofSeconds(5));
+                final TableHandle h2 = f2.getOrCancel(Duration.ofSeconds(5))) {
             assertThat(h1.exportId().toString().equals(h2.exportId().toString())).isEqualTo(expectEquals);
             try (
                     final TableHandle h3 = h1.updateView("K=ii");

java-client/session-dagger/src/test/java/io/deephaven/client/TableSpecTestBase.java

@@ -2,7 +2,7 @@
 
 import io.deephaven.client.impl.TableHandle;
 import io.deephaven.client.impl.TableHandle.TableHandleException;
-import io.deephaven.client.impl.TableServiceAsync.TableHandleFuture;
+import io.deephaven.client.impl.TableService;
 import io.deephaven.qst.table.TableSpec;
 import org.junit.Test;
 
@@ -44,7 +44,7 @@ public void serial() throws TableHandleException, InterruptedException {
 
     @Test(timeout = 10000)
     public void async() throws ExecutionException, InterruptedException {
-        try (final TableHandle handle = TableHandleFuture.get(session.executeAsync(table))) {
+        try (final TableHandle handle = session.executeAsync(table).getOrCancel()) {
             assertThat(handle.isSuccessful()).isTrue();
         }
     }

java-client/session-dagger/src/test/java/io/deephaven/client/impl/TableServicesAsyncTest.java → java-client/session-dagger/src/test/java/io/deephaven/client/impl/TableServiceAsyncTest.java

@@ -4,7 +4,7 @@
 package io.deephaven.client.impl;
 
 import io.deephaven.client.DeephavenSessionTestBase;
-import io.deephaven.client.impl.TableServiceAsync.TableHandleFuture;
+import io.deephaven.client.impl.TableService.TableHandleFuture;
 import io.deephaven.qst.table.TableSpec;
 import org.junit.Test;
 
@@ -16,7 +16,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-public class TableServicesAsyncTest extends DeephavenSessionTestBase {
+public class TableServiceAsyncTest extends DeephavenSessionTestBase {
 
     private static final Duration GETTIME = Duration.ofSeconds(15);
     private static final int CHAIN_OPS = 50;
@@ -26,23 +26,23 @@ public class TableServicesAsyncTest extends DeephavenSessionTestBase {
     public void longChainAsyncExportOnlyLast() throws ExecutionException, InterruptedException, TimeoutException {
         final List<TableSpec> longChain = createLongChain();
         final TableSpec longChainLast = longChain.get(longChain.size() - 1);
-        try (final TableHandle handle = get(session.tableServices().executeAsync(longChainLast))) {
+        try (final TableHandle handle = get(session.newStatefulTableService().executeAsync(longChainLast))) {
             checkSucceeded(handle);
         }
     }
 
     @Test(timeout = 20000)
     public void longChainAsyncExportAll() throws ExecutionException, InterruptedException, TimeoutException {
         final List<TableSpec> longChain = createLongChain();
-        final List<? extends TableHandleFuture> futures = session.tableServices().executeAsync(longChain);
+        final List<? extends TableHandleFuture> futures = session.newStatefulTableService().executeAsync(longChain);
         try {
             for (final TableHandleFuture future : futures) {
                 try (final TableHandle handle = get(future)) {
                     checkSucceeded(handle);
                 }
             }
         } catch (final Throwable t) {
-            TableHandleFuture.cancelOrClose(futures, true);
+            TableService.TableHandleFuture.cancelOrClose(futures, true);
             throw t;
         }
     }
@@ -51,9 +51,9 @@ public void longChainAsyncExportAll() throws ExecutionException, InterruptedExce
     public void longChainAsyncExportAllCancelAllButLast()
             throws ExecutionException, InterruptedException, TimeoutException {
         final List<TableSpec> longChain = createLongChain();
-        final List<? extends TableHandleFuture> futures = session.tableServices().executeAsync(longChain);
+        final List<? extends TableHandleFuture> futures = session.newStatefulTableService().executeAsync(longChain);
         // Cancel or close all but the last one
-        TableHandleFuture.cancelOrClose(futures.subList(0, futures.size() - 1), true);
+        TableService.TableHandleFuture.cancelOrClose(futures.subList(0, futures.size() - 1), true);
         try (final TableHandle lastHandle = get(futures.get(futures.size() - 1))) {
             checkSucceeded(lastHandle);
         }
@@ -64,10 +64,10 @@ public void immediatelyCompletedFromCachedTableServices()
             throws ExecutionException, InterruptedException, TimeoutException {
         final List<TableSpec> longChain = createLongChain();
         final TableSpec longChainLast = longChain.get(longChain.size() - 1);
-        final TableServices tableServices = session.tableServices();
-        try (final TableHandle ignored = get(tableServices.executeAsync(longChainLast))) {
+        final TableService tableService = session.newStatefulTableService();
+        try (final TableHandle ignored = get(tableService.executeAsync(longChainLast))) {
             for (int i = 0; i < 1000; ++i) {
-                try (final TableHandle handle = get(tableServices.executeAsync(longChainLast))) {
+                try (final TableHandle handle = get(tableService.executeAsync(longChainLast))) {
                     checkSucceeded(handle);
                 }
             }
@@ -76,7 +76,7 @@ public void immediatelyCompletedFromCachedTableServices()
 
     private static TableHandle get(TableHandleFuture future)
             throws ExecutionException, InterruptedException, TimeoutException {
-        return TableHandleFuture.get(future, GETTIME);
+        return future.getOrCancel(GETTIME);
     }
 
     private static void checkSucceeded(TableHandle x) {

java-client/session-examples/src/main/java/io/deephaven/client/examples/FilterTable.java

@@ -6,7 +6,6 @@
 import io.deephaven.api.filter.Filter;
 import io.deephaven.client.impl.Session;
 import io.deephaven.client.impl.TableHandle;
-import io.deephaven.client.impl.TableServiceAsync.TableHandleFuture;
 import io.deephaven.qst.table.TableSpec;
 import picocli.CommandLine;
 import picocli.CommandLine.ArgGroup;
@@ -39,7 +38,7 @@ enum Type {
     protected void execute(Session session) throws Exception {
         final Filter filter = type == Type.AND ? Filter.and(Filter.from(filters)) : Filter.or(Filter.from(filters));
         final TableSpec filtered = ticket.ticketId().table().where(filter);
-        try (final TableHandle handle = TableHandleFuture.get(session.tableServices().executeAsync(filtered))) {
+        try (final TableHandle handle = session.newStatefulTableService().executeAsync(filtered).getOrCancel()) {
             session.publish("filter_table_results", handle).get();
         }
     }

java-client/session-examples/src/main/java/io/deephaven/client/examples/UnreferenceableTableExample.java

@@ -33,7 +33,7 @@ protected void execute(Session session) throws Exception {
         final TableSpec rPlusOne = r.view("PlusOne=R + 1");
         final TableSpec rMinusOne = r.view("PlusOne=R - 1");
         final TableHandleManager manager = mode == null
-                ? session.tableServices()
+                ? session.newStatefulTableService()
                 : mode.batch
                         ? session.batch()
                         : session.serial();

java-client/session/src/main/java/io/deephaven/client/impl/Export.java

@@ -4,6 +4,7 @@
 package io.deephaven.client.impl;
 
 import io.deephaven.client.impl.ExportRequest.Listener;
+import io.deephaven.client.impl.ExportStates.State;
 import io.deephaven.qst.table.TableSpec;
 
 import java.util.Objects;
@@ -61,6 +62,10 @@ ExportStates exportStates() {
         return state.exportStates();
     }
 
+    State state() {
+        return state;
+    }
+
     /**
      * The table spec.
      *

java-client/session/src/main/java/io/deephaven/client/impl/ExportStates.java

@@ -23,13 +23,13 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -142,6 +142,8 @@ private ExportServiceRequest exportRequestImpl(ExportsRequest requests) {
             };
         }
         return new ExportServiceRequest() {
+            boolean sent;
+            boolean closed;
 
             @Override
             public List<Export> exports() {
@@ -150,27 +152,62 @@ public List<Export> exports() {
 
             @Override
             public void send() {
+                if (closed || sent) {
+                    return;
+                }
+                sent = true;
+                // After the user has called send, all handling of state needs to be handled by the respective
+                // io.deephaven.client.impl.ExportRequest.listener
                 send.run();
+
             }
 
             @Override
             public void close() {
-                lock.unlock();
+                if (closed) {
+                    return;
+                }
+                closed = true;
+                try {
+                    if (!sent) {
+                        cleanupUnsent();
+                    }
+                } finally {
+                    lock.unlock();
+                }
+            }
+
+            private void cleanupUnsent() {
+                for (Export result : results) {
+                    final State state = result.state();
+                    if (newStates.containsKey(state.exportId())) {
+                        // On brand new states that we didn't even send, we can simply remove them. We aren't
+                        // leaking anything, but we have incremented our export id creator state.
+                        removeImpl(state);
+                        continue;
+                    }
+                    result.release();
+                }
             }
         };
     }
 
-    private void release(State state) {
+
+    private void remove(State state) {
         lock.lock();
         try {
-            if (!exports.remove(state.table(), state)) {
-                throw new IllegalStateException("Unable to remove state");
-            }
+            removeImpl(state);
         } finally {
             lock.unlock();
         }
     }
 
+    private void removeImpl(State state) {
+        if (!exports.remove(state.table(), state)) {
+            throw new IllegalStateException("Unable to remove state");
+        }
+    }
+
     private Optional<State> lookup(TableSpec table) {
         return Optional.ofNullable(exports.get(table));
     }
@@ -224,7 +261,7 @@ class State {
         State(TableSpec table, int exportId) {
             this.table = Objects.requireNonNull(table);
             this.exportId = exportId;
-            this.children = new LinkedHashSet<>();
+            this.children = new CopyOnWriteArraySet<>();
         }
 
         Session session() {
@@ -258,7 +295,7 @@ synchronized void release(Export export) {
                 throw new IllegalStateException("Unable to remove child");
             }
             if (children.isEmpty()) {
-                ExportStates.this.release(this);
+                ExportStates.this.remove(this);
                 released = true;
                 sessionStub.release(
                         ReleaseRequest.newBuilder().setId(ExportTicketHelper.wrapExportIdInTicket(exportId)).build(),