sashapolo commented on code in PR #2500:
URL: https://github.com/apache/ignite-3/pull/2500#discussion_r1319515146
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java:
##########
@@ -2013,94 +1589,67 @@ public CompletableFuture<TableImpl>
tableImplAsync(String name) {
* @return Future representing pending completion of the {@code
TableManager#tableAsyncInternal} operation.
*/
public CompletableFuture<TableImpl> tableAsyncInternal(String name) {
- if (!busyLock.enterBusy()) {
- throw new IgniteException(new NodeStoppingException());
- }
+ return inBusyLockAsync(busyLock, () -> {
+ HybridTimestamp now = clock.now();
- try {
- return supplyAsync(() -> inBusyLock(busyLock, () ->
directTableId(name)), ioExecutor)
- .thenCompose(tableId -> inBusyLock(busyLock, () -> {
- if (tableId == null) {
+ return anyOf(schemaSyncService.waitForMetadataCompleteness(now),
stopManagerFuture)
+ .thenComposeAsync(unused -> inBusyLockAsync(busyLock, ()
-> {
+ CatalogTableDescriptor tableDescriptor =
catalogService.table(name, now.longValue());
+
+ // Check if the table has been deleted.
+ if (tableDescriptor == null) {
return completedFuture(null);
}
- return tableAsyncInternal(tableId, false);
+ return tableAsyncInternalBusy(tableDescriptor.id());
}));
- } finally {
- busyLock.leaveBusy();
- }
+ });
}
- /**
- * Internal method for getting table by id.
- *
- * @param id Table id.
- * @param checkConfiguration {@code True} when the method checks a
configuration before trying to get a table, {@code false}
- * otherwise.
- * @return Future representing pending completion of the operation.
- */
- public CompletableFuture<TableImpl> tableAsyncInternal(int id, boolean
checkConfiguration) {
- CompletableFuture<Boolean> tblCfgFut = checkConfiguration
- ? supplyAsync(() -> inBusyLock(busyLock, () ->
isTableConfigured(id)), ioExecutor)
- : completedFuture(true);
-
- return tblCfgFut.thenCompose(isCfg -> inBusyLock(busyLock, () -> {
- if (!isCfg) {
- return completedFuture(null);
- }
-
- TableImpl tbl = latestTablesById().get(id);
+ private CompletableFuture<TableImpl> tableAsyncInternalBusy(int tableId) {
+ TableImpl tableImpl = latestTablesById().get(tableId);
- if (tbl != null) {
- return completedFuture(tbl);
- }
+ if (tableImpl != null) {
+ return completedFuture(tableImpl);
+ }
- CompletableFuture<TableImpl> getTblFut = new CompletableFuture<>();
+ CompletableFuture<TableImpl> getLatestTableFuture = new
CompletableFuture<>();
- CompletionListener<Void> tablesListener = (token, v, th) -> {
- if (th == null) {
- CompletableFuture<Map<Integer, TableImpl>> tablesFut =
tablesByIdVv.get(token);
+ CompletionListener<Void> tablesListener = (token, v, th) -> {
+ if (th == null) {
+ CompletableFuture<Map<Integer, TableImpl>> tablesFuture =
tablesByIdVv.get(token);
- tablesFut.whenComplete((tables, e) -> {
- if (e != null) {
- getTblFut.completeExceptionally(e);
- } else {
- TableImpl table = tables.get(id);
+ tablesFuture.whenComplete((tables, e) -> {
+ if (e != null) {
+ getLatestTableFuture.completeExceptionally(e);
+ } else {
+ TableImpl table = tables.get(tableId);
- if (table != null) {
- getTblFut.complete(table);
- }
+ if (table != null) {
+ getLatestTableFuture.complete(table);
}
- });
- } else {
- getTblFut.completeExceptionally(th);
- }
- };
-
- assignmentsUpdatedVv.whenComplete(tablesListener);
+ }
+ });
+ } else {
+ getLatestTableFuture.completeExceptionally(th);
+ }
+ };
- // This check is needed for the case when we have registered
tablesListener,
- // but tablesByIdVv has already been completed, so listener would
be triggered only for the next versioned value update.
- tbl = latestTablesById().get(id);
+ assignmentsUpdatedVv.whenComplete(tablesListener);
- if (tbl != null) {
- assignmentsUpdatedVv.removeWhenComplete(tablesListener);
+ // This check is needed for the case when we have registered
tablesListener,
+ // but tablesByIdVv has already been completed, so listener would be
triggered only for the next versioned value update.
+ tableImpl = latestTablesById().get(tableId);
- return completedFuture(tbl);
- }
+ if (tableImpl != null) {
+ assignmentsUpdatedVv.removeWhenComplete(tablesListener);
- return getTblFut.whenComplete((unused, throwable) ->
assignmentsUpdatedVv.removeWhenComplete(tablesListener));
- }));
- }
+ return completedFuture(tableImpl);
+ }
- /**
- * Checks that the table is configured with specific id.
- *
- * @param id Table id.
- * @return True when the table is configured into cluster, false otherwise.
- */
- private boolean isTableConfigured(int id) {
- return configuredTablesCache.isTableConfigured(id);
+ return anyOf(getLatestTableFuture, stopManagerFuture)
+ .thenComposeAsync(o -> getLatestTableFuture, ioExecutor)
Review Comment:
> I thought that it would be better when one of the futures is completed
(from anyOf), the subsequent processing of the futures continued to be
performed in the inner pool
This is ok, but I think it's a bit redundant, because looks like there's not
a lot of things to execute, we wait for a future and then remove a listener
> I will try to explain: if we busyLock#enterBusy somewhere and do not
release it for some time, then in order to avoid blocking on the
TableManager#stop, we will simply immediately complete the future with an error.
Got it, thanks. I would suggest leaving a comment in the code about that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
