Semaphore details

Semaphore Basic usage scenarios

Semaphore The basic usage scenario of is to limit a certain number of threads to execute .

A simple example : A one-way tunnel can accommodate 10 A car or 5 Trucks pass (1 A truck is equivalent to 2 Cars ), The tunnel entrance records the equivalent proportion of vehicles that are currently in the tunnel . such as 1 A car and 1 A truck , The tunnel entrance displays 3. If tunnel entrance shows 10 It means it's full . When the car drove out of the tunnel , The number displayed at the tunnel entrance will be reduced accordingly . This example is suitable for the scenario Semaphore .

Semaphore In construction , You can pass in a int. Indicates how many licenses (permit). When the thread gets the lock , Tell the semaphore how many licenses to use ( Analogy with cars and trucks ), When the license to be used by the thread is insufficient , Then the calling thread will be blocked . You can have a preliminary understanding with the simple example above .

Semaphore - Semaphore

The following is a simple code demonstration

public static void main(String[] args) {
    //  Express 2 Permission .
    Semaphore sem = new Semaphore(2);
    for (int i = 0; i < 3; i++) {
        new Thread(() -> {
            try {
                //  One license is used by default .
                sem.acquire();
                System.out.println(Thread.currentThread() + " I get it.");
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread() + " I release it.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                sem.release();
            }
        }).start();
    }
}

Code output in :

Thread[Thread-0,5,main] I get it.
Thread[Thread-1,5,main] I get it.
Thread[Thread-1,5,main] I release it.
Thread[Thread-0,5,main] I release it.
Thread[Thread-2,5,main] I get it.
Thread[Thread-2,5,main] I release it.

The above can be roughly divided into the following three steps :

1. First step : The first thread 0 and 1, Get the lock . Threads 3 Blocked .
2. The second step : 3 Seconds later , Threads 0 And thread 1 Release the lock separately ,
3. The third step : Threads 2 You can get the lock .

Semaphore Acquire lock process

Semaphore There can be 4 There are three ways to get the lock .

1. acquire() Threads occupy a license .
2. acquire(int) Thread occupancy int Permission
3. acquireUninterruptibly() Threads occupy a license , The call cannot be interrupted
4. acquireUninterruptibliy(int) Thread occupancy int Permission , Call and do not interrupt

4 The two methods are only slightly different , Here we use acquire() Used for analysis , Others can be analyzed by themselves .

acquire Method

call Semaphore#acquire() Method , It is essentially called AQS#acquireSharedInterruptibly(int), Parameter is 1.

// arg  be equal to  1
public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //  About tryAcquireShared,Semaphore There are two realizations 
    //  One is fair lock , The other is unfair lock .  This analysis of unfair locks .
    if (tryAcquireShared(arg) < 0)
        //  call  AQS#doAcquireSharedInterruptibly(1)  Method 
        doAcquireSharedInterruptibly(arg);
}

In the above code , because AQS Regulations tryAcquireShared Methods should be overridden by the implementer . So in Semaphore There are two overrides in , One is the override of fair lock , The other is the override of unfair locks . Here choose to read with unfair lock . Because it is often used in daily life ( It could be unconscious , The constructor only needs to pass in one int).

// NonfairSync#tryAcquireShared  Method .
//  Be careful : NonfairSync extends Sync !!!
protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

// Sync#nonfairTryAcquireShared  Method 
int nonfairTryAcquireShared(int acquires) {
    //  When multithreading competition is fierce ,  The for The cycle will run many times .
    for (;;) {
        //  Get current status 
        int available = getState();
        //  Determine the remaining allowed threads 
        int remaining = available - acquires;
        //  adopt CAS Ensure multi-threaded operation .
        //  Finally, return the remaining .  Assume that the current remaining 2 individual .  To use 1 individual . 
        // if perform ( No other threads compete ) complete ,  And finally return to 1 individual .
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}

doAcquireSharedInterruptibly Method

Suppose the above method getState() Method returns 0, Expect to use 1 individual , Then calculate remaining = -1, And finally return to -1. Therefore, we will enter the following methods doAcquireSharedInterruptibly(int)

//  Suppose the parameter passed in is 1.
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    //  Encapsulate the calling thread into a shared Node,  Join the end of the queue of the two-way linked list 
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            //  Record node Former 
            final Node p = node.predecessor();
            //  The predecessor is the head node ,  Then try to get the lock 
            if (p == head) {
                int r = tryAcquireShared(arg);
                //  Locked successfully , Set the head node , And spread it 
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            //  Lock failed ,  Determine whether to sleep ,  If you don't sleep, go to the next cycle .
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

Summary of lock acquisition process

Semaphore The process of obtaining locks is summarized as follows :

1. Determine whether the lock acquisition conditions are met , Key methods nonfairTryAcquireShared.
2. If the lock is obtained successfully , Will also be modified state.
3. If getting lock fails , Key methods doAcquireSharedInterruptibly Blocked acquisition lock .
   1. Add to the bidirectional linked list
   2. If the head node follows , Attempt to acquire lock , Otherwise, it is judged to enter sleep and wait for awakening , Wake up and continue 3.2
   3. If you don't go to sleep , Then run directly to 3.2 Step

Semaphore Release lock process

Semaphore There are two ways to release the lock .

1. release() Release a license
2. release(int) Release int Permission

Both methods call AQS#releaseShared(int) Method , Use release() Method , Then the parameter is 1, Use release(int) Method , Then the parameter is int.

releaseShared Method

//  Release the shared lock 
public final boolean releaseShared(int arg) {
    //  call Semaphore#tryReleaseShared Method .
    if (tryReleaseShared(arg)) {
        // tryReleaseShared Release successful ,  Then release the two-way linked list head In the subsequent 
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared Method

// Semaphore#tryReleaseShared
protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        //  Get the current license ,  There's a concurrency problem 
        int current = getState();
        //  Calculate the number of licenses after release 
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        //  The spin ( adopt CAS) Set the state of .  If the setting is successful, it will return to true.
        if (compareAndSetState(current, next))
            return true;
    }
}

doReleaseShared Method

private void doReleaseShared() {
    for (;;) {
        //  Record the current head
        Node h = head;
        //  The queue contains waiting nodes 
        if (h != null && h != tail) {
            //  Record the waiting state of the header node 
            int ws = h.waitStatus;
            //  There is a next node that needs to wake up 
            if (ws == Node.SIGNAL) {
                // CAS  Set the state of ,  If it doesn't work ,  Concurrency leads to failure .
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;
                //  Wake up the subsequent .
                unparkSuccessor(h);
            }
            //  Concurrent , There may be wa by 0, The required status is PROPAGATE, Make sure you wake up 
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;
        }
        if (h == head)
            break;
    }
}

Release lock summary

Semaphore The process of releasing the lock is summarized as follows :

1. Release N Permission , Because there is concurrent release , need CAS Make sure to set the updated value .
2. Wake up the valid waiting nodes in the bidirectional linked list . ( There may be concurrency issues , introduce PROPAGATE state )
3. The awakened node invokes the process of acquiring locks .

The illustration Semaphore

public static void main(String[] args) throws InterruptedException {
    Semaphore sem = new Semaphore(2);
    for (int i = 0; i < 5; i++) {
        Thread thread = new Thread(() -> {
            try {
                sem.acquire();
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                sem.release();
            }
        });
        thread.start();
    }
}

The above procedure , View the whole process by adding non pause breakpoints and outputting log information . Split the output , Convenient view .

The first part of the log :

The number in the node represents the thread ID , by x Indicates that there is no thread recorded . On the far right is the node that grabs the lock , It records the thread number .

The second part of the log :

1. because Threads 0 and Threads 1 Start releasing the lock , And they are all updated state The state of
2. Threads 0 and Threads 1 At the same time, wake up the next valid node in the queue , There is a concurrency problem
3. Threads 0 Wake up successfully , Set thread 2 The node of waitStatus by 0, Threads 1 To wake up , It is found that someone has set , So set the thread 2 The node of waitStatus by PROPAGATE( spread ).
4. Threads 0 Wake up , Exit the lock release method . Threads 2 Successful lock snatching , And threads 1 Also exit the method of releasing the lock .

Threads 2 The picture after successfully grabbing the lock :

The third part of the log :

1. Threads 2 Wake up the thread 3 The node of , Threads 2 The wake-up task is over .
2. Threads 3 Lock acquired successfully , Threads 3 To wake up the thread 4. Threads 3 The wake-up task is over .

The fourth part of the log :

1. Threads 4 Trying to get a lock , Failure in the end .
2. Judge whether you can go to sleep , Discover your predecessor's waitStatus Not set to SIGNAL, So you can't sleep , Try it again .
3. The attempt failed , Go to sleep .

The fifth part of the log :

1. Threads 2 And thread 3 Release the lock in turn , And wake up the next thread in the queue .
2. Threads 2 Wake up the Threads 4, Threads 4 Go grab the lock , Threads 2 The wake-up task is over , Exit the lock release method .
3. Threads 4 Try to grab the lock , It is found that the lock grabbing is successful ( Later, you need to set the status in the queue , So it's not final ).
4. Threads 3 Because of the thread 2 Modified the header node , So threads 3 Set the head node status to PROPAGATE.
5. Threads 4 and 3 The wake-up task is over .

After the execution of the fifth part :

1. Threads 4 Get the lock .
2. The head node is therefore a thread 3 And thread 2 Wake up the threads in the queue concurrently , Cause threads 3 The first failure , And the second modification , Threads 4 The head of the first section has been changed , But it happens that there are no waiting nodes in the list , So the head node waitStatus by 0, So threads 3 Add the header node to the waitStatus Set to PROPAGATE.
3. Threads 4 After acquiring a lock , Will encapsulate threads 4 The thread in the node of is set to null, Convenient for GC Recycling .

The sixth part of the log :

1. Threads 4 Release the lock , Get into doReleaseShared Method , It is found that there are no nodes in the queue .

therefore AQS The last queue in is shown in the figure below :

Conclusion

1. understand Semaphore Lock release and acquisition process
2. understand Semaphore The underlying logic of
3. understand AQS The underlying shared lock mode

It's not easy to create , Thank you for watching , If you have questions and find mistakes in the text , Welcome message discussion !

---------------------
author ：Wuv1Up
source ：CSDN
original text ：https://blog.csdn.net/weixin_37150792/article/details/105692924
Copyright notice ： This is the author's original article , Please attach a link to the blog post ！
Content analysis By：CSDN,CNBLOG One click reprint plugin for blog posts