semaphore.java

来自「SRI international 发布的OAA框架软件」· Java 代码 · 共 809 行 · 第 1/3 页

            if (Thread.interrupted()) throw new InterruptedException();
            if (precheck()) return;
            WaitQueue.WaitNode w = new WaitQueue.WaitNode();
            w.doWait(this);
        }

        public boolean attempt(long nanos) throws InterruptedException {
            if (Thread.interrupted()) throw new InterruptedException();
            if (precheck()) return true;
            if (nanos <= 0) return false;

            WaitQueue.WaitNode w = new WaitQueue.WaitNode();
            return w.doTimedWait(this, nanos);
        }

        protected synchronized boolean precheck() {
            boolean pass = (permits_ > 0);
            if (pass) --permits_;
            return pass;
        }

        public synchronized boolean recheck(WaitQueue.WaitNode w) {
            boolean pass = (permits_ > 0);
            if (pass) --permits_;
            else       wq_.insert(w);
            return pass;
        }

        public void takeOver(WaitQueue.WaitNode n) {}

        protected synchronized WaitQueue.WaitNode getSignallee() {
            WaitQueue.WaitNode w = wq_.extract();
            if (w == null) ++permits_; // if none, inc permits for new arrivals
            return w;
        }

        public void release() {
            for (;;) {
                WaitQueue.WaitNode w = getSignallee();
                if (w == null) return;  // no one to signal
                if (w.signal(this)) return; // notify if still waiting, else skip
            }
        }

        /** Release N permits **/
        public void release(int n) {
            if (n < 0) throw new IllegalArgumentException("Negative argument");

            for (int i = 0; i < n; ++i) release();
        }

        public synchronized boolean hasQueuedThreads() {
            return wq_.hasNodes();
        }

        public synchronized int getQueueLength() {
            return wq_.getLength();
        }

        public synchronized Collection getQueuedThreads() {
            return wq_.getWaitingThreads();
        }
    }

    /**
     * Creates a <tt>Semaphore</tt> with the given number of
     * permits and nonfair fairness setting.
     * @param permits the initial number of permits available. This
     * value may be negative, in which case releases must
     * occur before any acquires will be granted.
     */
    public Semaphore(int permits) {
        impl = new NonfairImpl(permits);
    }

    /**
     * Creates a <tt>Semaphore</tt> with the given number of
     * permits and the given fairness setting.
     * @param permits the initial number of permits available. This
     * value may be negative, in which case releases must
     * occur before any acquires will be granted.
     * @param fair true if this semaphore will guarantee first-in
     * first-out granting of permits under contention, else false.
     */
    public Semaphore(int permits, boolean fair) {
        impl = (fair)? (Impl)new FairImpl(permits) : new NonfairImpl(permits);
    }

    /**
     * Acquires a permit from this semaphore, blocking until one is
     * available, or the thread is {@link Thread#interrupt interrupted}.
     *
     * <p>Acquires a permit, if one is available and returns immediately,
     * reducing the number of available permits by one.
     * <p>If no permit is available then the current thread becomes
     * disabled for thread scheduling purposes and lies dormant until
     * one of two things happens:
     * <ul>
     * <li>Some other thread invokes the {@link #release} method for this
     * semaphore and the current thread is next to be assigned a permit; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread.
     * </ul>
     *
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting
     * for a permit,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     *
     * @throws InterruptedException if the current thread is interrupted
     *
     * @see Thread#interrupt
     */
    public void acquire() throws InterruptedException {
        impl.acquire();
    }

    /**
     * Acquires a permit from this semaphore, blocking until one is
     * available.
     *
     * <p>Acquires a permit, if one is available and returns immediately,
     * reducing the number of available permits by one.
     * <p>If no permit is available then the current thread becomes
     * disabled for thread scheduling purposes and lies dormant until
     * some other thread invokes the {@link #release} method for this
     * semaphore and the current thread is next to be assigned a permit.
     *
     * <p>If the current thread
     * is {@link Thread#interrupt interrupted} while waiting
     * for a permit then it will continue to wait, but the time at which
     * the thread is assigned a permit may change compared to the time it
     * would have received the permit had no interruption occurred. When the
     * thread does return from this method its interrupt status will be set.
     *
     */
    public void acquireUninterruptibly() {
        boolean wasInterrupted = false;
        while (true) {
            try {
                impl.acquire();
                if (wasInterrupted) {
                    Thread.currentThread().interrupt();
                }
                return;
            }
            catch (InterruptedException e) {
                wasInterrupted = true;
            }
        }
    }

    /**
     * Acquires a permit from this semaphore, only if one is available at the
     * time of invocation.
     * <p>Acquires a permit, if one is available and returns immediately,
     * with the value <tt>true</tt>,
     * reducing the number of available permits by one.
     *
     * <p>If no permit is available then this method will return
     * immediately with the value <tt>false</tt>.
     *
     * <p>Even when this semaphore has been set to use a
     * fair ordering policy, a call to <tt>tryAcquire()</tt> <em>will</em>
     * immediately acquire a permit if one is available, whether or not
     * other threads are currently waiting.
     * This &quot;barging&quot; behavior can be useful in certain
     * circumstances, even though it breaks fairness. If you want to honor
     * the fairness setting, then use
     * {@link #tryAcquire(long, TimeUnit) tryAcquire(0, TimeUnit.SECONDS) }
     * which is almost equivalent (it also detects interruption).
     *
     * @return <tt>true</tt> if a permit was acquired and <tt>false</tt>
     * otherwise.
     */
    public boolean tryAcquire() {
        return impl.attempt();
    }

    /**
     * Acquires a permit from this semaphore, if one becomes available
     * within the given waiting time and the
     * current thread has not been {@link Thread#interrupt interrupted}.
     * <p>Acquires a permit, if one is available and returns immediately,
     * with the value <tt>true</tt>,
     * reducing the number of available permits by one.
     * <p>If no permit is available then
     * the current thread becomes disabled for thread scheduling
     * purposes and lies dormant until one of three things happens:
     * <ul>
     * <li>Some other thread invokes the {@link #release} method for this
     * semaphore and the current thread is next to be assigned a permit; or
     * <li>Some other thread {@link Thread#interrupt interrupts} the current
     * thread; or
     * <li>The specified waiting time elapses.
     * </ul>
     * <p>If a permit is acquired then the value <tt>true</tt> is returned.
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@link Thread#interrupt interrupted} while waiting to acquire
     * a permit,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     * <p>If the specified waiting time elapses then the value <tt>false</tt>
     * is returned.
     * If the time is less than or equal to zero, the method will not wait
     * at all.
     *
     * @param timeout the maximum time to wait for a permit
     * @param unit the time unit of the <tt>timeout</tt> argument.
     * @return <tt>true</tt> if a permit was acquired and <tt>false</tt>
     * if the waiting time elapsed before a permit was acquired.
     *
     * @throws InterruptedException if the current thread is interrupted
     *
     * @see Thread#interrupt
     *
     */
    public boolean tryAcquire(long timeout, TimeUnit unit)
        throws InterruptedException {
        return impl.attempt(unit.toNanos(timeout));
    }

    /**
     * Releases a permit, returning it to the semaphore.
     * <p>Releases a permit, increasing the number of available permits
     * by one.
     * If any threads are trying to acquire a permit, then one
     * is selected and given the permit that was just released.
     * That thread is (re)enabled for thread scheduling purposes.
     * <p>There is no requirement that a thread that releases a permit must
     * have acquired that permit by calling {@link #acquire}.
     * Correct usage of a semaphore is established by programming convention
     * in the application.
     */
    public void release() {
        impl.release(1);
    }

//    /**
//     * Acquires the given number of permits from this semaphore,
//     * blocking until all are available,
//     * or the thread is {@link Thread#interrupt interrupted}.
//     *
//     * <p>Acquires the given number of permits, if they are available,
//     * and returns immediately,
//     * reducing the number of available permits by the given amount.
//     *
//     * <p>If insufficient permits are available then the current thread becomes
//     * disabled for thread scheduling purposes and lies dormant until
//     * one of two things happens:
//     * <ul>
//     * <li>Some other thread invokes one of the {@link #release() release}
//     * methods for this semaphore, the current thread is next to be assigned
//     * permits and the number of available permits satisfies this request; or
//     * <li>Some other thread {@link Thread#interrupt interrupts} the current
//     * thread.
//     * </ul>
//     *
//     * <p>If the current thread:
//     * <ul>
//     * <li>has its interrupted status set on entry to this method; or
//     * <li>is {@link Thread#interrupt interrupted} while waiting
//     * for a permit,