【问题标题】:C++ template meta-programming: sort list of mutexes to enforce locking orderC ++模板元编程:对互斥体进行排序以强制执行锁定顺序
【发布时间】:2016-01-25 22:50:45
【问题描述】:

我有存储在 POSIX 共享内存中的数据结构(我们称它们为“资源”)。对每个资源的访问由每个资源互斥体介导。一个进程有时可能需要以原子方式更新多个资源。在更新/修改相关资源之前,此过程必须获取所有先决条件互斥锁。必须以明确定义的顺序获取互斥锁以避免典型的死锁情况。我想开发一种编译时方法来确保以正确的顺序获取锁。

每个资源以任意顺序单独映射到每个进程。因此,我无法按资源地址顺序获取资源。除了在编译时不会确定正确的顺序这一事实之外,资源地址的相对顺序可能会因进程而异,因为每个资源可能(甚至可能)映射到不同的虚拟地址。 幸运的是,由结构表示的每种资源类型都有一个 constexpr 定义的唯一整数 ID。我想按ID顺序获取资源。

假设每个数据结构看起来像这样:

template<typename ResourceStruct, int UniqueId>
struct SharedResource
{
    static constexpr int ID = UniqueId;
    ResourceStruct resource;
};

我有一个类似于 C++11 的 std::lock 的函数,它接收要锁定为模板参数的互斥锁列表。我相信应该可以在编译时根据每个资源的 ID 对这些模板参数进行排序。不幸的是,我一直在努力使用必要的模板元编程体操来实现它。我研究了几种排序模板参数的方法(例如,quicksort #1quicksort #2),但它们似乎都过于复杂。我是否过度思考我的问题?有没有更简单的方法?如果可能的话,我更喜欢纯 C++11 解决方案(我宁愿避免依赖 Boost)。

【问题讨论】:

  • 我想指出,如果该过程采取任何其他锁定(之前或之后),那么这一切都是徒劳的......
  • 当然可以,但这对于几乎任何协作同步方案都是危险的。出于调试目的,我可能会保留一个线程局部变量,用于标记线程何时拥有资源。嵌套请求可能会被拒绝(或至少被记录)。我的主要动机是推广一种应该(希望)避免错误的做法。第二个动机是效率:如果锁组以相反的顺序释放,调度程序抖动的机会就会减少。
  • 如果您准备好保留一个全局变量(线程本地或非线程本地),您可能有兴趣存储仍锁定在那里的最新 ID。然后,每当尝试获取锁时,您都可以检查资源的唯一ID,看看它是否被授权(注意:为了恢复之前锁定的ID,锁本身应该记住它)。
  • 这是个好主意。谢谢!
  • 编译时排序是一个有趣且具有挑战性的项目。但是,您的问题假定按顺序锁定是最好的算法。如果您的互斥锁具有try_lock 功能,则可能没有。请参阅howardhinnant.github.io/dining_philosophers.html,了解有序锁定算法与其他几种多锁定算法的性能比较。

标签: c++ templates c++11 synchronization metaprogramming


【解决方案1】:

带上你的身份证。将它们打包成一个模板序列。

将每个元素变成一对(值,索引)。

对这些元素进行编译时排序。我发现合并排序很容易编写,或者您可以使用冒泡排序或选择排序(假设计数很小)。快速排序可能有点矫枉过正。

现在,去掉索引。按排序顺序生成一组这些索引。

将您的原始参数包装在一个元组中,对它们执行std::get&lt;Is&gt;(其中Is 来自包含剥离索引的std::index_sequence)以获取新排序的参数,并调用一个函数。锁现在已订购。

手动编写排序是工作。如果你不想使用 boost 实现,你必须给自己写一个 sort。


template<class...Ts> struct types {using type=types;};
template<class types, size_t N> struct get_nth; // ::type is the nth element of types
template<class types, size_t N> struct remove_nth; // ::type is types without the nth
template<class types, class pred> struct min_index; // returns the index of the least element
template<class...Types> struct append; // appens the types in Types... into one types<?>
template<class types, class pred> struct selection_sort;
// if non-empty, gets the min element, generates
// a types<> containing just it, and appends it to the front of
// the remaining elements with the nth element removed, then sorted
// if types is empty, returns `types<>`.

最多应该是大约 100 行代码。 200 如果你喜欢回车。

使用索引位转换成对只是确保值可以遵循周围的类型。

【讨论】:

    【解决方案2】:

    鉴于我们讨论的是编译时排序,我假设参数的数量相当少。在这种情况下,我建议忘记实施通用解决方案,而只需使用Sorting Networks

    在 C++14 中会容易得多(由于自动返回类型推导),但是在 C++11 中仍然可以。模板深度限制甚至会非常好:

    template <typename T0, typename T1>
    struct cmp { static uint64_t const value = T0::ID < T1::ID; };
    
    //
    //  Sort 1
    //
    template <typename T0>
    struct sort_1 {
        typedef std::tuple<T0&> type;
        type sort(T0& t0) { return {t0}; }
    };
    
    //
    //  Sort 2
    //
    template <uint8_t C, typename T0, typename T1>
    struct sort_2_impl;
    
    //  T0 >= T1
    template <typename T0, typename T1>
    struct sort_2_impl<0, T0, T1> {
        typedef std::tuple<T1&, T0&> type;
        type sort(T0& t0, T1& t1) { return {t1, t0}; }
    };
    
    //  T0 < T1
    template <typename T0, typename T1>
    struct sort_2_impl<1, T0, T1> {
        typedef std::tuple<T0&, T1&> type;
        type sort(T0& t0, T1& t1) { return {t0, t1}; }
    };
    
    template <typename T0, typename T1>
    struct sort_2:
        sort_2_impl<
            cmp<T0,T1>::value,
            T0, T1
        > {};
    
    //
    //  Sort 3
    //
    template <uint8_t C, typename T0, typename T1, typename T2>
    struct sort_3_impl;
    
    //  0: T0 >= T1 & T0 >= T2 & T1 >= T2 -> T2 <= T1 <= T0
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<0, T0, T1, T2> {
        typedef std::tuple<T2&, T1&, T0&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t2, t1, t0}; }
    };
    
    //  1: T0 < T1 & T0 >= T2 & T1 >= T2 -> T2 <= T0 < T1
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<1, T0, T1, T2> {
        typedef std::tuple<T2&, T0&, T1&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t2, t0, t1}; }
    };
    
    //  2: T0 >= T1 & T0 < T2 & T1 >= T2 -> impossible
    
    //  3: T0 < T1 & T0 < T2 & T1 >= T2 -> T0 < T2 <= T1
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<3, T0, T1, T2> {
        typedef std::tuple<T0&, T2&, T1&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t0, t2, t1}; }
    };
    
    //  4: T0 >= T1 & T0 >= T2 & T1 < T2 -> T1 < T2 <= T0
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<4, T0, T1, T2> {
        typedef std::tuple<T1&, T2&, T0&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t1, t2, t0}; }
    };
    
    //  5: T0 < T1 & T0 >= T2 & T1 < T2 -> impossible
    
    //  6: T0 => T1 & T0 < T2 & T1 < T2 -> T1 <= T0 < T2
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<6, T0, T1, T2> {
        typedef std::tuple<T1&, T0&, T2&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t1, t0, t2}; }
    };
    
    //  7: T0 < T1 & T0 < T2 & T1 < T2 -> T0 < T1 < T2
    template <typename T0, typename T1, typename T2>
    struct sort_3_impl<7, T0, T1, T2> {
        typedef std::tuple<T0&, T1&, T2&> type;
        type sort(T0& t0, T1& t1, T2& t2) { return {t0, t1, t2}; }
    };
    
    template <typename T0, typename T1, typename T2>
    struct sort_3:
        sort_3_impl<
            (cmp<T0, T1>::value << 0) |
            (cmp<T0, T2>::value << 1) |
            (cmp<T1, T2>::value << 2),
            T0, T1, T2
        > {};
    

    哦,使用脚本来生成所有样板文件可能值得...

    【讨论】:

    • 使用代码生成器来实现上面的蛮力排序网络,生成支持多达 8 个元素的元组的网络会产生 370000+ 行(可读)代码。由于绝大多数生成的类型从未使用过,因此解析生成的代码是一个显着的编译时开销。 6 个元素的元组在大约 7000 行代码中是合理的。 7 个元素的元组似乎处于合理的边缘,约为 50000。
    【解决方案3】:

    按照 Yakk 概述的算法,并使用question 的答案,我设计了以下内容。我很感激反馈! (使用 clang++ 7.0.0 和 -std=c++11 编译)

    #include <iostream>
    #include <type_traits>
    #include <tuple>
    
    namespace detail
    {
        // pairs a tuple element with an index value
        template <typename TUP, std::size_t I>
        struct enumerated_tuple
        {
            static constexpr std::size_t index = I;
            using type = typename std::tuple_element<I, TUP>::type;
        };
    
    
        // implement index_sequence and make_index_sequence for C++11
        template <std::size_t...> struct index_sequence {};
    
        template <std::size_t N, std::size_t... Is>
        struct make_index_sequence : make_index_sequence<N - 1, N - 1, Is...> {};
    
        template <std::size_t... Is>
        struct make_index_sequence<0u, Is...> : index_sequence<Is...>
        {
             using type = index_sequence<Is...>;
        };
    
    
        // functions for binding an index to a tuple
    
        // returns a tuple of tuple elements paired with an index value
        template <typename TUP, std::size_t... I>
        auto make_indexed_tuple(TUP&& t, index_sequence<I...>) -> decltype(std::make_tuple(enumerated_tuple<TUP, I>()...))
        {
            return std::make_tuple(enumerated_tuple<TUP, I>()...);
        }
    
        // pairs each tuple element with an index
        template <typename TUP>
        struct indexed_tuple
        {
            using type = decltype(make_indexed_tuple(TUP(), typename make_index_sequence<std::tuple_size<TUP>::value>::type()));
        };
    
    
        // functions for generating index sequences used for removing a selected element from a tuple
    
        // join two sequences, with the second sequence shifted by Offset
        template <typename Seq1, std::size_t Offset, typename Seq2> struct concat_seq;
        template <std::size_t ... Is1, std::size_t Offset, std::size_t ... Is2>
        struct concat_seq<index_sequence<Is1...>, Offset, index_sequence<Is2...>>
        {
            using type = index_sequence<Is1..., (Offset + Is2)...>;
        };
    
        // generate a sequence 0..N without E, where E >= 0 and E <= N
        template <std::size_t N, std::size_t E>
        struct gen_seq
        {
            using type = typename detail::concat_seq<typename make_index_sequence<E>::type, E + 1, typename make_index_sequence<(N > E) ? (N - E - 1) : 0>::type>::type;
        };
    
    
        // generate a subtuple, picking out elements based upon the order and value of supplied integer sequence
        template <typename TUP, std::size_t... I>
        auto subtuple(TUP&& t, index_sequence<I...>) -> decltype(std::make_tuple(std::get<I>(t)...))
        {
            return std::make_tuple(std::get<I>(t)...);
        }
    
        // remove the nth element from a tuple
        template <typename TUP, std::size_t N>
        struct remove_nth
        {
            using type = decltype(subtuple(TUP(), typename gen_seq<std::tuple_size<TUP>::value, N>::type()));
        };
    
        // get the nth element from a tuple. (wrap std::tuple_element for the sake of consistency (flips template params))
        template <typename TUP, std::size_t N>
        struct get_nth
        {
            using type = typename std::tuple_element<N, TUP>::type;
        };
    
        // concatenates two tuples
        template <typename TUP1, typename TUP2>
        struct append
        {
            using type = decltype(std::tuple_cat(TUP1(), TUP2()));
        };
    
        // select the minimum type
        template <typename T0, typename T1, template<typename, typename> class CMP>
        struct select_min
        {
            using type = typename std::conditional<CMP<typename T0::type, typename T1::type>::value, T0, T1>::type;
        };
    
        // functions for finding the minimum element in a tuple
        template <typename TUP, std::size_t size, template<typename, typename> class CMP>
        struct min_tuple_element_helper
        {
            using type = typename select_min<typename get_nth<TUP, 0>::type, typename min_tuple_element_helper<typename remove_nth<TUP, 0>::type, size-1, CMP>::type, CMP>::type;
        };
    
        template <typename TUP, template<typename, typename> class CMP>
        struct min_tuple_element_helper<TUP, 1, CMP>
        {
            using type = typename std::tuple_element<0, TUP>::type;
        };
    
        // find the minimum tuple element, using the comparator CMP
        template <typename TUP, template<typename, typename> class CMP>
        struct min_tuple_element
        {
            using indexed = typename indexed_tuple<TUP>::type;
            using type_and_index = typename min_tuple_element_helper<indexed, std::tuple_size<indexed>::value, CMP>::type;
            using type = typename type_and_index::type;
            static constexpr std::size_t index = type_and_index::index;
        };
    
        template <typename TUP, std::size_t size, template<typename, typename> class CMP>
        struct selection_sort_helper
        {
            using index = typename indexed_tuple<TUP>::type;
            using selected = typename min_tuple_element<TUP, CMP>::type_and_index;
            using remaining = typename remove_nth<TUP, selected::index>::type;
            using remaining_sorted = typename selection_sort_helper<remaining, size-1, CMP>::type;
            using type = typename append<std::tuple<typename selected::type>, remaining_sorted>::type;
        };
    
        template <typename TUP, template<typename, typename> class CMP>
        struct selection_sort_helper<TUP, 1, CMP>
        {
            using type = TUP;
        };
    } // end namespace
    
    
    
    template <typename L, typename R>
    struct less_than
    {
        static constexpr bool value = (L::id < R::id);
    };
    
    template <typename L, typename R>
    struct greater_than
    {
        static constexpr bool value = (L::id > R::id);
    };
    
    // sort the elements in tuple, using the comparator CMP
    template <typename TUP, template<typename, typename> class CMP>
    struct selection_sort
    {
        using type = typename detail::selection_sort_helper<TUP, std::tuple_size<TUP>::value, CMP>::type;
    };
    
    // all tuple elements are a concrete type of this type
    template <std::size_t ID>
    struct Value
    {
        static constexpr std::size_t id = ID;
    };
    
    int main(void)
    {
        using example = typename std::tuple<Value<5>, Value<3>, Value<2>, Value<4>>;
        printf("unsorted tuple:\n");
        printf("%d\n", (int)std::tuple_element<0, example>::type::id);
        printf("%d\n", (int)std::tuple_element<1, example>::type::id);
        printf("%d\n", (int)std::tuple_element<2, example>::type::id);
        printf("%d\n", (int)std::tuple_element<3, example>::type::id);
    
        using min_sorted_example = typename selection_sort<example, less_than>::type;
        printf("min-sorted tuple:\n");
        printf("%d\n", (int)std::tuple_element<0, min_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<1, min_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<2, min_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<3, min_sorted_example>::type::id);
    
        using max_sorted_example = typename selection_sort<example, greater_than>::type;
        printf("max-sorted tuple:\n");
        printf("%d\n", (int)std::tuple_element<0, max_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<1, max_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<2, max_sorted_example>::type::id);
        printf("%d\n", (int)std::tuple_element<3, max_sorted_example>::type::id);
    
        return 0;
    }
    

    输出:

    unsorted tuple:
    5
    3
    2
    4
    min-sorted tuple:
    2
    3
    4
    5
    max-sorted tuple:
    5
    4
    3
    2
    

    【讨论】:

      猜你喜欢
      • 2022-01-08
      • 1970-01-01
      • 1970-01-01
      • 2013-11-13
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多