Use hashtab in forder()

Since range_str() runs a parallel OpenMP loop that may update the hash
table in a critical section, use a special form of hash_set that returns
the newly reallocated hash table instead of overwriting it in place.

Author: aitap

src/data.table.h

@@ -351,9 +351,11 @@ typedef struct hash_tab hashtab;
 // See vmaxget()/vmaxset() if you need to unprotect it manually.
 hashtab * hash_create(size_t n);
 // Inserts a new key-value pair into the hash, or overwrites an existing value.
-// Will raise an R error if inserting more than n elements.
+// Will grow the table in a thread-unsafe manner if needed.
 // Don't try to insert a null pointer, nothing good will come out of it.
 void hash_set(hashtab *, SEXP key, R_xlen_t value);
+// Same as hash_set, but returns the new hash table pointer, which the caller may assign atomically in a thread-safe manner.
+hashtab *hash_set_shared(hashtab *, SEXP key, R_xlen_t value);
 // Returns the value corresponding to the key present in the hash, otherwise returns ifnotfound.
 R_xlen_t hash_lookup(const hashtab *, SEXP key, R_xlen_t ifnotfound);
 

src/forder.c

@@ -291,28 +291,24 @@ static void cradix(SEXP *x, int n)
   free(cradix_xtmp);   cradix_xtmp=NULL;
 }
 
-static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max, int *out_na_count, bool *out_anynotascii, bool *out_anynotutf8, dhashtab * marks)
+static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max, int *out_na_count, bool *out_anynotascii, bool *out_anynotutf8, hashtab * marks)
 // group numbers are left in truelength to be fetched by WRITE_KEY
 {
   int na_count=0;
   bool anynotascii=false, anynotutf8=false;
   if (ustr_n!=0) internal_error_with_cleanup(__func__, "ustr isn't empty when starting range_str: ustr_n=%d, ustr_alloc=%d", ustr_n, ustr_alloc);  // # nocov
   if (ustr_maxlen!=0) internal_error_with_cleanup(__func__, "ustr_maxlen isn't 0 when starting range_str");  // # nocov
-  #pragma omp parallel for num_threads(getDTthreads(n, true))
+  #pragma omp parallel for num_threads(getDTthreads(n, true)) shared(marks)
   for(int i=0; i<n; i++) {
     SEXP s = x[i];
     if (s==NA_STRING) {
       #pragma omp atomic update
       na_count++;
       continue;
     }
-    // Why is it acceptable to call dhash_lookup when marks can be shared between threads?
-    // 1. We have rwlocks to avoid crashing on a pointer being invalidated by a different thread.
-    // 2. We check again after entering the critical section.
-    // 3. We only change the marks from zero to nonzero, so once a nonzero value is seen, it must be correct.
-    if (dhash_lookup(marks,s,0)<0) continue; // seen this group before
+    if (hash_lookup(marks,s,0)<0) continue; // seen this group before
     #pragma omp critical(range_str_write)
-    if (dhash_lookup(marks,s,0)>=0) {  // another thread may have set it while I was waiting, so check it again
+    if (hash_lookup(marks,s,0)>=0) {  // another thread may have set it while I was waiting, so check it again
       // now save unique SEXP in ustr so we can loop sort uniques when sorting too
       if (ustr_alloc<=ustr_n) {
         ustr_alloc = (ustr_alloc==0) ? 16384 : ustr_alloc*2;  // small initial guess, negligible time to alloc 128KiB (32 pages)
@@ -321,7 +317,9 @@ static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max
         if (ustr==NULL) STOP(_("Unable to realloc %d * %d bytes in range_str"), ustr_alloc, (int)sizeof(SEXP));  // # nocov
       }
       ustr[ustr_n++] = s;
-      dhash_set(marks, s, -ustr_n);  // unique in any order is fine. first-appearance order is achieved later in count_group
+      // Under the OpenMP memory model, if the hash table is expanded, we have to explicitly update the pointer.
+      #pragma omp atomic write
+      marks = hash_set_shared(marks, s, -ustr_n); // unique in any order is fine. first-appearance order is achieved later in count_group
       if (LENGTH(s)>ustr_maxlen) ustr_maxlen=LENGTH(s);
       if (!anynotutf8 &&    // even if anynotascii we still want to know if anynotutf8, and anynotutf8 implies anynotascii already
             !IS_ASCII(s)) { // anynotutf8 implies anynotascii and IS_ASCII will be cheaper than IS_UTF8, so start with this one
@@ -354,11 +352,11 @@ static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max
       if (LENGTH(s)>ustr_maxlen) ustr_maxlen=LENGTH(s);
     }
     cradix(ustr3, ustr_n);  // sort to detect possible duplicates after converting; e.g. two different non-utf8 map to the same utf8
-    dhash_set(marks, ustr3[0], -1);
+    hash_set(marks, ustr3[0], -1);
     int o = -1;
     for (int i=1; i<ustr_n; i++) {
       if (ustr3[i] == ustr3[i-1]) continue;  // use the same o for duplicates
-      dhash_set(marks, ustr3[i], --o);
+      hash_set(marks, ustr3[i], --o);
     }
     // now use the 1-1 mapping from ustr to ustr2 to get the ordering back into original ustr, being careful to reset tl to 0
     int *tl = malloc(sizeof(*tl) * ustr_n);
@@ -367,9 +365,9 @@ static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max
       STOP(_("Failed to alloc tl when converting strings to UTF8"));  // # nocov
     }
     const SEXP *tt = STRING_PTR_RO(ustr2);
-    for (int i=0; i<ustr_n; i++) tl[i] = dhash_lookup(marks, tt[i], 0);   // fetches the o in ustr3 into tl which is ordered by ustr
-    for (int i=0; i<ustr_n; i++) dhash_set(marks, ustr3[i], 0);    // reset to 0 tl of the UTF8 (and possibly non-UTF in ustr too)
-    for (int i=0; i<ustr_n; i++) dhash_set(marks, ustr[i], tl[i]); // put back the o into ustr's tl
+    for (int i=0; i<ustr_n; i++) tl[i] = hash_lookup(marks, tt[i], 0);   // fetches the o in ustr3 into tl which is ordered by ustr
+    for (int i=0; i<ustr_n; i++) hash_set(marks, ustr3[i], 0);    // reset to 0 tl of the UTF8 (and possibly non-UTF in ustr too)
+    for (int i=0; i<ustr_n; i++) hash_set(marks, ustr[i], tl[i]); // put back the o into ustr's tl
     free(tl);
     free(ustr3);
     UNPROTECT(1);
@@ -382,7 +380,7 @@ static void range_str(const SEXP *x, int n, uint64_t *out_min, uint64_t *out_max
       // that this is always ascending; descending is done in WRITE_KEY using max-this
       cradix(ustr, ustr_n);  // sorts ustr in-place by reference. assumes NA_STRING not present.
       for(int i=0; i<ustr_n; i++)     // save ordering in the CHARSXP. negative so as to distinguish with R's own usage.
-        dhash_set(marks, ustr[i], -i-1);
+        hash_set(marks, ustr[i], -i-1);
     }
     // else group appearance order was already saved to tl in the first pass
   }
@@ -573,7 +571,7 @@ SEXP forder(SEXP DT, SEXP by, SEXP retGrpArg, SEXP retStatsArg, SEXP sortGroupsA
         STOP(_("Item %d of order (ascending/descending) is %d. Must be +1 or -1."), col+1, sortType);
     }
     //Rprintf(_("sortType = %d\n"), sortType);
-    dhashtab * marks = NULL; // only used for STRSXP below
+    hashtab * marks = NULL; // only used for STRSXP below
     switch(TYPEOF(x)) {
     case INTSXP : case LGLSXP :  // TODO skip LGL and assume range [0,1]
       range_i32(INTEGER(x), nrow, &min, &max, &na_count);
@@ -610,8 +608,7 @@ SEXP forder(SEXP DT, SEXP by, SEXP retGrpArg, SEXP retStatsArg, SEXP sortGroupsA
       break;
     case STRSXP :
       // need2utf8 now happens inside range_str on the uniques
-      marks = dhash_create(4096); // relatively small to allocate, can grow exponentially later
-      PROTECT(marks->prot); n_protect++;
+      marks = hash_create(4096); // relatively small to allocate, can grow exponentially later
       range_str(STRING_PTR_RO(x), nrow, &min, &max, &na_count, &anynotascii, &anynotutf8, marks);
       break;
     default:
@@ -771,7 +768,7 @@ SEXP forder(SEXP DT, SEXP by, SEXP retGrpArg, SEXP retStatsArg, SEXP sortGroupsA
           if (nalast==-1) anso[i]=0;
           elem = naval;
         } else {
-          elem = -dhash_lookup(marks, xd[i], 0);
+          elem = -hash_lookup(marks, xd[i], 0);
         }
         WRITE_KEY
       }}

src/hash.c

@@ -73,17 +73,17 @@ static R_INLINE size_t hash_index2(SEXP key, uintptr_t multiplier) {
   return ((ptr & 0x0fffffff) * multiplier) | 1;
 }
 
-static R_INLINE void hash_rehash(hashtab *h) {
+static R_INLINE hashtab *hash_rehash(const hashtab *h) {
   size_t new_size = h->size * 2;
   hashtab *new_h = hash_create_(new_size, default_load_factor);
 
   for (size_t i = 0; i < h->size; ++i) {
     if (h->table[i].key) hash_set(new_h, h->table[i].key, h->table[i].value);
   }
-  *h = *new_h;
+  return new_h;
 }
 
-void hash_set(hashtab *h, SEXP key, R_xlen_t value) {
+static bool hash_set_(hashtab *h, SEXP key, R_xlen_t value) {
   size_t mask = h->size - 1;
   size_t h1 = hash_index1(key, h->multiplier1) & mask;
   size_t h2 = hash_index2(key, h->multiplier2) & mask;
@@ -99,18 +99,30 @@ void hash_set(hashtab *h, SEXP key, R_xlen_t value) {
       h->table[idx].key = key;
       h->table[idx].value = value;
       h->free--;
-      return;
+      return true;
     }
 
     if (h->table[idx].key == key) {
       h->table[idx].value = value;
-      return;
+      return true;
     }
   }
 
   // need to rehash
-  hash_rehash(h);
-  hash_set(h, key, value);
+  return false;
+}
+
+void hash_set(hashtab *h, SEXP key, R_xlen_t value) {
+  if (!hash_set_(h, key, value))
+    *h = *hash_rehash(h);
+  (void)hash_set_(h, key, value); // must succeed on the second try
+}
+
+hashtab *hash_set_shared(hashtab *h, SEXP key, R_xlen_t value) {
+  if (!hash_set_(h, key, value))
+    h = hash_rehash(h);
+  (void)hash_set_(h, key, value);
+  return h;
 }
 
 R_xlen_t hash_lookup(const hashtab *h, SEXP key, R_xlen_t ifnotfound) {