-
Notifications
You must be signed in to change notification settings - Fork 233
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Create a socket pool of multiple sockets bound to the same port #293
Open
JonathanLennox
wants to merge
19
commits into
jitsi:master
Choose a base branch
from
JonathanLennox:socketpool
base: master
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
19 commits
Select commit
Hold shift + click to select a range
933885a
Add socket pool, to allow sending to a single socket from multiple th…
JonathanLennox 8ef4273
Fix race: Return socket before releasing semaphore.
JonathanLennox dff4051
Add main() method to SocketPoolTest.
JonathanLennox 753eb8a
Change test sender to be more realistic.
JonathanLennox d2e792a
Remove blocking inside SocketPool.
JonathanLennox 0800279
More timing tests.
JonathanLennox 4e4afae
Improve timing test more.
JonathanLennox 5470f0a
Ktlint fix.
JonathanLennox af1c1e6
Simplify socket pool index calculation.
JonathanLennox 4403e8a
Command-line args for SocketPoolTest main().
JonathanLennox 5da22ee
Fix output
JonathanLennox 8f49368
Add warmup for testSendingOnce.
JonathanLennox 6182860
Make order of command-line arguments match output
JonathanLennox 46acf6a
Make send the API on socketpool, so socket choice can be internal.
JonathanLennox 7b3b5a9
Add experimental API annotation.
JonathanLennox f752308
Keep a lightweight count of which sockets are most heavily used.
JonathanLennox 0950c10
Minor cleanups.
JonathanLennox 902aca9
Change default pool size.
JonathanLennox c85bdc0
Use SocketPool in AbstractUdpListener.
JonathanLennox File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains 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 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 |
---|---|---|
@@ -0,0 +1,114 @@ | ||
/* | ||
* Copyright @ 2020 - Present, 8x8 Inc | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"); | ||
* you may not use this file except in compliance with the License. | ||
* You may obtain a copy of the License at | ||
* | ||
* http://www.apache.org/licenses/LICENSE-2.0 | ||
* | ||
* Unless required by applicable law or agreed to in writing, software | ||
* distributed under the License is distributed on an "AS IS" BASIS, | ||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
* See the License for the specific language governing permissions and | ||
* limitations under the License. | ||
*/ | ||
package org.ice4j.socket | ||
|
||
import java.net.DatagramPacket | ||
import java.net.DatagramSocket | ||
import java.net.DatagramSocketImpl | ||
import java.net.SocketAddress | ||
import java.nio.channels.DatagramChannel | ||
|
||
/** A pool of datagram sockets all bound on the same port. | ||
* | ||
* This is necessary to allow multiple threads to send packets simultaneously from the same source address, | ||
* in JDK 15 and later, because the [DatagramChannel]-based implementation of [DatagramSocketImpl] introduced | ||
* in that version locks the socket during a call to [DatagramSocket.send]. | ||
* | ||
* (The old [DatagramSocketImpl] implementation can be used by setting the system property | ||
* `jdk.net.usePlainDatagramSocketImpl` in JDK versions 15 through 17, but was removed in versions 18 and later.) | ||
* | ||
* This feature may also be useful on older JDK versions on non-Linux operating systems, such as macOS, | ||
* which block simultaneous writes through the same UDP socket at the operating system level. | ||
* | ||
* The sockets are opened such that packets will be _received_ on exactly one socket. | ||
*/ | ||
class SocketPool( | ||
/** The address to which to bind the pool of sockets. */ | ||
address: SocketAddress, | ||
/** The number of sockets to create for the pool. If this is set to zero (the default), the number | ||
* will be set automatically to an appropriate value. | ||
*/ | ||
requestedNumSockets: Int = 0 | ||
) { | ||
init { | ||
require(requestedNumSockets >= 0) { "RequestedNumSockets must be >= 0" } | ||
} | ||
|
||
internal class SocketAndIndex( | ||
val socket: DatagramSocket, | ||
var count: Int = 0 | ||
) | ||
|
||
val numSockets: Int = | ||
if (requestedNumSockets != 0) { | ||
requestedNumSockets | ||
} else { | ||
// TODO: set this to 1 in situations where pools aren't needed? | ||
2 * Runtime.getRuntime().availableProcessors() | ||
} | ||
|
||
private val sockets = buildList { | ||
val multipleSockets = numSockets > 1 | ||
var bindAddr = address | ||
for (i in 0 until numSockets) { | ||
val sock = DatagramSocket(null) | ||
if (multipleSockets) { | ||
sock.reuseAddress = true | ||
} | ||
sock.bind(bindAddr) | ||
if (i == 0 && multipleSockets) { | ||
bindAddr = sock.localSocketAddress | ||
} | ||
add(SocketAndIndex(sock, 0)) | ||
} | ||
} | ||
|
||
/** The socket on which packets will be received. */ | ||
val receiveSocket: DatagramSocket | ||
// On all platforms I've tested, the last-bound socket is the one which receives packets. | ||
// TODO: should we support Linux's flavor of SO_REUSEPORT, in which packets can be received on *all* the | ||
// sockets, spreading load? | ||
get() = sockets.last().socket | ||
|
||
fun send(packet: DatagramPacket) { | ||
val sendSocket = getSendSocket() | ||
sendSocket.socket.send(packet) | ||
returnSocket(sendSocket) | ||
} | ||
|
||
/** Gets a socket on which packets can be sent, chosen from among all the available send sockets. */ | ||
internal fun getSendSocket(): SocketAndIndex { | ||
if (numSockets == 1) { | ||
return sockets.first() | ||
} | ||
synchronized(sockets) { | ||
val min = sockets.minBy { it.count } | ||
min.count++ | ||
|
||
return min | ||
} | ||
} | ||
|
||
internal fun returnSocket(socket: SocketAndIndex) { | ||
synchronized(sockets) { | ||
socket.count-- | ||
} | ||
} | ||
|
||
fun close() { | ||
sockets.forEach { it.socket.close() } | ||
} | ||
} |
This file contains 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
Oops, something went wrong.
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.
I suggest logging the number we end up using.
Why does it default to twice the number of processors? Running your test on an A2 with 8 CPUs I see optimal performance for 8 sockets.