Wie erstelle ich einen richtigen Speicherpool für einen Multithread-Vektor (LFSV)?

Question

Unten ist eine alte Übung für eine Klasse, die nicht mehr an meiner Universität unterrichtet wird (Parallel Processing). Das Ziel besteht darin, eine Speicherbank zu erstellen und zu verwenden, um die Lock-Free Sorted Vector-Implementierung zu beschleunigen. Ich habe die Memory Bank selbst implementiert und das Ziel ist es, genug Speicher zur Verfügung zu stellen, damit ich nicht neue oder löschen im LFSV verwenden muss. Ich glaube, ich brauche eine Get() - Funktion, die die Adresse des Speichers zurückgibt (nicht sicher, wie der unbenutzte Speicher verfolgt wird), und Store sollte den Speicher freigeben (irgendwie als unbenutzt markieren).

Innerhalb LFSV (was vor meiner Intervention völlig in Ordnung war), erklärt die Übung, dass ich die neue ersetzen und löschen mit neuen Ersatz und Speicher (Speicher wollen wir befreit). Wie gehe ich vor beim Erstellen der Get (wenn dies falsch ist) oder der Store-Funktion, um wie eine richtige Speicherbank zu funktionieren? Ich werde auch irgendwelche Referenz- oder Speicherbankbeispiele online nehmen, von denen Sie vielleicht wissen, dass ich Probleme habe, gute Ressourcen in Bezug auf Speicherbänke und Multithreading zu finden.

Es gibt keine Fehler in diesem Programm, aber es kehrt als "FAIL" zurück, da ich die Speicherbank nicht richtig verwaltet habe.

#include <algorithm>//copy, random_shuffle 
#include <ctime> //std::time (NULL) to seed srand 
#include <iostream>  // std::cout 
#include <atomic>   // std::atomic 
#include <thread>   // std::thread 
#include <vector>   // std::vector 
#include <mutex>   // std::mutex 
#include <deque>   // std::deque 

class MemoryBank 
{ 
    std::deque< std::vector<int>* > slots; 
public: 
    MemoryBank() : slots(10000) 
    { 
     for (int i = 0; i<10000; ++i) 
     { 
      slots[i] = reinterpret_cast<std::vector<int>*>(new char[sizeof(std::vector<int>)]); 
     } 
    } 
    ~MemoryBank() 
    { 
     for (unsigned int i = 0; i < slots.size(); ++i) 
     { 
      delete slots[i]; 
     } 
     slots.clear(); 
    } 
    void * Get() 
    { 
     return &slots; 
    } 
    void Store(std::vector<int *> freeMemory) 
    { 
     return; 
    } 
}; 

class LFSV { 
    std::atomic< std::vector<int>* > pdata; 
    std::mutex wr_mutex; 
    MemoryBank mb; 

    public: 

    LFSV() : mb(), pdata(new (mb.Get()) std::vector<int>) {} 

    ~LFSV() 
    { 
     mb.~MemoryBank(); 
    } 

    void Insert(int const & v) { 
     std::vector<int> *pdata_new = nullptr, *pdata_old; 
     int attempt = 0; 
     do { 
      ++attempt; 
      delete pdata_new; 
      pdata_old = pdata; 
      pdata_new = new (mb.Get())std::vector<int>(*pdata_old); 

      std::vector<int>::iterator b = pdata_new->begin(); 
      std::vector<int>::iterator e = pdata_new->end(); 
      if (b==e || v>=pdata_new->back()) { pdata_new->push_back(v); } //first in empty or last element 
      else { 
       for (; b!=e; ++b) { 
        if (*b >= v) { 
         pdata_new->insert(b, v); 
         break; 
        } 
       } 
      } 
//   std::lock_guard<std::mutex> write_lock(wr_mutex); 
//   std::cout << "insert " << v << "(attempt " << attempt << ")" << std::endl; 
     } while (!(this->pdata).compare_exchange_weak(pdata_old, pdata_new )); 
     // LEAKing pdata_old since "delete pdata_old;" will cause errors 


//  std::lock_guard<std::mutex> write_lock(wr_mutex); 
//  std::vector<int> * pdata_current = pdata; 
//  std::vector<int>::iterator b = pdata_current->begin(); 
//  std::vector<int>::iterator e = pdata_current->end(); 
//  for (; b!=e; ++b) { 
//   std::cout << *b << ' '; 
//  } 
//  std::cout << "Size " << pdata_current->size() << " after inserting " << v << std::endl; 
    } 

    int const& operator[] (int pos) const { 
     return (*pdata)[ pos ]; 
    } 
}; 

LFSV lfsv; 

void insert_range(int b, int e) { 
    int * range = new int [e-b]; 
    for (int i=b; i<e; ++i) { 
     range[i-b] = i; 
    } 
    std::srand(static_cast<unsigned int>(std::time (NULL))); 
    std::random_shuffle(range, range+e-b); 
    for (int i=0; i<e-b; ++i) { 
     lfsv.Insert(range[i]); 
    } 
    delete [] range; 
} 

int reader(int pos, int how_many_times) { 
    int j = 0; 
    for (int i=1; i<how_many_times; ++i) { 
     j = lfsv[pos]; 
    } 
    return j; 
} 

std::atomic<bool> doread(true); 

void read_position_0() { 
    int c = 0; 
    while (doread.load()) { 
     std::this_thread::sleep_for(std::chrono::milliseconds(10)); 
     if (lfsv[0] != -1) { 
      std::cout << "not -1 on iteration " << c << "\n"; // see main - all element are non-negative, so index 0 should always be -1 
     } 
     ++c; 
    } 
} 

void test(int num_threads, int num_per_thread) 
{ 
    std::vector<std::thread> threads; 
    lfsv.Insert(-1); 
    std::thread reader = std::thread(read_position_0); 

    for (int i=0; i<num_threads; ++i) { 
     threads.push_back(std::thread(insert_range, i*num_per_thread, (i+1)*num_per_thread)); 
    } 
    for (auto& th : threads) th.join(); 

    doread.store(false); 
    reader.join(); 

    for (int i=0; i<num_threads*num_per_thread; ++i) { 
     //  std::cout << lfsv[i] << ' '; 
     if (lfsv[i] != i-1) { 
      std::cout << "Error\n"; 
      return; 
     } 
    } 
    std::cout << "All good\n"; 
} 

void test0() { test(1, 100); } 
void test1() { test(2, 100); } 
void test2() { test(8, 100); } 
void test3() { test(100, 100); } 

void (*pTests[])() = { 
    test0,test1,test2,test3//,test4,test5,test6,test7 
}; 


#include <cstdio> /* sscanf */ 
int main(int argc, char ** argv) { 
    if (argc==2) { //use test[ argv[1] ] 
     int test = 0; 
     std::sscanf(argv[1],"%i",&test); 
     try { 
      pTests[test](); 
     } catch(const char* msg) { 
      std::cerr << msg << std::endl; 
     } 
     return 0; 
    } 
} 

Answer 1

reinterpret_cast ist wirklich ein "Ich weiß, was ich tue, vertrau mir" Cast. Der Compiler wird Ihnen - wenn möglich - glauben.

In diesem Fall ist es jedoch völlig falsch. new char[] tut nicht geben Sie eine vector<int>* zurück.