Release the GIL around the TA-Lib C calls

README.md

@@ -553,6 +553,18 @@ latest = stream.SMA(close)
 assert (output[-1] - latest) < 0.00001
 ```
 
+## Threads
+
+The indicator functions release the GIL while the C code runs, so calls from several
+threads run concurrently. This applies to the Function API, the Streaming API and the
+Abstract API, which calls the same functions.
+
+Initialization, shutdown and the global settings stay under the GIL:
+`set_unstable_period`, `set_compatibility`, `_ta_set_candle_settings` and
+`_ta_restore_candle_default_settings`. Changing a setting while indicator calls are running
+in other threads is undefined behavior. Change settings when no call is in progress,
+and the next calls in every thread see the new value.
+
 ## Supported Indicators and Functions 📋
 
 We can show all the TA functions supported by TA-Lib, either as a `list` or

talib/_ta_lib.pxd

@@ -192,7 +192,7 @@ cdef extern from "ta-lib/ta_abstract.h":
 
     char* TA_FunctionDescriptionXML()
 
-cdef extern from "ta-lib/ta_func.h":
+cdef extern from "ta-lib/ta_func.h" nogil:
     TA_RetCode TA_ACCBANDS(int startIdx, int endIdx, const double inHigh[], const double inLow[], const double inClose[], int optInTimePeriod, int *outBegIdx, int *outNBElement, double outRealUpperBand[], double outRealMiddleBand[], double outRealLowerBand[])
     int TA_ACCBANDS_Lookback(int optInTimePeriod)
     TA_RetCode TA_ACOS(int startIdx, int endIdx, const double inReal[], int *outBegIdx, int *outNBElement, double outReal[])
@@ -461,8 +461,8 @@ cdef extern from "ta-lib/ta_func.h":
     int TA_ROCR_Lookback(int optInTimePeriod)
     TA_RetCode TA_ROCR100(int startIdx, int endIdx, const double inReal[], int optInTimePeriod, int *outBegIdx, int *outNBElement, double outReal[])
     int TA_ROCR100_Lookback(int optInTimePeriod)
-    TA_RetCode TA_RSI(int startIdx, int endIdx, const double inReal[], int optInTimePeriod, int *outBegIdx, int *outNBElement, double outReal[]) nogil
-    int TA_RSI_Lookback(int optInTimePeriod) nogil
+    TA_RetCode TA_RSI(int startIdx, int endIdx, const double inReal[], int optInTimePeriod, int *outBegIdx, int *outNBElement, double outReal[])
+    int TA_RSI_Lookback(int optInTimePeriod)
     TA_RetCode TA_SAR(int startIdx, int endIdx, const double inHigh[], const double inLow[], double optInAcceleration, double optInMaximum, int *outBegIdx, int *outNBElement, double outReal[])
     int TA_SAR_Lookback(double optInAcceleration, double optInMaximum)
     TA_RetCode TA_SAREXT(int startIdx, int endIdx, const double inHigh[], const double inLow[], double optInStartValue, double optInOffsetOnReverse, double optInAccelerationInitLong, double optInAccelerationLong, double optInAccelerationMaxLong, double optInAccelerationInitShort, double optInAccelerationShort, double optInAccelerationMaxShort, int *outBegIdx, int *outNBElement, double outReal[])

tests/test_threads.py

@@ -0,0 +1,63 @@
+from concurrent.futures import ThreadPoolExecutor
+
+import numpy as np
+from numpy.testing import assert_array_equal
+
+import talib
+from talib import abstract, func, stream
+
+THREADS = 8
+ROUNDS = 50
+
+
+def _calls(series, ford_2012):
+    o, h, lo, c = (ford_2012[k] for k in ('open', 'high', 'low', 'close'))
+    return [
+        lambda: func.EMA(series, timeperiod=30),
+        lambda: func.RSI(series),
+        lambda: func.MACD(series),
+        lambda: func.CDLDOJI(o, h, lo, c),
+        lambda: func.ATR(h, lo, c),
+        lambda: stream.EMA(series, timeperiod=30),
+        lambda: abstract.Function('BBANDS')(series),
+    ]
+
+
+def _flatten(result):
+    if isinstance(result, (tuple, list)):
+        return [np.asarray(r) for r in result]
+    return [np.asarray(result)]
+
+
+def test_functions_are_correct_when_called_concurrently(series, ford_2012):
+    # The C call runs with the GIL released, so several threads may be inside
+    # TA-Lib at once. Every result must still equal the single-threaded one.
+    calls = _calls(series, ford_2012)
+    expected = [_flatten(call()) for call in calls]
+
+    def worker(i):
+        call = calls[i % len(calls)]
+        return i % len(calls), _flatten(call())
+
+    with ThreadPoolExecutor(THREADS) as pool:
+        for which, got in pool.map(worker, range(THREADS * ROUNDS)):
+            for g, e in zip(got, expected[which]):
+                assert_array_equal(g, e)
+
+
+def test_settings_still_work_between_concurrent_calls(series):
+    # Global settings stay under the GIL. Changing one between batches of
+    # concurrent calls must apply to every later call.
+    talib.set_unstable_period('EMA', 0)
+    with ThreadPoolExecutor(THREADS) as pool:
+        base = list(pool.map(lambda _: func.EMA(series, timeperiod=30), range(THREADS)))
+    talib.set_unstable_period('EMA', 10)
+    try:
+        with ThreadPoolExecutor(THREADS) as pool:
+            shifted = list(pool.map(lambda _: func.EMA(series, timeperiod=30), range(THREADS)))
+    finally:
+        talib.set_unstable_period('EMA', 0)
+    for b, s in zip(base, shifted):
+        assert_array_equal(b, base[0])
+        assert_array_equal(s, shifted[0])
+    assert np.isnan(shifted[0]).sum() == np.isnan(base[0]).sum() + 10

tools/generate_func.py

@@ -372,7 +372,8 @@ def cleanup(name):
             else:
                 assert False, args
 
-    print('    retCode = lib.%s(' % name, end=' ')
+    print('    with nogil:')
+    print('        retCode = lib.%s(' % name, end=' ')
 
     for i, arg in enumerate(args):
         if i > 0:

tools/generate_stream.py

@@ -245,7 +245,8 @@ def cleanup(name):
             else:
                 assert False, args
 
-    print('    retCode = lib.%s(' % name, end=' ')
+    print('    with nogil:')
+    print('        retCode = lib.%s(' % name, end=' ')
 
     for i, arg in enumerate(args):
         if i > 0: