Code review von synchronized queue

knivil

Ich bastle an einer synchronized queue. Der Code ist hier http://ideone.com/17IYk6 .

#ifndef SYNC_H
#define SYNC_H

#include <basic/queue.h>
#include <boost/thread.hpp>

namespace af {
namespace sync {

template<typename T>
struct queue
{
    // for convenient
    typedef typename af::basic::queue<T> container;

    struct monitor_t
    {
        monitor_t() : q_(0) {};
        monitor_t(monitor_t& m) : q_(m.q_) {}
        ~monitor_t() {}

        template<typename F>
        void operator()(F& f) const
        {
            boost::lock_guard<boost::mutex> lock(q_->m_);
            q_->c_.notify_one();
            f(q_->q_);
        }

    private:
        queue* q_;
        // for setting reference
        friend monitor_t queue::monitor();

    };

    queue() : q_() {}

    bool empty() const
    {
        boost::lock_guard<boost::mutex> lock(m_);
        return q_.empty();
    }

    size_t size() const
    {
        boost::lock_guard<boost::mutex> lock(m_);
        return q_.size();
    }

    bool pop(T& msg, unsigned int ms)
    {
        boost::system_time const timeout
            = boost::get_system_time()
            + boost::posix_time::milliseconds(ms);
        boost::unique_lock<boost::mutex> lock(m_);
        while( q_.empty())
        {
            if (!c_.timed_wait(lock, timeout))
                return false;
        }

        q_.pop(msg);
        return true;
    }

    // consumes msg and moves it into the queue
    void push(T& msg)
    {
        boost::lock_guard<boost::mutex> lock(m_);
        q_.push(msg);
        c_.notify_one();
    }

    void clear()
    {
        boost::lock_guard<boost::mutex> lock(m_);
        q_.clear();
    }

    monitor_t monitor()
    {
        monitor_t tmp;
        tmp.q_ = this;
        return tmp;
    }

    ~queue()
    {
        clear();
    }

private:

    // do not copy
    queue(queue const&);
    queue& operator=(queue const&);

    af::basic::queue<T> q_;
    mutable boost::mutex m_;
    boost::condition_variable c_;
};

}
}

Bei der Implementierung folge ich dem Beispiel aus http://www.youtube.com/playlist?list=PL1835A90FC78FF8BE . Aber da ich ab und an mehr als nur pop und push einzeln aufrufen moechte sondern als "Transaktion", kann ich bei Bedarf ein Monitorobjekt benutzen. Dabei folge ich dem Beispiel monitor<T> aus http://isocpp.org/blog/2013/01/c-concurrency-herb-sutter (sollte in die Playlist) angepasst auf meine Queue.

Nun zu meinem Anliegen: Gibt es etwas zu verbessern oder zu aendern, sind Fehler enthalten? Fuer Kommentare bin ich dankbar. Das public Interface von af::basic::queue weicht von std::queue etwas ab und sieht folgendermassen aus:

template<typename T>
struct queue
{
    queue();
    bool empty() const;
    size_t size() const;
    void pop(T& msg);
    void push(T& msg);
    void clear();
    ~queue();
};

Es kompiliert gcc 4.72 und einfache Tests sind erfolgreich. Ich kann kein C++11 benutzen und bei boost bin ich auf Version 1.49.0 beschraenkt. Danke.

edit: code von ideone hierhergetragen.

cooky451

Shade Of Mine

Warum unique_lock in pop() ?

Ansonsten ist mir das notify_one in monitor nicht ganz geheuer. Aber das ist nur ein Gefühl ohne dass ich es belegen kann. Ich habe das Video von Herb Sutter dazu jetzt aber auch nicht gesehen.

Eine Optimierungsmöglichkeit wären 2 Arten von Locks: read und write. size() und empty() brauchen nur readlocks während pop und push write Locks brauchen. Ob das eine optimierung ist, weiß ich aber nicht. Es wäre aber auf jedenfall eine Option die man sich anschauen könnte.

uU wäre ein non blocking pop auch interessant. Also eins ohne timed_wait. Wobei man das wahrscheinlich mit einem 0 als Parameter simulieren könnte. Gefühlt könnte man hier aber mit einem eigenen empty test in pop etwas rausholen.

Sieht aber eigentlich alles OK aus.

hustbaer

Ich mag keine absoluten Timeouts. Ist zwar (viel) einfacher "korrekt" zum implementieren als mit relativen (vor allem weil die Standard-Library keine "timeout clock" Klasse zur Verfügung stellt -- oder gibt es sowas?), aber dafür ist es eben auch nur "korrekt" und nicht korrekt: bei Änderungen der Systemzeit kann es zu Problemen kommen.

~~
Und die Monitor-Sache könnte problematisch werden, wenn die "gemonitorte" Funktion [c]pop[/c] aufruft. Dann wäre die Mutex nämlich doppelt gelockt, und könnte daher durch das [c]timed_lock[/c] nicht mehr entsperrt werden.

Eine Möglichkeit das zu lösen wäre z.B. direkten Zugriff auf die Mutex zu geben, und einen pop-Overload zu machen der ein [c]unique_lock[/c] Objekt als Parameter nimmt.
Oder die Queue selbst lockbar machen, so dass man ein [c]unique_lock<my_queue>[/c] verwenden kann.
Oder dem Funktor den der Monitor ausführt als Parameter das [c]unique_lock[/c] mitgeben - irgendsowas.~~
EDIT: ich sehe gerade du übergibst ja direkt ne Referenz auf die interne "nicht threadsafe" queue an die Funktion. Dadurch gibt es das von mir oben beschriebene Problem natürlich nicht mehr.

hustbaer

Shade Of Mine schrieb:

Warum unique_lock in pop() ?

Vermutlich weil man einen lock_guard nicht "entlocken" kann (oder geht das?) -- siehe timed_lock Aufruf.

hustbaer

Shade Of Mine schrieb:

Ansonsten ist mir das notify_one in monitor nicht ganz geheuer.

Ja, da müsste wohl ein notify_all hin.
Denn die aufgerufene Funktion kann ja mehr als einen Eintrag einfügen. Es würde dann aber nur einer der "popper" aufgeweckt, und das ist vermutlich nicht im Sinne des Erfinders.

knivil

Ich habe das Video von Herb Sutter dazu jetzt aber auch nicht gesehen.

Kommt auch so nicht vor, da es im Video sehr einfach gehalten wurde.

hustbaer schrieb:

Ja, da müsste wohl ein notify_all hin.
Denn die aufgerufene Funktion kann ja mehr als einen Eintrag einfügen. Es würde dann aber nur einer der "popper" aufgeweckt, und das ist vermutlich nicht im Sinne des Erfinders.

Korrekt, das wird im Monitor geaendert. Wie sieht das bei push aus, wenn 2 Elemente mittels push ohne Kontextwechsel eingefuegt werden und zwei Konsumer warten. Wird dann auch nur einer aufgeweckt? Ich glaube da muss ich nochmal im Standard zu pthread und boost nachlesen.

wäre ein non blocking pop auch interessant. Also eins ohne timed_wait.

Ja, darueber habe ich auch nachgedacht. Wollte das Interface mit try_pop (non blocking) und pop (blockierend ohne timeout) nicht ueberladen.

Ich mag keine absoluten Timeouts.

Ich auch nicht. Aber das C++11 Threadinterface mit wait_for steht mir leider nicht zur Verfuegung. Somit wird das relative Timeout am Interface in ein absolutes gewandelt. Der Benutzer der Klasse sieht nur das relative. Und ja, Systemzeit aendern ist fatal.

Warum unique_lock in pop() ?

Das Interface von boost::condition_variable aber auch von C++11s std::condition_variable sieht es vor. Und ja, lock_guard hat nur Konstruktor und Destruktor, kann also nicht unlocken.

Eine Optimierungsmöglichkeit wären 2 Arten von Locks: read und write. size() und empty() brauchen nur readlocks während pop und push write Locks brauchen.

Darueber werde ich nachdenken.

Dann wäre die Mutex nämlich doppelt gelockt,

Ja, ich haette ein Beispiel fuer die Benutzung angeben sollen. Die Funktion/Funktor bekommt ein af::sync::queue<T>::container (also hier af::basic::queue<T> ) und nicht af::sync::queue<T> .

Auf alle Faelle: Ich danke euch!

knivil

Nachtrag: Ich habe mich entschieden das Monitorobjekt aufzugeben und einfach eine kleine Methode synchronized(F f) anzubieten, die das gleiche macht.

template<typename F>
void synchronized(F& f)
{
    boost::lock_guard<boost::mutex> lock(m_);
    c_.notify_all();
    f(q_);
}

hustbaer

knivil schrieb:

Korrekt, das wird im Monitor geaendert. Wie sieht das bei push aus, wenn 2 Elemente mittels push ohne Kontextwechsel eingefuegt werden und zwei Konsumer warten. Wird dann auch nur einer aufgeweckt? Ich glaube da muss ich nochmal im Standard zu pthread und boost nachlesen.

Ich bin mir 95% sicher dass das kein Problem macht.
Aber lies es im Standard nach, garantieren kann ich es dir nicht.

Falls der Standard doch doof sein sollte kann man als Workaround den aufgeweckten Popper vor dem Unlock nochmal checken lassen ob die Queue noch Elemente enthält, und falls ja nochmal notify_one aufrufen lassen.

In Fällen wo immer viele Popper warten, aber die neuen Einträge immer nur langsam dahertröpfeln könnte das performanceschonender sein als ein notify_all.
(Wobei es natürlich in Fällen wo nur ein Popper wartet sehr gute Chancen hat eine Spur langsamer zu sein.)

dot

Falls es da immer noch um die Message Queue von hier geht und du durch diese Queue einfach nur Threads mit Arbeit versorgen willst, möchte ich auch mal die Möglichkeit in den Raum werfen, die Queue völlig lock-free als einfachen Ring Buffer zu implementieren.