2
$\begingroup$

So, I need a data structure that would support the following operations:

  • Add an element to the structure and get some "handle" for that element
  • Retrieve an element by a handle
  • Remove an element by a handle
  • Iterate over all elements

Note that the elements are not expected to have any particular order in iteration. Preferably it should also be cache-friendly and don't use too many memory allocations. The language I have in mind is C++, if that matters.

A doubly-linked list satisfies the requirements (with a "handle" being just a pointer/iterator to a list node). However,

  • It uses a memory allocation per each new element, which is a bit too much
  • Iteration over a list is extremely cache-unfriendly (as compared to e.g. iteration over contiguous array)

Another possible implementation is to use a hand-written slab allocator over a contiguous chunk of memory, with each cell being either free of occupied. Occupied cells store the actual elements, while free cells maintain a singly-linked list of free cells by storing the index of the next cell in the list (or some predefined null value to mark the end of the list). A "handle" in this case would be an index of a cell. Coupled with some extra per-cell data to decide whether a cell is free or occupied, and an index of the head of the list of free cells, this structure allows

  • Fast amortized insertion: put the new element to the head of the list of free cells, updating the head index (and reallocating the cells array if no free cell is available)
  • Fast element access (just an index into array)
  • Fast removal: add the element's cell to the list of free cells
  • Fast iteration: iterate over an array, checking if the current cell is occupied

However, I am not entirely happy with this solution, since

  • It has to explicitly maintain the cells state (free/occupied) - this is mostly a technical inconvenience and not a serious issue
  • Iteration is still not branch-predictor-friendly (due to checking whether the cell is free or occupied) and might waste a lot of time skipping free cells (thanks to Andrej Bauer for pointing this explicitly), so it is an improvement only if the number of free cells is sufficiently low.

Yet another solution is to implement any kind of binary key-value tree stored inside a contiguous array (auto-expanding, like std::vector), with array indices used instead of tree node pointers, and keys serving as handles for element access/removal (their only purpose being implementing handles). This way we have

  • Amortized $O(\log n)$ insertion - append a tree node to the end of the array, altering the tree structure accordingly (or even $O(1)$ if the tree supports $O(1)$ insertion to the end, like a splay tree, - the data structure generates keys itself, so it can guarantee that they only increase for new elements)
  • Amortized $O(\log n)$ removal - remove the node from the tree structure, then swap it with the last node in the array (fixing the indices stored in the last node's children and parent), and decrement the array size
  • $O(\log n)$ element access via the tree structure on keys
  • Ideal iteration: simply iterate over the whole array (even better if we keep values in a separate array, synchronized with the array for the tree structure, since the latter is not needed for iteration)

This solution trades iteration speed for $O(\log n)$ other operations.

I am mostly interested in increasing the iteration performance, while still retaining fast amortized insertion/deletion. I'm ok with spending some extra $O(n)$ memory, shall the need arise.

So, here's my question: are there better options for a data structure with these requirements?

$\endgroup$
12
  • $\begingroup$ Which of the operations do you expect to work faster with your suggestion, in comparison with using a linked list and standard allocation of memory that your programming language provides? Note that in your suggested solution iteration need not be fast, as it may waste a lot of time skipping over empty cells (and must always scan the entire block). $\endgroup$ Nov 26, 2020 at 9:39
  • $\begingroup$ @AndrejBauer Thanks for your comment, I'll update the question. I'm mostly concerned with the iteration performance (while still keeping insert/remove O(1)). $\endgroup$
    – lisyarus
    Nov 26, 2020 at 9:43
  • $\begingroup$ A linked list satisfies your requirements because allocations are amortized O(1). Is this really about cache? Do you know that your list fits in cache? $\endgroup$ Nov 26, 2020 at 9:48
  • $\begingroup$ @AndrejBauer Iteration over linked list is slower, since there's no guarantee that successive nodes are tightly packed in memory. Iteration over my solution is fast (as is generally the case with iteration over contiguous memory), but might waste time skipping empty cells, as you've already noted. I'm asking whether there is a better way. $\endgroup$
    – lisyarus
    Nov 26, 2020 at 9:56
  • 1
    $\begingroup$ @AndrejBauer maybe have a look at this stackoverflow thread: stackoverflow.com/questions/1402483/…. The other list container std::forward_list is slightly faster, but ultimately suffers from the same problems. $\endgroup$
    – STanja
    Nov 26, 2020 at 10:12

2 Answers 2

0
$\begingroup$

This is my implementation of a list structure.

It is not dynamically allocated, and all allocations are done when it is constructed.

  • Iteration is at worst case O(n), n being the max size of the list
  • Insert is O(1)
  • Remove is O(1), except if using sort optimization(tradeoff: slows remove but potentially helps iteration)

If sorting optimization is used, any new elements that are added will be given an index towards the beginning of the list to fill in gaps better, making better use of the cache and reducing free cells in between. However this is called in the erase function, so removal will be O(log n).

Sorting could be called at a certain threshold (like once every 100 erase calls OR every second or so). This way it will not have as large of an impact on removal time complexity.

Also sorting refers to sorting the element indices that were erased and added back to the array of available indices.

template <typename T> struct zList{

#define _setBit(__byteArray, __bitPosition)   (__byteArray[(__bitPosition)/8] |=  (0x01<<((__bitPosition)%8)))
#define _clearBit(__byteArray, __bitPosition) (__byteArray[(__bitPosition)/8] &= ~(0x01<<((__bitPosition)%8)))
#define _getBit(__byteArray, __bitPosition) (__byteArray[(__bitPosition)/8] & (0x01<<((__bitPosition)%8)))
#define divideAndRoundUp(a, b) ((((a)+(b))-1)/(b))

/*
    notes:

    -cache friendly iteration
    -no allocations for insert or erase (unless in the deconstructor of the type)

    -out of range access returns a valid reference (extra element at the end to the max size)
        might be better to just return an exception/assert ?
        can also check if it exists before accessing "list.exists(HANDLE);"

    -sort_descending() this function is used to sort the array indices that were removed
        this may or may not be an optimization depending on the use case.

        sorting helps fill in gaps after erasing elements then adding new elements (the index will be the min index(closest to 0) and not a random one))
            this helps the iteration by looping through less empty elements

        if you don't sort then erase() will be O(1), but iteration *might* suffer
            also sort can be called every x elements erased, or on a timer (maybe once per second, either manually call it or the next time erase it called it can check the time and decide if it needs to sort)


    -Insertion is O(1)
    -Removal is O(1) unless sorting is used O(log n)

    -Iteration: worst case O(n 'n being the maxSize of the array'),    it will iterate the max size of the list and iterate by 1 each time if the unused elements are every other
                    best case O(>=list.size()), an internal minimum index is stored so it knows where the first element is located and how many elements there are
                        so it knows when to start and stop



    HOW TO USE-------------------------------------------------------------------

        //can use Custom class/struct, or a base type
            struct SomeType{
                private:
                int allocated;

                public:
                std::string someString;
                int someValue;
                unsigned char* someAllocation;

                //default constructors will be called if not provided
                SomeType(){ //all constructors are called upon list initialization
                    allocated = 0;
                }
                ~SomeType(){ //all deconstructors are called when an element is erased
                    if(allocated){
                        free(someAllocation);
                    }
                }

                void alloc(){ //if you need to allocate memory, make sure to free it in the deconstructor so it will be cleaned up when erased or zList goes out of scope
                    if(!allocated){
                        allocated = 1;
                        someAllocation = (unsigned char*)malloc(100000000);
                    }
                }

            };

        //Initialize the list
            zList<SomeType> list(10);

        //Clear and or resire the list
            list.clear(); //clear
            list.clear(15); //clear and make new size 15

        //Insert - a little awkward syntax but whatever
            int handleIndex = -1;
            if(!list.isFull()){
                handleIndex = list.newIndex(); //Get an availible index(or handle) in the list
                if(handleIndex==-1){} //You can skip list.isFull() and check if the handle is -1 instead (means it is full)

                //Setting up the element, [] operator returns a refrence to the element
                list[handleIndex].alloc(); //Allocate memory
                list[handleIndex].someString = "test";
                list[handleIndex].someValue = 128;
            }

        //Remove/erase
            list.erase(handleIndex); //erase the element (this calls the deconstructor and then the constructor to reset the memory)


        //Itteration - syntax like standard containers
            //you can pass in a starting handle/index to begin()
            for(auto it = list.begin(); it!=list.end(); ++it){
                SomeType& st_0 = (*it); //dereferencing returns reference to the element
                SomeType& st_1 = list[it.currentIndex]; //or get reference by using the index
            }

            //If you need to erase an element during iteration
            for(auto it = list.begin(); it!=list.end();){ //remove ++it from the statement
                SomeType& st_0 = (*it); //dereferencing returns reference to the element
                SomeType& st_1 = list[it.currentIndex]; //or get reference by using the index

                int needToErase;
                if(needToErase){
                    list.erase(it); //Must erase using iterator and not index here so it can move the iterator to the next element
                }else ++it; //iterate to the next element if we didn't erase
            }

*/

private:
    unsigned long long lastSortTime;
    int minimumLoopIndex; //keeps track of the min index that is being occuied
    int sortIndex; //Helps to lower the amount of indices we need to sort
    int* availIndex; //available indices that we can use
    unsigned char* usedIndicies; //a bitfield representing the array indices that are currently occupied
    T* data; //Array of the elements, this array is allocated with an additional refrence element. This is returned if the array is indexed out of range
    int availIndexCnt; //number of elements we can add to the list
    int maxSize; //max number of elements we can add to the list

public:
    struct Iterator {
        private:
            T* dataPtr;
            int _maxSize;
            unsigned char* _usedIndicies;
            int elementsFound;
            int currentIndicieCount;
        public:
            int currentIndex;

            Iterator(int startIndex, T* _dataPtr, int __maxSize, int _currentIndicieCount, unsigned char* __usedIndicies) : currentIndex(startIndex), dataPtr(_dataPtr), _maxSize(__maxSize), _usedIndicies(__usedIndicies), elementsFound(1), currentIndicieCount(_currentIndicieCount) {}
            Iterator& operator++(){

                //If we have found all the elements then we can stop checking the rest just return end() now
                if(elementsFound>=currentIndicieCount){
                    currentIndex = _maxSize; 
                    return *this;
                }

                currentIndex++; 

                //Find the next element that is being used
                do{
                    //std::cout << currentIndex << " bb\n";
                    if(_getBit(_usedIndicies, currentIndex)){
                        ++elementsFound;
                        break;
                    }else{
                        unsigned long long testBits = *(unsigned long long*)&_usedIndicies[currentIndex/8]; //Get the next 8 bytes (that includes the current index)
                        
                        //See notes in begin()
                        if(!(testBits              )){ currentIndex += 64-(currentIndex&0x3F); continue; } //advance upto 64 empty elements
                        if(!(testBits&0xFFFFFFFFull)){ currentIndex += 32-(currentIndex&0x1F); continue; } //advance upto 32 empty elements
                        if(!(testBits&0xFFFFull    )){ currentIndex += 16-(currentIndex&0x0F); continue; } //advance upto 16 empty elements
                        if(!(testBits&0xFFull      )){ currentIndex +=  8-(currentIndex&0x07); continue; } //advance upto  8 empty elements
                        if(!(testBits&0xFull       )){ currentIndex +=  4-(currentIndex&0x03); continue; } //advance upto  4 empty elements
                        if(!(testBits&0x3ull       )){ currentIndex +=  2-(currentIndex&0x01); continue; } //advance upto  2 empty elements

                        currentIndex++; //move just 1 element
                    }
                }while(currentIndex < _maxSize);
                if(currentIndex>_maxSize) currentIndex = _maxSize; //make == to end()

                return *this;
            }
            Iterator& operator+=(int steps){
                int newIndex = currentIndex + steps;
                currentIndex = (newIndex < _maxSize) ? newIndex : _maxSize;
                return *this;
            }
            bool operator==(const Iterator& other) const {
                return currentIndex == other.currentIndex;
            }
            bool operator!=(const Iterator& other) const {
                return currentIndex != other.currentIndex;
            }
            T& operator*() {
                return *(&dataPtr[currentIndex]);
            }
    };

    zList(int _maxSize){
        if(_maxSize < 0) _maxSize = 1; //don't allow negative sizes
        maxSize = _maxSize;
        //data = tnew<T>(_maxSize+1, TMC::ZLIST_CONTAINER);
        data = (T*)malloc((_maxSize+1)*sizeof(T)); //make sure to construct all of them initialy
        int i=-1; while(++i < (_maxSize+1)) new(&data[i]) T();

        availIndex = (int*)malloc(_maxSize*sizeof(int));
        int g=-1; while(++g < _maxSize) availIndex[(_maxSize-1)-g] = g;
        availIndexCnt = _maxSize;
        usedIndicies = (unsigned char*)malloc((divideAndRoundUp(_maxSize,8)+16)*sizeof(unsigned char)); //added 16 to be able to access out of range bits
        memset(usedIndicies, 0, (divideAndRoundUp(_maxSize,8)+16)); //zero memory
        sortIndex = maxSize;
        lastSortTime = 0;
        minimumLoopIndex = maxSize; //this will just make the iterator start at the end()
    };
    ~zList(){
        int i=-1; while(++i<(maxSize+1)) data[i].~T(); //call the deconstuctor, all of them will be constructed to be able to safely do this
        free(availIndex);
        free(usedIndicies);
        free(data);
    }

    bool isFull(){ return !availIndexCnt; } //if count is 0 then it flips to 1 and returns as true
    bool isEmpty(){ return availIndexCnt == maxSize; }
    unsigned int size(){ return maxSize-availIndexCnt; }
    void exists(unsigned int arrayIndex){ return ( (arrayIndex<maxSize) && (!_getBit(usedIndicies, arrayIndex)) );} //check if the index if being used

    //Erases all the elements, if a newSize is provided, then it will clear and then resize the list
    void clear(int newSize = 0){ //if newSize is not 0 then delete the old list and create a new one
        for (auto it = begin(); it != end();) erase(it); //erase all the elements

        int g=-1; while(++g < maxSize) availIndex[(maxSize-1)-g] = g;
        memset(usedIndicies, 0, tsize(usedIndicies));
        availIndexCnt = maxSize;
        sortIndex = maxSize;
        lastSortTime = 0;
        minimumLoopIndex = maxSize;

        if(newSize>0){
            if(newSize!=maxSize){
                this->~zList(); // Call the destructor for the current object
                new (this) zList(newSize);
            }
        }
    }
    
    //Sorts the availIndex array starting from the sort index to the last index of the total count available
    void sort_descending(){
        int n = availIndexCnt - sortIndex; //number of ints to sort from the sort index to the current count
        if(!n) return;
        int* tmp = &availIndex[sortIndex]; //point to the sort index

        std::sort(tmp, tmp+n, std::greater()); //sort the array in descending order

        //-i is the number of sequential numbers from the sort index
        //-k is the index that is expected next, sub 1 each after each check itteration
        int i=-1, k=(maxSize-sortIndex)-1 ; while(++i<n){
            if((k--)!=tmp[i]) break;
        }
        sortIndex += i; //move the sort index by the num of sequential indicies
    }

    //Get an availible index we can use
    int newIndex(){
        if(!availIndexCnt) return -1; //-1 means it is full
        int arrayIndex = availIndex[--availIndexCnt];
        _setBit(usedIndicies, arrayIndex);
        sortIndex--;

        if(arrayIndex<minimumLoopIndex) minimumLoopIndex = arrayIndex; //if our index we got is less than then set the minindex so the loop knows to begin their
        return arrayIndex;
    }

    //erase by using the index we want to remove
    void erase(unsigned int arrayIndex){
        if(arrayIndex>=maxSize || (!_getBit(usedIndicies, arrayIndex))) return;

        //if we are adding back to the array next to the sort index, increment the sort index if the array index is 1 less
        if((sortIndex==availIndexCnt) && (((maxSize-sortIndex)-1)==arrayIndex)) sortIndex++;

        //if the array index matches the min index we can safely move it up 1
        if(arrayIndex==minimumLoopIndex) minimumLoopIndex++; 
        

        availIndex[availIndexCnt++] = arrayIndex; //add index back to be reused
        _clearBit(usedIndicies, arrayIndex);
        data[arrayIndex].~T(); //call the deconstuctor
        new(&data[arrayIndex]) T(); //then the constructor again
        
        //Optional - see notes
            //sort_descending();

        /*
        TODO:
        set a timer, or call sort every x times
        if(lastSortTime){ //if a timer was set, sort if time passed
            if(getNetTime()>lastSortTime){
                lastSortTime = 0; //all sorted so don't check again till we remove another one
            }
        }else{
            lastSortTime = getNetTime()+1000; //sort in 1 second from now
        }
        */
    }

    //erase by using the itterator, then increment the it to the next one
    void erase(Iterator& it){
        int arrayIndex = it.currentIndex;
        ++it; //move itterator to the next one BEFORE we remove the current node
        erase(arrayIndex); //erase by index
    }

    
    //Return a refrence to the element in the list
    FORCEINLINE T& operator[](unsigned int arrayIndex){
        //Takes the array index as an unsigned to avoid negative checks, and return the maxSize index if we try to access outside the range (0 to (maxSize-1))
        return *(&data[min(arrayIndex, maxSize)]);
    }

    // Begin function returns an iterator to the head
    Iterator begin(int startIndex = 0){
        int useCustomStart = startIndex!=0;
        if(!useCustomStart) startIndex = minimumLoopIndex; //if automatic start, begin at the minIndex
        
        //Start at a valid starting index, make sure that start bit is actually set
        do{
            //std::cout << startIndex << " aa\n";
            if(_getBit(usedIndicies, startIndex)){
                break;
            }else{
                unsigned long long testBits = *(unsigned long long*)&usedIndicies[startIndex/8]; //type casting to 64bit int wont cause segfault because usedIndicies is padded with 16 additional bytes

                /*
                    these checks allow us to skip empty cells in the array
                    
                    depending on the sparsity of the empty cells/ application use cases, some checks may not be necessary
                    for example if the empty cells are very sparse (5 or 6) per occupied element, just test the first 4 and 2 bits
                
                    TODO: 
                        automatically determine this with a rough guess

                */ 

                if(!(testBits              )){ startIndex += 64-(startIndex&0x3F); continue; } //advance upto 64 empty elements
                if(!(testBits&0xFFFFFFFFull)){ startIndex += 32-(startIndex&0x1F); continue; } //advance upto 32 empty elements
                if(!(testBits&0xFFFFull    )){ startIndex += 16-(startIndex&0x0F); continue; } //advance upto 16 empty elements
                if(!(testBits&0xFFull      )){ startIndex +=  8-(startIndex&0x07); continue; } //advance upto  8 empty elements
                if(!(testBits&0xFull       )){ startIndex +=  4-(startIndex&0x03); continue; } //advance upto  4 empty elements
                if(!(testBits&0x3ull       )){ startIndex +=  2-(startIndex&0x01); continue; } //advance upto  2 empty elements

                startIndex++; //move just 1 element (This is required if the checks above fail)
            }

        }while(startIndex < maxSize);
        if(startIndex>maxSize) startIndex = maxSize; //make == to end()
        if(!useCustomStart) minimumLoopIndex = startIndex;

        return Iterator(startIndex, data, maxSize, size(), usedIndicies);
    }

    // End function returns an iterator to a null pointer (end of the list)
    Iterator end() {
        return Iterator(maxSize, data, maxSize, size(), usedIndicies);
    }
};
$\endgroup$
1
  • 1
    $\begingroup$ (C++ doesn't call them deconstructors.) Can you point out distinguishing features of the implementation presented? $\endgroup$
    – greybeard
    Oct 27, 2023 at 10:15
0
$\begingroup$

A simple method is a resizable array with a bitmap of which entries are used, and an index where to look for the next unused entry. Resize the array when it’s 90% full. Since you can typically check 64 bits for the next unset bit, finding an empty entry is very fast.

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.