Move TCP connection to thread, fully unregister completed search #3348
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,7 +12,9 @@ | |
| import java.net.InetSocketAddress; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
| import java.util.function.BiFunction; | ||
|
|
@@ -58,7 +60,7 @@ public class PVAClient implements AutoCloseable | |
| private final ConcurrentHashMap<Integer, PVAChannel> channels_by_id = new ConcurrentHashMap<>(); | ||
|
|
||
| /** TCP handlers by server address */ | ||
| private final ConcurrentHashMap<InetSocketAddress, Future<ClientTCPHandler>> tcp_handlers = new ConcurrentHashMap<>(); | ||
|
|
||
| private final AtomicInteger request_ids = new AtomicInteger(); | ||
|
|
||
|
|
@@ -89,20 +91,24 @@ public PVAClient() throws Exception | |
|
|
||
| // TCP traffic is handled by one ClientTCPHandler per address (IP, socket). | ||
| // Pass helper to channel search for getting such a handler. | ||
| final BiFunction<InetSocketAddress, Boolean, Future<ClientTCPHandler>> tcp_provider = (the_addr, use_tls) -> | ||
| tcp_handlers.computeIfAbsent(the_addr, addr -> | ||
| { | ||
| // If absent, create with initial empty GUID | ||
| final CompletableFuture<ClientTCPHandler> create_tcp = new CompletableFuture<>(); | ||
| create_tcp.completeAsync(() -> | ||
| { | ||
| try | ||
| { | ||
| return new ClientTCPHandler(this, addr, Guid.EMPTY, use_tls); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| logger.log(Level.WARNING, "Cannot connect to TCP " + addr, ex); | ||
| } | ||
| return null; | ||
| }); | ||
| return create_tcp; | ||
| }); | ||
| search = new ChannelSearch(udp, udp_search_addresses, tcp_provider, name_server_addresses); | ||
|
|
||
|
|
@@ -250,32 +256,52 @@ void handleSearchResponse(final int channel_id, final InetSocketAddress server, | |
| channel.setState(ClientChannelState.FOUND); | ||
| logger.log(Level.FINE, () -> "Reply for " + channel + " from " + (tls ? "TLS " : "TCP ") + server + " " + guid); | ||
|
|
||
| // TCP connection can be slow, especially when blocked by firewall, so move to thread | ||
| Thread.ofVirtual() | ||
|
There was a problem hiding this comment. One alternative to this way of implementing the functionality, would be to create the thread only for the computation that computes the value of the I'm thinking of something along the lines of: This is just an idea to discuss. (Also, I have not compiled this code, it's just a sketch.) There was a problem hiding this comment. Good idea. Go ahead and update the branch like that, since this also addresses your other concern about There was a problem hiding this comment. I have implemented this idea now and pushed the implementation to the branch of this pull request. (Commit: f982551) There was a problem hiding this comment. So the conclusion is that it doesn't seem that we can avoid creating many threads this way, but at least we can create virtual threads instead of OS-level threads. |
||
| .name("TCP connect " + server) | ||
| .start(() -> | ||
| { | ||
| final Future<ClientTCPHandler> tcp_future = tcp_handlers.computeIfAbsent(server, addr -> | ||
| { | ||
| final CompletableFuture<ClientTCPHandler> create_tcp = new CompletableFuture<>(); | ||
| create_tcp.completeAsync(() -> | ||
|
There was a problem hiding this comment. In fact, will this not spawn an OS-level thread? The documentation of Completes this CompletableFuture with the result of the given Supplier function invoked from an asynchronous task using the default executor. |
||
| { | ||
| try | ||
| { | ||
| return new ClientTCPHandler(this, addr, guid, tls); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| logger.log(Level.WARNING, "Cannot connect to TCP " + addr, ex); | ||
| } | ||
| return null; | ||
| }); | ||
| return create_tcp; | ||
| }); | ||
| ClientTCPHandler tcp; | ||
| try | ||
| { | ||
| tcp = tcp_future.get(); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| logger.log(Level.WARNING, "Cannot connect to " + server, ex); | ||
| tcp = null; | ||
| } | ||
| // In case of connection errors, tcp will be null | ||
| if (tcp == null) | ||
| { // Cannot connect to server on provided port? Likely a server or firewall problem. | ||
| // On the next search, that same server might reply and then we fail the same way on connect. | ||
| // Still, no way around re-registering the search so we succeed once the server is fixed. | ||
| search.register(channel, false /* not "now" but eventually */); | ||
| } | ||
| else | ||
| { | ||
| if (tcp.updateGuid(guid)) | ||
| logger.log(Level.FINE, "Search-only TCP handler received GUID, now " + tcp); | ||
| channel.registerWithServer(tcp); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /** Called by {@link ClientTCPHandler} when connection is lost or closed because unused | ||
|
|
@@ -288,7 +314,18 @@ void handleSearchResponse(final int channel_id, final InetSocketAddress server, | |
| void shutdownConnection(final ClientTCPHandler tcp) | ||
| { | ||
| // Forget this connection | ||
| final Future<ClientTCPHandler> tcp_future = tcp_handlers.remove(tcp.getRemoteAddress()); | ||
| final ClientTCPHandler removed; | ||
| try | ||
| { | ||
| removed = tcp_future == null ? null : tcp_future.get(); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| logger.log(Level.WARNING, "Cannot obtain TCP client to close for " + tcp, ex); | ||
| return; | ||
| } | ||
|
|
||
| if (removed != tcp) | ||
| logger.log(Level.WARNING, "Closed unknown " + tcp, new Exception("Call stack")); | ||
|
|
||
|
|
@@ -352,8 +389,15 @@ public void close() | |
| } | ||
|
|
||
| // Stop TCP and UDP threads | ||
| for (Future<ClientTCPHandler> handler : tcp_handlers.values()) | ||
| try | ||
| { | ||
| handler.get().close(true); | ||
| } | ||
| catch (Exception ex) | ||
| { | ||
| logger.log(Level.WARNING, "PVA Client error getting channel to close", ex); | ||
| } | ||
|
|
||
| udp.close(); | ||
| } | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LinkedList.remove()only removes the first occurrence of an element. Can an element occur multiple times in abucket, and if so, should all the occurrences of the element be removed?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
boostonly adds it once, so needs to be like this in all places that add?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah, I see. I wasn't aware it was checked when it was added.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Well, it's checked in one place, need to see if it's enforced in all places.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is perhaps outside the scope of this pull request, but if we change the type of
bucketto be aSet[1], then we can likely get bothIf the order of insertion is important, an implementation like, e.g.,
LinkedHashSetcould be used.(Edited to update the link to point to the documentation of Java version 17.)
[1] https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/Set.html
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's now a Set