Repository navigation
Expand file tree
/
Copy paththreadpoolctl.py
More file actions
2051 lines (1698 loc) · 79 KB
/
Copy paththreadpoolctl.py
File metadata and controls
2051 lines (1698 loc) · 79 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""threadpoolctl
This module provides utilities to introspect native libraries that relies on
thread pools (notably BLAS and OpenMP implementations) and dynamically set the
maximal number of threads they can use.
"""
# License: BSD 3-Clause
# The code to introspect dynamically loaded libraries on POSIX systems is
# adapted from code by Intel developer @anton-malakhov available at
# https://github.com/IntelPython/smp (Copyright (c) 2017, Intel Corporation)
# and also published under the BSD 3-Clause license
from __future__ import annotations
import ctypes
import itertools
import os
import re
import sys
import warnings
from abc import ABC, abstractmethod
from collections.abc import Callable
from contextlib import ContextDecorator
from dataclasses import dataclass, field
from functools import lru_cache, wraps
from threading import Lock, Thread, current_thread
from threading import local as thread_local
from typing import Literal, TypeVar, final
# ctypes.util is not imported on Linux: on CPython 3.14 it allocates a
# process-lifetime CFUNCTYPE callback that is not fork-safe with some libffi
# builds (#225). dllist also uses dl_iterate_phdr internally (#239), which we
# already avoid on Linux via /proc/self/maps on older versions of Python.
dllist = None
if sys.platform != "emscripten" and (
# Python 3.15 doesn't have the CFUNCTYPE anymore:
sys.platform == "linux" and sys.version_info[:2] >= (3, 15)
):
try:
from ctypes.util import dllist
except ImportError:
# CPython before 3.14 does not provide dll inspection.
dllist = None
# Ensure we don't get spurious cache invalidation the first time
# ThreadpoolController is created, as the Windows dll listing methods import
# from there:
if sys.platform == "win32":
from ctypes import wintypes
del wintypes
__version__ = "3.8.0.dev0"
__all__ = [
"LibController",
"ThreadpoolController",
"get_cached_controller",
"register",
"reset_cached_controller",
"threadpool_info",
"threadpool_limits",
]
# One can get runtime errors or even segfaults due to multiple OpenMP libraries
# loaded simultaneously which can happen easily in Python when importing and
# using compiled extensions built with different compilers and therefore
# different OpenMP runtimes in the same program. In particular libiomp (used by
# Intel ICC) and libomp used by clang/llvm tend to crash. This can happen for
# instance when calling BLAS inside a prange. Setting the following environment
# variable allows multiple OpenMP libraries to be loaded. It should not degrade
# performances since we manually take care of potential over-subscription
# performance issues, in sections of the code where nested OpenMP loops can
# happen, by dynamically reconfiguring the inner OpenMP runtime to temporarily
# disable it while under the scope of the outer OpenMP parallel section.
os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "True")
# Hard limit for Windows library paths resolved through GetModuleFileNameExW.
# Kept below unlimited for security reasons; see CHANGES.md and PR #189.
_WINDOWS_MAX_LIBRARY_PATH_LENGTH = 2600
# Structure to cast the info on dynamically loaded library. See
# https://linux.die.net/man/3/dl_iterate_phdr for more details.
_SYSTEM_UINT = ctypes.c_uint64 if sys.maxsize > 2**32 else ctypes.c_uint32
_SYSTEM_UINT_HALF = ctypes.c_uint32 if sys.maxsize > 2**32 else ctypes.c_uint16
_HAS_PROCFS = (
# Only available on Linux:
sys.platform == "linux"
# Make sure /proc is mounted:
and os.path.exists("/proc/self")
)
class _dl_phdr_info(ctypes.Structure):
_fields_ = [
("dlpi_addr", _SYSTEM_UINT), # Base address of object
("dlpi_name", ctypes.c_char_p), # path to the library
("dlpi_phdr", ctypes.c_void_p), # pointer on dlpi_headers
("dlpi_phnum", _SYSTEM_UINT_HALF), # number of elements in dlpi_phdr
]
# The RTLD_NOLOAD flag for loading shared libraries is not defined on Windows.
try:
_RTLD_NOLOAD = os.RTLD_NOLOAD
except AttributeError:
_RTLD_NOLOAD = ctypes.DEFAULT_MODE
# What scope does the API affect?
_ThreadLimitScope = Literal[
# Using the API sets a limit only on the current thread.
"current_thread",
# Using the API sets a limit for every thread in the process; whether or
# not it's a shared process-wide pool or per-thread limit needs to be
# determined some other way.
"process",
# Something else, unexpected; perhaps another variant, perhaps information
# can't be determined under the current configuration.
"unknown",
]
def _determine_thread_limit_scope(
get_n_threads: Callable[[], int], set_n_threads: Callable[[int], None]
) -> _ThreadLimitScope:
"""
Run some experiments to determine the scope of the given get/set API.
This function might not work if you only have one core available.
This function might not work if you set a limit on a library with an
environment variable.
The function works by changing the number of threads in loaded controllers,
which can be a process-wide change. As such, it is not always thread-safe.
An attempt will be made to restore all settings to their previous state,
but the result may be subtly different, e.g. if "unset" has different
semantics than "set to the default returned value".
"""
previous = get_n_threads()
# Some plausible constraints we need to keep in mind:
#
# 1. The API might not allow setting more than the number of (available, or
# physical) cores.
# 2. Some hard limit on number of threads.
try:
# Choose a desired number of threads that is different than the current
# number, and hopefully achievable under the current configuration:
if previous < 2:
expected = 2
else:
# It's 2 or more, so shrink it slightly:
expected = previous - 1
thread_result = []
def get_and_set() -> None:
set_n_threads(expected)
thread_result.append(get_n_threads())
thread = Thread(target=get_and_set)
thread.start()
thread.join()
OpenBLASController.clear_controlling_thread(force=True)
# First, getting in the same thread as a set should always give same
# number, if it's a number in a reasonable range. A possible exception
# fo failing this is if the number of thread is limited by available
# CPU, and only one CPU is available. In that case we can't empirically
# determine how the API works. We try to not reach that point here, but
# you can imagine a thread pool implementation that is aware of
# cgroups, in which case a Docker container limited to one core will
# pass the safety check at the start of the function. Perhaps
# cpu_count() from loky should be moved into this package...
if thread_result != [expected]:
return "unknown"
# Now, check this thread:
if get_n_threads() == expected:
# Setting modified this thread's results too:
return "process"
elif get_n_threads() == previous:
# Setting modified the other thread, but not this one:
return "current_thread"
else:
# No idea what's going on:
return "unknown"
finally:
set_n_threads(previous)
OpenBLASController.clear_controlling_thread(force=True)
class LibController(ABC):
"""Abstract base class for the individual library controllers
A library controller must expose the following class attributes:
- user_api : str
Usually the name of the library or generic specification the library
implements, e.g. "blas" is a specification with different implementations.
- internal_api : str
Usually the name of the library or concrete implementation of some
specification, e.g. "openblas" is an implementation of the "blas"
specification.
- filename_prefixes : tuple
Possible prefixes of the shared library's filename that allow to
identify the library. e.g. "libopenblas" for libopenblas.so.
and implement the following methods: `get_num_threads`, `set_num_threads` and
`get_version`.
Threadpoolctl loops through all the loaded shared libraries and tries to match
the filename of each library with the `filename_prefixes`. If a match is found, a
controller is instantiated and a handler to the library is stored in the `dynlib`
attribute as a `ctypes.CDLL` object. It can be used to access the necessary symbols
of the shared library to implement the above methods.
The following information will be exposed in the info dictionary:
- user_api : standardized API, if any, or a copy of internal_api.
- internal_api : implementation-specific API.
- num_threads : the current thread limit.
- prefix : prefix of the shared library's filename.
- filepath : path to the loaded shared library.
- version : version of the library (if available).
In addition, each library controller may expose internal API specific entries. They
must be set as attributes in the `set_additional_attributes` method.
"""
@final
def __init__(self, *, filepath=None, prefix=None, parent=None):
"""This is not meant to be overriden by subclasses."""
self.parent = parent
self.prefix = prefix
self.filepath = filepath
self.dynlib = _CDLL_CACHE.get_cdll(filepath)
self._symbol_prefix, self._symbol_suffix = self._find_affixes()
self.version = self.get_version()
self.set_additional_attributes()
def info(self, debugging_info: bool = False):
"""Return relevant info wrapped in a dict.
Parameters
----------
debugging_info : bool
Include extra fields which require more intrusive actions to
obtain, and which can't always reliably be determined.
"""
hidden_attrs = ("dynlib", "parent", "_symbol_prefix", "_symbol_suffix")
result = {
"user_api": self.user_api,
"internal_api": self.internal_api,
"num_threads": self.num_threads,
**{k: v for k, v in vars(self).items() if k not in hidden_attrs},
}
if debugging_info:
result["thread_limit_scope"] = _determine_thread_limit_scope(
self.get_num_threads, self.set_num_threads
)
return result
def set_additional_attributes(self):
"""Set additional attributes meant to be exposed in the info dict"""
@property
def num_threads(self):
"""Exposes the current thread limit as a dynamic property
This is not meant to be used or overriden by subclasses.
"""
return self.get_num_threads()
@abstractmethod
def get_num_threads(self):
"""Return the maximum number of threads available to use"""
@abstractmethod
def set_num_threads(self, num_threads):
"""Set the maximum number of threads to use
When possible, implementations of this method should choose a thread
limiting API that only applies to the current thread.
"""
@abstractmethod
def get_version(self):
"""Return the version of the shared library"""
def _find_affixes(self):
"""Return the affixes for the symbols of the shared library"""
return "", ""
def _get_symbol(self, name):
"""Return the symbol of the shared library accounding for the affixes"""
return getattr(
self.dynlib, f"{self._symbol_prefix}{name}{self._symbol_suffix}", None
)
# Internal marker for missing cache lookups:
_MISSING = object()
_T = TypeVar("_T")
@dataclass
class _CDLLCache:
"""
Cache CDLL instances and their associated method results, as well as which
``LibController`` subclass to use for a given shared library file path.
"""
_controller_cache: dict[str, type[LibController] | None] = field(
default_factory=dict
)
_cdll_cache: dict[str, ctypes.CDLL] = field(default_factory=dict)
_method_result_cache: dict[tuple[type[LibController], str, ctypes.CDLL], object] = (
field(default_factory=dict)
)
def _check_prefix(
self, library_basename: str, filename_prefixes: list[str]
) -> str | None:
"""Return the prefix library_basename starts with
Return None if none matches.
"""
for prefix in filename_prefixes:
if library_basename.startswith(prefix):
return prefix
return None
def create_controller(
self, filepath: str, parent: ThreadpoolController
) -> LibController | None:
"""
Create the associated controller, if there is one, relying on cached
info for the given ``filepath``.
"""
result = self._controller_cache.get(filepath, _MISSING)
if result is not _MISSING:
controller_class, filepath, prefix = result
if controller_class is None:
return None
return controller_class(filepath=filepath, prefix=prefix, parent=parent)
# It's not in the cache, so proceed to actually search for it.
# Required to resolve symlinks
original_filepath = filepath
filepath = _realpath(filepath)
# `lower` required to take account of OpenMP dll case on Windows
# (vcomp, VCOMP, Vcomp, ...)
filename = os.path.basename(filepath).lower()
# Loop through supported libraries to find if this filename corresponds
# to a supported one.
for controller_class in _ALL_CONTROLLERS:
# check if filename matches a supported prefix
prefix = self._check_prefix(filename, controller_class.filename_prefixes)
# filename does not match any of the prefixes of the candidate
# library. move to next library.
if prefix is None:
continue
# Legacy workaround for BLAS libraries that conda-forge used to expose
# on Windows as libblas.dll, disambiguated via implementation-specific
# symbols. Current conda-forge stacks no longer load libblas.dll (e.g.
# MKL is exposed as mkl_rt.<version>.dll instead), so this path is
# kept for older installs but cannot be exercised in today's CI.
if prefix == "libblas":
if filename.endswith(".dll"):
libblas = ctypes.CDLL(filepath, _RTLD_NOLOAD)
if not any(
hasattr(libblas, func)
for func in controller_class.check_symbols
):
continue
else:
# Non-Windows libblas DSOs (e.g. from openblas) lack the symbols
# needed to instantiate a controller and would duplicate entries.
continue
# filename matches a prefix. Now we check if the library has the symbols we
# are looking for. If none of the symbols exists, it's very likely not the
# expected library (e.g. a library having a common prefix with one of the
# our supported libraries). Otherwise, create and store the library
# controller.
try:
lib_controller = controller_class(
filepath=filepath, prefix=prefix, parent=parent
)
except OSError:
# Probably because we couldn't load the filepath as a CDLL.
continue
if not hasattr(controller_class, "check_symbols") or any(
hasattr(lib_controller.dynlib, func)
for func in controller_class.check_symbols
):
self._controller_cache[original_filepath] = (
controller_class,
filepath,
prefix,
)
return lib_controller
# Didn't find any matching libraries.
self._controller_cache[original_filepath] = (None, "", "")
return None
def get_cdll(self, filepath: str) -> ctypes.CDLL:
"""
Get the ``CDLL`` for a path, loading if necessary, using a cached
version if it was already loaded.
"""
result = self._cdll_cache.get(filepath, _MISSING)
if result is _MISSING:
result = ctypes.CDLL(filepath, mode=_RTLD_NOLOAD)
self._cdll_cache[filepath] = result
return result
def cache_method_on_dynlib(
self, method: Callable[[LibController], _T]
) -> Callable[[LibController], _T]:
"""
Caching decorator for idempotent read-only methods of
``LibController``, with the cache shared across instances that have the
same ``CDLL`` instance.
"""
cache = self._method_result_cache
name = method.__name__
@wraps(method)
def wrapper(self):
key = (self.__class__, name, self.dynlib)
result = cache.get(key, _MISSING)
if result is _MISSING:
result = method(self)
cache[key] = result
return result
return wrapper
_CDLL_CACHE = _CDLLCache()
class OpenBLASController(LibController):
"""Controller class for OpenBLAS.
When OpenBLAS uses the OpenMP threading layer, each thread that calls into
OpenBLAS gets its own OpenBLAS thread pool, and correspondingly the limit
set by ``threadpoolctl`` is per-thread. Each Python thread can therefore
set its own limit without consideration for other Python threads.
In constrast, when the OpenMP threading layer is not used, there is a
process-wide shared thread pool, so having multiple Python threads change
the limit at the same time makes no sense. Therefore:
1. Thread pool size can only be set by a single thread, the first
thread to call ``set_num_threads()``. In practice this will usually
be the main thread.
2. Other threads silently do nothing.
3. ``clear_controlling_thread()`` can be used to reset the thread that
can meaningfully use the limiting API.
``with threadpool_limits()`` and the function decorator
``threadpool_limits.wrap()`` will automatically clear the controlling
thread when they're done, including when nested.
"""
user_api = "blas"
internal_api = "openblas"
filename_prefixes = (
"libopenblas",
"libblas", # legacy conda-forge Windows shim, see _make_controller_from_path
"libscipy_openblas",
"openblas", # Windows conda package use openblas.dll
)
_symbol_prefixes = ("", "scipy_")
_symbol_suffixes = ("", "64_", "_64")
# All variations of "openblas_get_num_threads", accounting for the affixes
check_symbols = tuple(
f"{prefix}openblas_get_num_threads{suffix}"
for prefix, suffix in itertools.product(_symbol_prefixes, _symbol_suffixes)
)
@_CDLL_CACHE.cache_method_on_dynlib
def _find_affixes(self):
for prefix, suffix in itertools.product(
self._symbol_prefixes, self._symbol_suffixes
):
if hasattr(self.dynlib, f"{prefix}openblas_get_num_threads{suffix}"):
return prefix, suffix
def set_additional_attributes(self):
self.threading_layer = self._get_threading_layer()
self.architecture = self._get_architecture()
def get_num_threads(self):
# See discussion in set_num_threads for details:
if self.threading_layer == "openmp":
symbol = "omp_get_max_threads"
else:
symbol = "openblas_get_num_threads"
get_num_threads_func = self._get_symbol(symbol)
if get_num_threads_func is not None:
return get_num_threads_func()
return None
class _SHARED_STATE:
"""Shared state across instances."""
controlling_thread: Thread | None = None
lock = Lock()
@classmethod
def clear_controlling_thread(cls, force: bool = False):
"""Reset which thread can set limits (on non-OpenMP threading layer).
It's up the caller to ensure the current controlling thread is done.
By default, only the controlling thread has an effect when calling this
method. You can have any thread clear the controlling thread by
setting ``force=True``.
"""
with cls._SHARED_STATE.lock:
if force or cls._SHARED_STATE.controlling_thread is current_thread():
cls._SHARED_STATE.controlling_thread = None
def set_num_threads(self, num_threads):
if self.threading_layer == "openmp":
# Per-thread limits, so no need to be constraining:
return self._real_set_num_threads(num_threads)
# A shared process-wide thread pool, so don't want different Python
# threads fighting over limits:
this_thread = current_thread()
match self._SHARED_STATE.controlling_thread:
case None:
# No thread controls OpenBLAS, so try to take control:
with self._SHARED_STATE.lock:
# Another thread might have changed it, so ensure that's
# not the case:
if self._SHARED_STATE.controlling_thread is None:
self._SHARED_STATE.controlling_thread = this_thread
# Controlling thread should now be set, to either this thread
# or another, so try again:
return self.set_num_threads(num_threads)
case controlling_thread if controlling_thread is this_thread:
# This thread controls OpenBLAS:
return self._real_set_num_threads(num_threads)
case _:
# Some other thread controlls OpenBLAS:
return
def _real_set_num_threads(self, num_threads):
# The OpenBLAS limiting API is process-wide, and we want current thread
# limit if possible. When OpenBLAS is backed by OpenMP, using the
# OpenMP API allows for current thread limiting when OpenMP has that
# behavior. That is the case for libgomp, libomp, and libiomp, what you
# would find on Linux or macOS.
#
# On Windows the Visual C++ OpenMP API is process-wide, unfortunately,
# though this may be fixed if the /openmp:llvm flag is used:
# https://github.com/joblib/threadpoolctl/issues/230
#
# Also worth knowing that before v0.3.34, the OpenBLAS limiting API is
# broken when using OpenMP threading:
# https://github.com/OpenMathLib/OpenBLAS/issues/5806
if self.threading_layer == "openmp":
symbol = "omp_set_num_threads"
else:
symbol = "openblas_set_num_threads"
set_num_threads_func = self._get_symbol(symbol)
if set_num_threads_func is not None:
return set_num_threads_func(num_threads)
return None
@_CDLL_CACHE.cache_method_on_dynlib
def get_version(self):
# None means OpenBLAS is not loaded or version < 0.3.4, since OpenBLAS
# did not expose its version before that.
get_version_func = self._get_symbol("openblas_get_config")
if get_version_func is not None:
get_version_func.restype = ctypes.c_char_p
config = get_version_func().split()
if config[0] == b"OpenBLAS":
return config[1].decode("utf-8")
return None
return None
@_CDLL_CACHE.cache_method_on_dynlib
def _get_threading_layer(self):
"""Return the threading layer of OpenBLAS"""
get_threading_layer_func = self._get_symbol("openblas_get_parallel")
if get_threading_layer_func is not None:
threading_layer = get_threading_layer_func()
if threading_layer == 2:
return "openmp"
elif threading_layer == 1:
return "pthreads"
return "disabled"
return "unknown"
@_CDLL_CACHE.cache_method_on_dynlib
def _get_architecture(self):
"""Return the architecture detected by OpenBLAS"""
get_architecture_func = self._get_symbol("openblas_get_corename")
if get_architecture_func is not None:
get_architecture_func.restype = ctypes.c_char_p
return get_architecture_func().decode("utf-8")
return None
class _ControllingThreadOwnership(thread_local):
"""Count how many paired limit-unlimits have modified OpenBLAS.
Clear controlling thread if it hits zero.
In the future this could potentialy also apply to non-OpenBLAS libraries
that have a shared process pool.
"""
count = 0
def increment(self):
"""Increment ownership."""
self.count += 1
def decrement(self):
"""Decrement ownership."""
self.count -= 1
if self.count <= 0:
OpenBLASController.clear_controlling_thread()
CONTROLLING_THREAD_OWNERSHIP = _ControllingThreadOwnership()
class BLISController(LibController):
"""Controller class for BLIS"""
user_api = "blas"
internal_api = "blis"
filename_prefixes = (
"libblis",
"libblas",
) # libblas: legacy conda-forge Windows shim
check_symbols = (
"bli_thread_get_num_threads",
"bli_thread_set_num_threads",
"bli_info_get_version_str",
"bli_info_get_enable_openmp",
"bli_info_get_enable_pthreads",
"bli_arch_query_id",
"bli_arch_string",
)
def set_additional_attributes(self):
self.threading_layer = self._get_threading_layer()
self.architecture = self._get_architecture()
def get_num_threads(self):
get_func = getattr(self.dynlib, "bli_thread_get_num_threads", lambda: None)
num_threads = get_func()
# by default BLIS is single-threaded and get_num_threads
# returns -1. We map it to 1 for consistency with other libraries.
return 1 if num_threads == -1 else num_threads
def set_num_threads(self, num_threads):
set_func = getattr(
self.dynlib, "bli_thread_set_num_threads", lambda num_threads: None
)
return set_func(num_threads)
@_CDLL_CACHE.cache_method_on_dynlib
def get_version(self):
get_version_ = getattr(self.dynlib, "bli_info_get_version_str", None)
if get_version_ is None:
return None
get_version_.restype = ctypes.c_char_p
return get_version_().decode("utf-8")
@_CDLL_CACHE.cache_method_on_dynlib
def _get_threading_layer(self):
"""Return the threading layer of BLIS"""
if getattr(self.dynlib, "bli_info_get_enable_openmp", lambda: False)():
return "openmp"
elif getattr(self.dynlib, "bli_info_get_enable_pthreads", lambda: False)():
return "pthreads"
return "disabled"
@_CDLL_CACHE.cache_method_on_dynlib
def _get_architecture(self):
"""Return the architecture detected by BLIS"""
bli_arch_query_id = getattr(self.dynlib, "bli_arch_query_id", None)
bli_arch_string = getattr(self.dynlib, "bli_arch_string", None)
if bli_arch_query_id is None or bli_arch_string is None:
return None
# the true restype should be BLIS' arch_t (enum) but int should work
# for us:
bli_arch_query_id.restype = ctypes.c_int
bli_arch_string.restype = ctypes.c_char_p
return bli_arch_string(bli_arch_query_id()).decode("utf-8")
class FlexiBLASController(LibController):
"""Controller class for FlexiBLAS"""
user_api = "blas"
internal_api = "flexiblas"
filename_prefixes = ("libflexiblas",)
check_symbols = (
"flexiblas_get_num_threads",
"flexiblas_set_num_threads",
"flexiblas_get_version",
"flexiblas_list",
"flexiblas_list_loaded",
"flexiblas_current_backend",
)
@property
def loaded_backends(self):
return self._get_backend_list(loaded=True)
@property
def current_backend(self):
return self._get_current_backend()
def info(self, debugging_info: bool = False):
"""Return relevant info wrapped in a dict"""
# We override the info method because the loaded and current backends
# are dynamic properties
exposed_attrs = super().info(debugging_info=debugging_info)
exposed_attrs["loaded_backends"] = self.loaded_backends
exposed_attrs["current_backend"] = self.current_backend
return exposed_attrs
def set_additional_attributes(self):
self.available_backends = self._get_backend_list(loaded=False)
def get_num_threads(self):
get_func = getattr(self.dynlib, "flexiblas_get_num_threads", lambda: None)
num_threads = get_func()
# by default BLIS is single-threaded and get_num_threads
# returns -1. We map it to 1 for consistency with other libraries.
return 1 if num_threads == -1 else num_threads
def set_num_threads(self, num_threads):
set_func = getattr(
self.dynlib, "flexiblas_set_num_threads", lambda num_threads: None
)
return set_func(num_threads)
@_CDLL_CACHE.cache_method_on_dynlib
def get_version(self):
get_version_ = getattr(self.dynlib, "flexiblas_get_version", None)
if get_version_ is None:
return None
major = ctypes.c_int()
minor = ctypes.c_int()
patch = ctypes.c_int()
get_version_(ctypes.byref(major), ctypes.byref(minor), ctypes.byref(patch))
return f"{major.value}.{minor.value}.{patch.value}"
def _get_backend_list(self, loaded=False):
"""Return the list of available backends for FlexiBLAS.
If loaded is False, return the list of available backends from the FlexiBLAS
configuration. If loaded is True, return the list of actually loaded backends.
"""
func_name = f"flexiblas_list{'_loaded' if loaded else ''}"
get_backend_list_ = getattr(self.dynlib, func_name, None)
if get_backend_list_ is None:
return None
n_backends = get_backend_list_(None, 0, 0)
backends = []
for i in range(n_backends):
backend_name = ctypes.create_string_buffer(1024)
get_backend_list_(backend_name, 1024, i)
if backend_name.value.decode("utf-8") != "__FALLBACK__":
# We don't know when to expect __FALLBACK__ but it is not a real
# backend and does not show up when running flexiblas list.
backends.append(backend_name.value.decode("utf-8"))
return backends
def _get_current_backend(self):
"""Return the backend of FlexiBLAS"""
get_backend_ = getattr(self.dynlib, "flexiblas_current_backend", None)
if get_backend_ is None:
return None
backend = ctypes.create_string_buffer(1024)
get_backend_(backend, ctypes.sizeof(backend))
return backend.value.decode("utf-8")
def switch_backend(self, backend):
"""Switch the backend of FlexiBLAS
Parameters
----------
backend : str
The name or the path to the shared library of the backend to switch to. If
the backend is not already loaded, it will be loaded first.
"""
if backend not in self.loaded_backends:
if backend in self.available_backends:
load_func = getattr(self.dynlib, "flexiblas_load_backend", lambda _: -1)
else: # assume backend is a path to a shared library
load_func = getattr(
self.dynlib, "flexiblas_load_backend_library", lambda _: -1
)
res = load_func(str(backend).encode("utf-8"))
if res == -1:
raise RuntimeError(
f"Failed to load backend {backend!r}. It must either be the name of"
" a backend available in the FlexiBLAS configuration "
f"{self.available_backends} or the path to a valid shared library."
)
# Invalidate the global cached controller:
global _CACHED_CONTROLLER
with _CACHE_LOCK:
_CACHED_CONTROLLER = None
# Trigger a new search of loaded shared libraries since loading a new
# backend caused a dlopen.
self.parent._load_libraries()
switch_func = getattr(self.dynlib, "flexiblas_switch", lambda _: -1)
idx = self.loaded_backends.index(backend)
res = switch_func(idx)
if res == -1:
raise RuntimeError(f"Failed to switch to backend {backend!r}.")
class MKLController(LibController):
"""Controller class for MKL"""
user_api = "blas"
internal_api = "mkl"
filename_prefixes = (
"libmkl_rt",
"mkl_rt",
"libblas", # legacy conda-forge Windows shim, see _make_controller_from_path
)
check_symbols = (
"MKL_Get_Max_Threads",
"MKL_Set_Num_Threads_Local",
"MKL_Get_Version_String",
"MKL_Set_Threading_Layer",
)
def set_additional_attributes(self):
self.threading_layer = self._get_threading_layer()
def get_num_threads(self):
get_func = getattr(self.dynlib, "MKL_Get_Max_Threads", lambda: None)
return get_func()
def set_num_threads(self, num_threads):
set_func = getattr(
self.dynlib, "MKL_Set_Num_Threads_Local", lambda num_threads: None
)
return set_func(num_threads)
@_CDLL_CACHE.cache_method_on_dynlib
def get_version(self):
if not hasattr(self.dynlib, "MKL_Get_Version_String"):
return None
res = ctypes.create_string_buffer(200)
self.dynlib.MKL_Get_Version_String(res, 200)
version = res.value.decode("utf-8")
group = re.search(r"Version ([^ ]+) ", version)
if group is not None:
version = group.groups()[0]
return version.strip()
@_CDLL_CACHE.cache_method_on_dynlib
def _get_threading_layer(self):
"""Return the threading layer of MKL"""
# The function mkl_set_threading_layer returns the current threading
# layer. Calling it with an invalid threading layer allows us to safely
# get the threading layer
set_threading_layer = getattr(
self.dynlib, "MKL_Set_Threading_Layer", lambda layer: -1
)
layer_map = {
0: "intel",
1: "sequential",
2: "pgi",
3: "gnu",
4: "tbb",
-1: "not specified",
}
return layer_map[set_threading_layer(-1)]
class OpenMPController(LibController):
"""Controller class for OpenMP"""
user_api = "openmp"
internal_api = "openmp"
filename_prefixes = ("libiomp", "libgomp", "libomp", "vcomp")
check_symbols = (
"omp_get_max_threads",
"omp_get_num_threads",
)
def get_num_threads(self):
get_func = getattr(self.dynlib, "omp_get_max_threads", lambda: None)
return get_func()
def set_num_threads(self, num_threads):
set_func = getattr(self.dynlib, "omp_set_num_threads", lambda num_threads: None)
return set_func(num_threads)
def get_version(self):
# There is no way to get the version number programmatically in OpenMP.
return None
# Controllers for the libraries that we'll look for in the loaded libraries.
# Third party libraries can register their own controllers.
_ALL_CONTROLLERS = [
OpenBLASController,
BLISController,
MKLController,
OpenMPController,
FlexiBLASController,
]
# Helpers for the doc and test names
_ALL_USER_APIS = list({lib.user_api for lib in _ALL_CONTROLLERS})
_ALL_INTERNAL_APIS = [lib.internal_api for lib in _ALL_CONTROLLERS]
_ALL_PREFIXES = list(
{prefix for lib in _ALL_CONTROLLERS for prefix in lib.filename_prefixes}
)
_ALL_BLAS_LIBRARIES = [
lib.internal_api for lib in _ALL_CONTROLLERS if lib.user_api == "blas"
]
_ALL_OPENMP_LIBRARIES = OpenMPController.filename_prefixes
def register(controller):
"""Register a new controller"""
_ALL_CONTROLLERS.append(controller)
_ALL_USER_APIS.append(controller.user_api)
_ALL_INTERNAL_APIS.append(controller.internal_api)
_ALL_PREFIXES.extend(controller.filename_prefixes)
def _format_docstring(*args, **kwargs):
def decorator(o):
if o.__doc__ is not None:
o.__doc__ = o.__doc__.format(*args, **kwargs)
return o
return decorator
@lru_cache(maxsize=10000)
def _realpath(filepath):
"""Small caching wrapper around os.path.realpath to limit system calls"""
return os.path.realpath(filepath)
@_format_docstring(USER_APIS=list(_ALL_USER_APIS), INTERNAL_APIS=_ALL_INTERNAL_APIS)
def threadpool_info(debugging_info: bool = False):
"""Return the maximal number of threads for each detected library.
Return a list with all the supported libraries that have been found. Each
library is represented by a dict with the following information: