How Do I Make This Container Thread-Safe?

[MutantJohn]

So, I've attempted to recreate CGAL's Compact_Container. I haven't looked at their source, really. I'm just trying to go by what's in their papers. You can find a pretty good explanation of the container in Section 3.

A quick explanation of the container is basically, we maintain a linked list of blocks. A block is a contiguous allocation of elements. An element is a node that contains user data and a pointer.

The pointer of the element determines its state. Because of some weird architecture hocus pocus, I guess all addresses are 4-byte multiples so the last two bits are guaranteed to be 0 and we can use those two bits to store the state.

00 = occupied
10 = free
11 = boundary

So, when a container wants to write an element, it uses a free list. When a new block is allocated and added into the linked list, it creates a blank array and the elements of the new block can be appended to the free list.

The free list is really a singly-linked list. If an element contains a pointer whose last two bits are 10, the "actual"value of the pointer is the next element in the free list.

So when we insert into the container, we really just pop off the head and continue on. If we clear the queue, we allocate a new block which adds to the free list again.

Now, in that paper they claim they make it thread-safe by just making sure each thread has its own free list. I'm curious how to implement this. I want to make the container transparent to the user.

Is it possible to get the id of the currently running thread? I was thinking that if I could id the currently executing thread, I'd be able to manage whether or not it needs a free list or it already has one.

Adding blocks will be an atomic operation but that shouldn't hinder any performance because of how blocks are added to the list.

I was also wondering if I could get any feedback on my attempt at a partial implementation of the container. Right now, you can only insert elements but I did make a draft of an iterator and I think my way of managing the free list and adding blocks is working too.

#ifndef CONTAINER_H_
#define CONTAINER_H_


#include <iostream>
#include <vector>
#include <array>
#include <cassert>
#include <memory>
#include <utility>
#include <cstddef>
#include <list>





/*  This code is an attempt at replicating CGAL's memory
 *  container. It's probably not as good.
 *
 */


template<class T> class Element;
template<class T> class Block;
template<class T> class Container;
template<class T> class ContainerIterator;


/*


    For an element, the last two bits of the next_
    pointer define the state


    0 = 00 = occupied
    2 = 10 = free
    3 = 11 = block boundary


    The value of the non-masked pointer is then
    equal to either the next pointer in the free list
    or a pointer to the start of another block


 */


template<class T>
class Element {
  public:
    T data;


  private:
    Element *next_;


  public:
    Element(void) {
      next_ = (Element* ) 2;
      new(&data) T();
    }


    Element(T cpy) {
      next_ = (Element* ) 2;
      new(&data) T(cpy);
    }


    ~Element(void) {
      data.~T();
    }


    void SetNext(Element *ptr) {
      size_t p = (size_t ) ptr | ((size_t ) next_ & 3);
      next_ = (Element* ) p;
    }


    void MakeNull(void) {
      next_ = nullptr;
    }


    Element* GetNext(void) {
      size_t n = (size_t ) next_ & ~3;
      return (Element* ) n;
    }


    void MakeBoundary(void) {
      size_t n = (size_t ) next_ | 3;
      next_ = (Element* ) n;
    }


    void MakeBoundary(Element* other) {
      size_t n = (size_t ) other | 3;
      next_ = (Element* ) n;
    }


    bool IsBoundary(void) const {
      return ((size_t ) next_ & 3) == 3;
    }


    void MakeOccupied(void) {
      size_t n = (size_t ) next_ & ~3;
      next_ = (Element* ) n;
    }


    bool IsOccupied(void) const {
      return ((size_t ) next_ & 3) == 0;
    }
};


template<class T>
class Block {
  private:
    Element<T> *elements_;
    size_t size_;


    void init(size_t num_elems) {
      elements_ = (Element<T>* ) malloc(num_elems * sizeof(*elements_));
      size_ = num_elems;


      elements_[0].MakeNull();
      elements_[size_ - 1].MakeNull();


      elements_[0].MakeBoundary();
      elements_[size_ - 1].MakeBoundary();
    }


  public:
    Block(size_t num_elems) {
      init(num_elems);


      for (size_t i = 1; i < (size_ - 1); ++i) {
        new(elements_ + i) Element<T>();
        elements_[i].SetNext(elements_ + i + 1);
      }


      elements_[size_ - 2].SetNext(nullptr);
    }


    Block(size_t num_elems, const T data) {
      init(num_elems);
      for (size_t i = 1; i < size_ - 1; ++i) {
        new(elements_ + i) Element<T>(data);
      }
      elements_[size_ - 2].SetNext(nullptr);
    }


    ~Block(void) {
      for (size_t i = 0; i < size_; ++i) {
        elements_[i].~Element<T>();
      }
      free(elements_);
    }


    Element<T>* Data(const size_t idx) {
      return elements_ + idx;
    }


    size_t size(void) const {
      return size_;
    }
};


template<class T>
class Container {
  public:
    typedef ContainerIterator<T> iterator;


  private:
    friend class ContainerIterator<T>;


    std::list<Block<T>*> blocks_;


    Element<T> *first_;
    Element<T> *last_;


    Element<T> *free_list_;


    size_t block_size_;


    void PushBack(void) {
      block_size_ += 16;


      Block<T> *new_block = new Block<T>(block_size_);
      auto last = blocks_.back();


      blocks_.push_back(new_block);    
      last->Data(last->size() - 1)->SetNext(blocks_.back()->Data(0));
      blocks_.back()->Data(0)->SetNext(last->Data(last->size() - 1));


      free_list_ = blocks_.back()->Data(1);  
      last_ = blocks_.back()->Data(block_size_ - 1);
    }


  public:
    Container(void) {
      block_size_ = 16;


      Block<T> *new_block = new Block<T>(block_size_);


      blocks_.push_back(new_block);
      Block<T>* &first_block = blocks_.front();


      free_list_ = first_block->Data(1);
      first_ = first_block->Data(0);
      last_ = first_block->Data(first_block->size() - 1);
    }


    ~Container(void) {
      blocks_.remove_if([](Block<T> *b) {
          delete b;
          return true;
      });
    }


    void Insert(T tmp) {
      if (free_list_ == nullptr) {
        PushBack();
      }


      free_list_->data = tmp;
      free_list_->MakeOccupied();
      free_list_ = free_list_->GetNext();
    }


    ContainerIterator<T> Begin(void) {
      auto tmp = ContainerIterator<T>(*this);
      tmp.SetToFirst();
      tmp.FindNextOccupied();
      return tmp;
    }


    ContainerIterator<T> End(void) {
      auto tmp = ContainerIterator<T>(*this);
      tmp.SetToLast();
      return tmp;
    }
};


template<class T>
class ContainerIterator {
  private:
    Container<T> &container_;
    Element<T> *element_;
    Element<T> *prev_;


  public:
    ContainerIterator(Container<T> &container) : container_(container)  {
      element_ = nullptr;
      prev_ = nullptr;
    }


    void SetToFirst(void) {
      element_ = container_.first_;
      assert(element_->GetNext() == nullptr);
    }


    void SetToLast(void) {
      element_ = container_.last_;
      assert(element_->GetNext() == nullptr);
      assert(container_.last_ != container_.first_);
    }


    void FindNextOccupied(void) {
      while (!element_->IsOccupied() || element_ == prev_) {
        if (element_->IsBoundary()) {
          auto next = element_->GetNext();


          if (next == nullptr) {
            if (element_ == container_.last_) {
              return;
            } else {
              ++element_;
              continue;
            }
          } else {
            element_ = next;
          }
        }


        ++element_;
      }
      prev_ = element_;
    }


    void operator++(void) {
      FindNextOccupied();
    }


    void FindPrevOccupied(void) {
      while (!element_->IsOccupied()) {


      }
    }


    void operator--(void) {
      FindPrevOccupied();
    }


    bool operator==(const ContainerIterator &other) const {
      return element_ == other.element_;
    }


    bool operator!=(const ContainerIterator &other) const {
      return !(element_ == other.element_);
    }


    T& operator*(void) {
      return element_->data;
    }
};


#endif // CONTAINER_H_

#include "include/container.h"


int main(void) {
  Container<int> container;


  for (int i = 0; i < 128; ++i) {
    container.Insert(i);
  }


  for (auto it = container.Begin(); it != container.End(); ++it) {
    std::cout << *it << std::endl;
  }


  return 0;
}

[MutantJohn]

I've updated the code to now support deleting elements and bi-directional iteration.

The interface isn't as clean as I was hoping though.

If you want to do a full span over the container in reverse, you need :

for (auto it = container.rEnd(); it !=rBegin(); --it)
    // ...

But just so long as the iterator is kept in the range between rBegin() and End(), the iterator can be incremented and decremented safely without any repercussions. I think. I think the nomenclature is confusing... It's tough designing these interfaces!

It's because I want Begin() to start somewhere useful and End() needs to be the actual end-of-data and vice versa hence rBegin() and rEnd(). rEnd() starts you somewhere useful while rBegin() is the absolute end-of-data.

Update code:

#ifndef CONTAINER_H_
#define CONTAINER_H_


#include "globals.h"


/*  This code is an attempt at replicating CGAL's memory
 *  container. It's probably not as good.
 *
 */


template<class T> class Element;
template<class T> class Block;
template<class T> class Container;
template<class T> class ContainerIterator;


/*


    For an element, the last two bits of the next_
    pointer define the state


    0 = 00 = occupied
    2 = 10 = free
    3 = 11 = block boundary


    The value of the non-masked pointer is then
    equal to either the next pointer in the free list
    or a pointer to the start of another block


 */


template<class T>
class Element {
  public:
    T data;


  private:
    Element *next_;


  public:
    Element(void) {
      next_ = (Element* ) 2;
      new(&data) T();
    }


    Element(T cpy) {
      next_ = (Element* ) 2;
      new(&data) T(cpy);
    }


    ~Element(void) {
      data.~T();
    }


    void SetNext(Element *ptr) {
      size_t p = (size_t ) ptr | ((size_t ) next_ & 3);
      next_ = (Element* ) p;
    }


    void MakeNull(void) {
      next_ = nullptr;
    }


    Element* GetNext(void) {
      size_t n = (size_t ) next_ & ~3;
      return (Element* ) n;
    }


    void MakeBoundary(void) {
      size_t n = (size_t ) next_ | 3;
      next_ = (Element* ) n;
    }


    void MakeBoundary(Element* other) {
      size_t n = (size_t ) other | 3;
      next_ = (Element* ) n;
    }


    bool IsBoundary(void) const {
      return ((size_t ) next_ & 3) == 3;
    }


    void MakeOccupied(void) {
      size_t n = (size_t ) next_ & ~3;
      next_ = (Element* ) n;
    }


    bool IsOccupied(void) const {
      return ((size_t ) next_ & 3) == 0;
    }


    void MakeFree(void) {
      next_ = (Element* ) 2;
      std::memset(&data, 0, sizeof(T));
    }


    bool IsFree(void) const {
      return ((size_t ) next_ & 3) == 2;
    }
};


template<class T>
class Block {
  private:
    Element<T> *elements_;
    size_t size_;


    void init(size_t num_elems) {
      elements_ = (Element<T>* ) malloc(num_elems * sizeof(*elements_));
      size_ = num_elems;


      elements_[0].MakeNull();
      elements_[size_ - 1].MakeNull();


      elements_[0].MakeBoundary();
      elements_[size_ - 1].MakeBoundary();
    }


  public:
    Block(size_t num_elems) {
      init(num_elems);


      for (size_t i = 1; i < (size_ - 1); ++i) {
        new(elements_ + i) Element<T>();
        elements_[i].SetNext(elements_ + i + 1);
      }


      elements_[size_ - 2].SetNext(nullptr);
    }


    Block(size_t num_elems, const T data) {
      init(num_elems);
      for (size_t i = 1; i < size_ - 1; ++i) {
        new(elements_ + i) Element<T>(data);
      }
      elements_[size_ - 2].SetNext(nullptr);
    }


    ~Block(void) {
      for (size_t i = 0; i < size_; ++i) {
        elements_[i].~Element<T>();
      }
      free(elements_);
    }


    Element<T>* Data(const size_t idx) {
      return elements_ + idx;
    }


    size_t size(void) const {
      return size_;
    }
};


template<class T>
class Container {
  public:
    typedef ContainerIterator<T> iterator;


  private:
    friend class ContainerIterator<T>;


    std::list<Block<T>*> blocks_;


    Element<T> *first_;
    Element<T> *last_;


    Element<T> *free_list_;


    size_t block_size_;


    void PushBack(void) {
      block_size_ += 16;


      Block<T> *new_block = new Block<T>(block_size_);
      auto last = blocks_.back();


      blocks_.push_back(new_block);    
      last->Data(last->size() - 1)->SetNext(blocks_.back()->Data(0));
      blocks_.back()->Data(0)->SetNext(last->Data(last->size() - 1));


      free_list_ = blocks_.back()->Data(1);  
      last_ = blocks_.back()->Data(block_size_ - 1);
    }


  public:
    Container(void) {
      block_size_ = 16;


      Block<T> *new_block = new Block<T>(block_size_);


      blocks_.push_back(new_block);
      Block<T>* &first_block = blocks_.front();


      free_list_ = first_block->Data(1);
      first_ = first_block->Data(0);
      last_ = first_block->Data(first_block->size() - 1);
    }


    ~Container(void) {
      blocks_.remove_if([](Block<T> *b) {
          delete b;
          return true;
      });
    }


    void Insert(T tmp) {
      if (free_list_ == nullptr) {
        PushBack();
      }


      free_list_->data = tmp;
      free_list_->MakeOccupied();
      free_list_ = free_list_->GetNext();
    }


    void Delete(iterator pos) {
      auto ptr = pos.element();
      ptr->MakeFree();
      ptr->SetNext(free_list_);
      free_list_ = ptr;
    }


    iterator Begin(void) {
      auto tmp = iterator(*this);
      tmp.SetToFirst();
      tmp.FindNextOccupied();
      return tmp;
    }


    iterator End(void) {
      auto tmp = iterator(*this);
      tmp.SetToLast();
      return tmp;
    }


    iterator rBegin(void) {
      auto tmp = iterator(*this);
      tmp.SetToFirst();
      return tmp;
    }


    iterator rEnd(void) {
      auto tmp = iterator(*this);
      tmp.SetToLast();
      tmp.FindPrevOccupied();
      return tmp;
    }
};


template<class T>
class ContainerIterator {
  private:
    Container<T> &container_;
    Element<T> *element_;
    Element<T> *prev_;


  public:
    ContainerIterator(Container<T> &container) : container_(container) {
      element_ = nullptr;
      prev_ = nullptr;
    }


    Element<T>* element(void) {
      return element_;
    }


    void SetToFirst(void) {
      element_ = container_.first_;
      assert(element_->GetNext() == nullptr);
    }


    void SetToLast(void) {
      element_ = container_.last_;
      assert(element_->GetNext() == nullptr);
      assert(container_.last_ != container_.first_);
    }


    void FindNextOccupied(void) {
      while (!element_->IsOccupied() || element_ == prev_) {
        if (element_->IsBoundary()) {
          auto next = element_->GetNext();


          if (next == nullptr) {
            if (element_ == container_.last_) {
              return;
            } else {
              ++element_;
              continue;
            }
          } else {
            element_ = next;
          }
        }


        ++element_;
      }
      prev_ = element_;
    }


    void operator++(void) {
      FindNextOccupied();
    }


    void FindPrevOccupied(void) {
      while (!element_->IsOccupied() || element_ == prev_) {
        if (element_->IsBoundary()) {
          auto next = element_->GetNext();


          if (next == nullptr) {
            if (element_ == container_.first_) {
              return;
            } else {
              --element_;
              continue;
            }
          } else {
            element_ = next;
          }
        }


        --element_;
      }
      prev_ = element_;
    }


    void operator--(void) {
      FindPrevOccupied();
    }


    bool operator==(const ContainerIterator &other) const {
      return element_ == other.element_;
    }


    bool operator!=(const ContainerIterator &other) const {
      return !(element_ == other.element_);
    }


    T& operator*(void) {
      return element_->data;
    }


    T* operator->() {
      return &(element_->data);
    }
};


#endif // CONTAINER_H_

[Codeplug]

Why does Element use placement new like that?
Block using malloc/free and placement new also seems strange.

I'm no expert on rvalue references or move semantics, but it seems some of that could be used here.

gg

[Codeplug]

struct gg
{
    gg() { std::cout << 'C'; }
    ~gg() { std::cout << 'D'; }
};

int main(void) {
    Container<gg> container;
    container.Insert(gg());
    return 0;
}

CCCCCCCCCCCCCCCCCCCCCCCCCCCCCDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDDD

Ouch.

gg

[MutantJohn]

Ouch? Why is that ouch? XD

Regardless, I see what you mean... Here's an updated version.

I think you're also onto something with the move semantics. I hadn't even though about that... It might be a good performance boost, actually. Thanks for the advice!

Edit :

#include "include/container.h"


struct gg {
  int *data;


  gg(void) {
    std::cout << "C" << std::endl;
    data = new int();
  }
  ~gg(void) {
    std::cout << "D" << std::endl;
    delete data;
  }
};


int main(void) {
  Container<gg> c;


  c.Insert(gg());


  return 0;
}

valgrind:

==15684== Memcheck, a memory error detector
==15684== Copyright (C) 2002-2013, and GNU GPL'd, by Julian Seward et al.
==15684== Using Valgrind-3.10.0.SVN and LibVEX; rerun with -h for copyright info
==15684== Command: ./bin/regulus
==15684== 
C
D
==15684== 
==15684== HEAP SUMMARY:
==15684==     in use at exit: 0 bytes in 0 blocks
==15684==   total heap usage: 4 allocs, 4 frees, 300 bytes allocated
==15684== 
==15684== All heap blocks were freed -- no leaks are possible

[Codeplug]

I get CDD.

I would remove the memset and the explicit destructor in Element.

gg

[King Mir]

Now, in that paper they claim they make it thread-safe by just making sure each thread has its own free list. I'm curious how to implement this. I want to make the container transparent to the user.

You make it a thread local variable, with c++11's thread_local storage specifier. It's not transparent to the user; each thread has only one copy of the container, and does not see insertion from other threads.

If you want to implement a real thread safe container, don't start by optimizing the serial case, because what you really need to optimize is the synchronization.

[MutantJohn]

I get CDD.

I would remove the memset and the explicit destructor in Element.

gg

Are you also getting a double free then? I swear, I only saw the destructor being called once in my code. Maybe I posted an old version...

[phantomotap]

Here's an updated version.

O_o

template<class T>
class Element {
  public:
    T data;

  private:
    Element *next_;

  public:
    Element(void) {
      next_ = (Element* ) 2;
      std::memset(&data, 0, sizeof(T));
    }

Nope.

    ~Element(void) {
         std::cout << "overwriting " << &data << "\n";
      data.~T();
    }

Nope.

template<class T>
class Block {
  private:
    Element<T> *elements_;
    size_t size_;

    void init(size_t num_elems) {
      elements_ = (Element<T>* ) malloc(num_elems * sizeof(*elements_));
      size_ = num_elems;

      elements_[0].MakeNull();
      elements_[size_ - 1].MakeNull();

      elements_[0].MakeBoundary();
      elements_[size_ - 1].MakeBoundary();
    }

  public:
    Block(size_t num_elems) {
      init(num_elems);

      for (size_t i = 1; i < (size_ - 1); ++i) {
        elements_[i].MakeFree();
        elements_[i].SetNext(elements_ + i + 1);
      }

      elements_[size_ - 2].SetNext(nullptr);
    }

Nope.

    ~Block(void) {
      free(elements_);
    }

Nope.

Your code is terrible. I could continue for some time.

Your code is terrible in several different ways!

You are not manually calling constructors when should be manually calling constructors because you didn't use the `new` operator.

You are overwriting memory which doesn't belong to you.

You would be overwriting still more memory which doesn't belong to you, but the memory you overwrite is incorrectly never initialized because you failed to manually call still more constructors.

You fail to account for `malloc` failing.

You fail to account for `new` failing.

You fail to account for the constructors failing when you do manually call constructors.

You would fail to manually unwind some constructed elements, but you failed to construct the relevant elements in any event.

You perform operations which are lost in the event of exceptions in multiple places leaving the object in any inconsistent state.

I'm going to stop before I start crying.

You should toss the garbage you have written. You should then start by working on only one component until you actually have a working implementation.

Elysia and I, probably several others, have told you time and again not to play with `placement new` and relevant problems until you have more experience with idioms and patterns which prevent or at least help solve the issues being discussed.

At the very least, you should never be writing a "naked" `malloc` or `free` call; you need an RAII component to help you get things right.

Soma

[MutantJohn]

Well then...

Would you mind elaborating on some of the problem areas you noticed?

[phantomotap]

Would you mind elaborating on some of the problem areas you noticed?

O_o

I don't really have the time what with trying to get ahead on work before "Fallout 4" is released in my area.

I'd spend more time explaining the problems than just fixing the code, and you already know I'm not just going to fix your code.

On the other hand, I didn't post those snippets in particular by accident; if you spend a few minutes walking through each snippet, you should understand.

Soma

[Dave11]

you don't need to call new and the destructor , they are called automatically.

when your class contain inner objects, their constructor is called automatically on the constructor of the object that wraps them , and their destructor is called when the destructor of the wrapper object is called.

you can override this behaviour if the contained objects doesn't have default constructor or you want to intialize them with other constructor then the default.
in any chance you dont use placement new.

the constructor which gets T as argument, it is also plain wrong. the correct way is either copy the variable, wither move it. but anyway, not need for placement new here :

Element (const T& rhs) : t(rhs.t) , next(nullptr){}
Element (T&& rhs) : t(std::move(rhs.t)) , next(nullptr) { //safely invalidate rhs.t in some way }

next, the line

(Element* ) 2;

doesn't mean anything. you cast the number 2 into a pointer? do you know what sits on the 3 byte of your computer's RAM? it's obsurd.

[MutantJohn]

Does the pointer cast really not do anything? I assumed it was better than setting the pointer to nullptr and then​ setting the second bit.

Edit: Also, I don't use new[] because I don't want the elements to be constructed. I want it to be like resize() vs reserve().

resize() probably just calls new[] but I want reserve() which allocates but does not construct.

This is because I doubt there's any one constructor I can use to initialize an entire block. Blocks need to have their first and last Elements set to boundaries and the inner elements needs to be free and have their unmasked pointers point to the next element in the free list.

So, how would any default constructor define that behavior? That's why I opted to do it manually.

Edit edit: Also phantom, I don't get why you want Fallout 4. Metal Gear Solid V already came out.

[MutantJohn]

Is this slightly better?

#include <iostream>
#include <cstdlib>


class Point {
  private:
    double c_[3];


  public:
    Point(void) {
      std::cout << "Calling default constructor" << std::endl;
    };
};


int main(void) {
  size_t n = 16;
  n *= sizeof(Point);


  Point *points = static_cast<Point*>(::operator new(n));
  ::operator delete(points);


  return 0;
}

Why... Why does this work the way it does?

I've never seen this ":perator" syntax before. What is it? Why does it not call the constructors or destructors? Is it really that different than malloc or free then?

[phantomotap]

Is this slightly better?

O_o

You aren't addressing any of the issues with the implementation I referenced because you haven't implemented any of relevant the mechanisms.

Why does this work the way it does?

As I've explained to so very many, the `new` operator isn't the same as the `operator new` function.

Why does it not call the constructors or destructors?

The `new` operator is where the rollback and construction mechanism lives.

Is it really that different than malloc or free then?

No. The `operator new` function is the C++ standard interface for memory allocation used by the `new` operator.

You are just using a different standard interface for memory allocation.

Furthermore, you would continue to have all the same issues I referenced if you only changed the interface for allocation you have used. The problems your code has doesn't live in the `malloc` function. The problems your code has is the result of you misusing and abusing "placement new" without accounting for any exceptions which may arise in several places.

[Edit]

Also phantom, I don't get why you want Fallout 4. Metal Gear Solid V already came out.

If you want to know what I have to say about "Fallout 4" or "Metal Gear Solid 5", you'll have to see if you can find a snapshot from IRC logs.
[/Edit]

Soma