Implementing semaphore by using mutex operations and primitives

Question

Some time ago had an interview and was asked to implement Semaphore by using mutex operations and primitives only (he allowed int to be considered as atomic). I came with solution below. He did not like busy/wait part -- while (count >= size) {} -- and asked to implement locking instead by using more primitive types and mutexes. I did not manage to come with improved solution. Any ideas how it could be done?

struct Semaphore {
int size;
atomic<int> count;
mutex updateMutex;

Semaphore(int n) : size(n) { count.store(0); }

void aquire() {
    while (1) {
        while (count >= size) {}
        updateMutex.lock();
        if (count >= size) {
            updateMutex.unlock();
            continue;
        }
        ++count;
        updateMutex.unlock();
        break;
    }
}

void release() {
    updateMutex.lock();
    if (count > 0) {
        --count;
    } // else log err
    updateMutex.unlock();
}
};

Libel

Use a condition variable to remove the busy wait. en.cppreference.com/w/cpp/thread/condition_variable Also why are you not using RAII to do the lock/unlock. – Teetotal

Answer 1

I'd wager this is not possible to implement without a busy-loop using mutexes only.

If not busy-looping, you have to block somewhere. The only blocking primitive you've got is a mutex. Hence, you have to block on some mutex, when the semaphore counter is zero. You can be woken up only by the single owner of that mutex. However, you should woken up whenever an arbitrary thread returns a counter to the semaphore.

Now, if you are allowed condition variables, it's an entirely different story.

Carrizales

"You can be woken up only by the single owner of that mutex." Why? Isn't the whole point of mutexes that you let non-owning threads lock and unlock it? – Zeuxis

@SebastianHojas A mutex can only block a thread if it is locked. A locked mutex can only be unlocked by the thread that owns it. Hence, a thread that is blocked on a mutex can only be woken up by some one particular thread. You can't use a mutex if you have a case where a thread is blocked and either of two threads might do something that needs to unblock it. – Saprophyte

That is correct. This is an impossible task and, presumably, the correct answer is to explain why it can't be done. You can suggest improving the busy waiting using various platform-specific features (like "rep nop" or "pause" on x86 and related CPUs). You can suggest using condition variables. – Saprophyte

@DavidSchwartz It's been a while since I asked this question, but I really appreciate reading the answere here now. Thanks! – Zeuxis

That is true for C++. However, in other languages like Go you can indeed unlock a mutex from another threat (goroutine), and you can easily implement a binary semaphore with a mutex. See here: play.golang.org/p/jkBWIoLqnTY – Auvil

No, you cannot unlock a "mutex". You an unlock the Go specific type "sync.Mutex", which is just a badly named type, as it goes against the convention in every other major API. – Carrizales

Could this be done if only two threads access the semaphore, but one of them always increase it and the other always decrease it (think of a writer/reader pattern)? – Luo

Answer 2

That's true because technically there are some parts in your code that have no need to exist. 1- you used atomic datatypes atomic<int> count; which will take very few more cycles in execution and it is useless as long as incrementing and decrementing are locked by updateMutex.lock(); code so there is no other thread can change it during the locked state.

2- you put while (count >= size) {} which is also useless because you checked count again after the spinlock statement which is necessary and the one important here. "remember spinlock is a while(1)" when the mutex is taken by another thread.

besides if you decided to use int count; with some compiler's optimizations, maybe your code won't re-read count value!! for optimization, remember your semaphore is supposed to be used by different threads!! so you need to make it volatile, to avoid this problem.

at last, let me rewrite your code in a more performant way.

struct Semaphore {
int size;
volatile int count;
mutex updateMutex;

Semaphore(int n) : size(n), count(0) {}

void aquire() {
    while (1) {
        updateMutex.lock();
        if (count >= size) {
            updateMutex.unlock();
            continue;
        }
        ++count;
        updateMutex.unlock();
        break;
    }
}

void release() {
    updateMutex.lock();
    if (count > 0) {
        --count;
    } // else log err
    updateMutex.unlock();
    }

 };

Momentarily

Answer 3

Mutex and Condition_Variable are two most fundamental synchronization primitives. Mutex is used to exclusively access resource while Condition_Variable is used to signal. They are defined in C++11. The Semaphore is more advanced synchronization primitive. They are only defined until C++20.

If I were asked to implement semaphore, I would do

// C++20 has provided the semaphore in <semaphore> head
class Semaphore
{
    size_t avail;
    std::mutex mtx;
    std::condition_variable cv;

public:
    Semaphore(int avail = 1) : avail(avail){}

    void acquire() // P(x): named from the Dutch word proberen, which means to test.
    {
        std::unique_lock<std::mutex> lk(mtx);
        cv.wait(lk, [this](){return avail > 0;});
        avail--;
    }

    void release() // V(x): named from the Dutch word verhogen, which means to increment.
    {
        std::unique_lock<std::mutex> lk(mtx);
        avail++;
        cv.notify_one();
    }
};

Middy

Answer 4

as @chill pointet out, the solution I did write down here will not work, as locks have unique ownership. I guess in the end you will revert to busy wait (if you are not allowed to use condition variables). I leave it here if ppl. have the same idea they see that this DOES NOT WORK ;)

struct Semaphore {
int size;
atomic<int> count;
mutex protection;
mutex wait;

Semaphore(int n) : size(n) { count.store(0); }

void aquire() {
    protection.lock();
    --count;
    if (count < -1) {
        protection.unlock();
        wait.lock();
    }
    protection.unlock();
}

void release() {
    protection.lock();
    ++count;
    if (count > 0) {
        wait.unlock();
    }
    protection.unlock();
}
};

Uriniferous

It is worth noting that this solution does not actually need count to be atomic, since all required synchronization is done with the protection mutex. – Petie

No, this does not work. Only whoever executed 'wait.lock()' in 'acquire' can successfully call 'wait.unlock()' in release, but well, it can't, because it's blocked in 'acquire' and any other thread will get an error of not being the mutex owner. – Carrizales

for me it was the same as @Petie : I was not aware that mutexes in C++11 have that property. It makes sense sind usually you would use a condition variable for that. – Uriniferous

This actually works if you have a specific thread that take care of all the locking and unlocking of 'wait' and you simply ask this thread to lock and unlock (ie implement mutex that anyone can acquiere or release). – Sargasso

This is quite an interesting idea, although I can not really imagine how to make that work currently. I mean what I could imagine would be that the semaphore is its own thread, Client Threads call acquire and always immediately go to sleep, if the acquire op. works, it is awoken by the thread, else they wait. However, how to I guarantee that I do not miss that one wakeup if the acquire works? Or I am completely on a wrong track now? – Uriniferous

I just realized that I changed my answer to precisely this x) But can't a mutex be unlocked by other threads? I thought that was the difference between mutex and lock. The problem states "mutex" not "lock". – Bahrain

Answer 5

Who said you could only use one or two mutexes?

class Semaphore {
    private member tickets: Cell with int = Cell(0);
    private member waiting: List with Mutex = List();

    public trywaitfor(): bool {
        for (;;) {
            local n = tickets.value;
            if (n <= 0)
                return false;
            if (tickets.cmpxch(n, n - 1) === n)
                return true;
        }
    }

    public waitfor() {
        for (;;) {
            if (trywaitfor())
                return;
            local m = Mutex();

            /* First acquire won't block */
            m.acquire();

            /* Schedule our mutex as waiting */
            waiting.append(m);

            /* Prevent race condition in case a
             * ticket was posted in the mean-time */
            if (trywaitfor()) {
                if (!waiting.removeif(e -> e === m)) {
                    /* Race condition: our mutex was removed
                     * from the list of waiters by another
                     * thread calling `release' and picking
                     * us (even though we didn't actually end
                     * up waiting). To prevent the loss of a
                     * post-signal, forward said signal to
                     * the next waiting thread. */
                    wakeone();
                }
                return;
            }

            /* Second acquire blocks until released */
            m.acquire();
        }
    }

    private wakeone() {
        local m;
        try {
            m = waiting.popfront();
        } catch (...) {
            /* Catch handler for when popfront()
             * fails because the list is empty */
            return;
        }
        /* Wake up a waiting thread */
        m.release();
    }

    public post() {
        local n;
        /* Add a new ticket */
        do {
            n = tickets.value;
        } while (!tickets.cmpxch(n, n + 1));

        /* Wake up at most 1 waiting thread */
        wakeone();
    }
}

This assumes that:

• The Mutex type isn't recursive
• The thread calling Mutex.release() doesn't have to be the same one that called Mutex.acquire() (Mutex implementations that aren't recursive usually allow this as well, since disallowing it would already require identifying the thread that did the acquire, which is wholly unnecessary if you don't want to be recursive)
  • Yes: I know that means that it's not technically a "Mutex", but rather a "Lock", but come on: pleanty of people use those two terms interchangably.
• All operations on the Cell (an atomic object container) and List (it's a list, duh) types are atomic (but even if they're weren't, you can just add another mutex around all calls to tickets and waiting)

I guess your interviewer meant lists/vectors/arrays when he was talking about "more primitive types"

Polymer

Answer 6

EDIT - use a second mutex for queuing intstead of threads

Since a mutex already have proper thread-support, it can be used to queue the threads (instead of doing it explicitly as I first had tried to do). Unless the mutex is restricted to only let the owner unlock it (a lock?), then this solution doesn't work.

I found the solution in Anthony Howe's pdf that I came across when searching. There are two more solutions given there as well. I changed the names to make more sense for this example.

more or less pseudo code:

Semaphore{
    int n;
    mutex* m_count;         //unlocked initially
    mutex* m_queue;         //locked initially
};

void wait(){
    m_count.lock();
    n = n-1;
    if(n < 0){
        m_count.unlock();
        m_queue.lock();     //wait
    }
    m_count.unlock();       //unlock signal's lock
}

void signal(){
    m_count.lock();
    n = n+1;

    if(n <= 0){
        m_queue.unlock();   //leave m_count locked
    }
    else{
        m_count.unlock();
    }
}

Bahrain

This won't work. If you have a thread wait(), then get context switched by the OS after it has release m1 but before it blocks then you have have another thread call signal(), see the thread in the queue, signal that thread to wake up (which it will). But then the first thread's the next instruction to block so it will sleep immediately. – Citral

edited my answer. Realized it's the same now as another answer here. – Bahrain

I think this is still broken. Let's say t1 starts by signalling. Then, 'm_count' becomes locked, and m_queue becomes unlocked. Now, have t2 wait. It gets stalled waiting to enter the 'wait()' function because 'count' is locked. Further, 'm_count' can never become unlocked unless you never signal before waiting, which is not the behaviour you are after. – Citral

@Citral Sorry I made a typo. m_count is unlocked at start of wait (will edit that now). Furthermore, m_queue will only be unlocked if n<=0, meaning a thread must be waiting. m_count is unlocked from the start so wait can be called before signal. Also, it makes no sense to start by signalling? This must be avoided and is usually left to the coder from what I've seen. – Bahrain

Answer 7

lemme try this

`

# number of threads/workers
w = 10
# maximum concurrency
cr = 5
r_mutex = mutex()
w_mutex = [mutex() for x in range(w)]

# assuming mutex can be locked and unlocked by anyone 
# (essentially we need a binary semaphore)

def acquire(id):

    r_mutex.lock()
    cr -= 1
    # r_mutex.unlock()
    
    # if exceeding maximum concurrency
    if cr < 0:
        # lock twice to be waken up by someone
        w_mutex[id].lock()
        r_mutex.unlock()
        w_mutex[id].lock()
        w_mutex[id].unlock()
        return

    r_mutex.unlock()

    
    
       
def release(id):

    r_mutex.lock()
    cr += 1
    # someone must be waiting if cr < 0
    if cr <= 0:
        # maybe you can do this in a random order
        for w in w_mutex:
            if w.is_locked():
                w.unlock()
                break
    r_mutex.unlock()

`

Kachine

@Sneftel care to put forth the scenario? – Kachine

A thread runs acquire and is preempted just after releasing r_mutex. Another thread runs release but gets through the loop without finding the first thread to wake up. – Harriman

Additionally, this isn’t how mutexes work. Neither POSIX nor Widows mutexes, nor standard C or C++ mutexes, can be released by a thread that does not own them. – Harriman

@Sneftel, thanks for pointing out. and I already pointed out that what I needed was a binary semaphore – Kachine