[Boost] semaphores equivalent?

[Darkvater]

Ik ben momenteel mijn applicatie aan het porteren naar andere OS's en gebruik hierbij Boost voor threading - voorheen Windows API.

Ik zit nog met het omzetten van de Windows API calls van CreateSemaphore / ReleaseSemaphore / WaitForSingleObject te knoeien. Ik denk dat boost::interprocess::interprocess_semaphore is wat ik nodig heb.

ReleaseSemaphore is dan boost::interprocess::interprocess_semaphore::post()
WaitForSingleObject is dan boost::interprocess::interprocess_semaphore::wait()

Maar hoe zit het precies met CreateSemaphore? In de Windows API kan ik ze aanmaken met een initial count en een maximum count. De constructor van Boost heeft alleen maar een initial count.

oude code:

C++:

int main() {
  int len = 10;
  HANDLE empty = CreateSemaphore(NULL, len, len, NULL);
  HANDLE full  = CreateSemaphore(NULL, 0,   len, NULL);
}

nieuwe code:

C++:

using boost::interprocess;
int main() {
  int len = 10;
  interprocess_semaphore empty(len);
  interprocess_semaphore full(0);
}

Maar nu heb ik geen maximum die een error geeft als mijn programma helemaal de weg kwijt raakt. Of zie ik iets over het hoofd?

[Soultaker]

In het algemeen heeft een semaphore ook helemaal geen maximum count. Waar heb je die voor nodig?

[Zoijar]

Heb je echt interprocess communicatie nodig? Want dan moet je je semaphores ook in shared memory aanmaken etc. Of is het gewoon een op zichzelfstaande applicatie? In dat geval kan je beter gewoon Boost.Thread gebruiken.

(ik zie tegenwoordig meer condition variables dan semaphores. Dit is een voorbeeld van een hele simpele producer consumer met boost threads; misschien dat het je verder op weg helpt:)

C++:

    // producer/consumer buffer, using a list
    template <typename T>
    class ProducerConsumer : private boost::noncopyable {
    public:
        ProducerConsumer() {}

        ~ProducerConsumer() {}

        bool empty() const {
            return m_items.empty();
        }

        bool full() const {
            return m_items.size() > 100;
        }

        void put(const T& item) {
            {
                boost::unique_lock<boost::mutex> lock(m_mutex);
                while(full()) {
                    m_cond_full.wait(lock);
                }
                m_items.push_back(item);
            }
            m_cond_empty.notify_one();
        }

        T get() {
            boost::unique_lock<boost::mutex> lock(m_mutex);
            while (empty()) {
                m_cond_empty.wait(lock);
            }

            T result = m_items.front();
            m_items.pop_front();

            m_cond_full.notify_one();
            return result;
        }

    private:
        boost::condition_variable m_cond_empty, m_cond_full;
        boost::mutex m_mutex;
        std::list<T> m_items;
    };

for completeness... met interprocess en managed shared memory:

C++:

#ifndef PRODUCER_CONSUMER_H_
#define PRODUCER_CONSUMER_H_

#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
#include <boost/interprocess/allocators/allocator.hpp>
#include <memory>

namespace FS {

template <typename T = char, typename Alloc = std::allocator<T> >
class SingleProducerConsumerBuffer {
public:
    // this is required because the shared memory maps to a different virtual pointer
    // in each process; i.e., regular pointers into shared memory can't be shared among processes
    // to solve this an offset pointer is used, which represents an offset from the start of the
    // shared memory -- wherever that is for each process.
    typedef typename boost::interprocess::offset_ptr<T> pointer;    

    enum AcquireMode {READ, WRITE};

    struct AcquireSlot {
        AcquireSlot(SingleProducerConsumerBuffer<T, Alloc>& x, AcquireMode am, bool block = true) : pcb(x), mode(am) {
            if (mode == READ) {
                ptr = pcb.acquire_read_slot(block, slot_nr);
            } else {
                ptr = pcb.acquire_write_slot(block, slot_nr);
            }
        }

        ~AcquireSlot() {
            if (ptr) {
                if (mode == READ) {
                    pcb.release_read_slot();
                } else {
                    pcb.release_write_slot();
                }
            }
        }

        T* ptr;
        size_t slot_nr;
    private:
        SingleProducerConsumerBuffer<T, Alloc>& pcb;
        AcquireMode mode;
    private:
        AcquireSlot(const AcquireSlot& src);
        AcquireSlot& operator=(const AcquireSlot& src);
    };

    typedef boost::shared_ptr<AcquireSlot> SlotPtr;

    SlotPtr acquireSlot(AcquireMode am, bool block = true) {
        return SlotPtr(new AcquireSlot(*this, am, block));
    }

    SingleProducerConsumerBuffer(size_t nr_slots, size_t slot_size, Alloc a = Alloc()) :
        sem_full(0), sem_empty(nr_slots), alloc(a), read_index(0), write_index(0), slots(nr_slots), slot_sz(slot_size)
    {
        storage = alloc.allocate(slots*slot_sz);
    }

    ~SingleProducerConsumerBuffer() {
        alloc.deallocate(storage.get(), slots*slot_sz);
    }

    size_t getSlotSize() const {
        return slot_sz;
    }

    size_t getNrSlots() const {
        return slots;
    }
    
protected:
    T* acquire_write_slot(bool block, size_t& slot_out) {
        if (block) {
            sem_empty.wait();
        } else if (!sem_empty.try_wait()) {
            return 0;
        }

        T* result = &storage[write_index*slot_sz];
        slot_out = write_index;

        advance(write_index);

        return result;
    }

    void release_write_slot() {
        sem_full.post();
    }

    T* acquire_read_slot(bool block, size_t& slot_out) {
        if (block) {
            sem_full.wait();
        } else if (!sem_full.try_wait()) {
            return 0;
        }

        T* result = &storage[read_index*slot_sz];
        slot_out = read_index;

        advance(read_index);
        return result;
    }

    void release_read_slot() {
        sem_empty.post();
    }

private:
    void advance(size_t& i) const {
        i = (i+1)%slots;
    }

private:
    boost::interprocess::interprocess_semaphore sem_full, sem_empty;
    Alloc alloc;
    size_t read_index, write_index, slots, slot_sz;
    pointer storage;
};

template <typename T, typename Segment = boost::interprocess::managed_shared_memory>
struct SHMTypes {
    typedef T type;
    typedef typename Segment::segment_manager segment_manager;
    typedef typename boost::interprocess::allocator<type, segment_manager> allocator;
    typedef typename FS::SingleProducerConsumerBuffer<type, allocator> ProducerConsumerBuffer;
};

typedef SHMTypes<char>::ProducerConsumerBuffer ShmPCB;
typedef SHMTypes<char>::allocator ShmAlloc;

typedef SingleProducerConsumerBuffer<char, std::allocator<char> > HeapPCB;

} // namespace FS

#endif

[ Voor 129% gewijzigd door Zoijar op 28-11-2009 21:55 ]

[Soultaker]

@Zoijar: ik vind dat toch niet echt winst ten opzichte van:

C++:

void put(const T& item) {
  m_sem_not_full.wait();
  m_items.push_back(item);
  m_sem_not_empty.post();
}

T get() {
  m_sem_not_empty.wait();
  T result = m_items.front();
  m_items.pop_front();
  m_sem_not_full.post();
  return result;
}

Niet alleen is 't veel minder code (de full() en empty() methods heb je ook niet meer nodig) maar de logica is (naar mijn mening) ook simpeler te doorzien. Natuurlijk hebben condition variables ook wel hun nut (je kunt er complexere logica mee implementeren, zoals bijvoorbeeld een buffer met variabele grootte) maar als je framework nu semaphores ondersteunt en je hebt een situatie zoals deze waarin je ze goed kunt gebruiken, waarom zou je dat dan niet doen? Wat is het voordeel van jouw (complexere) oplossing?

[Zoijar]

Soultaker schreef op zaterdag 28 november 2009 @ 22:38:
Wat is het voordeel van jouw (complexere) oplossing?

Dat Boost.Threads geen semaphore class heeft Overigens vind ik conditie variabelen makkelijker lezen en minder foutgevoelig. Die semaphore oplossing (zoals ik die ook in mijn 2e stuk code gebruik) ondersteunt overigens ook geen meerdere producers en consumers; de conditie variabele methode wel. Maar dat kan je natuurlijk oplossen door er een mutex omheen te zetten. Maar zo zie je hoe snel je een foutje maakt met semaphores.

[H!GHGuY]

Je kan natuurlijk ook een semaphore maken mbh mutex+condition variable.

code:

class semaphore
{
  boost::mutex m_Mutex;
  boost::condition_variable m_condvar;
  int m_count;

  void wait()
  {
     boost::unique_lock<boost::mutex> lock(m_Mutex);
     --m_count;
     while (m_count < 0)
       m_condvar.wait(lock);
  }
  void post()
  {
     boost::unique_lock<boost::mutex> lock(m_Mutex)
     ++m_count;
     m_condvar.signal();
  }
}

(code zomaar uit de hand, verwacht niet dat het zomaar compilet en werkt)

[Soultaker]

@H!GHGuY: die code is incorrect (even los van het feit dat je een constructor mist). Voorbeeld: als twee threads wait() aanroepen staat de count op -2. Als dan een thread post() aanroept gaat de count naar -1 maar wordt geen van beide threads wakker. De correcte aanpak is om de counter pas te decrementen wanneer wait() returnt:

C++:

void wait()
{
    boost::unique_lock<boost::mutex> lock(m_Mutex);
    while (m_count <= 0)
        m_condvar.wait(lock);
    --m_count;
}

(Eventueel kun je ook notify_one() gebruiken om precies één wachtende thread wakker te maken natuurlijk, en dan kan de while in wait() een if worden.)

[Zoijar]

Soultaker schreef op zondag 29 november 2009 @ 12:05:
en dan kan de while in wait() een if worden.)

Nee, want je kan volgens mij nog steeds spurious wake-ups krijgen in conditie vars.

[xos]

Als dat gebeurd gaat een while je toch ook niet helpen? Dan kan de situatie ontstaan dat twee threads uit de while loop breken (al dan niet door een signal of een spurious wakeup) omdat --m_count nog niet is gedaan.

[Soultaker]

@Zoijar: Ah ok, dan niet. Daar zou je de documentatie voor moeten lezen.

(POSIX semaphores hebben daar ook last van geloof ik.)

@xos: Jawel; nadat de condition gesignaled wordt, wordt automatisch de mutex weer verkregen. Het kan dus zijn dat twee threads wakker worden, maar dan kan er maar één tegelijk uitvoeren. De eerste verlaagt dan de counter en returnt; de tweede constateert daarna dat de count toch nog nul is, en gaat weer waiten.

[ Voor 57% gewijzigd door Soultaker op 29-11-2009 12:22 ]

[Darkvater]

Ik wilde het programma gewoon een-op-een overzetten zonder functionele verandering. De maximum count van de semaphore wordt gebruikt voor de buffer-lengte.

Als ik zo rondkijk in boost::interprocess::semaphore dan lijken ze het met een mutex en een counter te doen; vandaar mijn vraag. Maar ja, als het niet kan, zal ik het wel veranderen naar signal+mutex.

[Soultaker]

Een simpele semafoorklasse bouwen zoals H!GHGuY voorstelde is ook niet verkeerd; dan kun je de rest van je code houden zoals die is.

[Darkvater]

Soultaker schreef op zondag 29 november 2009 @ 13:18:
Een simpele semafoorklasse bouwen zoals H!GHGuY voorstelde is ook niet verkeerd; dan kun je de rest van je code houden zoals die is.

Ja, uiteraard, maar dat is dan niet equivalent met de native Windows API code. Bij elke andere verandering die ik heb gedaan tijdens het porteren is dat wel het geval. Na optimalisatie, etc. wordt (bijna) precies dezelfde code aangeroepen die ik voorheen zelf had staan voor windows.

[Soultaker]

Waarschijnlijk kun je die Windows API ook eenvoudig emuleren als je zou willen, maar wat je wil natuurlijk. De boost aanpak is waarschijnlijk toch zinniger in een C++ programma omdat je dan niet expliciet hoeft te locken/unlocken (en je niet kunt "vergeten" te unlocken als er een C++ exception geraised wordt).

Had je trouwens die maximum count nog ergens voor nodig? (Daar was het topic tenslotte om begonnen!)

[ Voor 33% gewijzigd door Soultaker op 29-11-2009 13:36 ]

[H!GHGuY]

Soultaker schreef op zondag 29 november 2009 @ 12:05:
@H!GHGuY: die code is incorrect (even los van het feit dat je een constructor mist).

Dat had ik ook wel verwacht
Daarom ook dat het onder de code nog even speciaal staat aangegeven dat de code nog moet gecorrigeerd worden.

Onder het motto: we zijn geen afhaalbalie, maar we helpen je graag op weg, dus.

[ Voor 10% gewijzigd door H!GHGuY op 29-11-2009 15:12 ]

[Darkvater]

Soultaker schreef op zondag 29 november 2009 @ 13:35:
Had je trouwens die maximum count nog ergens voor nodig? (Daar was het topic tenslotte om begonnen!)

De maximum count van de semaphore wordt gebruikt voor de buffer-lengte.

hier

[Soultaker]

Oef, slecht gelezen, mijn excuses.

Maar ik begrijp dan nog steeds niet waar je dat maximum voor nodig hebt; tenminste niet in een standaard producer/consumer scenario. De count voor je empty semafoor wordt geinitialiseerd op het aantal vrije sloten; dat zorgt er al voor dat je nooit meer sloten kunt vullen en vormt dus gelijk het maximum voor de full semafoor (die je immers pas verhoogt nadat je de ander verlaagt, en verlaagt voordat je de ander verhoogt).

[.oisyn]

Even tussendoor:

Soultaker schreef op zondag 29 november 2009 @ 12:20:
@xos: Jawel; nadat de condition gesignaled wordt, wordt automatisch de mutex weer verkregen. Het kan dus zijn dat twee threads wakker worden, maar dan kan er maar één tegelijk uitvoeren.

De code waar het over ging:

C++:

void wait() 
{ 
    boost::unique_lock<boost::mutex> lock(m_Mutex); 
    while (m_count <= 0) 
        m_condvar.wait(lock); 
    --m_count; 
}

Je wilt zeggen dat als een thread de wait op m_condvar doet, dat hij dan m_Mutex released? Is dat een feature van boost.thread (zelf nooit mee gewerkt)? Want in Windows is het zo dat als je een mutex lockt en je gaat dan op iets anders wachten, dat je de mutex doodleuk gelockt houdt. En dat lijkt me hier ook logisch, je hebt immers een scoped lock, en je bent nog niet buiten scope, dus geef je de lock ook niet vrij.

[Soultaker]

Ja, dat is een gebruikelijke feature van condition variables (in ieder geval ken ik ze niet anders). Je moet eerst een mutex locken, en vervolgens kun je op één (of in het geval van pthreads meerdere) conditions wachten; tijdens het wachten is de mutex releaset, maar zodra de call op de condition variable returnt hebt je de mutex weer gelockt.

De truc is natuurlijk dat het unlocken van de de mutex en het wachten op de conditie atomair uitgevoerd wordt; dat maakt condition variables juist zo krachtig. Als je zelf de mutex zou moeten locken/unlocken, zou je een race condition krijgen waarbij de condition gesignaled wordt net ná je de mutex releaset maar vóór je op de condition variable wacht en dan zit je (potentieel) voor altijd vast. Vandaar dat condition variables (voor zover ik ze ken) altijd functies hebben om ze atomair met een lock op een mutex uit te wisselen.

_{DISCLAIMER
Mijn random posts zijn absoluut geen surrogaat voor het lezen van de manual die bij je multiprogramming library zit. Ik blaat maar wat gebaseerd op mijn (gebrekkige) ervaring met pthreads, en eerder betrapte Zoijar me al op een fout, dus je weet wat mijn uitspraken waard zijn.}

edit:
Ik zie dat Microsoft sinds Windows Vista ook condition variables support met zo op het zicht vergelijkbare semantics, al geloof ik niet dat ze ook mutexen supporten, maar wat POSIX mutexes noemt zijn in Windows toch meer te vergelijken met critical sections.

[ Voor 26% gewijzigd door Soultaker op 30-11-2009 00:20 ]

[.oisyn]

Ah ja, het was me even ontgaan dat het lock object ook aan de wait() wordt meegegeven. Ik interpreteerde het even als "als je een wait doet, dan release je alle mutices die je ownt", wat me nogal vaag leek

[ Voor 6% gewijzigd door .oisyn op 30-11-2009 00:21 ]

[Zoijar]

.oisyn schreef op maandag 30 november 2009 @ 00:21:
Ah ja, het was me even ontgaan dat het lock object ook aan de wait() wordt meegegeven. Ik interpreteerde het even als "als je een wait doet, dan release je alle mutices die je ownt", wat me nogal vaag leek

Ik had eerlijk gezegd precies hetzelfde de eerste keer dat ik dit soort code doornam... maar idd, condition variables nemen een lock mee en releasen/re-acquiren die.

[H!GHGuY]

.oisyn schreef op maandag 30 november 2009 @ 00:21:
Ah ja, het was me even ontgaan dat het lock object ook aan de wait() wordt meegegeven. Ik interpreteerde het even als "als je een wait doet, dan release je alle mutices die je ownt", wat me nogal vaag leek

Zie hier voor de glibc broncode: http://sourceware.org/git...d15960962674912d3;hb=HEAD

Altijd plezant om eens te zien hoe ze dit nou implementeren.