Sdílení objektů mezi vlákny v C++20 (II)

Tento článek je volným pokračování předchozího článku Sdílení objektů mezi vlákny jednoduše v C++20. Tentokrát se podíváme na jiné řešení, které má ambice se dostat do normy C++26

Synchronized_value na CppConu

Krátké představení nového návrhu najdete ve videu Lightning Talk: Thread Safety With synchronized_value in C++ – Jørgen Fogh – CppCon 2023. Video stojí za shlédnutí, protože má něco přes 3 minuty, včetně upoutávky na CppCon 2024 na začátku videa

Pro ty, kteří si nemohou prohlížet videa mám krátke STLDW (Still Too Long, Didn't Watch)

Myšlenka je taková, že mám nějakou třídu „Foo“, která nepoužívá zámky ani jiný druh synchronizace. Nová šablona synchronized_value umožňuje třídu ochránit zámkem, takže jakýkoliv přístup do třídy bude vždy zajištěn automatickým zamknutí a odemknutí po použití

synchronized_value<Foo> foo(args...);
foo->method();
foo->bar(a,b,c);

Rozdíl mezi přímým voláním metod na Foo a voláním přes šablonu synchronized_value je v tom, že místo tečky . je použit operátor ukazatele ->. Tato „slabina“ je dána tím, že C++ stále ještě neumí tvořit wrapovací třídy, což by mohl umožnit jiný návrh na možnost přetížit operator tečky a tím zajistit forwardování volání metod do jiné třídy (šablona by byla jen wrapper). Škoda, musí nám stačit šipka. I tak to je velká pomoc a čitelnost kód určitě neutrpí.

Proč čekat do 2026?

Anthony Williams vypracoval svůj návrh už v roce 2014 a mohl tak v návrhu počítat s normou maximálně C++14 ale spíš C++11. O deset let později už máme k dispozici mnohem lepší nástroje, takže si myslím, že je zbytečné čekat až do roku 2026 a zkusme si něco podobného napsat už teď (a tady).

Synchronized_value (na rootu)

Začneme návrhem třídy 

template<typename T, typename Lock = std::mutex>
class synchronized_value {
public:
    //uvidíme
private:
    T payload;
    mutable Lock mx;
};

Šablona předpokládá, že uživatel bude moci dodat vlastní třídu, která implementuje zámek, a už teď mě napadá možnost, co kdybych chtěl použít shared_mutex? To by umožnilo zamykat objekt ve sdíleném režimu při const přístupu. Velice podobně to mám udělané v předchozím článku u shared_lockable_ptr. Budeme tedy předpokládat, že 

Lock = std::shared_mutex

Ale podporu pro std::mutex by jsme neměli opomenout, protože std::mutex je menší a v některých situacích i rychlejší, než shared_mutex. Záleží na situaci. Každopádně na podporu mutexu se podíváme až na konci.

Konstruktor

Předpokládejme, že konstruktor šablony bude mít stejné argumenty jako konstruktor třídy, kterou takto obalujeme.

template<typename ... Args>
requires std::is_constructible_v<T, Args ...>
synchronized_value(Args && ... args)
     :payload(std::forward<Args>(args)...) {}

Pro ty z vás, kteří poprvé v životě vidí použití konceptu, tak tento konstruktor je šablona s libovolným počtem argumentů, ale vynucuje si (requires), že argumenty lze použít ke konstrukci typu T (payloadu). Pokud by to nešlo, překladač nahlásí chybu už na tomto místě

Samotná konstrukce payloadu používá forwardování argumentů do konstruktoru

Implementace operator->

Tohle nebude tak jednoduché  jak se na první pohled zdá. Je sice pravdou, že před vrácením výsledku tohoto operátoru můžeme mutex zamknout, ale jakým způsobem zajistíme jeho odemčení po návratu z volané metody? Následující implementace je špatně

T *operator->() {
    mx.lock();
    return &payload; //ŠPATNĚ

toto je taky špatně

T *operator->() {
    std::lock_guard lk(mx);
    return &payload; //ŠPATNĚ

K nalezení správného řešení si musíme vzpomenout na způsob, jakým C++ vyhodnocuje operátor šipky. Návratová hodnota totiž nemusí byt nutně ukazatel. Vrácení takové hodnoty není chybu, ale na vrácenou hodnotu je opět aplikován stejný operátor… tak dlouho, až je výsledkem čistý ukazatel.

To co tady potřebujeme je: foo->(něco)->method() … kde to (něco) je objekt, který existuje po dobu volání metody a zanikne na konci výrazu

Mohl by to být chytrý ukazatel?

chytrý_ukazatel operator->()

Tento chytrý_ukazatel definuje vlastní operator ->, který však už vrací pointer, a také by mohl umožňovat specifikovat vlastní způsob uvolnění reference, ve které provedeme odemčení. 

A nebo můžeme použít std::unique_ptr s vlastním deleterem. Pojďme si to napsat:

    struct access_deleter_mutable {
        Lock *mx;
        void operator()(auto){mx->unlock();}
    };

    struct access_deleter_const {
        Lock *mx;
        void operator()(auto){mx->unlock_shared();}
    };

Opět pro připomenutí: Deleter je funkce nebo objekt imitující funkci, která se volá s jedním parametrem a to typicky s pointerem na objekt, který je předmětem chytrého ukazatele. Funkcí deleteru je objekt zničit. My tady ale nic s objektem dělat nebudeme, budeme pouze odemykat zámek. Parametr se tedy ignoruje. Ukazatel na zámek musí být dodán při konstrukci chytrého ukazatele a jeho deleteru.

Napsal jsem dvě varianty tohoto deleteru, jedna je pro mutable přístup, kdy se zamyká exkluzivně, druhá je pro const přístup, kdy se zamyká ve sdíleném režimu. Pojďmě si ještě napsat konstrukce těchto chytrých ukazatelů – jako samostatné metody

using access_ptr_mutable = std::unique_ptr<T, access_deleter_mutable>;
using access_ptr_const = std::unique_ptr<const T, access_deleter_const>;

access_ptr_mutable lock() {
        mx.lock();
        return {&payload, {&mx}};
}

access_ptr_const lock() const {
        mx.lock_shared();
        return {&payload, {&mx}};
}

access_ptr_const lock_shared() const {return lock();}

Operace lock() je rozdělena na mutable a const variantu a podle toho vrací patřičný chytrý ukazatel. Vždy nejprve zamkneme daný mutex a pak je vrácen chytrý ukazatel včetně initializace pointeru na zámek, aby ho bylo možné odemknout.

Metoda lock_shared()   umožňuje přístup na objekt ve sdíleném režimu i přesto, že máme jeho mutable variantu. Samozřejmě, že nám to dovolí přistupovat jen na const metody.

Implementace operátoru šipky je už triviální:

auto operator->() {return lock();}
auto operator->() const {return lock();}

A to je vše přátelé… 

Tedy ne úplně, musíme utáhnout několik volných konců

Funkce apply()

Nárvh p0290r4 přidává funkci apply() s možností zavolat dodanou lambdu pod zámkem

synchronized_value<std::string> s;

std::string read_value(){
    return apply([](auto& x){return x;},s);
}

void set_value(std::string const& new_val){
    apply([&](auto& x){x=new_val;},s);
}

Hlavní výhodou (není vidět v příkladu) je, že funkce apply lze volat s vícero synchronized_value s, podobně jako  std::visit

apply(fn, a, b, c...);

To znamená, že funkce se sama postará o bezpečné zamknutí všech synchronizovaných hodnot a jejich obsah pak předá do funkce fn. Hlavním benefitem je, že zamykání řeší potencionální deadlocky, které by jinak nastaly v důsledku nedodržení pořadí zamykání. Používá to std::scope_lock. 

Tuto funkci musíme napsat jako friend protože musí mít přístup do soukromé části šablony

template<typename Fn, typename ... Values, typename ... Locks>
friend auto apply(Fn &&fn, synchronized_value &v1, synchronized_value<Values,Locks> &... vals) {
     std::scoped_lock lk(v1.mx, vals.mx...);
     return fn(v1.payload, vals.payload...);
}
template<typename Fn, typename ... Values, typename ... Locks>
friend auto apply(Fn &&fn, const synchronized_value &v1, const synchronized_value<Values,Locks> &... vals)  {

     std::tuple<std::shared_lock<Lock>,std::shared_lock<Locks>... > lst(
                        std::shared_lock<Lock>(v1.mx,std::defer_lock),
                        std::shared_lock<Locks>(vals.mx,std::defer_lock)...);

     auto lk = std::apply([](auto &...ll){return std::scoped_lock(ll...);},lst);

     return fn(v1.payload, vals.payload...);
}

Máme zde variantu jako pro mutable, tak const proměnné. Přitom bych upozornil, že lze funkci použít pouze pokud jsou všechny mutable, nebo všechny const, nelze to míchat. Obě šablony kromě funkce, která se má zavolat, mají vynucený minimálně jeden argument. To proto, že funkce apply(fn), bez dalších argumentů jednak nedává smysl a jednak by mohla mít význam v jiném kontextu a došlo by ke kolizi definic.

Mutable-varianta je realizována přímočaře. Jednoduše necháme zamknout všechny zámky a pak se zavolá funkce. 

Oproti tomu const-varianta je maličko komplikovanější v tom, že std::scoped_lock neexistuje v sdílené variantě. To lze obejít tím, že všechny zámky nejprve zabalíme do std::shared_lock (s defer lock). Tyto objekty pak lze zamknout pomocí std::scoped_lock se stejnou úrovní bezpečnost, pouze ve sdíleném režimu.

Konstrukce kopie, přesunu, přiřazení

Je otázkou, zda potřebujeme řešit kopii, přesun a přiřazení atd. Zcela určitě mutexy nelze přesouvat, ale payload ano. 

synchronized_value(const synchronized_value &other):payload(*other.lock_shared()) {}
synchronized_value(synchronized_value &&other):payload(std::move(*other.lock())) {}

Přiřazení hodnoty lze zařídit bez nutné konstrukce dočasné  synchronized_value

template<typename X>
requires std::is_assignable_v<T&, X>
synchronized_value &operator=(X &&val) {
    std::unique_lock _(mx);
    payload = std::forward<X>(val);
    return *this;
}

Ostatní přiřazení:

synchronized_value &operator=(const synchronized_value &other) {
    if (this != &other) {
        std::shared_lock other_lock(other.mx, std::defer_lock);
        std::scoped_lock lk(mx, other_lock);
        payload = other.payload;
    }
    return *this;
}
synchronized_value &operator=( synchronized_value &&other) {
    if (this != &other) {
        std::scoped_lock lk(mx, other.mx);
        payload = std::move(other.payload);
    }
    return *this;
}

I tady musíme dát pozor na zamykání dvou zámků a speciálně u přiřazení kopií je nutné hromadně zamknout jeden zámek ve sdíleném a druhý v exklusivním režimu

Podpora std::mutex

Šablona je navržena pro použití shared_mutex, který má metody pro zamykání ve sdíleném režimu. Třída mutex tyto metody nemá, takže šablona nepůjde s mutexem použít. Mohli bysme mutex zabalit do pomocné třídy, která by shared_lock() přemapovala na lock() a podobně pro unlock a try_lock. A nebo můžeme upravit šablonu aby detekovala použití mutexu a podle toho se zařídila

Jak poznáme mutex? No spíš jak poznáme objekt, který umí zamykání ve sdíleném režimu. Napíšeme si koncept

template<typename T>
concept is_shared_lockable = requires(T val) {
    {val.lock_shared()};
    {val.unlock_shared()};
    {val.try_lock_shared()};
};

Na základě tohoto konceptu můžeme vybrat, který wrapper se bude používat pro zamykání ve sdíleném režimu. Můžeme si vybrat mezi std::unique_lock a  std::shared_lock

 using shared_wrapper = std::conditional_t<
                  is_shared_lockable<Lock>,
                  std::shared_lock<Lock>,
                  std::unique_lock<Lock> >;

Tento wrapper budeme používat všude tam, kde jsme doposud používali std::shared_lock, tedy jak ve funkci apply, tak třeba v přiřazení

Upravíme ještě lock_shared() s deleterem

struct access_deleter_const {
    Lock *mx;
    void operator()(auto){shared_wrapper wrp(*mx, std::adopt_lock);}
};

Víme, že funkce se volá, když je zámek v zamčeném stavu, takže jen konstruujeme wrapper, adoptujeme zámek a destruktor tohoto wrapperu jej odemkne

access_ptr_const lock() const {
    shared_wrapper wrp(mx);
    return {&payload, {wrp.release()}};
}

Zde se použije funkce release() pro uvolnění zámku z wrapperu bez odemčení a předání na deleter chytrého ukazatel

A to je víceméně vše, pak už by šablona měla zvládat jak mutex, tak shared_mutex přičemž mutex zamyká vždy v exkluzivním režimu, zatímco shared_mutex „umí“ i sdílený režim

Celá třída

template<typename T>
concept is_shared_lockable = requires(T val) {
    {val.lock_shared()};
    {val.unlock_shared()};
    {val.try_lock_shared()};
};


template<typename T>
concept is_lockable = requires(T &val) {
    {val.lock()};
    {val.unlock()};
    {val.try_lock()};
};


template<typename T, is_lockable Lock = std::mutex>
class synchronized_value {
public:


using shared_wrapper = std::conditional_t<is_shared_lockable<Lock>, std::shared_lock<Lock>, std::unique_lock<Lock> >;


template<typename ... Args>
requires std::is_constructible_v<T, Args ...>
synchronized_value(Args && ... args)
        :payload(std::forward<Args>(args)...) {}


synchronized_value(const synchronized_value &other):payload(*other.lock_shared()) {}
synchronized_value(synchronized_value &&other):payload(std::move(*other.lock())) {}
synchronized_value &operator=(const synchronized_value &other) {        
    if (this != &other) {
         shared_wrapper other_lock(other.mx, std::defer_lock);
         std::scoped_lock lk(mx, other_lock);
         payload = other.payload;
    }
    return *this;        
}
synchronized_value &operator=( synchronized_value &&other) {
    if (this != &other) {
        std::scoped_lock lk(mx, other.mx);
        payload = std::move(other.payload);
    }
    return *this;
}

struct access_deleter_mutable {
     Lock *mx;
     void operator()(auto){mx->unlock();}
}; 


struct access_deleter_const {
     Lock *mx;
     void operator()(auto){shared_wrapper wrp(*mx, std::adopt_lock);}
}; 


using access_ptr_mutable = std::unique_ptr<T, access_deleter_mutable>;
using access_ptr_const = std::unique_ptr<const T, access_deleter_const>;

access_ptr_mutable lock() {
     mx.lock();
     return {&payload, {&mx}};
}

access_ptr_const lock() const {
     shared_wrapper wrp(mx);        
     return {&payload, {wrp.release()}};
}

access_ptr_const lock_shared() const {return lock();}
access_ptr_mutable operator->() {return lock();}
access_ptr_const operator->() const {return lock();}

const synchronized_value &make_const() const {return *this;}

template<typename X>
requires std::is_assignable_v<T&, X>
synchronized_value &operator=(X &&val) {
        std::unique_lock _(mx);
        payload = std::forward<X>(val);
        return *this;
}


template<typenameFn, typename ... Values, typename ... Locks>
friend auto apply(Fn &&fn, synchronized_value &v1, synchronized_value<Values,Locks> &... vals) {
        std::scoped_lock lk(v1.mx, vals.mx...);
        return fn(v1.payload, vals.payload...);
}
template<typenameFn, typename ... Values, typename ... Locks>
friend auto apply(Fn &&fn, constsynchronized_value &v1, constsynchronized_value<Values,Locks> &... vals)  {

       std::tuple<shared_wrapper,typenamesynchronized_value<Values,Locks>::shared_wrapper... > lst(
             shared_wrapper(v1.mx,std::defer_lock),
             typename synchronized_value<Values,Locks>::shared_wrapper(vals.mx,std::defer_lock)...);            

       auto lk = std::apply([](auto &...ll){return std::scoped_lock(ll...);},lst);

       return fn(v1.payload, vals.payload...);
}


protected:
    T payload;
    mutable Lock mx;
};

Ukázku naživo si můžete vyzkoušet zde (pro mutex zde). V ukázce se používá imitace zámků, které pouze vypisují do konzole jaký druh operace by se měl provádět (VerboseLock)

Závěr

Cílem bylo naimplementovat něco, co se teprve do normy C++ chystá. Ukázat, že není třeba čekat, až se to schválí. Doufám, že tento jednoduchý nástroj najde uplatnění i ve vašich programech. Jakékoliv dotazy a komentáře vítám.