BUG: allow replacement in the dispatch cache (numpy#26693)

Allow identical replacement in dispatch cache, since this can be hit with freethreaded Python

* TST: add a failing test for dispatch cache thread safety

* TST: test replace=True error with new semantics

First version added a lock, but maybe some information is interesting:

The test code I added raises an unhandled thread exception every time I run it using a linux x86 laptop without the change to dispatching.c in this PR.

Still not sure why this failure is only hit under gcc.

Here's what I think is happening: right now the locking in the dispatch cache is only internal to the cache itself. The current locking strategy allows a race condition where two threads simultaneously see the cache is empty early in dispatch and then try fill it. The second thread to fill it sees the cache is filled and raises an exception, because the replace parameter for PyArrayIdentityHash_SetItem is 0, so replacements raise an exception.

I don't think it's possible to support this replace feature without moving the lock from the dispatch cache struct to somewhere in the dispatching logic in dispatching.c. We'd need to lock around all the spots we check for an entry in the dispatch cache and then later insert an entry into the cache. Happy to try that approach if it turns out replacing entries in this cache is problematic for some reason. I didn't want to do that since this code is hit every time a ufunc is called so I don't want to add even larger blocks of code that need to be locked around.

In practice, I don't think it's problematic to simply replace entries when this happens, at least not any more problematic than the current approach, since the dispatch cache holds borrowed references to ArrayMethod instances.

Fixes numpy#26690.

Author: ngoldbaum

numpy/_core/src/common/npy_hashtable.c

@@ -210,10 +210,13 @@ _resize_if_necessary(PyArrayIdentityHash *tb)
  * @param value Normally a Python object, no reference counting is done.
  *        use NULL to clear an item.  If the item does not exist, no
  *        action is performed for NULL.
- * @param replace If 1, allow replacements.
+ * @param replace If 1, allow replacements. If replace is 0 an error is raised
+ *        if the stored value is different from the value to be cached. If the
+ *        value to be cached is identical to the stored value, the value to be
+ *        cached is ignored and no error is raised.
  * @returns 0 on success, -1 with a MemoryError or RuntimeError (if an item
- *        is added which is already in the cache).  The caller should avoid
- *        the RuntimeError.
+ *        is added which is already in the cache and replace is 0).  The
+ *        caller should avoid the RuntimeError.
  */
 NPY_NO_EXPORT int
 PyArrayIdentityHash_SetItem(PyArrayIdentityHash *tb,
@@ -228,10 +231,10 @@ PyArrayIdentityHash_SetItem(PyArrayIdentityHash *tb,
 
     PyObject **tb_item = find_item(tb, key);
     if (value != NULL) {
-        if (tb_item[0] != NULL && !replace) {
+        if (tb_item[0] != NULL && tb_item[0] != value && !replace) {
             UNLOCK_TABLE(tb);
             PyErr_SetString(PyExc_RuntimeError,
-                    "Identity cache already includes the item.");
+                    "Identity cache already includes an item with this key.");
             return -1;
         }
         tb_item[0] = value;

numpy/_core/tests/test_hashtable.py

@@ -24,7 +24,12 @@ def test_identity_hashtable(key_length, length):
     res = identityhash_tester(key_length, keys_vals, replace=True)
     assert res is expected
 
-    # check that ensuring one duplicate definitely raises:
-    keys_vals.insert(0, keys_vals[-2])
+    if length == 1:
+        return
+
+    # add a new item with a key that is already used and a new value, this
+    # should error if replace is False, see gh-26690
+    new_key = (keys_vals[1][0], object())
+    keys_vals[0] = new_key
     with pytest.raises(RuntimeError):
         identityhash_tester(key_length, keys_vals)

numpy/_core/tests/test_multithreading.py

@@ -1,4 +1,5 @@
 import concurrent.futures
+import threading
 
 import numpy as np
 import pytest
@@ -30,13 +31,29 @@ def func(seed):
 def test_parallel_ufunc_execution():
     # if the loop data cache or dispatch cache are not thread-safe
     # computing ufuncs simultaneously in multiple threads leads
-    # to a data race
+    # to a data race that causes crashes or spurious exceptions
     def func():
         arr = np.random.random((25,))
         np.isnan(arr)
 
     run_threaded(func, 500)
 
+    # see gh-26690
+    NUM_THREADS = 50
+
+    b = threading.Barrier(NUM_THREADS)
+
+    a = np.ones(1000)
+
+    def f():
+        b.wait()
+        return a.sum()
+
+    threads = [threading.Thread(target=f) for _ in range(NUM_THREADS)]
+
+    [t.start() for t in threads]
+    [t.join() for t in threads]
+
 def test_temp_elision_thread_safety():
     amid = np.ones(50000)
     bmid = np.ones(50000)