问题

I want to use std::sort for an opaque data type with a run-time record length. Here is an answer to a similar question.

我想在运行时记录长度的不透明数据类型上使用std::sort。以下是一个类似问题的答案。

Attempts

I have tried indirectly sorting record pointers and indices, but these methods have terrible cache locality because the comparisons need to access randomly distributed rows of the dataset.

我尝试间接地对记录指针和索引进行排序，但这些方法的缓存局部性很差，因为比较需要访问数据集中随机分布的行。

I also tried falling back to the C functions qsort which allows for run-time record length. This is better, but it requires a callback function for the comparison. This has two drawbacks, 1) performance suffers compared to using a lambda which can be inlined, and 2) the interface ergonomics are poor because you (probably) have to create a structure to hold a functor for the comparison.

我还尝试回退到允许运行时记录长度的C函数qsort。这更好，但它需要一个用于比较的回调函数。这有两个缺点，1) 性能不如使用可以内联的lambda函数，2) 接口人机工程学很差，因为你（可能）必须创建一个用于比较的函数对象的结构。

Possible Idea

The previously mentioned answer suggested a,

先前提到的答案建议了一个，

pod block pseudo reference iterator with a specialized swap,

but did not give any hints about how that might work.

但没有提供关于它如何工作的任何提示。

Summary

To summarize, I want a way to call std::sort passing in a lambda for the comparison for an array of opaque data types with run-time record length as the following code demonstrates,

总之，我想找到一种调用std::sort的方法，以便为具有运行时记录长度的不透明数据类型数组传递一个用于比较的lambda函数，如下面的代码所示，

#include <iostream>
#include <random>

using std::cout, std::endl;

int main(int argc, const char *argv[]) {
    // We have 100 records each with 50 bytes.
    int nrecords = 100, nbytes = 50;
    std::vector<uint8_t> data(nrecords * nbytes);

    // Generate random data for testing.
    std::uniform_int_distribution<uint8_t> d;
    std::mt19937_64 rng;
    std::generate(data.begin(), data.end(), [&]() { return d(rng); });

    int key_index = 20;
    auto cmp = [&](uint8_t *a, uint8_t *b) {
        int aval = *(int*)(a + key_index), bval = *(int*)(b + key_index);
        return aval < bval;
    };

    // How can I call std::sort with run-time record length?
    // std::sort(data.begin(), ...)
    return 0;
}

Iterator Attempt

I attempted to write an iterator based on the previous suggestion, but I have not been able to get it to compile yet. My iterator-foo is sub-par and I feel like I am making a conceptual error.

我尝试根据之前的建议编写一个迭代器，但是我还没有能够成功编译它。我的iterator-foo不够优秀，我觉得我可能在概念上出现了错误。

Here is the code,

#include <algorithm>
#include <random>

struct MemoryIterator {
    using iterator_category = std::forward_iterator_tag;
    using difference_type = std::ptrdiff_t;
    using value_type = uint8_t *;
    using pointer = value_type*;
    using reference = value_type&;

    MemoryIterator(value_type ptr, size_t element_size)
        : ptr_(ptr)
        , element_size_(element_size) {
    }

    reference operator*() {
        return ptr_;
    }

    MemoryIterator& operator++() {
        ptr_ += element_size_;
        return *this;
    }

    MemoryIterator& operator--() {
        ptr_ -= element_size_;
        return *this;
    }

    MemoryIterator operator++(int) {
        auto tmp = *this;
        ++(*this);
        return tmp;
    }

    MemoryIterator operator--(int) {
        auto tmp = *this;
        --(*this);
        return tmp;
    }

    MemoryIterator& operator+=(size_t n) {
        ptr_ += n * element_size_;
        return *this;
    }

    MemoryIterator operator+(size_t n) {
        auto r = *this;
        r += n;
        return r;
    }

    MemoryIterator& operator-=(size_t n) {
        ptr_ -= n * element_size_;
        return *this;
    }

    MemoryIterator operator-(size_t n) {
        auto r = *this;
        r -= n;
        return r;
    }
    friend bool operator==(const MemoryIterator& a, const MemoryIterator& b) {
        return a.ptr_ == b.ptr_;
    }

    friend difference_type operator-(const MemoryIterator& a, const MemoryIterator& b) {
        return a.ptr_ - b.ptr_;
    }

    friend void swap(MemoryIterator a, MemoryIterator b) {
    }

private:
    value_type ptr_;
    size_t element_size_;
};

int main(int argc, const char *argv[]) {
    int nrecords = 100, nbytes = 50;
    std::vector<uint8_t> data(nrecords * nbytes);

    std::uniform_int_distribution<uint8_t> d;
    std::mt19937_64 rng;
    std::generate(data.begin(), data.end(), [&]() { return d(rng); });

    int key_index = 20;
    auto cmp = [&](uint8_t *a, uint8_t *b) {
        int aval = *(int*)(a + key_index), bval = *(int*)(b + key_index);
        return aval < bval;
    };

    MemoryIterator begin(data.data(), nbytes);
    MemoryIterator end(data.data() + data.size(), nbytes); 
    std::sort(begin, end, cmp);

    return 0;
}

It complains that it cannot compare two MemoryIterator's which is true -- it needs to use the comparison function.

它抱怨它无法比

#include <iostream>
#include <random>

using std::cout, std::endl;

int main(int argc, const char *argv[]) {
    // We have 100 records each with 50 bytes.
    int nrecords = 100, nbytes = 50;
    std::vector<uint8_t> data(nrecords * nbytes);

    // Generate random data for testing.
    std::uniform_int_distribution<uint8_t> d;
    std::mt19937_64 rng;
    std::generate(data.begin(), data.end(), [&]() { return d(rng); });

    int key_index = 20;
    auto cmp = [&](uint8_t *a, uint8_t *b) {
        int aval = *(int*)(a + key_index), bval = *(int*)(b + key_index);
        return aval < bval;
    };

    // How can I call std::sort with run-time record length?
    // std::sort(data.begin(), ...)
    return 0;
}

#include <algorithm>
#include <random>

struct MemoryIterator {
    using iterator_category = std::forward_iterator_tag;
    using difference_type = std::ptrdiff_t;
    using value_type = uint8_t *;
    using pointer = value_type*;
    using reference = value_type&;

    MemoryIterator(value_type ptr, size_t element_size)
        : ptr_(ptr)
        , element_size_(element_size) {
    }

    reference operator*() {
        return ptr_;
    }

    MemoryIterator& operator++() {
        ptr_ += element_size_;
        return *this;
    }

    MemoryIterator& operator--() {
        ptr_ -= element_size_;
        return *this;
    }

    MemoryIterator operator++(int) {
        auto tmp = *this;
        ++(*this);
        return tmp;
    }

    MemoryIterator operator--(int) {
        auto tmp = *this;
        --(*this);
        return tmp;
    }

    MemoryIterator& operator+=(size_t n) {
        ptr_ += n * element_size_;
        return *this;
    }

    MemoryIterator operator+(size_t n) {
        auto r = *this;
        r += n;
        return r;
    }

    MemoryIterator& operator-=(size_t n) {
        ptr_ -= n * element_size_;
        return *this;
    }

    MemoryIterator operator-(size_t n) {
        auto r = *this;
        r -= n;
        return r;
    }
    friend bool operator==(const MemoryIterator& a, const MemoryIterator& b) {
        return a.ptr_ == b.ptr_;
    }

    friend difference_type operator-(const MemoryIterator& a, const MemoryIterator& b) {
        return a.ptr_ - b.ptr_;
    }

    friend void swap(MemoryIterator a, MemoryIterator b) {
    }

private:
    value_type ptr_;
    size_t element_size_;
};

int main(int argc, const char *argv[]) {
    int nrecords = 100, nbytes = 50;
    std::vector<uint8_t> data(nrecords * nbytes);

    std::uniform_int_distribution<uint8_t> d;
    std::mt19937_64 rng;
    std::generate(data.begin(), data.end(), [&]() { return d(rng); });

    int key_index = 20;
    auto cmp = [&](uint8_t *a, uint8_t *b) {
        int aval = *(int*)(a + key_index), bval = *(int*)(b + key_index);
        return aval < bval;
    };

    MemoryIterator begin(data.data(), nbytes);
    MemoryIterator end(data.data() + data.size(), nbytes); 
    std::sort(begin, end, cmp);

    return 0;
}

/opt/local/libexec/llvm-16/bin/../include/c++/v1/__algorithm/sort.h:533:21: error:
invalid operands to binary expression ('MemoryIterator' and 'MemoryIterator')
if (__i >= __j)
~~~ ^  ~~~
/opt/local/libexec/llvm-16/bin/../include/c++/v1/__algorithm/sort.h:639:8: note:
in instantiation of function template specialization
'std::__introsort<std::_ClassicAlgPolicy, (lambda at
sort0.cpp:92:16) &, MemoryIterator>' requested here
std::__introsort<_AlgPolicy, _Compare>(__first, __last, __comp, __depth_limit);
^
/opt/local/libexec/llvm-16/bin/../include/c++/v1/__algorithm/sort.h:699:10: note:
in instantiation of function template specialization 'std::__sort<(lambda at
sort0.cpp:92:16) &, MemoryIterator>' requested here
std::__sort<_WrappedComp>(std::__unwrap_iter(__first), std::__unwrap_iter(__last), _...

答案1

得分: 3

以下是翻译好的部分：

这里是使std::sort按照您的要求工作所需的 Ingredient：

代理引用

代理引用是用户定义的类型，尽可能模仿C++引用。通过它们，字节数组中的元素实际上是由std::sort移动的。

请注意，对于更健壮的实现，也许执行的操作不仅仅是sort，您还会提供类似于标准库容器中的iterator和const_iterator的ConstRecordReference。

struct RecordReference {
    explicit RecordReference(void* data, size_t element_size):
        _data(data), _size(element_size) {}
    
    RecordReference(Record&);
    
    RecordReference(RecordReference const& rhs):
        _data(rhs.data()), _size(rhs.size()) {}
    
    /// 因为这表示引用，所以赋值代表引用的值的复制，而不是引用的复制
    RecordReference& operator=(RecordReference const& rhs) {
        assert(size() == rhs.size());
        std::memcpy(data(), rhs.data(), size());
        return *this;
    }
    
    RecordReference& operator=(Record const& rhs);
    
    /// 还有`swap`通过引用进行交换
    friend void swap(RecordReference a, RecordReference b) {
        assert(a.size() == b.size());
        size_t size = a.size();
        // 如果您真的很注重标准兼容性，也许不要使用alloca
        auto* buffer = (void*)alloca(size);
        std::memcpy(buffer, a.data(), size);
        std::memcpy(a.data(), b.data(), size);
        std::memcpy(b.data(), buffer, size);
    }
    
    void* data() const { return _data; }
    
    size_t size() const { return _size; }
    
private:
    void* _data;
    size_t _size;
};

/// 在Record的定义之后：

RecordReference::RecordReference(Record& rhs):
    _data(rhs.data()), _size(rhs.size()) {}

RecordReference& RecordReference::operator=(Record const& rhs) {
    assert(size() == rhs.size());
    std::memcpy(data(), rhs.data(), size());
    return *this;
}

用于表示记录的类类型

令人讨厌的是，我们甚至需要这个。我们需要它是因为libstdc++的std::sort实现（可能是任何实现）会创建一些栈分配的值副本，因此我们需要为我们的迭代器的value_type提供一些合理的东西，实际上表示一个记录。

我们提供了内联缓冲区以避免堆分配，直到一个合理的大小。您可以调整InlineSize参数以适应您的记录的预期大小。

struct Record {
    static constexpr size_t InlineSize = 56; 
    
    Record(RecordReference ref): _size(ref.size()) {
        if (is_inline()) {
            std::memcpy(&inline_data, ref.data(), size());
        }
        else {
            heap_data = std::malloc(size());
            std::memcpy(heap_data, ref.data(), size());
        }
    }

    Record(Record&& rhs) noexcept: _size(rhs.size()) {
        if (is_inline()) {
            std::memcpy(&inline_data, &rhs.inline_data, size());
        }
        else {
            heap_data = rhs.heap_data;
            rhs.heap_data = nullptr;
        }
    }
    
    ~Record() {
        if (!is_inline()) {
            std::free(heap_data);
        }
    }
    
    void* data() { return (void*)((Record const*)this)->data(); }
    
    void const* data() const { return is_inline() ? inline_data : heap_data; }
    
    size_t size() const { return _size; }
    
private:
    bool is_inline() const { return _size <= InlineSize; }
    
    size_t _size;
    union {
        char inline_data[InlineSize];
        void* heap_data;
    };
};

然而，似乎至少libc++的std::sort实现一次只会创建一个此类型的对象。这意味着我们可以预先分配一个thread_local缓冲区在栈上保存该单个对象。这样，我们可以防止为任何记录大小分配堆，但这依赖于std::sort的一个实现细节，并且可能在将来的构建中创建运行时错误，因此我不建议这样做。此外，这使得这些类型对许多其他算法都无用。

最后是您的迭代器类

正如注释中所指出的，它需要定义关系比较运算符（以满足_随机访问迭代器_的要求，或者仅仅是因为sort逻辑上需要它们）。

它通过value_type和reference typedefs公开我们的Record和RecordReference类型。

struct MemoryIterator {
    using iterator_category = std::random_access_iterator_tag;
    using difference_type = std::ptrdiff_t;
    using value_type = Record;
    using pointer = void*;
    using reference = RecordReference;

    MemoryIterator(pointer ptr, size_t size)
        : ptr_((char*)ptr)
        , size_(size) {
    }

    reference operator*() const {
        return reference(ptr_, size_);
    }

    reference operator[](size_t index) const {
        return *(*this + index);
    }

    MemoryIterator& operator++() {
        ptr_ += size_;
        return *this;
    }

    MemoryIterator& operator--() {
        ptr_ -= size_;
        return *this;
    }

    MemoryIterator operator++(int) {
        auto tmp = *this;
        ++(*this);
        return tmp;
    }

    MemoryIterator operator--(int) {
        auto tmp = *this;
        --(*this);
        return tmp;
    }

    MemoryIterator& operator+=(size_t n) {
        ptr_ += n * size_;
        return *this;
    }

    MemoryIterator operator+(size_t n) const {
        auto r = *this;
        r += n;
        return r;
    }

    MemoryIterator& operator-=(size_t n) {
        ptr_ -= n * size_;
        return *this;
    }

    MemoryIterator operator-(size_t n) const {
        auto r = *this;
        r -= n;
        return r;
    }
    
    friend bool operator==(const MemoryIterator& a,

<details>
<summary>英文:</summary>

Here are the ingredients we need to make `std::sort` work the way you desire:

## Proxy references

Proxy references are user defined types mimicking C++ references (as much as possible). Through them the elements in the byte array are actually moved around by `std::sort`.

Note that for a more robust implementation that perhaps does more than `sort`, you would also supply a `ConstRecordReference`, akin to `iterator` and `const_iterator` in standard library containers.

struct RecordReference {
explicit RecordReference(void* data, size_t element_size):
_data(data), _size(element_size) {}

RecordReference(Record&);
RecordReference(RecordReference const& rhs):
_data(rhs.data()), _size(rhs.size()) {}
/// Because this represents a reference, an assignment represents a copy of
/// the referred-to value, not of the reference
RecordReference& operator=(RecordReference const& rhs) {
assert(size() == rhs.size());
std::memcpy(data(), rhs.data(), size());
return *this;
}
RecordReference& operator=(Record const& rhs);
/// Also `swap` swaps 'through' the reference
friend void swap(RecordReference a, RecordReference b) {
assert(a.size() == b.size());
size_t size = a.size();
// Perhaps don't use alloca if you're really serious
// about standard conformance
auto* buffer = (void*)alloca(size);
std::memcpy(buffer, a.data(), size);
std::memcpy(a.data(), b.data(), size);
std::memcpy(b.data(), buffer, size);
}
void* data() const { return _data; }
size_t size() const { return _size; }

private:
void* _data;
size_t _size;
};

/// After the definition of Record (see below):

RecordReference::RecordReference(Record& rhs):
_data(rhs.data()), _size(rhs.size()) {}

RecordReference& RecordReference::operator=(Record const& rhs) {
assert(size() == rhs.size());
std::memcpy(data(), rhs.data(), size());
return *this;
}

## Class type to represent your records
It is rather annoying that we even need this. We need it because libstdc++'s `std::sort` implementation (and probably any implementation) creates some stack allocated copies of our values, so we need to supply something sensible for our iterators `value_type` that actually represents a record.
We provide an inline buffer to avoid heap allocations up to a reasonable size. You can tune the `InlineSize` parameter to the expected size of your records.

struct Record {
static constexpr size_t InlineSize = 56;

Record(RecordReference ref): _size(ref.size()) {
if (is_inline()) {
std::memcpy(&inline_data, ref.data(), size());
}
else {
heap_data = std::malloc(size());
std::memcpy(heap_data, ref.data(), size());
}
}
Record(Record&& rhs) noexcept: _size(rhs.size()) {
if (is_inline()) {
std::memcpy(&inline_data, &rhs.inline_data, size());
}
else {
heap_data = rhs.heap_data;
rhs.heap_data = nullptr;
}
}
~Record() {
if (!is_inline()) {
std::free(heap_data);
}
}
void* data() { return (void*)((Record const*)this)->data(); }
void const* data() const { return is_inline() ? inline_data : heap_data; }
size_t size() const { return _size; }

private:
bool is_inline() const { return _size <= InlineSize; }

size_t _size;
union {
char inline_data[InlineSize];
void* heap_data;
};

};

It seems however that at least libc++'s `std::sort` implementation only ever creates one object of this type at a single time. This means that we could preallocate a `thread_local` buffer on the stack to hold that single object. This way we prevent heap allocations for any record size, at the expense of relying on an implementation detail of `std::sort` and potentially creating _runtime_ errors in future builds, so I would not recommend this. Also this renders the types useless for many other algorithms.  
## And finally your iterator class
As noted in the comments it needs to have relational comparison operators defined (to satisfy the _random access iterator_ requirements or just because `sort` logically needs them).
It exposes our `Record` and `RecordReference` types via the `value_type` and `reference` typedefs.

struct MemoryIterator {
using iterator_category = std::random_access_iterator_tag;
using difference_type = std::ptrdiff_t;
using value_type = Record;
using pointer = void*;
using reference = RecordReference;

MemoryIterator(pointer ptr, size_t size)
: ptr_((char*)ptr)
, size_(size) {
}
reference operator*() const {
return reference(ptr_, size_);
}
reference operator[](size_t index) const {
return *(*this + index);
}
MemoryIterator& operator++() {
ptr_ += size_;
return *this;
}
MemoryIterator& operator--() {
ptr_ -= size_;
return *this;
}
MemoryIterator operator++(int) {
auto tmp = *this;
++(*this);
return tmp;
}
MemoryIterator operator--(int) {
auto tmp = *this;
--(*this);
return tmp;
}
MemoryIterator& operator+=(size_t n) {
ptr_ += n * size_;
return *this;
}
MemoryIterator operator+(size_t n) const {
auto r = *this;
r += n;
return r;
}
MemoryIterator& operator-=(size_t n) {
ptr_ -= n * size_;
return *this;
}
MemoryIterator operator-(size_t n) const {
auto r = *this;
r -= n;
return r;
}
friend bool operator==(const MemoryIterator& a, const MemoryIterator& b) {
assert(a.size_ == b.size_);
return a.ptr_ == b.ptr_;
}
friend std::strong_ordering operator<=>(const MemoryIterator& a, const MemoryIterator& b) {
assert(a.size_ == b.size_);
return a.ptr_ <=> b.ptr_;
}
friend difference_type operator-(const MemoryIterator& a, const MemoryIterator& b) {
assert(a.size_ == b.size_);
return (a.ptr_ - b.ptr_) / a.size_;
}

private:
char* ptr_;
size_t size_;
};

[Here you can find a live demo of the code.][1]
[1]: https://godbolt.org/z/aah7z9zcv
</details>
# 答案2
**得分**: 1
这里是一些看起来有效的简单概念验证代码。它通过构建指向原始数据中的“记录”的结构体的第二个向量，然后对其进行排序来实现。通过精心制作合适的拷贝构造函数和拷贝赋值运算符，可以说服`std::sort`执行正确的操作。我不知道这是否是最佳方法，代码肯定可以更好地重构。
好的，让我们开始：
```cpp
#include <iostream>
#include <cstring>
#include <algorithm>
#include <vector>
using std::cout, std::endl;
struct sortable_record
{
sortable_record(int record_size, uint8_t *data, uint8_t *swap_buf) :
record_size(record_size), data(data), swap_buf(swap_buf) { }
sortable_record(const sortable_record &other)
{
data = other.swap_buf;
operator=(other);
}
sortable_record &operator=(const sortable_record &other)
{
record_size = other.record_size;
swap_buf = other.swap_buf;
memcpy(data, other.data, record_size);
return *this;
}
int record_size = 0;
uint8_t *data = nullptr;
uint8_t *swap_buf = nullptr;
};
int main() {
int nrecords = 10, nbytes = 50;
std::vector<uint8_t> data(nrecords * nbytes);
// 生成（简化的）测试数据。
for (int i = 0; i < nrecords; ++i)
{
std::fill(&data[i * nbytes], &data[i * nbytes] + nbytes, 'z' - i);
*(&data[i * nbytes] + nbytes - 1) = 0;
}
// 填充 sortable_record 的向量
std::vector<uint8_t> swap_buf(nbytes);
std::vector<sortable_record> sort_me;
sort_me.reserve(nrecords);
for (int i = 0; i < nrecords; ++i)
sort_me.emplace_back(nbytes, &data[i * nbytes], swap_buf.data());
int key_index = 20;
auto cmp = [key_index](const sortable_record &a, const sortable_record &b) {
uint8_t aval = a.data[key_index];
uint8_t bval = b.data[key_index];
return aval < bval;
};
std::sort(sort_me.begin(), sort_me.end(), cmp);
for (int i = 0; i < nrecords; ++i)
std::cout << (const char *)&data[i * nbytes] << endl;
}

请注意，如果没有sort_me.reserve(nrecords);，会发生糟糕的事情，因此构建sort_me的代码应该全部封装在一个好的工厂函数中。

在线演示

#include <iostream>
#include <cstring>
#include <algorithm>
using std::cout, std::endl;
struct sortable_record
{
sortable_record (int record_size, uint8_t *data, uint8_t *swap_buf) :
record_size (record_size), data (data), swap_buf (swap_buf) { }
sortable_record (const sortable_record &other)
{
data = other.swap_buf;
operator= (other);
}
sortable_record &operator= (const sortable_record &other)
{
record_size = other.record_size;
swap_buf = other.swap_buf;
memcpy (data, other.data, record_size);
return *this;
}
int record_size = 0;
uint8_t *data = nullptr;
uint8_t *swap_buf = nullptr;
};
int main() {
int nrecords = 10, nbytes = 50;
std::vector<uint8_t> data(nrecords * nbytes);
// Generate (simplified!) data for testing.
for (int i = 0; i < nrecords; ++i)
{
std::fill (&data [i * nbytes], &data [i * nbytes] + nbytes, 'z' - i);
*(&data [i * nbytes] + nbytes - 1) = 0;
}
// Populate vector of sortable_record's
std::vector<uint8_t> swap_buf (nbytes);
std::vector<sortable_record> sort_me;
sort_me.reserve (nrecords);
for (int i = 0; i < nrecords; ++i)
sort_me.emplace_back (nbytes, &data [i * nbytes], swap_buf.data ());
int key_index = 20;
auto cmp = [key_index](const sortable_record &a, const sortable_record &b) {
uint8_t aval = a.data [key_index];
uint8_t bval = b.data [key_index];
return aval < bval;
};
std::sort (sort_me.begin(), sort_me.end(), cmp);
for (int i = 0; i < nrecords; ++i)
std::cout << (const char *) &data [i * nbytes] << endl;
}