integration/luprex/core/cpp/idalloc.cpp

#include "idalloc.hpp"
#include <iostream>
#include <map>


static int64_t nthbatch(int64_t n) {
    return int64_t(0x0001000000000000) + n*256;
}

static bool ranges_equal(const std::deque<int64_t> &dq, int64_t a, int64_t b, int64_t c) {
    if (dq.size() != 3) return false;
    if (dq[0] != a) return false;
    if (dq[1] != b) return false;
    if (dq[2] != c) return false;
    return true;
}

IdGlobalPool::IdGlobalPool() {
    salvaged_.clear();
    next_batch_ = 0;
    next_id_ = 0;
}

IdGlobalPool::~IdGlobalPool() {
}

void IdGlobalPool::init_master() {
    salvaged_.clear();
    next_batch_ = 0x0001000000000000;
    next_id_ = 0x0010000000000000;
}

void IdGlobalPool::init_synch() {
    salvaged_.clear();
    next_batch_ = 0;
    next_id_ = 0x001E000000000000;
}

int64_t IdGlobalPool::get_one() {
    return next_id_++;
}

int64_t IdGlobalPool::get_batch() {
    int64_t batch;
    if (salvaged_.empty()) {
        if (next_batch_ == 0) {
            batch = 0;
        } else {
            batch = next_batch_;
            next_batch_ += 256;
        }
    } else {
        batch = salvaged_.back();
        salvaged_.pop_back();
    }
    return batch;
}

void IdGlobalPool::salvage(int64_t batch) {
    if (batch == 0) return;
    if (next_batch_ == 0) return;
    if ((batch & 0xFF) >= 128) return;
    salvaged_.push_back(batch);
}

void IdGlobalPool::salvage_thread(lua_State *L) {
    salvage(lua_getnextid(L));
    lua_setnextid(L, 0);
}

int64_t IdGlobalPool::alloc_id_for_thread(lua_State *L) {
    int64_t batch = lua_getnextid(L);
    if (batch != 0) {
        int64_t id = batch;
        batch += 1;
        if ((batch & 0xFF) == 0) batch = 0;
        lua_setnextid(L, batch);
        return id;
    } else {
        return get_one();
    }
}

void IdGlobalPool::serialize(StreamBuffer *sb) {
    sb->write_int64(next_batch_);
    sb->write_int64(next_id_);
    sb->write_uint32(salvaged_.size());
    for (int64_t batch : salvaged_) {
        sb->write_int64(batch);
    }
}

void IdGlobalPool::deserialize(StreamBuffer *sb) {
    next_batch_ = sb->read_int64();
    next_id_ = sb->read_int64();
    uint32_t salvaged_size = sb->read_uint32();
    salvaged_.resize(salvaged_size);
    for (int i=0; i < int(salvaged_size); i++) {
        salvaged_[i] = sb->read_int64();
    }
}

IdPlayerPool::IdPlayerPool(IdGlobalPool *g) {
    global_ = g;
    fifo_capacity_ = 0;
}

IdPlayerPool::~IdPlayerPool() {
}

void IdPlayerPool::set_fifo_capacity(int n) {
    assert((n >= 0) && (n <= 250));
    fifo_capacity_ = n;
    while (int(ranges_.size()) > n) {
        global_->salvage(ranges_.back());
        ranges_.pop_back();
    }
}

void IdPlayerPool::refill() {
    while (int(ranges_.size()) < fifo_capacity_) {
        int64_t batch = global_->get_batch();
        if (batch == 0) break;
        ranges_.push_back(batch);
    }
}

void IdPlayerPool::test_push_back(int64_t range) {
    ranges_.push_back(range);
}

void IdPlayerPool::test_pop_front() {
    ranges_.pop_front();
}

void IdPlayerPool::test_clear_ranges() {
    ranges_.clear();
}

int64_t IdPlayerPool::get_batch() {
    while (int(ranges_.size()) < fifo_capacity_ + 1) {
        int64_t batch = global_->get_batch();
        if (batch == 0) break;
        ranges_.push_back(batch);
    }
    if (ranges_.empty()) {
        return 0;
    } else {
        int64_t batch = ranges_.front();
        ranges_.pop_front();
        return batch;
    }
}

void IdPlayerPool::salvage_thread(lua_State *L) {
    global_->salvage_thread(L);
}

void IdPlayerPool::prepare_thread(lua_State *L) {
    global_->salvage_thread(L);
    lua_setnextid(L, get_batch());
}

void IdPlayerPool::serialize(StreamBuffer *sb) {
    sb->write_uint8(fifo_capacity_);
    sb->write_uint8(ranges_.size());
    for (int64_t batch : ranges_) {
        sb->write_int64(batch);
    }
}

void IdPlayerPool::deserialize(StreamBuffer *sb) {
    fifo_capacity_ = sb->read_uint8();
    int ranges_size = sb->read_uint8();
    ranges_.resize(ranges_size);
    for (int i=0; i < ranges_size; i++) {
        ranges_[i] = sb->read_int64();
    }
}

bool IdPlayerPool::exactly_equal(const IdPlayerPool &other) const {
    if (fifo_capacity_ != other.fifo_capacity_) return false;
    if (ranges_.size() != other.ranges_.size()) return false;
    for (int i = 0; i < int(ranges_.size()); i++) {
        if (ranges_[i] != other.ranges_[i]) {
            return false;
        }
    }
    return true;
}

bool IdPlayerPool::valid() const {
    if ((fifo_capacity_ < 0) || (fifo_capacity_ > 250)) return false;
    if (int(ranges_.size()) > fifo_capacity_) return false;
    return true;
}

bool IdPlayerPool::make_patch(const IdPlayerPool &auth, StreamBuffer *sb) const {
    assert(valid());
    assert(auth.valid());

    // The fifo capacity cannot be 255, so we use this as special
    // to indicate that no changes are present.
    if (exactly_equal(auth)) {
        sb->write_uint8(255);
        return false;
    }
    
    // Write the fifo capacity and nranges
    sb->write_uint8(auth.fifo_capacity_);
    sb->write_uint8(auth.ranges_.size());
    
    // Build up an index of the known IDs.
    std::map<int64_t, int> index;
    for (int i = 0; i < int(ranges_.size()); i++) {
        index[ranges_[i]] = i;
    }

    // Write the ranges, but encode known IDs in one byte.
    for (int i = 0; i < int(auth.ranges_.size()); i++) {
        int64_t n = auth.ranges_[i];
        auto iter = index.find(n);
        if (iter == index.end()) {
            sb->write_uint8(255);
            sb->write_int64(n);
        } else {
            int slot = iter->second;
            sb->write_uint8(slot);
        }
    }
    return true;
}

void IdPlayerPool::apply_patch(StreamBuffer *sb) {
    // read the header byte
    int fifo_cap = sb->read_uint8();
    if (fifo_cap == 255) {
        return;
    }
    fifo_capacity_ = fifo_cap;
    int nranges = sb->read_uint8();
    std::deque<int64_t> old = std::move(ranges_);
    ranges_.clear();
    for (int i = 0; i < nranges; i++) {
        int index = sb->read_uint8();
        if (index < 255) {
            assert(index < int(old.size()));
            ranges_.push_back(old[index]);
        } else {
            ranges_.push_back(sb->read_int64());
        }
    }
}

LuaDefine(unittests_idalloc, "c") {
    IdGlobalPool gp;
    IdPlayerPool pp(&gp);
    IdGlobalPool gpds;
    IdPlayerPool ppds(&gpds);
    StreamBuffer sb;

    // Synchronous pools produce IDs starting at 0x001E000000000000
    gp.init_synch();
    LuaAssert(L, gp.get_one() == 0x001E000000000000);
    LuaAssert(L, gp.get_one() == 0x001E000000000001);
    LuaAssert(L, gp.get_one() == 0x001E000000000002);

    // Master pools produce IDs starting at 0x0010000000000000
    gp.init_master();
    LuaAssert(L, gp.get_one() == 0x0010000000000000);
    LuaAssert(L, gp.get_one() == 0x0010000000000001);
    LuaAssert(L, gp.get_one() == 0x0010000000000002);

    // Synchronous pools produce only null batches.
    gp.init_synch();
    LuaAssert(L, gp.get_batch() == 0);
    LuaAssert(L, gp.get_batch() == 0);
    gp.salvage(nthbatch(5));
    LuaAssert(L, gp.get_batch() == 0);
    
    // Simple fetch batches with a few salvages.
    gp.init_master();
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    LuaAssert(L, gp.get_batch() == nthbatch(1));
    LuaAssert(L, gp.get_batch() == nthbatch(2));
    gp.salvage(nthbatch(182));
    gp.salvage(nthbatch(183));
    LuaAssert(L, gp.get_batch() == nthbatch(183));
    LuaAssert(L, gp.get_batch() == nthbatch(182));
    LuaAssert(L, gp.get_batch() == nthbatch(3));
    
    // Salvage of a zero-batch does nothing.
    gp.init_master();
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    LuaAssert(L, gp.get_batch() == nthbatch(1));
    gp.salvage(0);
    LuaAssert(L, gp.get_batch() == nthbatch(2));

    // Salvage of a partial batch.
    gp.init_master();
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    LuaAssert(L, gp.get_batch() == nthbatch(1));
    gp.salvage(nthbatch(142) + 10);
    LuaAssert(L, gp.get_batch() == nthbatch(142) + 10);
    LuaAssert(L, gp.get_batch() == nthbatch(2));

    // Salvage of a half-empty batch does nothing.
    gp.init_master();
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    LuaAssert(L, gp.get_batch() == nthbatch(1));
    gp.salvage(nthbatch(142) + 145);
    LuaAssert(L, gp.get_batch() == nthbatch(2));

    // In the synchronous model, refill should do nothing.
    pp.test_clear_ranges();
    gp.init_synch();
    pp.set_fifo_capacity(3);
    pp.refill();
    LuaAssert(L, pp.size() == 0);
    LuaAssert(L, pp.get_batch() == 0);
    LuaAssert(L, pp.size() == 0);
    LuaAssert(L, pp.get_batch() == 0);

    // In the master model, with fifo disabled.  Fifo should remain
    // empty, but batches should be returned.
    gp.init_master();
    pp.test_clear_ranges();
    pp.set_fifo_capacity(0);
    pp.refill();
    LuaAssert(L, pp.size() == 0);
    LuaAssert(L, pp.get_batch() == nthbatch(0));
    LuaAssert(L, pp.size() == 0);
    LuaAssert(L, pp.get_batch() == nthbatch(1));

    // Test refill from master (with enabled fifo).
    gp.init_master();
    pp.test_clear_ranges();
    pp.set_fifo_capacity(3);
    pp.refill();
    LuaAssert(L, ranges_equal(pp.ranges_, nthbatch(0), nthbatch(1), nthbatch(2)));

    // Now test that get_batch keeps the pool filled from master.
    LuaAssert(L, pp.get_batch() == nthbatch(0));
    LuaAssert(L, ranges_equal(pp.ranges_, nthbatch(1), nthbatch(2), nthbatch(3)));

    // Test unqueueing the batches.
    LuaAssert(L, gp.get_batch() == nthbatch(4));
    LuaAssert(L, gp.get_batch() == nthbatch(5));
    pp.set_fifo_capacity(0);
    LuaAssert(L, gp.get_batch() == nthbatch(1));
    LuaAssert(L, gp.get_batch() == nthbatch(2));
    LuaAssert(L, gp.get_batch() == nthbatch(3));
    LuaAssert(L, gp.get_batch() == nthbatch(6));

    // Try preparing a thread and salvaging a thread.
    gp.init_master();
    pp.test_clear_ranges();
    pp.set_fifo_capacity(3);
    lua_setnextid(L, 0);
    pp.prepare_thread(L);
    LuaAssert(L, lua_getnextid(L) == nthbatch(0));
    lua_setnextid(L, 0);
    pp.prepare_thread(L);
    LuaAssert(L, lua_getnextid(L) == nthbatch(1));

    // Try salvaging the pool from the thread.
    pp.salvage_thread(L);
    LuaAssert(L, lua_getnextid(L) == 0);
    LuaAssert(L, gp.get_batch() == nthbatch(1));

    // Allocate IDs from inside a thread.
    gp.init_master();
    lua_setnextid(L, 0xFD);
    LuaAssert(L, gp.alloc_id_for_thread(L) == 0xFD);
    LuaAssert(L, gp.alloc_id_for_thread(L) == 0xFE);
    LuaAssert(L, gp.alloc_id_for_thread(L) == 0xFF);
    LuaAssert(L, gp.alloc_id_for_thread(L) == 0x0010000000000000);
    LuaAssert(L, lua_getnextid(L) == 0);
    
    // Serialize and deserialize a global pool.
    gp.init_master();
    gpds.init_master();
    LuaAssert(L, gp.get_one() == 0x0010000000000000);
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    gp.salvage(nthbatch(182));
    gp.salvage(nthbatch(183));
    gp.serialize(&sb);
    gpds.deserialize(&sb);
    LuaAssert(L, gpds.get_one() == 0x0010000000000001);
    LuaAssert(L, gpds.get_batch() == nthbatch(183));
    LuaAssert(L, gpds.get_batch() == nthbatch(182));
    LuaAssert(L, gpds.get_batch() == nthbatch(1));

    // Serialize and deserialize a player pool.
    gp.init_master();
    gpds.init_synch();
    LuaAssert(L, gp.get_batch() == nthbatch(0));
    pp.test_clear_ranges();
    pp.set_fifo_capacity(3);
    pp.refill();
    ppds.test_clear_ranges();
    pp.serialize(&sb);
    ppds.deserialize(&sb);
    LuaAssert(L, ppds.get_fifo_capacity()==3);
    LuaAssert(L, ppds.size() == 3);
    LuaAssert(L, ppds.get_batch() == nthbatch(1));
    LuaAssert(L, ppds.get_batch() == nthbatch(2));
    LuaAssert(L, ppds.get_batch() == nthbatch(3));

    // Difference transmit compare two empty pools.
    gp.init_master();
    gpds.init_master();
    pp.test_clear_ranges();
    ppds.test_clear_ranges();
    pp.set_fifo_capacity(3);
    ppds.set_fifo_capacity(3);
    
    // Check case: no differences.
    sb.clear();
    LuaAssert(L, !ppds.make_patch(pp, &sb));
    ppds.apply_patch(&sb);
    LuaAssert(L, ppds.exactly_equal(pp));

    // Add some values to master pool
    pp.test_push_back(123);
    pp.test_push_back(456);

    // transmit and compare. Add extra bytes
    sb.clear();
    LuaAssert(L, ppds.make_patch(pp, &sb));
    sb.write_uint32(0);
    ppds.apply_patch(&sb);
    LuaAssert(L, sb.write_count() - sb.read_count() == 4);
    LuaAssert(L, ppds.exactly_equal(pp));

    // Pop a value from master pool
    pp.test_pop_front();
    pp.test_push_back(789);

    // transmit and compare.
    sb.clear();
    LuaAssert(L, ppds.make_patch(pp, &sb));
    ppds.apply_patch(&sb);
    LuaAssert(L, ppds.exactly_equal(pp));

    return 0;
}