We are sharing our understanding on Selector and how it works below, what we have understood that though some aspects of Selector are thread safe, the real action where a thread gets the ready events and processes them, there today only on thread can come into action at any given point time, which is not enabling us to achieve concurrency to get and process these readiness events concurrently on multiple threads.
Please help figure out how to achieve the above.
- Selector can be used to monitor Selectable Java NIO Channels, whenever a channel[s] is ready for event[s] which can be Read, Write , Accept, Connect (as per registered interest), Selector will notify us.
- We register our Socket with the Selector and specify the operations which we are interested (Read/Write/Connect/Accept) in to get the Ready event using register Api.
- The register call returns us a Selection Key which represents relationship between Socket and Selector.
- ‘Selection Key’ contains information regarding the selector, channel, Interested Set (Operations which channel is Interested In) and Ready Set (Set of Operations which are Ready and can be performed without blocking, sub-set of interested set).
- ‘Selector’ maintains three key sets.
- Cancelled Key Set:
a. Set of Selection Key’s which have deregistered from the selector.
b. Canceled Key Set is not thread safe. - Registered Key Set:
a. Set of Selection Key’s which have registered with the selector.
b. Registered Key Set is Thread safe. - Selected Key Set:
a. Set of Selection Key’s which have at least one Interested Operation in ready state.
b. Selected Key Set is not Thread Safe.
- We block our thread (NETWORK_IO_IN) on the selector artefact select () call, which will unblock when at least one of the registered channels is ready to perform the interested operation.
- select call is thread safe, it internally involves three steps:
- First it iterates over the cancelled key set and free the resources w.r.t channel/socket which deregistered from the selector.
- In the second steps, it polls the OS which all channels are ready to perform the IO based on their Interested Set.
- In the third step it populates it’s selected key set.
- Note : All these steps are internally synchronizing on the selector then on the cancel and selected key set.
- To obtain the Selected key set which contains the selection key (which holds info about the channel) which are ready to perform IO we use selectedKeys (), which returns us a reference of the
selected key set which is a private member of the selector.
- We Iterate over the selected key set, identify which Interested operation is ready (Read/Write/Connect/Accept) and perform that IO.
Adding the Code Flow below, where we have not done any synchronization by ourselves.
- Here 10 threads are blocked on the selector artefact select call.
- We continuously send the data to the IP:Port on which channel/socket is listening on and is registered with the selector.
- We get Read events whenever the channel/socket is ready to read data from OS buffer.
- Observation:
-
All the threads are blocked on the select call and came out one after the another.
-
The thread which comes out of the select call let’s say T1 gets the selected key set and operates over it.
-
Considering another thread T2 which is blocked on the select call and is on step mentioned in 7.c, is also operating on the same selected key set.
-
Then we get a Concurrent Modification Exception as both thread tries to perform modification at same time to the selected key set.
public fun AsynIOArtefactLoop () {
<code>while (true) {println ("Thread " + Thread.currentThread().id + "blocked on select")// Threads will get blocked on the select calls until there is at least one channel /socket which is ready.selector.select ()println ("Thread " + Thread.currentThread().id + "comes out of select")// Returns the reference of the selected key set.val selected_key_set = selector.selectedKeys()val iterator = selected_key_set.iterator()println ("Thread " + Thread.currentThread().id + “operating on the selected key set”)while (iterator.hasNext()) {// Extract the key (selection key object ) and increment the iterator.val key = iterator.next()// Remove the Seleciton Key object from the selected key set.iterator.remove()println (" Event receive corrsponding to descriptor id " + key.attachment() as Int)when {// Channel/Socket is ready for Receive.key.isReadable -> {…// Buffer in which to copy the receive data.var buffer = ByteBuffer.allocate(1024)// Get the channel from Selection key object and call the receive.(key.channel() as DatagramChannel).receive(buffer)…// Print the Receive data .println ("Message Received: ${String(receivedData)}")}key.isWritable -> println("Write Ready")}}}</code><code>while (true) { println ("Thread " + Thread.currentThread().id + "blocked on select") // Threads will get blocked on the select calls until there is at least one channel /socket which is ready. selector.select () println ("Thread " + Thread.currentThread().id + "comes out of select") // Returns the reference of the selected key set. val selected_key_set = selector.selectedKeys() val iterator = selected_key_set.iterator() println ("Thread " + Thread.currentThread().id + “operating on the selected key set”) while (iterator.hasNext()) { // Extract the key (selection key object ) and increment the iterator. val key = iterator.next() // Remove the Seleciton Key object from the selected key set. iterator.remove() println (" Event receive corrsponding to descriptor id " + key.attachment() as Int) when { // Channel/Socket is ready for Receive. key.isReadable -> { … // Buffer in which to copy the receive data. var buffer = ByteBuffer.allocate(1024) // Get the channel from Selection key object and call the receive. (key.channel() as DatagramChannel).receive(buffer) … // Print the Receive data . println ("Message Received: ${String(receivedData)}") } key.isWritable -> println("Write Ready") } } } </code>while (true) { println ("Thread " + Thread.currentThread().id + "blocked on select") // Threads will get blocked on the select calls until there is at least one channel /socket which is ready. selector.select () println ("Thread " + Thread.currentThread().id + "comes out of select") // Returns the reference of the selected key set. val selected_key_set = selector.selectedKeys() val iterator = selected_key_set.iterator() println ("Thread " + Thread.currentThread().id + “operating on the selected key set”) while (iterator.hasNext()) { // Extract the key (selection key object ) and increment the iterator. val key = iterator.next() // Remove the Seleciton Key object from the selected key set. iterator.remove() println (" Event receive corrsponding to descriptor id " + key.attachment() as Int) when { // Channel/Socket is ready for Receive. key.isReadable -> { … // Buffer in which to copy the receive data. var buffer = ByteBuffer.allocate(1024) // Get the channel from Selection key object and call the receive. (key.channel() as DatagramChannel).receive(buffer) … // Print the Receive data . println ("Message Received: ${String(receivedData)}") } key.isWritable -> println("Write Ready") } } }}
Now adding the code flow, where we have done synchronization by ourselves.
- Here also 10 threads are blocked on the selector artefact select call.
- We continuously send the data to the IP:Port on which channel/socket is listening on and is registered with the selector.
- We get Read events whenever the channel/socket is ready to read data from OS buffer.
- Observation:
- All the threads are blocked on the select call and came out one after the another.
- The thread which comes out of the select call let’s say T1 gets the lock on the selected_key_set, whose reference is returned by selectedKeys () call.
- Considering another thread T2 which is blocked on the select call and is on step mentioned in 7.c, if already one of the thread have taken lock over the selected_key_set it will not get hold to operate over it which synchronizes the entire iteration process with select () call.
public fun select () {
<code> while (true) {
println ("Thread " + Thread.currentThread().id + "blocked on select")
selector.select ()
println ("Thread " + Thread.currentThread().id + "comes out of select")
val selected_key_set = selector.selectedKeys()
// Here we have synchronize on the selected_key_set which ensures that thread which are blocked on the select call will not be able to use it if another thread is already operating over it.
synchronized (selected_key_set) {
println ("Thread " + Thread.currentThread().id + "blocked on selected key set")
val iterator = selected_key_set.iterator()
while (iterator.hasNext()) {
val key = iterator.next()
println (" Event receive corrsponding to descriptor id " + key.attachment() as Int)
iterator.remove()
when {
key.isReadable -> {
counter++;
var buffer = ByteBuffer.allocate(1024)
(key.channel() as DatagramChannel).receive(buffer)
buffer.flip()
val receivedData = ByteArray(buffer.remaining())
buffer.get(receivedData)
println ("Recv Event Number: "+ counter)
println ("Message Received: ${String(receivedData)}")
}
key.isWritable -> println("Write Ready")
}
}
}
println ("Thread " + Thread.currentThread().id + "comes out of selected key set")
}
</code>
<code> while (true) {
println ("Thread " + Thread.currentThread().id + "blocked on select")
selector.select ()
println ("Thread " + Thread.currentThread().id + "comes out of select")
val selected_key_set = selector.selectedKeys()
// Here we have synchronize on the selected_key_set which ensures that thread which are blocked on the select call will not be able to use it if another thread is already operating over it.
synchronized (selected_key_set) {
println ("Thread " + Thread.currentThread().id + "blocked on selected key set")
val iterator = selected_key_set.iterator()
while (iterator.hasNext()) {
val key = iterator.next()
println (" Event receive corrsponding to descriptor id " + key.attachment() as Int)
iterator.remove()
when {
key.isReadable -> {
counter++;
var buffer = ByteBuffer.allocate(1024)
(key.channel() as DatagramChannel).receive(buffer)
buffer.flip()
val receivedData = ByteArray(buffer.remaining())
buffer.get(receivedData)
println ("Recv Event Number: "+ counter)
println ("Message Received: ${String(receivedData)}")
}
key.isWritable -> println("Write Ready")
}
}
}
println ("Thread " + Thread.currentThread().id + "comes out of selected key set")
}
</code>
while (true) {
println ("Thread " + Thread.currentThread().id + "blocked on select")
selector.select ()
println ("Thread " + Thread.currentThread().id + "comes out of select")
val selected_key_set = selector.selectedKeys()
// Here we have synchronize on the selected_key_set which ensures that thread which are blocked on the select call will not be able to use it if another thread is already operating over it.
synchronized (selected_key_set) {
println ("Thread " + Thread.currentThread().id + "blocked on selected key set")
val iterator = selected_key_set.iterator()
while (iterator.hasNext()) {
val key = iterator.next()
println (" Event receive corrsponding to descriptor id " + key.attachment() as Int)
iterator.remove()
when {
key.isReadable -> {
counter++;
var buffer = ByteBuffer.allocate(1024)
(key.channel() as DatagramChannel).receive(buffer)
buffer.flip()
val receivedData = ByteArray(buffer.remaining())
buffer.get(receivedData)
println ("Recv Event Number: "+ counter)
println ("Message Received: ${String(receivedData)}")
}
key.isWritable -> println("Write Ready")
}
}
}
println ("Thread " + Thread.currentThread().id + "comes out of selected key set")
}
}
Here we are doing synchronization on the Selected Key Set when we are doing processing of the Ready Events (Read/Write/Accept/Connect), which only allows only one thread to operate at a time and blocks the other threads.
If I have a multi-core CPU I am not able to take the performance advantage because only one of my thread will be doing processing at a time and other will wait for it.