int (int maxStoragesToProcess) { int successfullyCleaned = 0; for (int i = 0; i < maxStoragesToProcess; i++) { Command command = commandsQueue.poll(); if (command == null) { break; } if (command instanceof PostCloseStorageCleanupCommand) { final PostCloseStorageCleanupCommand closeStorageCommand = (PostCloseStorageCleanupCommand)command; final PagedFileStorageWithRWLockedPageContent storage = closeStorageCommand.storageToClose; final CompletableFuture<?> futureToFinalize = closeStorageCommand.onFinish; if (!storage.isClosed()) { AssertionError error = new AssertionError( "Code bug: storage " + storage + " must be closed before PostCloseStorageCleanupCommand is queued"); futureToFinalize.completeExceptionally(error); throw error; } PagesTable pagesTable = storage.pages(); // RC: Actually, we don't need to _reclaim_ pages of .close()-ed storage -- we need to .flush() // them, but the pages themselves are owned by cache, not storage. So it may be // reasonable to just leave pages there they are, until they are either reclaimed by // a regular reclamation path (because their utility is completely decayed under no // use), or re-used by a new storage opened for the same file. IMHO, for our use-cases, // the ability to re-use pages for re-opened storages is not very important, but the // opportunity to simplify .close() codepath may be worth it. //TODO RC: catch UncheckedIOException from page.flush() calls (in entombPageAndGetPageBuffer), // and collect them. Do limited retries: count unsuccessful page.flush(), and give up after // N tries -- remove task from queue, rise an exception leading to app restart // request (not worth to allow user continue working if we know work can't be // saved) boolean somePagesStillInUse = tryToReclaimAll(pagesTable); if (!somePagesStillInUse) { Path file = storage.getFile(); try { //TODO RC: this better to be responsibility of the storage (why cache should even know // about channels cache?) PageCacheUtils.CHANNELS_CACHE.closeChannel(file); } catch (Throwable t) { LOG.error("Can't close channel for " + file, t); futureToFinalize.completeExceptionally(t); } successfullyCleaned++; synchronized (pagesPerFile) { Path absolutePath = storage.getFile().toAbsolutePath(); PagesTable removed = pagesPerFile.remove(absolutePath); assert removed != null : "Storage for [" + absolutePath + "] must exists"; } futureToFinalize.complete(null); } else { //return command to the queue, to be re-tried later: commandsQueue.offer(command); } } } return successfullyCleaned; }