Problems persist in KQueueSelectorProvider (Mac) in 7u6 ea

[Jason T. Greene]

On 8/13/12 4:24 PM, David Schlosnagle wrote:
> On Mon, Aug 13, 2012 at 3:13 PM, Jason T. Greene
> <jason.greene at redhat.com> wrote:
>> I threw together a quick patch (attached below against 7u-dev), and it
>> appears to resolve the issues on 6 and 7. Although it needs more testing.
>>
>> The preClose problem disappears as well with this change. My hunch is that
>> the tight spin loop on kqueue makes it impossible for close/dup2 to ever
>> complete.
>
> Hi Jason,
>
> A few minor comments/questions in KQueueArrayWrapper.java:

Thanks!

> Should the updateList and Update instance fields all be final?

Sure they could be.

> Is there any reason not to avoid the continue in updateRegistrations()?
>      while ((u = updateList.poll()) != null) {
>          SelChImpl ch = u.channel;
>          if (ch.isOpen()) {
>              register0(kq, ch.getFDVal(), u.events & POLLIN, u.events & POLLOUT);
>          }
>      }

No reason, but then again there is also no reason to avoid it either. 
It's just a stylistic difference.

> Do you foresee any lock contention issues with the synchronization on
> updateList?

The NIO design is such that this is necessary, you need to ensure atomic 
and consistent changes to the underlying queue/poll. The lock is only 
held for a short period of time.

> Would using a java.util.concurrent.LinkedBlockingQueue
> rather than the LinkedList with synchronization violate the API
> semantics of batching the updates?

Before the selector is ran, you need to process the entire queue. Else 
there is no guarantee the poll implementation's state is consistent with 
the NIO API's state. So you would have to use the drainTo operation to 
get proper locking semantics, and that requires a needless copy into 
another collection. Even if you didn't have the copy the memory overhead 
of LinkedBlockingQueue is quite a bit more than a LinkedList. There's 
also a small difference in that JDK locks (synchronized) have 
optimization paths in openjdk that do not exist for ReentrantLock (and 
they use far less memory).

An ArrayDeque on the other hand would likely be more efficient (memory + 
cpu) than the LinkedList. I was planning to change the patch to use it 
instead, although at the expected queue size, I doubt there will be much 
of a real world significance.

  I was thinking something along the
> lines of the KQueueArrayWrapper.java diff below would work (but I
> haven't tested at all).
>
> Thanks,
> Dave
>
>
> diff --git a/KQueueArrayWrapper.java b/KQueueArrayWrapper.java
> index 5a7020c..936f086 100644
> --- a/KQueueArrayWrapper.java
> +++ b/KQueueArrayWrapper.java
> @@ -34,7 +34,9 @@ package sun.nio.ch;
>   import sun.misc.*;
>   import java.io.IOException;
>   import java.io.FileDescriptor;
> -
> +import java.util.Iterator;
> +import java.util.Queue;
> +import java.util.concurrent.LinkedBlockingQueue;
>
>   /*
>    * struct kevent {           // 32-bit    64-bit
> @@ -85,6 +87,9 @@ class KQueueArrayWrapper {
>       // The fd of the interrupt line coming in
>       private int incomingInterruptFD;
>
> +    // The queue of file descriptor registration updates to process
> +    private final Queue<Update> updateQueue = new LinkedBlockingQueue<>();
> +
>       static {
>           initStructSizes();
>           String datamodel = java.security.AccessController.doPrivileged(
> @@ -100,6 +105,16 @@ class KQueueArrayWrapper {
>           kq = init();
>       }
>
> +    // Used to update file description registrations
> +    private static class Update {
> +        final SelChImpl channel;
> +        final int events;
> +        Update(SelChImpl channel, int events) {
> +            this.channel = channel;
> +            this.events = events;
> +        }
> +    }
> +
>       void initInterrupt(int fd0, int fd1) {
>           outgoingInterruptFD = fd1;
>           incomingInterruptFD = fd0;
> @@ -137,12 +152,31 @@ class KQueueArrayWrapper {
>           }
>       }
>
> -    void setInterest(int fd, int events) {
> -        register0(kq, fd, events & POLLIN, events & POLLOUT);
> +    void setInterest(SelChImpl channel, int events) {
> +        // update existing registration
> +        updateQueue.add(new Update(channel, events));
>       }
>
> -    void release(int fd) {
> -        register0(kq, fd, 0, 0);
> +    void release(SelChImpl channel) {
> +        // flush any pending updates
> +        for (Iterator<Update> it = updateQueue.iterator(); it.hasNext();) {
> +            if (it.next().channel == channel) {
> +                it.remove();
> +            }
> +        }
> +
> +        // remove
> +        register0(kq, channel.getFDVal(), 0, 0);
> +    }
> +
> +    void updateRegistrations() {
> +        Update u = null;
> +        while ((u = updateQueue.poll()) != null) {
> +            SelChImpl ch = u.channel;
> +            if (ch.isOpen()) {
> +                register0(kq, ch.getFDVal(), u.events & POLLIN,
> u.events & POLLOUT);
> +            }
> +        }
>       }
>
>       void close() throws IOException {
> @@ -157,6 +191,7 @@ class KQueueArrayWrapper {
>       }
>
>       int poll(long timeout) {
> +        updateRegistrations();
>           int updated = kevent0(kq, keventArrayAddress, NUM_KEVENTS, timeout);
>           return updated;
>       }
>


-- 
Jason T. Greene
JBoss AS Lead / EAP Platform Architect
JBoss, a division of Red Hat