<commit_before>/*
  Provides a software solution to the problem of applying atomic operations using
  128-bit wide pointers via compression. The algorithm used for compression (and
  decompression) rely on the number locales. In majority of cases where numLocales
  is within a 16 bit range, we can encode it in the upper 16 bits of the address
  as only 2^48 bits of the virtual address space is ever used. In the cases where
  numLocales is between 2^17 and 2^32, we use a novel but simple technique of
  maintaining 'tables' that can be indexed into via a 32-bit integer, allowing the
  other 32-bits to be used for the locale id. The table can be indexed without
  need of synchronization, but as a Chapel array cannot be operated on while it's
  domain is being updated, we use a simple `sync` variable locally and maintain
  32 segments that are allocated on demand. While descriptors are valid, the segment
  being indexed into will be safe to use without synchronization. A simple skip list
  is utilized to efficiently hash objects to their unique index.
*/

use CyclicDist;
use Random;
use BigInteger;

// In the case where you want to use the descriptor table approach for testing or
// because there's some architecture which the assumption that only 48 bits are used
// for the virtual address space is incorrect, you can disable it here.
config param usePointerCompression = true;


// We require this to create a wide pointer after decompression.
extern type wide_ptr_t;
extern type c_nodeid_t;
extern proc chpl_return_wide_ptr_node(c_nodeid_t, c_void_ptr) : wide_ptr_t;

// Hashes a class instance by pointer. We ignore the locale id portion because
// this is used for pointers that refer into our address space.
proc hash(obj) : uint where isClass(obj.type) {
  var key =  __primitive("cast", uint(64), __primitive("_wide_get_addr", obj));
  key = (~key) + (key << 21); // key = (key << 21) - key - 1;
  key = key ^ (key >> 24);
  key = (key + (key << 3)) + (key << 8); // key * 265
  key = key ^ (key >> 14);
  key = (key + (key << 2)) + (key << 4); // key * 21
  key = key ^ (key >> 28);
  key = key + (key << 31);
  return key;
}

proc hash(obj) {
  halt("Obj to hash *must* be a class instance...");
}


/*
  The core `GlobalAtomicObject` which manages compression and decompression
  for atomic objects. It should be noted that if numLocales >= 2^17, then any
  object being operated on is added to the descriptor table, which will leak data
  if never freed with '_delete' (although if the memory is reused,
  so will the memory managed by this object). The descriptor table will be cleaned
  up automatically when this goes out of scope.
*/
record GlobalAtomicObject {
  type objType;
  var _atomicVar: atomic uint(64);

  // We maintain a SkipList to manage memory (which is built on top of the descriptor
  // table itself). Currently, as the SkipList is not concurrent (requires Hazard Pointers
  // which requires thread-local storage, or task-local storage in this case).
  var perLocaleDomain = LocaleSpace dmapped Cyclic(startIdx=LocaleSpace.low);
  var descriptorTables : [perLocaleDomain] SkipList(objType);
  var descriptorTableLocks : [perLocaleDomain] sync bool;

  /*
    Compresses an object into a descriptor.
  */
  proc compress(obj:objType) : uint {
    if obj == nil then return 0;

    // If we have less than 2^16 locales, we can perform a faster compression
    // by packing the 48 usable bits of the virtual address with 16 bits of the
    // locale/node id.
    if usePointerCompression && (numLocales & 0xFFFF == numLocales) {
      var locId = chpl_nodeFromLocaleID(__primitive("_wide_get_locale", obj)) : uint;
      var addr = __primitive("cast", uint, __primitive("_wide_get_addr", obj));
      return locId << 48 | addr & 0x0000ffffffffffff;
    }

    // Fallback: We use descriptor tables
    var retval : uint;
    on obj {
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first] = true;
      var node = descriptorTables[descriptorTables.domain.localSubdomain().first].insert(obj);
      var localeId = here.id;
      if (localeId & 0xFFFFFFFF) != localeId then halt("LocaleID > 2^32");
      var idx = node.idx;
      if (idx & 0xFFFFFFFF) != idx then halt("Idx > 2^32");
      retval = localeId << 32 | idx;
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first];
    }

    return retval;
  }

  /*
    Decompresses a descriptor into the wide pointer object.
  */
  proc decompress(descr:uint) : objType {
    if descr == 0 then return nil;


    // If we have less than 2^16 locales, then we know we performed the
    // faster compression method so we need to decompress it in the same way...
    if usePointerCompression && (numLocales & 0xFFFF == numLocales) {
      var locId = descr >> 48;
      var addr = descr & 0x0000ffffffffffff;
      var wideptr = chpl_return_wide_ptr_node(locId, __primitive("cast", c_void_ptr, addr));

      // We've created the wide pointer, but unfortunately Chapel does not support
      // the ability to cast it to the actual object, so we have to do some
      // trickery to get it to work. What we do is we allocate a wide pointer on
      // the stack and memcpy our wideptr into the other. This is needed so we
      // have the same type.
      var newObj : objType;
      on Locales[here.id] do newObj = nil;

      // Warning: If this is *not* a wide pointer, it will overwrite part of the stack.
      // If it is, it works. Dangerous, but only way to do so at the moment.
      c_memcpy(c_ptrTo(newObj), c_ptrTo(wideptr), 16);
      return newObj;
    }

    var localeId = descr >> 32;
    var idx = descr & 0xFFFFFFFF;
    return descriptorTables[localeId : int].memPool.access(idx : int).key;
  }

  inline proc read() : objType {
    return decompress(_atomicVar.read());
  }

  inline proc write(obj:objType) {
    _atomicVar.write(compress(obj));
  }

  inline proc exchange(obj:objType) : objType {
    return decompress(_atomicVar.exchange(compress(obj)));
  }

  // TODO: Save time by comparing the actual objects, not just the hash... if both
  // objects are remote, this is a *significant* cost.
  //  var currDescr = _atomicVar.read();
  //  var currObj = decompress(currDescr);
  //  if currObj == expectedObj then _atomicVar.compareExchangeStrong(currDescr, compress(newObj));
  inline proc compareExchange(expectedObj:objType, newObj:objType) : bool {
    return _atomicVar.compareExchangeStrong(compress(expectedObj), compress(newObj));
  }

  /*
    Deletes the object from our table so it may be recycled.
  */
  inline proc _delete(obj:objType) {
    if obj != nil then on obj {
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first] = true;
      descriptorTables[descriptorTables.domain.localSubdomain().first].remove(obj);
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first];
    }
  }
}


class SkipListNode {
  type keyType;
  var hash : uint(64);
  var key : keyType;
  var idx : uint;
  var forwardDom : domain(int);
  var forward : [forwardDom] SkipListNode(keyType);
}

/*
  TODO: This needs to be made concurrent or else this will be the bottleneck
  for our implementation. Hazard Pointers likely needed (this also would be a
  viable example for a need of per-locale garbage collection. Perhaps make it
  possible to declare a zone such that no wide-pointers can be generated and that
  data cannot escape such a zone. It'd be very useful for lock-free/wait-free
  data structures, and should make garbage collection possible if mark-and-sweep
  stop-the-world is *only* for the task inside of the zone. Then again, hazard
  pointers work too...)
*/
record SkipList {
  type keyType;
  var maxLevel = 4;
  var level = 1;
  var header : SkipListNode(keyType);
  var memPool : FlatObjectPool(SkipListNode(keyType));

  proc SkipList(type keyType) {
    memPool = new FlatObjectPool(SkipListNode(keyType), lambda () { return new SkipListNode(keyType); });
    header = makeNode(level);
  }

  // RNG to create randomized level
  var randStr = makeRandomStream(real);

  inline proc randomLevel() {
    var _level = 1;
    while randStr.getNext() < 0.5 && _level < maxLevel {
      _level = _level + 1;
    }
    return _level;
  }

  // Recycle and clean before usage...
  inline proc makeNode(_level, hash : uint = 0, key : keyType = nil) : SkipListNode(keyType) {
    var (idx, node) = memPool.alloc();
    node.idx = idx : uint;
    node.hash = hash;
    node.key = key;
    node.forwardDom = { 1 .. maxLevel };
    node.forward = nil;

    return node;
  }

  proc remove(key : keyType) {
    // We hash the object to remove from the list. Note: We do not take into
    // account the localeId, so the object *must* not be a wide pointer.
    var hashKey = hash(key);
    var update : [{1 .. maxLevel}] SkipListNode(keyType);
    var currNode = header;
    var currLevel = level;

    // Start at higher level first...
    while currLevel >= 1 {
      // Maintain same level so long ourHashKey > theirHashKey
      while currNode.forward[currLevel] != nil && currNode.forward[currLevel].hash < hashKey {
        currNode = currNode.forward[currLevel];
      }

      // At this point, ourHashKey < theirHashKey meaning we can't travel along this level.
      // However, because we are before them, that means that we would forward to them (if
      // we were to insert ourselves)
      update[currLevel] = currNode;
      currLevel = currLevel - 1;
    }

    // At this point, we cannot traverse any further than the bottom level.
    currNode = currNode.forward[1];
    if currNode == nil || currNode.hash != hashKey {
      return;
    }

    // It is a match, prepare to delete it by bridging the gap...
    for i in 1 .. level {
      if update[i].forward[i] != currNode then break;
      update[i].forward[i] = currNode.forward[i];
    }

    // Recycle node
    memPool.dealloc(currNode.idx);

    // If we are the highest node in the list, then we can reduce it.
    while level > 1 && header.forward[level] == nil {
      level = level - 1;
    }
  }

  proc insert(key : keyType) : SkipListNode(keyType) {
    // We hash the object to insert into the list. Note: We do not take into
    // account the localeId, so the object *must* not be a wide pointer.
    var hashKey = hash(key);
    var update : [{1 .. maxLevel}] SkipListNode(keyType);
    var currNode = header;
    var currLevel = level;

    // Start at higher level first...
    while currLevel >= 1 {
      // Maintain same level so long ourHashKey > theirHashKey
      while currNode.forward[currLevel] != nil && currNode.forward[currLevel].hash < hashKey {<commit_msg>Made necessary refactoring for GlobalAtomicObject<commit_after>/*
  Provides a software solution to the problem of applying atomic operations using
  128-bit wide pointers via compression. The algorithm used for compression (and
  decompression) rely on the number locales. In majority of cases where numLocales
  is within a 16 bit range, we can encode it in the upper 16 bits of the address
  as only 2^48 bits of the virtual address space is ever used. In the cases where
  numLocales is between 2^17 and 2^32, we use a novel but simple technique of
  maintaining 'tables' that can be indexed into via a 32-bit integer, allowing the
  other 32-bits to be used for the locale id. The table can be indexed without
  need of synchronization, but as a Chapel array cannot be operated on while it's
  domain is being updated, we use a simple `sync` variable locally and maintain
  32 segments that are allocated on demand. While descriptors are valid, the segment
  being indexed into will be safe to use without synchronization. A simple skip list
  is utilized to efficiently hash objects to their unique index.
*/

use BlockDist;
use Random;
use BigInteger;

// In the case where you want to use the descriptor table approach for testing or
// because there's some architecture which the assumption that only 48 bits are used
// for the virtual address space is incorrect, you can disable it here.
config param usePointerCompression = true;


// We require this to create a wide pointer after decompression.
extern type wide_ptr_t;
extern type c_nodeid_t;
extern proc chpl_return_wide_ptr_node(c_nodeid_t, c_void_ptr) : wide_ptr_t;

// Hashes a class instance by pointer. We ignore the locale id portion because
// this is used for pointers that refer into our address space.
inline proc hash(obj) : uint where isClass(obj.type) {
  var key =  __primitive("cast", uint(64), __primitive("_wide_get_addr", obj));
  key = (~key) + (key << 21); // key = (key << 21) - key - 1;
  key = key ^ (key >> 24);
  key = (key + (key << 3)) + (key << 8); // key * 265
  key = key ^ (key >> 14);
  key = (key + (key << 2)) + (key << 4); // key * 21
  key = key ^ (key >> 28);
  key = key + (key << 31);
  return key;
}

proc hash(obj) {
  halt("Obj to hash *must* be a class instance...");
}


/*
  The core `GlobalAtomicObject` which manages compression and decompression
  for atomic objects. It should be noted that if numLocales >= 2^17, then any
  object being operated on is added to the descriptor table, which will leak data
  if never freed with '_delete' (although if the memory is reused,
  so will the memory managed by this object). The descriptor table will be cleaned
  up automatically when this goes out of scope.
*/

const compressedAddrMask = 0x0000FFFFFFFFFFFF;
const compressedLocaleIdMask = 0xFFFF;
const tableLocaleIdMask = 0xFFFFFFFF;
const tableIdxMask = 0xFFFFFFFF;
const compressedLocIdOffset = 48;
const tableLocIdOffset = 32;

record GlobalAtomicObject {
  type objType;
  var _atomicVar : atomic uint(64);

  // We maintain a SkipList to manage memory (which is built on top of the descriptor
  // table itself). Currently, as the SkipList is not concurrent (requires Hazard Pointers
  // which requires thread-local storage, or task-local storage in this case).
  var perLocaleDomain = LocaleSpace dmapped Block(boundingBox=LocaleSpace);
  var descriptorTables : [perLocaleDomain] SkipList(objType);
  var descriptorTableLocks : [perLocaleDomain] sync bool;

  // Only if number of locales fits within 16 bits...
  inline proc shouldCompress {
    return usePointerCompression && (numLocales & compressedLocaleIdMask == numLocales);
  }

  inline proc getLocaleId(obj) {
    return chpl_nodeFromLocaleID(__primitive("_wide_get_locale", obj)) : uint;
  }

  inline proc getAddr(obj) {
    return __primitive("cast", uint, __primitive("_wide_get_addr", obj));
  }

  inline proc ourLocaleIdx {
    return perLocaleDomain.localSubdomain().first;
  }

  inline proc maxDescriptorTableLocaleCheck(locId) {
    if (locId & tableLocaleIdMask) != locId {
      halt("LocaleID: ", locId, " > 2^32");
    }
  }

  inline proc maxDescriptorTableIndexCheck(idx) {
    if (idx & tableIdxMask) != idx {
      halt("Idx: ", idx, " > 2^32");
    }
  }

  inline proc widePointerCheck(obj) {
    if !__primitive("is wide pointer", obj) {
      halt(
        "Dummy object created was not a wide pointer!",
        " Assumption: Inside a 'local' block"
      );
    }
  }

  inline proc uintToCVoidPtr(addr) {
    return __primitive("cast", c_void_ptr, addr);
  }

  /*
    Compresses an object into a descriptor.
  */
  proc compress(obj:objType) : uint {
    if obj == nil then return 0;

    // Perform a faster compression by packing the 48 usable bits of the virtual
    // address with 16 bits of the locale/node id.
    if shouldCompress {
      var locId = getLocaleId(obj);
      var addr = getAddr(obj);
      return (locId << compressedLocIdOffset) | (addr & compressedAddrMask);
    }

    // Fallback: We use descriptor tables
    var descriptor : uint;
    on obj {
      // Lock
      descriptorTableLocks[ourLocaleIdx] = true;

      // The skip list returns the node, which contains the table index.
      // We return a node instead of just the index for potential testing purposes.
      var node = descriptorTables[ourLocaleIdx].insert(obj);
      var locId = here.id;
      maxDescriptorTableLocaleCheck(locId);
      var idx = node.idx;
      maxDescriptorTableIndexCheck(idx);
      descriptor = locId << compressedLocIdOffset | idx;

      // Unlock
      descriptorTableLocks[ourLocaleIdx];
    }

    return descriptor;
  }

  /*
    Decompresses a descriptor into the wide pointer object.
  */
  proc decompress(descr:uint) : objType {
    if descr == 0 then return nil;


    // If we have less than 2^16 locales, then we know we performed the
    // faster compression method so we need to decompress it in the same way...
    if shouldCompress {
      var locId = descr >> compressedLocIdOffset;
      var addr = descr & compressedAddrMask;
      var wideptr = chpl_return_wide_ptr_node(locId, uintToCVoidPtr(addr));

      // We've created the wide pointer, but unfortunately Chapel does not support
      // the ability to cast it to the actual object, so we have to do some
      // trickery to get it to work. What we do is we allocate a wide pointer on
      // the stack and memcpy our wideptr into the other. This is needed so we
      // have the same type.
      var newObj : objType;
      on Locales[here.id] do newObj = nil;
      widePointerCheck(newObj);
      c_memcpy(c_ptrTo(newObj), c_ptrTo(wideptr), 16);

      return newObj;
    }

    // Fallback: Descriptor table was used, and we can directly index into the
    // the skip list's memory pool.
    var locId = descr >> tableLocIdOffset;
    var idx = descr & tableIdxMask;
    return descriptorTables[locId : int].memPool.access(idx : int).key;
  }

  inline proc read() : objType {
    return decompress(_atomicVar.read());
  }

  inline proc write(obj:objType) {
    _atomicVar.write(compress(obj));
  }

  inline proc exchange(obj:objType) : objType {
    return decompress(_atomicVar.exchange(compress(obj)));
  }

  // TODO: Save time by comparing the actual objects, not just the hash... if both
  // objects are remote, this is a *significant* cost.
  //  var currDescr = _atomicVar.read();
  //  var currObj = decompress(currDescr);
  //  if currObj == expectedObj then _atomicVar.compareExchangeStrong(currDescr, compress(newObj));
  inline proc compareExchange(expectedObj:objType, newObj:objType) : bool {
    return _atomicVar.compareExchangeStrong(compress(expectedObj), compress(newObj));
  }

  /*
    Deletes the object from our table so it may be recycled.
  */
  inline proc _delete(obj:objType) {
    if obj != nil then on obj {
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first] = true;
      descriptorTables[descriptorTables.domain.localSubdomain().first].remove(obj);
      descriptorTableLocks[descriptorTableLocks.domain.localSubdomain().first];
    }
  }
}

/*
  Skip List implementation

  TODO: This needs to be made concurrent or else this will be the bottleneck
  for our implementation. Hazard Pointers likely needed (this also would be a
  viable example for a need of per-locale garbage collection. Perhaps make it
  possible to declare a zone such that no wide-pointers can be generated and that
  data cannot escape such a zone. It'd be very useful for lock-free/wait-free
  data structures, and should make garbage collection possible if mark-and-sweep
  stop-the-world is *only* for the task inside of the zone. Then again, hazard
  pointers work too...)
*/

param globalAtomicObjectSkipListMaxLevel = 4;

class SkipListNode {
  type keyType;
  var hash : uint(64);
  var key : keyType;
  var idx : uint;
  var forward : globalAtomicObjectSkipListMaxLevel * SkipListNode(keyType);
}

record SkipList {
  type keyType;
  var level = 1;
  var header : SkipListNode(keyType);
  var memPool : FlatObjectPool(SkipListNode(keyType));

  proc SkipList(type keyType) {
    memPool = new FlatObjectPool(SkipListNode(keyType), lambda () { return new SkipListNode(keyType); });
    header = makeNode(level);
  }

  inline proc randomLevel() {
    var levelRNG = makeRandomStream(real);
    var _level = 1;
    while levelRNG.getNext() < 0.5 && _level < globalAtomicObjectSkipListMaxLevel {
      _level = _level + 1;
    }
    return _level;
  }

  // Recycle and clean before usage...
  inline proc makeNode(_level, hash : uint = 0, key : keyType = nil) : SkipListNode(keyType) {
    var (idx, node) = memPool.alloc();
    node.idx = idx : uint;
    node.hash = hash;
    node.key = key;
    for i in 1 .. globalAtomicObjectSkipListMaxLevel do node.forward[i] = nil;

    return node;
  }

  proc remove(key : keyType) {
    // We hash the object to remove from the list. Note: We do not take into
    // account the localeId, so the object *must* not be a wide pointer.
    var hashKey = hash(key);
    var update : globalAtomicObjectSkipListMaxLevel * SkipListNode(keyType);
    var currNode = header;
    var currLevel = level;

    // Start at higher level first...
    while currLevel >= 1 {
      // Maintain same level so long ourHashKey > theirHashKey
      while currNode.forward[currLevel] != nil && currNode.forward[currLevel].hash < hashKey {
        currNode = currNode.forward[currLevel];
      }

      // At this point, ourHashKey < theirHashKey meaning we can't travel along this level.
      // However, because we are before them, that means that we would forward to them (if
      // we were to insert ourselves)
      update[currLevel] = currNode;
      currLevel = currLevel - 1;
    }

    // At this point, we cannot traverse any further than the bottom level.
    currNode = currNode.forward[1];
    if currNode == nil || currNode.hash != hashKey {
      return;
    }

    // It is a match, prepare to delete it by bridging the gap...
    for i in 1 .. level {
      if update[i].forward[i] != currNode then break;
      update[i].forward[i] = currNode.forward[i];
    }

    // Recycle node
    memPool.dealloc(currNode.idx);

    // If we are the highest node in the list, then we can reduce it.
    while level > 1 && header.forward[level] == nil {
      level = level - 1;
    }
  }

  proc insert(key : keyType) : SkipListNode(keyType) {
    // We hash the object to insert into the list. Note: We do not take into
    // account the localeId, so the object *must* not be a wide pointer.
    var hashKey = hash(key);
    var update : globalAtomicObjectSkipListMaxLevel * SkipListNode(keyType);
    var currNode = header;
    var currLevel = level;

    // Start at higher level first...
    while currLevel >= 1 {
      // Maintain same level so long ourHashKey > theirHashKey
      while currNode.forward[currLevel] != nil && currNode.forward[currLevel].hash < hashKey {
{'fraction_non_alphanumeric': 0.09070034443168772, 'fraction_numerical': 0.009538108275192086, 'mean_word_length': 3.197775718257646, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 8, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '333235', 'n_tokens_mistral': 6850, 'n_tokens_neox': 6533, 'n_words': 2613}