【问题标题】:algorithm to combine std::unique with a reduce step?将 std::unique 与减少步骤相结合的算法?
【发布时间】:2021-08-03 13:29:35
【问题描述】:

有人能想出一个干净(快速)的解决方案来解决以下问题吗:

  • 我有一个条目序列,基本上包含一个键和一个值,比如一个
struct Value {
    int index = 0;
    int cost = 0;
}
  • 我现在想合并条目,这样每个键只包含一次,但值应该合并 - 即每个 index 应该只包含在序列中一次,并且每个重复索引的 cost 应该累积.

我想出的基本解决方案是对序列进行排序,当在传递给std::sortBinaryPredicate 中检测到相等的条目时,cost 将与lhs 相加。然后rhs 的成本将设置为 0。然后是 remove_if,它删除了 0 成本值。示例见此处:

#include <cstdlib>
#include <vector>
#include <algorithm>
#include <iostream>

struct Value
{
    int index = 0;
    int cost = 0;
};

// generate a bunch of random values in a vector
// values will have indices in range [0..10]
std::vector<Value> generator()
{
    std::vector<Value> v(20);
    std::generate(v.begin(), v.end(), []() { return Value{std::rand() % 10, std::rand() % 10}; });
    return v;
}

void print(const std::vector<Value> &values)
{
    for (auto v : values)
        std::cout << "{i=" << v.index << ", c=" << v.cost << "}, ";
    std::cout << "\n";
}

// 
void merge(std::vector<Value> &values)
{
    // sort values and merge costs
    std::sort(values.begin(), values.end(), [](auto &lhs , auto &rhs) {
        if (lhs.index == rhs.index) {
            lhs.cost += rhs.cost;
            rhs.cost = 0;
        }
        return lhs.index < rhs.index;
    });
    // remove entries with empty cost
    auto it = std::remove_if(values.begin(), values.end(), [](const auto &v) { return v.cost == 0; });
    values.erase(it, values.end());
}

int main()
{
    auto v = generator();
    std::cout << "generated values: ";
    print(v);

    merge(v);
    std::cout << "merged values: ";
    print(v);

}

Live on Compiler Explorer

问题是:虽然上面的示例产生了正确的结果,但我可以看出它不符合 C++ 标准。 BinaryPredicate“不应通过取消引用的迭代器应用任何非常量函数”http://eel.is/c++draft/algorithms.requirements#8.sentence-4。比较是一个 BinaryPredicate。 http://eel.is/c++draft/alg.sorting#general-2.sentence-1)

这是否意味着我唯一的选择是推出自定义 inplace_unique_reduce 或类似的,或者是否有其他优雅的方法来解决这个问题?我宁愿不必为此编写自己的重要算法。

谢谢

【问题讨论】:

  • 请内嵌示例代码,而不仅仅是通过指向外部易失资源的链接。
  • std::sort里面做积累是个坏主意。
  • 您可以先排序,然后处理相等的键(现在应该连续存储),最后删除所有重复的键。当然,第二步和第三步可以同时完成。但是,第 2 步和第 3 步的复杂度为 O(n)。因此,我不会太在意 1 或 2 次额外的循环运行。
  • 绝对需要就地完成吗?地图将是一个非常合适的解决方案,imo。另外,您希望解决方案有多通用?在您提供的答案中,该解决方案非常通用,但通过指定一些提取函数参数(projection)可能会更加通用。
  • 我同意 Scheff 的观点,即在排序之后,大多数 STL 解决方案可能是两阶段(for_each+remove_if)。然后你可以使用Value* prev = nullptr 来跟踪当前的 value.index。对于大量数据,您可以改用自定义合并排序,合并也可以进行归约。

标签: c++ algorithm stl


【解决方案1】:

这是迄今为止我能想到的最好的,但我仍然想知道是否有不需要自定义算法而是结合现有算法的解决方案:


template <class ForwardIt, class BinaryPredicateCompare, class BinaryOpReduce>
ForwardIt inplace_unique_reduce(ForwardIt first, ForwardIt last, BinaryPredicateCompare cmp, BinaryOpReduce reduce)
{
    if (first == last)
        return last;

    ForwardIt result = first;
    while (++first != last) {
        if (cmp(*result, *first)) {
            reduce(*result, *first);
        } else if (++result != first) {
            *result = std::move(*first);
        }
    }
    return ++result;
}

【讨论】:

  • 嗯,如果有两个具有相同indexValue 对象不相邻,这不会失败吗?
  • @Fureeish 显然是的,就像普通的std::unique 一样,您必须先运行std::sort
【解决方案2】:

假设您可以接受额外的分配,我会使用std::map(或std::unordered_map):

auto merge_entries(std::vector<Value>& original_values) {
    auto values = std::map<int, int>();

    for (const auto [index, cost] : original_values) {
        values[index] += cost;
    }

    const auto end_of_merged_values = std::transform(
            values.cbegin(), values.cend(), original_values.begin(),
            [](const auto entry) {
                return Value{entry.first, entry.second};
            }
    );

    original_values.erase(end_of_merged_values, original_values.end());
}

除了一个for() 循环(在我看来,可以std::for_each 代替,尽管这样的改变会引入不必要的样板,导致代码更难阅读,在我看来),这个解决方案只使用STL。

我们首先使用映射合并所有条目,然后我们覆盖一些元素,以便我们原来的std::vector 保存合并的条目。超级方便的是std::transform 返回一个指向插入范围末尾的迭代器。为什么它对我们有益?因为除了不发生合并的不太可能的情况之外,与最初传入的元素相比,我们的元素更少。使用该迭代器,我们可以erase 向量的其余部分(未覆盖的元素)保持其干净,类似 STL 的风格。


假设您可以接受额外分配,但您可以接受加强迭代器要求(双向),我会使用 std::partial_sumstd::unique

template <class BiDirIt, class BinaryPredicateCompare, class BinaryOpReduce>
auto inplace_unique_reduce(
        BiDirIt first, BiDirIt last,
        BinaryPredicateCompare cmp,
        BinaryOpReduce reduce
) {
    std::partial_sum(
            std::make_reverse_iterator(last), std::make_reverse_iterator(first),
            std::make_reverse_iterator(last),
            [cmp, reduce](auto acc, const auto& elem) {
                if (cmp(acc, elem)) {
                    return reduce(acc, elem);
                } else {
                    acc = elem;
                }
                return acc;
            }
    );

    return std::unique(first, last, cmp);
}

这样使用:

auto values = std::vector<Value>{
        {1, 1}, {2, 2}, {2, 7}, {0, 5},
        {3, 3}, {1, 2}, {3, 10}
};
auto comparator = [](const auto& lhs, const auto& rhs) {
    return lhs.index == rhs.index;
};
auto reducer = [](const auto& lhs, const auto& rhs) {
    return Value{lhs.index, lhs.cost + rhs.cost};
};

auto to_remove = inplace_unique_reduce(
        values.begin(), values.end(),
        comparator,
        reducer
);

values.erase(to_remove, values.end());

for (const auto[index, cost] : values) {
    std::cout << index << ' ' << cost << '\n';
}

就像您的原始答案一样,这不会合并不相邻的元素,但要做到这一点,您必须从我的答案的第一部分按index 对它们进行排序或使用map 之类的东西。

std::make_reverse_iterator 调用是必要的,因为std::partial_sum 将合并的元素累积在给定的一组连续等效元素的最右侧。另一方面,std::unique 仅保留此类组中的第一个元素。因此,您希望以与您将成为std::unique-ing 的顺序相反的顺序合并元素。


您对复制或移动成本高昂的情况提出了一些担忧 - 在这种情况下,您要么使用考虑到独特约束的自定义解决方案,要么放宽约束。在这里,我们移动分配合并的条目,但这就是潜在的瓶颈。如果您的移动分配运算符很昂贵,我担心没有标准解决方案适合您,您必须自己动手,就像在您的答案中一样。

【讨论】:

  • 从可读性 POV 来看,这个解决方案很好,但遗憾的是,由于每个映射条目的额外内存分配,它比我的解决方案更糟糕。此外,请注意,在更复杂的场景中,不容易像您正在做的那样将值拆分为键和值(例如,当值是仅移动时)。或者这样做可能代价高昂。
  • @milianw imo 你的问题有点缺乏明确的要求。我准备了一个包含自定义键提取、自定义值提取、自定义减速器等的解决方案,但我只是不知道您目标的边缘情况。如果您展示更多示例,那就太好了(例如,当您的struct 中有第三个元素时会发生什么 - 减少时会发生什么?)此外,我要求在您的答案下方进行澄清 - 您的解决方案没有合并不相邻的条目,而这个条目。应该吗?
  • 酷。是否可以通过使用反向迭代器来消除对 reverse 的需求 - 例如在打电话给partial_sum?无论如何,不​​要为此付出太多汗水,我很乐意将赏金交给你,除非其他人提出了我们迄今为止没有想到的奇特解决方案。就个人而言,我上面写的自定义解决方案似乎是最好的,因为它做的工作量最少。真的很遗憾,STL 算法很难在这里结合起来提供有效的解决方案
  • @milianw 有趣的是,我刚刚完成了我提到的编辑,您绝对可以使用反向迭代器执行您刚才建议的操作(摆脱反向迭代器,但仍然需要双向迭代器)。一旦我回到我的电脑,我可能会将该编辑应用到代码中。至于赏金,还是等别人有机会吧。这并不意味着什么 - 我很乐意帮助并深入研究算法:)
【解决方案3】:

我个人会更乐意修改remove_ifPredicate 中的对象,因为它是一种比排序更简单、更直接的算法。不过,它确实需要跟踪输出索引:

void merge(std::vector<Value> &values)
{
    std::sort(values.begin(), values.end(), [](auto &lhs , auto &rhs) { return lhs.index < rhs.index; });

    std::size_t o_i = 0;
    auto it = std::remove_if(values.begin(), values.end(), [&](const auto &v) { 
        if(o_i > 0 && values[o_i-1].index == v.index)
        {
            values[o_i-1].cost += v.cost;
            return true;
        }
        o_i++;
        return false; });
    values.erase(it, values.end());
}

这是另一个使用 range-v3 库的版本(std::ranges 还没有 group_by)。

#include <range/v3/all.hpp>

void merge(std::vector<Value> &values)
{
    std::sort(values.begin(), values.end(), [](auto &lhs, auto &rhs) { return lhs.index < rhs.index; });
    
    auto merged = values
        | ranges::v3::view::group_by([](Value &lhs, Value &rhs) { return lhs.index == rhs.index; })
        | ranges::v3::view::transform([](auto &&vs){
                int index, cost = 0;
                for(auto& v : vs) { index=v.index; cost+=v.cost;}
                return Value{index, cost};
            });
    auto it = ranges::v3::copy(merged.begin(), merged.end(), values.begin()).second;
    values.erase(it, values.end());
}

另外,不要忘记普通的旧 for 循环:

void merge(std::vector<Value> &values)
{
    std::sort(values.begin(), values.end(), [](auto &lhs , auto &rhs) { return lhs.index < rhs.index; });

    std::size_t j = 0;
    for(std::size_t i = 0; i < values.size(); i++)
    {
        if(j>0 && values[j-1].index == values[i].index)
        {
            values[j-1].cost += values[i].cost;
            continue;
        }
        if(j!=i) values[j]=std::move(values[i]);
        j++;
    }
    values.erase(values.begin()+j, values.end());
}

【讨论】:

  • 谢谢,remove_if 版本肯定比我原来的版本好看。但它也有同样的问题——根据标准,谓词应该是常量:eel.is/c++draft/algorithms.requirements#7
  • 基于范围的版本不是就地的,是吗? IE。 merged 有分配,不是吗?
  • 好问题!我认为它是就地的,但我对 group_by 或 range(也不是它们的实现)不够熟悉,无法确定。
  • j = 0; for(i = 1; i &lt; size; ++i) { if(values[i].index == values[j].index) { values[j].cost += values[i].cost; } else if(++j != i) { values[j] = std::move(values[i]); } } 跳过了很多操作。
  • group_by 不分配任何东西,因为它只对连续的元素进行分组。如果您想在所有情况下真正合并所有元素,此解决方案还需要排序。
猜你喜欢
  • 2017-02-16
  • 2021-10-03
  • 2016-12-17
  • 1970-01-01
  • 1970-01-01
  • 2017-01-28
  • 1970-01-01
  • 1970-01-01
  • 2015-07-12
相关资源
最近更新 更多