세마포어

카운팅 세마포어는 특정 자원이나 특정 연산을 동시에 사용하거나 호출할 수 있는 스레드의 수를 제한하고자 할 때 사용한다.
카운팅 세마포어의 이런 기능을 활용하면 자원 풀이나 컬렉션의 크기에 제한을 두고자 할 때 유용하다.

 

public Semaphore(int permits) {  
    sync = new NonfairSync(permits);  
}

Semaphore 클래스는 가상의 퍼밋을 만들어 내부 상태를 관리하며, Semaphore를 생성할 때 생성 메소드에 최초로 생성할 퍼밋의 수를 넘겨준다. 외부 스레드는 퍼밋을 요청해 확보하거나, 이전에 확보한 퍼밋을 반납할 수도 있다.

 

public boolean tryAcquire(int permits, long timeout, TimeUnit unit)  
    throws InterruptedException {  
    if (permits < 0) throw new IllegalArgumentException();  
    return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));  
}

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)  
        throws InterruptedException {  
    if (!Thread.interrupted()) {  
        if (tryAcquireShared(arg) >= 0)  
            return true;  
        if (nanosTimeout <= 0L)  
            return false;  
        int stat = acquire(null, arg, true, true, true,  
                           System.nanoTime() + nanosTimeout); 
        if (stat > 0)  
            return true;  
        if (stat == 0)  
            return false;  
    }  
    throw new InterruptedException();  
}

final int acquire(Node node, int arg, boolean shared,  
                  boolean interruptible, boolean timed, long time) {  
    Thread current = Thread.currentThread();  
    byte spins = 0, postSpins = 0;   // retries upon unpark of first thread  
    boolean interrupted = false, first = false;  
    Node pred = null;                // predecessor of node when enqueued 

    // for 루프를 남는 퍼밋이 생기거나, 인터럽트가 걸리거나, 지정한 시간을 넘겨 타임아웃이 걸리기 전까지 반복한다.
    for (;;) {  
        if (!first && (pred = (node == null) ? null : node.prev) != null &&  
            !(first = (head == pred))) {  
            if (pred.status < 0) {  
                cleanQueue();           // predecessor cancelled  
                continue;  
            } else if (pred.prev == null) {  
                Thread.onSpinWait();    // ensure serialization  
                continue;  
            }  
        }  
        if (first || pred == null) {  
            boolean acquired;  
            try {  
                if (shared)  
                    acquired = (tryAcquireShared(arg) >= 0);  
                else  
                    acquired = tryAcquire(arg);  
            } catch (Throwable ex) {  
                cancelAcquire(node, interrupted, false);  
                throw ex;  
            }  
            if (acquired) {  
                if (first) {  
                    node.prev = null;  
                    head = node;  
                    pred.next = null;  
                    node.waiter = null;  
                    if (shared)  
                        signalNextIfShared(node);  
                    if (interrupted)  
                        current.interrupt();  
                }  
                return 1;  
            }  
        }  
        if (node == null) {                 // allocate; retry before enqueue  
            if (shared)  
                node = new SharedNode();  
            else  
                node = new ExclusiveNode();  
        } else if (pred == null) {          // try to enqueue  
            node.waiter = current;  
            Node t = tail;  
            node.setPrevRelaxed(t);         // avoid unnecessary fence  
            if (t == null)  
                tryInitializeHead();  
            else if (!casTail(t, node))  
                node.setPrevRelaxed(null);  // back out  
            else  
                t.next = node;  
        } else if (first && spins != 0) {  
            --spins;                        // reduce unfairness on rewaits  
            Thread.onSpinWait();  
        } else if (node.status == 0) {  
            node.status = WAITING;          // enable signal and recheck  
        } else {  
            long nanos;  
            spins = postSpins = (byte)((postSpins << 1) | 1);  
            if (!timed)  
                LockSupport.park(this);  
            else if ((nanos = time - System.nanoTime()) > 0L)  
                LockSupport.parkNanos(this, nanos);  
            else  
                break;  
            node.clearStatus();  
            if ((interrupted |= Thread.interrupted()) && interruptible)  
                break;  
        }  
    }  
    return cancelAcquire(node, interrupted, interruptible);  
}

tryAcqure 메서드를 활용하여 퍼밋을 요청한다. 현재 사용할 수 있는 남은 퍼밋이 없는 경우, acquire 메소드는 남는 퍼밋이 생기거나, 인터럽트가 걸리거나, 지정한 시간을 넘겨 타임아웃이 걸리기 전까지 대기한다.

 

public void release(int permits) {  
    if (permits < 0) throw new IllegalArgumentException();  
    sync.releaseShared(permits);  
}

public final boolean releaseShared(int arg) {  
    if (tryReleaseShared(arg)) {  
        signalNext(head);  
        return true;  
    }  
    return false;  
}

release 메소드는 확보했던 퍼밋을 다시 세마포어에게 반납하는 기능을 한다.

 

이진 세마포어

카운팅 세마포어를 좀 더 간단한 형태로 살펴보면 이진 세마포어를 생각할 수 있다. 이진 세마포어는 초기 퍼밋 값이 1로 지정된 카운팅 세마포어이다. 이진 세마포어는 비재진입 락의 역할을 하는 뮤텍스로 활용할 수 있다. 이진 세마포어의 퍼밋을 갖고 있는 스레드가 뮤텍스를 확보한 것이다.