free()：下一个尺寸无效（快速）：使用时 posix_memalign

Question

我使用 posix_memalign() 来分配放置我的指针的地方，但是当我尝试释放这个地方时，会出现这样的情况：

*** `/home/liqiaochu/lockless_rb/test_lockless_rb' 中的错误：free()：下一个大小无效（快速）：0x000000000060b000 ***

我的代码正在运行：当调用 rb.destroy() 时，它将中止

lockless_rb.h

#pragma once
#include <stdint.h>
#include <atomic>

enum rb_sync_type
{
    RB_SYNC_ST, // single thread 
    RB_SYNC_MT, // multi thread
};

struct rb_headtail
{
    volatile std::atomic<uint64_t> head{0};
    volatile std::atomic<uint64_t> tail{0};
};

class lockless_rb
{
    public:
        // ret：success：0 failed：-1
        int create(int count);
        void destroy();
        int enqueue_one_elem(void* elem, int rb_sync_type);
        int dequeue_one_elem(void** elem, int rb_sync_type);

    private:
        // one blk a time
        int __move_prod_head(uint64_t* old_head, int rb_sync_type);
        void __enqueue_one_elem_data(uint64_t old_head, void* elem);
        void __update_prod_tail(uint64_t old_head, int rb_sync_type);

        int __move_cons_head(uint64_t* old_head, int rb_sync_type);
        void __dequeue_one_elem_data(uint64_t old_head, void** elem);
        void __update_cons_tail(uint64_t old_head, int rb_sync_type);

        void* ptr_mem_ = nullptr;
        rb_headtail prod_;
        rb_headtail cons_;
        uint64_t size_ = 0;
        uint64_t mask_ = 0;
};

lockless_rb.cpp

#include "include/lock_less_rb.h"

#include <stdio.h>
#include <stdlib.h>
#include <emmintrin.h>

int lockless_rb::create(int count)
{
    int ret = 0;
    size_t page_sz = 4096;
    size_t page_cnt = (count * sizeof(void*)) / page_sz + 1;
    ret = posix_memalign(&ptr_mem_, page_sz, page_cnt);
    printf("%d\n", ret);

    // ptr_mem_ = malloc(sizeof(void*) * count);
    // if (nullptr == ptr_mem_)
    //     return -1;
    size_ = count;
    mask_ = count - 1;
    return 0;
}

void lockless_rb::destroy()
{
    free(ptr_mem_);
}

int lockless_rb::enqueue_one_elem(void* elem, int rb_sync_type)
{
    uint64_t prod_head;
    int n = __move_prod_head(&prod_head, rb_sync_type);
    if (0 == n)
        return -1;
    __enqueue_one_elem_data(prod_head, elem);
    __update_prod_tail(prod_head, rb_sync_type);
    return 0;
}

int lockless_rb::dequeue_one_elem(void** elem, int rb_sync_type)
{
    uint64_t cons_head;
    int n = __move_cons_head(&cons_head, rb_sync_type);
    if (0 == n)
        return -1;
    __dequeue_one_elem_data(cons_head, elem);
    __update_cons_tail(cons_head, rb_sync_type);
    return 0;
}

// private：
int lockless_rb::__move_cons_head(uint64_t* old_head, int rb_sync_type)
{
    bool success = false;
    int free_entries = 0;
    while (success == false) {
        *old_head = cons_.head;
        asm volatile ("" : : : "memory");
        free_entries = (prod_.tail - *old_head);
        if (free_entries < 1)
            return 0; // success read nums
        if (rb_sync_type == 0) {
            cons_.head = *old_head + 1;
            asm volatile ("" : : : "memory");
            success = true;
        } else {
            success = cons_.head.compare_exchange_weak(*old_head, *old_head + 1);
        }
    }
    return 1;
}

void lockless_rb::__dequeue_one_elem_data(uint64_t old_head, void** elem)
{
    uint64_t i;
    uint64_t idx = old_head & mask_;
    uint64_t* ptr_mem64 = (uint64_t*)ptr_mem_;
    uint64_t* elem_64 = (uint64_t*)(elem);
    *elem = reinterpret_cast<void*>(ptr_mem64[idx]);
}

void lockless_rb::__update_cons_tail(uint64_t old_head, int rb_sync_type)
{
    asm volatile ("" : : : "memory");
    if (0 != rb_sync_type) {
        while (cons_.tail != old_head) { // faster than while(1)
            _mm_pause();
        }
    }
    cons_.tail = old_head + 1;
}

int lockless_rb::__move_prod_head(uint64_t* old_head, int rb_sync_type)
{
    bool success = false;
    int free_entries = 0;
    while (success == false) {
        *old_head = prod_.head;
        asm volatile ("" : : : "memory");
        free_entries = (size_ + cons_.tail - *old_head);
        if (free_entries < 1)
            return 0;
        if (rb_sync_type == 0) {
            prod_.head = *old_head + 1;
            asm volatile ("" : : : "memory");
            success = true;
        } else {
            success = prod_.head.compare_exchange_weak(*old_head, *old_head + 1);
        }
    }
    return 1;
}

void lockless_rb::__enqueue_one_elem_data(uint64_t old_head, void* elem)
{
    uint64_t i;
    uint64_t idx = old_head & mask_;
    uint64_t* ptr_mem64 = (uint64_t*)ptr_mem_;
    ptr_mem64[idx] = reinterpret_cast<uint64_t>(elem);
}

void lockless_rb::__update_prod_tail(uint64_t old_head, int rb_sync_type)
{
    asm volatile ("" : : : "memory");
    if (0 != rb_sync_type) {
        while (prod_.tail != old_head) {
            _mm_pause();
        }
    }
    prod_.tail = old_head + 1;
}

测试代码：

#include "include/lock_less_rb.h"

#include <thread>
#include <vector>
#include <iostream>

int main()
{
    lockless_rb rb;
    rb.create(4); // rb size = 4

    int cons_num = 2;
    int enqueue_num = 0;
    int loop_times = 100;
    std::atomic_int dequeue_num(0);
    char msg[20] = "hello_world";

    std::vector<std::thread> thrs;
    for (auto i = 0; i < cons_num; i++) {
        thrs.emplace_back([&]{
            while (1) {
                if (loop_times == dequeue_num) {
                    printf("dequeue %d\n", dequeue_num.load());
                    break; 
                }                
                void* msg1;
                if (rb.dequeue_one_elem(&msg1, rb_sync_type::RB_SYNC_MT) != 0)
                    continue;
                else {
                    dequeue_num++;
                }
            }
        });
    }

    thrs.emplace_back([&]{
        while (1) {
            if (loop_times == enqueue_num) {
                printf("enqueue %d\n", enqueue_num);
                break;
            }
            if (rb.enqueue_one_elem(msg, rb_sync_type::RB_SYNC_ST) != 0)
                continue;
            else {
                // printf("%s, %d\n", (char*)(msg), enqueue_num);
                enqueue_num++;
            }
        }
    });

    for (auto& t:thrs) {
        t.join();
    }

    rb.destroy();
    return 0;
}

谢谢！！！

Answer 1

在这次通话中：

rb.create(4);

create成员函数的参数count的值为4。根据

size_t page_cnt = (count * sizeof(void*)) / page_sz + 1;

page_cnt 的值为 1。在

ret = posix_memalign(&ptr_mem_, page_sz, page_cnt);

你要求posix_memalign分配page_cnt字节，因此，只是1个字节，根据documentation:

int posix_memalign(void **memptr, size_t alignment, size_t size);

The function posix_memalign() allocates size bytes...

您不想分配 page_cnt 页吗？如果是这样，请将调用更改为：

ret = posix_memalign(&ptr_mem_, page_sz, page_cnt * page_sz);

free()：下一个尺寸无效（快速）：使用时 posix_memalign

free(): invalid next size (fast): when use posix_memalign

c++

linux

lockless