Java example source code file (EventPortWrapper.java)

This example Java source code file (EventPortWrapper.java) is included in the alvinalexander.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Learn more about this Java project at its project page.

Java - Java tags/keywords

allocatednativeobject, bitset, eventportwrapper, ignore, initial_pending_update_size, internalerror, ioexception, max_update_array_size, object, offsetof_object, open_max, poll_max, port_source_fd, security, sizeof_port_event, util

The EventPortWrapper.java Java example source code

/*
 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package sun.nio.ch;

import java.io.IOException;
import java.security.AccessController;
import java.util.BitSet;
import java.util.HashMap;
import java.util.Map;

import sun.misc.Unsafe;
import sun.security.action.GetIntegerAction;
import static sun.nio.ch.SolarisEventPort.*;

/**
 * Manages a Solaris event port and manipulates a native array of pollfd structs
 * on Solaris.
 */

class EventPortWrapper {
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final int addressSize = unsafe.addressSize();

    // Maximum number of open file descriptors
    static final int   OPEN_MAX     = IOUtil.fdLimit();

    // Maximum number of events to retrive in one call to port_getn
    static final int   POLL_MAX     =  Math.min(OPEN_MAX-1, 1024);

    // initial size of the array to hold pending updates
    private final int INITIAL_PENDING_UPDATE_SIZE = 256;

    // maximum size of updateArray
    private static final int MAX_UPDATE_ARRAY_SIZE = AccessController.doPrivileged(
        new GetIntegerAction("sun.nio.ch.maxUpdateArraySize", Math.min(OPEN_MAX, 64*1024)));

    // special update status to indicate that it should be ignored
    private static final byte IGNORE = -1;

    // port file descriptor
    private final int pfd;

    // the poll array (populated by port_getn)
    private final long pollArrayAddress;
    private final AllocatedNativeObject pollArray;

    // required when accessing the update* fields
    private final Object updateLock = new Object();

    // the number of pending updates
    private int updateCount;

    // queue of file descriptors with updates pending
    private int[] updateDescriptors = new int[INITIAL_PENDING_UPDATE_SIZE];

    // events for file descriptors with registration changes pending, indexed
    // by file descriptor and stored as bytes for efficiency reasons. For
    // file descriptors higher than MAX_UPDATE_ARRAY_SIZE (unlimited case at
    // least then the update is stored in a map.
    private final byte[] eventsLow = new byte[MAX_UPDATE_ARRAY_SIZE];
    private Map<Integer,Byte> eventsHigh;
    // Used by release and updateRegistrations to track whether a file
    // descriptor is registered with /dev/poll.
    private final BitSet registered = new BitSet();

    // bit set to indicate if a file descriptor has been visited when
    // processing updates (used to avoid duplicates calls to port_associate)
    private BitSet visited = new BitSet();

    EventPortWrapper() throws IOException {
        int allocationSize = POLL_MAX * SIZEOF_PORT_EVENT;
        pollArray = new AllocatedNativeObject(allocationSize, true);
        pollArrayAddress = pollArray.address();
        this.pfd = port_create();
        if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)
            eventsHigh = new HashMap<>();
    }

    void close() throws IOException {
        port_close(pfd);
        pollArray.free();
    }

    private short getSource(int i) {
        int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_SOURCE;
        return pollArray.getShort(offset);
    }

    int getEventOps(int i) {
        int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_EVENTS;
        return pollArray.getInt(offset);
    }

    int getDescriptor(int i) {
        int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_OBJECT;
        if (addressSize == 4) {
            return pollArray.getInt(offset);
        } else {
            return (int) pollArray.getLong(offset);
        }
    }

    private void setDescriptor(int i, int fd) {
        int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_OBJECT;
        if (addressSize == 4) {
            pollArray.putInt(offset, fd);
        } else {
            pollArray.putLong(offset, fd);
        }
    }

    private void setUpdate(int fd, byte events) {
        if (fd < MAX_UPDATE_ARRAY_SIZE) {
            eventsLow[fd] = events;
        } else {
            eventsHigh.put(Integer.valueOf(fd), Byte.valueOf(events));
        }
    }

    private byte getUpdate(int fd) {
        if (fd < MAX_UPDATE_ARRAY_SIZE) {
            return eventsLow[fd];
        } else {
            Byte result = eventsHigh.get(Integer.valueOf(fd));
            // result should never be null
            return result.byteValue();
        }
    }

    int poll(long timeout) throws IOException {
        // update registrations prior to poll
        synchronized (updateLock) {

            // process newest updates first
            int i = updateCount - 1;
            while (i >= 0) {
                int fd = updateDescriptors[i];
                if (!visited.get(fd)) {
                    short ev = getUpdate(fd);
                    if (ev != IGNORE) {
                        if (ev == 0) {
                            if (registered.get(fd)) {
                                port_dissociate(pfd, PORT_SOURCE_FD, (long)fd);
                                registered.clear(fd);
                            }
                        } else {
                            if (port_associate(pfd, PORT_SOURCE_FD, (long)fd, ev)) {
                                registered.set(fd);
                            }
                        }

                    }
                    visited.set(fd);
                }
                i--;
            }
            updateCount = 0;
        }

        // poll for events
        int updated = port_getn(pfd, pollArrayAddress, POLL_MAX, timeout);

        // after polling we need to queue all polled file descriptors as they
        // are candidates to register for the next poll.
        synchronized (updateLock) {
            for (int i=0; i<updated; i++) {
                if (getSource(i) == PORT_SOURCE_USER) {
                    interrupted = true;
                    setDescriptor(i, -1);
                } else {
                    // the default is to re-associate for the next poll
                    int fd = getDescriptor(i);
                    registered.clear(fd);
                    setInterest(fd);
                }
            }
        }

        return updated;
    }

    private void setInterest(int fd) {
        assert Thread.holdsLock(updateLock);

        // record the file descriptor and events, expanding the
        // respective arrays first if necessary.
        int oldCapacity = updateDescriptors.length;
        if (updateCount >= oldCapacity) {
            int newCapacity = oldCapacity + INITIAL_PENDING_UPDATE_SIZE;
            int[] newDescriptors = new int[newCapacity];
            System.arraycopy(updateDescriptors, 0, newDescriptors, 0, oldCapacity);
            updateDescriptors = newDescriptors;
        }
        updateDescriptors[updateCount++] = fd;
        visited.clear(fd);
    }

    void setInterest(int fd, int mask) {
        synchronized (updateLock) {
            setInterest(fd);
            setUpdate(fd, (byte)mask);
            assert getUpdate(fd) == mask;
        }
    }

    void release(int fd) {
        synchronized (updateLock) {
            if (registered.get(fd)) {
                try {
                    port_dissociate(pfd, PORT_SOURCE_FD, (long)fd);
                } catch (IOException ioe) {
                    throw new InternalError(ioe);
                }
                registered.clear(fd);
            }
            setUpdate(fd, IGNORE);
        }
    }

    // -- wakeup support --

    private boolean interrupted;

    public void interrupt() {
        try {
            port_send(pfd, 0);
        } catch (IOException ioe) {
            throw new InternalError(ioe);
        }
    }

    boolean interrupted() {
        return interrupted;
    }

    void clearInterrupted() {
        interrupted = false;
    }
}