Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes15kdownloads
test_simd.py1346 linesDownload Raw Back to tests
1# NOTE: Please avoid the use of numpy.testing since NPYV intrinsics
2# may be involved in their functionality.
3import itertools
4import math
5import operator
6import re
7
8import pytest
9
10from numpy._core._multiarray_umath import __cpu_baseline__
11from numpy._core._simd import clear_floatstatus, get_floatstatus, targets
12
13
14def check_floatstatus(divbyzero=False, overflow=False,
15                      underflow=False, invalid=False,
16                      all=False):
17    #define NPY_FPE_DIVIDEBYZERO  1
18    #define NPY_FPE_OVERFLOW      2
19    #define NPY_FPE_UNDERFLOW     4
20    #define NPY_FPE_INVALID       8
21    err = get_floatstatus()
22    ret = (all or divbyzero) and (err & 1) != 0
23    ret |= (all or overflow) and (err & 2) != 0
24    ret |= (all or underflow) and (err & 4) != 0
25    ret |= (all or invalid) and (err & 8) != 0
26    return ret
27
28class _Test_Utility:
29    # submodule of the desired SIMD extension, e.g. targets["AVX512F"]
30    npyv = None
31    # the current data type suffix e.g. 's8'
32    sfx = None
33    # target name can be 'baseline' or one or more of CPU features
34    target_name = None
35
36    def __getattr__(self, attr):
37        """
38        To call NPV intrinsics without the attribute 'npyv' and
39        auto suffixing intrinsics according to class attribute 'sfx'
40        """
41        return getattr(self.npyv, attr + "_" + self.sfx)
42
43    def _x2(self, intrin_name):
44        return getattr(self.npyv, f"{intrin_name}_{self.sfx}x2")
45
46    def _data(self, start=None, count=None, reverse=False):
47        """
48        Create list of consecutive numbers according to number of vector's lanes.
49        """
50        if start is None:
51            start = 1
52        if count is None:
53            count = self.nlanes
54        rng = range(start, start + count)
55        if reverse:
56            rng = reversed(rng)
57        if self._is_fp():
58            return [x / 1.0 for x in rng]
59        return list(rng)
60
61    def _is_unsigned(self):
62        return self.sfx[0] == 'u'
63
64    def _is_signed(self):
65        return self.sfx[0] == 's'
66
67    def _is_fp(self):
68        return self.sfx[0] == 'f'
69
70    def _scalar_size(self):
71        return int(self.sfx[1:])
72
73    def _int_clip(self, seq):
74        if self._is_fp():
75            return seq
76        max_int = self._int_max()
77        min_int = self._int_min()
78        return [min(max(v, min_int), max_int) for v in seq]
79
80    def _int_max(self):
81        if self._is_fp():
82            return None
83        max_u = self._to_unsigned(self.setall(-1))[0]
84        if self._is_signed():
85            return max_u // 2
86        return max_u
87
88    def _int_min(self):
89        if self._is_fp():
90            return None
91        if self._is_unsigned():
92            return 0
93        return -(self._int_max() + 1)
94
95    def _true_mask(self):
96        max_unsig = getattr(self.npyv, "setall_u" + self.sfx[1:])(-1)
97        return max_unsig[0]
98
99    def _to_unsigned(self, vector):
100        if isinstance(vector, (list, tuple)):
101            return getattr(self.npyv, "load_u" + self.sfx[1:])(vector)
102        else:
103            sfx = vector.__name__.replace("npyv_", "")
104            if sfx[0] == "b":
105                cvt_intrin = "cvt_u{0}_b{0}"
106            else:
107                cvt_intrin = "reinterpret_u{0}_{1}"
108            return getattr(self.npyv, cvt_intrin.format(sfx[1:], sfx))(vector)
109
110    def _pinfinity(self):
111        return float("inf")
112
113    def _ninfinity(self):
114        return -float("inf")
115
116    def _nan(self):
117        return float("nan")
118
119    def _cpu_features(self):
120        target = self.target_name
121        if target == "baseline":
122            target = __cpu_baseline__
123        else:
124            target = target.split('__')  # multi-target separator
125        return ' '.join(target)
126
127class _SIMD_BOOL(_Test_Utility):
128    """
129    To test all boolean vector types at once
130    """
131    def _nlanes(self):
132        return getattr(self.npyv, "nlanes_u" + self.sfx[1:])
133
134    def _data(self, start=None, count=None, reverse=False):
135        true_mask = self._true_mask()
136        rng = range(self._nlanes())
137        if reverse:
138            rng = reversed(rng)
139        return [true_mask if x % 2 else 0 for x in rng]
140
141    def _load_b(self, data):
142        len_str = self.sfx[1:]
143        load = getattr(self.npyv, "load_u" + len_str)
144        cvt = getattr(self.npyv, f"cvt_b{len_str}_u{len_str}")
145        return cvt(load(data))
146
147    def test_operators_logical(self):
148        """
149        Logical operations for boolean types.
150        Test intrinsics:
151            npyv_xor_##SFX, npyv_and_##SFX, npyv_or_##SFX, npyv_not_##SFX,
152            npyv_andc_b8, npvy_orc_b8, nvpy_xnor_b8
153        """
154        data_a = self._data()
155        data_b = self._data(reverse=True)
156        vdata_a = self._load_b(data_a)
157        vdata_b = self._load_b(data_b)
158
159        data_and = [a & b for a, b in zip(data_a, data_b)]
160        vand = getattr(self, "and")(vdata_a, vdata_b)
161        assert vand == data_and
162
163        data_or = [a | b for a, b in zip(data_a, data_b)]
164        vor = getattr(self, "or")(vdata_a, vdata_b)
165        assert vor == data_or
166
167        data_xor = [a ^ b for a, b in zip(data_a, data_b)]
168        vxor = self.xor(vdata_a, vdata_b)
169        assert vxor == data_xor
170
171        vnot = getattr(self, "not")(vdata_a)
172        assert vnot == data_b
173
174        # among the boolean types, andc, orc and xnor only support b8
175        if self.sfx not in ("b8"):
176            return
177
178        data_andc = [(a & ~b) & 0xFF for a, b in zip(data_a, data_b)]
179        vandc = self.andc(vdata_a, vdata_b)
180        assert data_andc == vandc
181
182        data_orc = [(a | ~b) & 0xFF for a, b in zip(data_a, data_b)]
183        vorc = self.orc(vdata_a, vdata_b)
184        assert data_orc == vorc
185
186        data_xnor = [~(a ^ b) & 0xFF for a, b in zip(data_a, data_b)]
187        vxnor = self.xnor(vdata_a, vdata_b)
188        assert data_xnor == vxnor
189
190    def test_tobits(self):
191        data2bits = lambda data: sum(int(x != 0) << i for i, x in enumerate(data, 0))
192        for data in (self._data(), self._data(reverse=True)):
193            vdata = self._load_b(data)
194            data_bits = data2bits(data)
195            tobits = self.tobits(vdata)
196            bin_tobits = bin(tobits)
197            assert bin_tobits == bin(data_bits)
198
199    def test_pack(self):
200        """
201        Pack multiple vectors into one
202        Test intrinsics:
203            npyv_pack_b8_b16
204            npyv_pack_b8_b32
205            npyv_pack_b8_b64
206        """
207        if self.sfx not in ("b16", "b32", "b64"):
208            return
209        # create the vectors
210        data = self._data()
211        rdata = self._data(reverse=True)
212        vdata = self._load_b(data)
213        vrdata = self._load_b(rdata)
214        pack_simd = getattr(self.npyv, f"pack_b8_{self.sfx}")
215        # for scalar execution, concatenate the elements of the multiple lists
216        # into a single list (spack) and then iterate over the elements of
217        # the created list applying a mask to capture the first byte of them.
218        if self.sfx == "b16":
219            spack = [(i & 0xFF) for i in (list(rdata) + list(data))]
220            vpack = pack_simd(vrdata, vdata)
221        elif self.sfx == "b32":
222            spack = [(i & 0xFF) for i in (2 * list(rdata) + 2 * list(data))]
223            vpack = pack_simd(vrdata, vrdata, vdata, vdata)
224        elif self.sfx == "b64":
225            spack = [(i & 0xFF) for i in (4 * list(rdata) + 4 * list(data))]
226            vpack = pack_simd(vrdata, vrdata, vrdata, vrdata,
227                               vdata,  vdata,  vdata,  vdata)
228        assert vpack == spack
229
230    @pytest.mark.parametrize("intrin", ["any", "all"])
231    @pytest.mark.parametrize("data", (
232        [-1, 0],
233        [0, -1],
234        [-1],
235        [0]
236    ))
237    def test_operators_crosstest(self, intrin, data):
238        """
239        Test intrinsics:
240            npyv_any_##SFX
241            npyv_all_##SFX
242        """
243        data_a = self._load_b(data * self._nlanes())
244        func = eval(intrin)
245        intrin = getattr(self, intrin)
246        desired = func(data_a)
247        simd = intrin(data_a)
248        assert not not simd == desired
249
250class _SIMD_INT(_Test_Utility):
251    """
252    To test all integer vector types at once
253    """
254    def test_operators_shift(self):
255        if self.sfx in ("u8", "s8"):
256            return
257
258        data_a = self._data(self._int_max() - self.nlanes)
259        data_b = self._data(self._int_min(), reverse=True)
260        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
261
262        for count in range(self._scalar_size()):
263            # load to cast
264            data_shl_a = self.load([a << count for a in data_a])
265            # left shift
266            shl = self.shl(vdata_a, count)
267            assert shl == data_shl_a
268            # load to cast
269            data_shr_a = self.load([a >> count for a in data_a])
270            # right shift
271            shr = self.shr(vdata_a, count)
272            assert shr == data_shr_a
273
274        # shift by zero or max or out-range immediate constant is not
275        # applicable and illogical
276        for count in range(1, self._scalar_size()):
277            # load to cast
278            data_shl_a = self.load([a << count for a in data_a])
279            # left shift by an immediate constant
280            shli = self.shli(vdata_a, count)
281            assert shli == data_shl_a
282            # load to cast
283            data_shr_a = self.load([a >> count for a in data_a])
284            # right shift by an immediate constant
285            shri = self.shri(vdata_a, count)
286            assert shri == data_shr_a
287
288    def test_arithmetic_subadd_saturated(self):
289        if self.sfx in ("u32", "s32", "u64", "s64"):
290            return
291
292        data_a = self._data(self._int_max() - self.nlanes)
293        data_b = self._data(self._int_min(), reverse=True)
294        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
295
296        data_adds = self._int_clip([a + b for a, b in zip(data_a, data_b)])
297        adds = self.adds(vdata_a, vdata_b)
298        assert adds == data_adds
299
300        data_subs = self._int_clip([a - b for a, b in zip(data_a, data_b)])
301        subs = self.subs(vdata_a, vdata_b)
302        assert subs == data_subs
303
304    def test_math_max_min(self):
305        data_a = self._data()
306        data_b = self._data(self.nlanes)
307        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
308
309        data_max = [max(a, b) for a, b in zip(data_a, data_b)]
310        simd_max = self.max(vdata_a, vdata_b)
311        assert simd_max == data_max
312
313        data_min = [min(a, b) for a, b in zip(data_a, data_b)]
314        simd_min = self.min(vdata_a, vdata_b)
315        assert simd_min == data_min
316
317    @pytest.mark.parametrize("start", [-100, -10000, 0, 100, 10000])
318    def test_reduce_max_min(self, start):
319        """
320        Test intrinsics:
321            npyv_reduce_max_##sfx
322            npyv_reduce_min_##sfx
323        """
324        vdata_a = self.load(self._data(start))
325        assert self.reduce_max(vdata_a) == max(vdata_a)
326        assert self.reduce_min(vdata_a) == min(vdata_a)
327
328
329class _SIMD_FP32(_Test_Utility):
330    """
331    To only test single precision
332    """
333    def test_conversions(self):
334        """
335        Round to nearest even integer, assume CPU control register is set to rounding.
336        Test intrinsics:
337            npyv_round_s32_##SFX
338        """
339        features = self._cpu_features()
340        if not self.npyv.simd_f64 and re.match(r".*(NEON|ASIMD)", features):
341            # very costly to emulate nearest even on Armv7
342            # instead we round halves to up. e.g. 0.5 -> 1, -0.5 -> -1
343            _round = lambda v: int(v + (0.5 if v >= 0 else -0.5))
344        else:
345            _round = round
346        vdata_a = self.load(self._data())
347        vdata_a = self.sub(vdata_a, self.setall(0.5))
348        data_round = [_round(x) for x in vdata_a]
349        vround = self.round_s32(vdata_a)
350        assert vround == data_round
351
352class _SIMD_FP64(_Test_Utility):
353    """
354    To only test double precision
355    """
356    def test_conversions(self):
357        """
358        Round to nearest even integer, assume CPU control register is set to rounding.
359        Test intrinsics:
360            npyv_round_s32_##SFX
361        """
362        vdata_a = self.load(self._data())
363        vdata_a = self.sub(vdata_a, self.setall(0.5))
364        vdata_b = self.mul(vdata_a, self.setall(-1.5))
365        data_round = [round(x) for x in list(vdata_a) + list(vdata_b)]
366        vround = self.round_s32(vdata_a, vdata_b)
367        assert vround == data_round
368
369class _SIMD_FP(_Test_Utility):
370    """
371    To test all float vector types at once
372    """
373    def test_arithmetic_fused(self):
374        vdata_a, vdata_b, vdata_c = [self.load(self._data())] * 3
375        vdata_cx2 = self.add(vdata_c, vdata_c)
376        # multiply and add, a*b + c
377        data_fma = self.load([a * b + c for a, b, c in zip(vdata_a, vdata_b, vdata_c)])
378        fma = self.muladd(vdata_a, vdata_b, vdata_c)
379        assert fma == data_fma
380        # multiply and subtract, a*b - c
381        fms = self.mulsub(vdata_a, vdata_b, vdata_c)
382        data_fms = self.sub(data_fma, vdata_cx2)
383        assert fms == data_fms
384        # negate multiply and add, -(a*b) + c
385        nfma = self.nmuladd(vdata_a, vdata_b, vdata_c)
386        data_nfma = self.sub(vdata_cx2, data_fma)
387        assert nfma == data_nfma
388        # negate multiply and subtract, -(a*b) - c
389        nfms = self.nmulsub(vdata_a, vdata_b, vdata_c)
390        data_nfms = self.mul(data_fma, self.setall(-1))
391        assert nfms == data_nfms
392        # multiply, add for odd elements and subtract even elements.
393        # (a * b) -+ c
394        fmas = list(self.muladdsub(vdata_a, vdata_b, vdata_c))
395        assert fmas[0::2] == list(data_fms)[0::2]
396        assert fmas[1::2] == list(data_fma)[1::2]
397
398    def test_abs(self):
399        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
400        data = self._data()
401        vdata = self.load(self._data())
402
403        abs_cases = ((-0, 0), (ninf, pinf), (pinf, pinf), (nan, nan))
404        for case, desired in abs_cases:
405            data_abs = [desired] * self.nlanes
406            vabs = self.abs(self.setall(case))
407            assert vabs == pytest.approx(data_abs, nan_ok=True)
408
409        vabs = self.abs(self.mul(vdata, self.setall(-1)))
410        assert vabs == data
411
412    def test_sqrt(self):
413        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
414        data = self._data()
415        vdata = self.load(self._data())
416
417        sqrt_cases = ((-0.0, -0.0), (0.0, 0.0), (-1.0, nan), (ninf, nan), (pinf, pinf))
418        for case, desired in sqrt_cases:
419            data_sqrt = [desired] * self.nlanes
420            sqrt = self.sqrt(self.setall(case))
421            assert sqrt == pytest.approx(data_sqrt, nan_ok=True)
422
423        # load to truncate precision
424        data_sqrt = self.load([math.sqrt(x) for x in data])
425        sqrt = self.sqrt(vdata)
426        assert sqrt == data_sqrt
427
428    def test_square(self):
429        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
430        data = self._data()
431        vdata = self.load(self._data())
432        # square
433        square_cases = ((nan, nan), (pinf, pinf), (ninf, pinf))
434        for case, desired in square_cases:
435            data_square = [desired] * self.nlanes
436            square = self.square(self.setall(case))
437            assert square == pytest.approx(data_square, nan_ok=True)
438
439        data_square = [x * x for x in data]
440        square = self.square(vdata)
441        assert square == data_square
442
443    @pytest.mark.parametrize("intrin, func", [("ceil", math.ceil),
444    ("trunc", math.trunc), ("floor", math.floor), ("rint", round)])
445    def test_rounding(self, intrin, func):
446        """
447        Test intrinsics:
448            npyv_rint_##SFX
449            npyv_ceil_##SFX
450            npyv_trunc_##SFX
451            npyv_floor##SFX
452        """
453        intrin_name = intrin
454        intrin = getattr(self, intrin)
455        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
456        # special cases
457        round_cases = ((nan, nan), (pinf, pinf), (ninf, ninf))
458        for case, desired in round_cases:
459            data_round = [desired] * self.nlanes
460            _round = intrin(self.setall(case))
461            assert _round == pytest.approx(data_round, nan_ok=True)
462
463        for x in range(0, 2**20, 256**2):
464            for w in (-1.05, -1.10, -1.15, 1.05, 1.10, 1.15):
465                data = self.load([(x + a) * w for a in range(self.nlanes)])
466                data_round = [func(x) for x in data]
467                _round = intrin(data)
468                assert _round == data_round
469
470        # test large numbers
471        for i in (
472            1.1529215045988576e+18, 4.6116860183954304e+18,
473            5.902958103546122e+20, 2.3611832414184488e+21
474        ):
475            x = self.setall(i)
476            y = intrin(x)
477            data_round = [func(n) for n in x]
478            assert y == data_round
479
480        # signed zero
481        if intrin_name == "floor":
482            data_szero = (-0.0,)
483        else:
484            data_szero = (-0.0, -0.25, -0.30, -0.45, -0.5)
485
486        for w in data_szero:
487            _round = self._to_unsigned(intrin(self.setall(w)))
488            data_round = self._to_unsigned(self.setall(-0.0))
489            assert _round == data_round
490
491    @pytest.mark.parametrize("intrin", [
492        "max", "maxp", "maxn", "min", "minp", "minn"
493    ])
494    def test_max_min(self, intrin):
495        """
496        Test intrinsics:
497            npyv_max_##sfx
498            npyv_maxp_##sfx
499            npyv_maxn_##sfx
500            npyv_min_##sfx
501            npyv_minp_##sfx
502            npyv_minn_##sfx
503            npyv_reduce_max_##sfx
504            npyv_reduce_maxp_##sfx
505            npyv_reduce_maxn_##sfx
506            npyv_reduce_min_##sfx
507            npyv_reduce_minp_##sfx
508            npyv_reduce_minn_##sfx
509        """
510        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
511        chk_nan = {"xp": 1, "np": 1, "nn": 2, "xn": 2}.get(intrin[-2:], 0)
512        func = eval(intrin[:3])
513        reduce_intrin = getattr(self, "reduce_" + intrin)
514        intrin = getattr(self, intrin)
515        hf_nlanes = self.nlanes // 2
516
517        cases = (
518            ([0.0, -0.0], [-0.0, 0.0]),
519            ([10, -10],  [10, -10]),
520            ([pinf, 10], [10, ninf]),
521            ([10, pinf], [ninf, 10]),
522            ([10, -10], [10, -10]),
523            ([-10, 10], [-10, 10])
524        )
525        for op1, op2 in cases:
526            vdata_a = self.load(op1 * hf_nlanes)
527            vdata_b = self.load(op2 * hf_nlanes)
528            data = func(vdata_a, vdata_b)
529            simd = intrin(vdata_a, vdata_b)
530            assert simd == data
531            data = func(vdata_a)
532            simd = reduce_intrin(vdata_a)
533            assert simd == data
534
535        if not chk_nan:
536            return
537        if chk_nan == 1:
538            test_nan = lambda a, b: (
539                b if math.isnan(a) else a if math.isnan(b) else b
540            )
541        else:
542            test_nan = lambda a, b: (
543                nan if math.isnan(a) or math.isnan(b) else b
544            )
545        cases = (
546            (nan, 10),
547            (10, nan),
548            (nan, pinf),
549            (pinf, nan),
550            (nan, nan)
551        )
552        for op1, op2 in cases:
553            vdata_ab = self.load([op1, op2] * hf_nlanes)
554            data = test_nan(op1, op2)
555            simd = reduce_intrin(vdata_ab)
556            assert simd == pytest.approx(data, nan_ok=True)
557            vdata_a = self.setall(op1)
558            vdata_b = self.setall(op2)
559            data = [data] * self.nlanes
560            simd = intrin(vdata_a, vdata_b)
561            assert simd == pytest.approx(data, nan_ok=True)
562
563    def test_reciprocal(self):
564        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
565        data = self._data()
566        vdata = self.load(self._data())
567
568        recip_cases = ((nan, nan), (pinf, 0.0), (ninf, -0.0), (0.0, pinf), (-0.0, ninf))
569        for case, desired in recip_cases:
570            data_recip = [desired] * self.nlanes
571            recip = self.recip(self.setall(case))
572            assert recip == pytest.approx(data_recip, nan_ok=True)
573
574        data_recip = self.load([1 / x for x in data])  # load to truncate precision
575        recip = self.recip(vdata)
576        assert recip == data_recip
577
578    def test_special_cases(self):
579        """
580        Compare Not NaN. Test intrinsics:
581            npyv_notnan_##SFX
582        """
583        nnan = self.notnan(self.setall(self._nan()))
584        assert nnan == [0] * self.nlanes
585
586    @pytest.mark.parametrize("intrin_name", [
587        "rint", "trunc", "ceil", "floor"
588    ])
589    def test_unary_invalid_fpexception(self, intrin_name):
590        intrin = getattr(self, intrin_name)
591        for d in [float("nan"), float("inf"), -float("inf")]:
592            v = self.setall(d)
593            clear_floatstatus()
594            intrin(v)
595            assert check_floatstatus(invalid=True) is False
596
597    @pytest.mark.parametrize('py_comp,np_comp', [
598        (operator.lt, "cmplt"),
599        (operator.le, "cmple"),
600        (operator.gt, "cmpgt"),
601        (operator.ge, "cmpge"),
602        (operator.eq, "cmpeq"),
603        (operator.ne, "cmpneq")
604    ])
605    def test_comparison_with_nan(self, py_comp, np_comp):
606        pinf, ninf, nan = self._pinfinity(), self._ninfinity(), self._nan()
607        mask_true = self._true_mask()
608
609        def to_bool(vector):
610            return [lane == mask_true for lane in vector]
611
612        intrin = getattr(self, np_comp)
613        cmp_cases = ((0, nan), (nan, 0), (nan, nan), (pinf, nan),
614                     (ninf, nan), (-0.0, +0.0))
615        for case_operand1, case_operand2 in cmp_cases:
616            data_a = [case_operand1] * self.nlanes
617            data_b = [case_operand2] * self.nlanes
618            vdata_a = self.setall(case_operand1)
619            vdata_b = self.setall(case_operand2)
620            vcmp = to_bool(intrin(vdata_a, vdata_b))
621            data_cmp = [py_comp(a, b) for a, b in zip(data_a, data_b)]
622            assert vcmp == data_cmp
623
624    @pytest.mark.parametrize("intrin", ["any", "all"])
625    @pytest.mark.parametrize("data", (
626        [float("nan"), 0],
627        [0, float("nan")],
628        [float("nan"), 1],
629        [1, float("nan")],
630        [float("nan"), float("nan")],
631        [0.0, -0.0],
632        [-0.0, 0.0],
633        [1.0, -0.0]
634    ))
635    def test_operators_crosstest(self, intrin, data):
636        """
637        Test intrinsics:
638            npyv_any_##SFX
639            npyv_all_##SFX
640        """
641        data_a = self.load(data * self.nlanes)
642        func = eval(intrin)
643        intrin = getattr(self, intrin)
644        desired = func(data_a)
645        simd = intrin(data_a)
646        assert not not simd == desired
647
648class _SIMD_ALL(_Test_Utility):
649    """
650    To test all vector types at once
651    """
652    def test_memory_load(self):
653        data = self._data()
654        # unaligned load
655        load_data = self.load(data)
656        assert load_data == data
657        # aligned load
658        loada_data = self.loada(data)
659        assert loada_data == data
660        # stream load
661        loads_data = self.loads(data)
662        assert loads_data == data
663        # load lower part
664        loadl = self.loadl(data)
665        loadl_half = list(loadl)[:self.nlanes // 2]
666        data_half = data[:self.nlanes // 2]
667        assert loadl_half == data_half
668        assert loadl != data  # detect overflow
669
670    def test_memory_store(self):
671        data = self._data()
672        vdata = self.load(data)
673        # unaligned store
674        store = [0] * self.nlanes
675        self.store(store, vdata)
676        assert store == data
677        # aligned store
678        store_a = [0] * self.nlanes
679        self.storea(store_a, vdata)
680        assert store_a == data
681        # stream store
682        store_s = [0] * self.nlanes
683        self.stores(store_s, vdata)
684        assert store_s == data
685        # store lower part
686        store_l = [0] * self.nlanes
687        self.storel(store_l, vdata)
688        assert store_l[:self.nlanes // 2] == data[:self.nlanes // 2]
689        assert store_l != vdata  # detect overflow
690        # store higher part
691        store_h = [0] * self.nlanes
692        self.storeh(store_h, vdata)
693        assert store_h[:self.nlanes // 2] == data[self.nlanes // 2:]
694        assert store_h != vdata  # detect overflow
695
696    @pytest.mark.parametrize("intrin, elsizes, scale, fill", [
697        ("self.load_tillz, self.load_till", (32, 64), 1, [0xffff]),
698        ("self.load2_tillz, self.load2_till", (32, 64), 2, [0xffff, 0x7fff]),
699    ])
700    def test_memory_partial_load(self, intrin, elsizes, scale, fill):
701        if self._scalar_size() not in elsizes:
702            return
703        npyv_load_tillz, npyv_load_till = eval(intrin)
704        data = self._data()
705        lanes = list(range(1, self.nlanes + 1))
706        lanes += [self.nlanes**2, self.nlanes**4]  # test out of range
707        for n in lanes:
708            load_till = npyv_load_till(data, n, *fill)
709            load_tillz = npyv_load_tillz(data, n)
710            n *= scale
711            data_till = data[:n] + fill * ((self.nlanes - n) // scale)
712            assert load_till == data_till
713            data_tillz = data[:n] + [0] * (self.nlanes - n)
714            assert load_tillz == data_tillz
715
716    @pytest.mark.parametrize("intrin, elsizes, scale", [
717        ("self.store_till", (32, 64), 1),
718        ("self.store2_till", (32, 64), 2),
719    ])
720    def test_memory_partial_store(self, intrin, elsizes, scale):
721        if self._scalar_size() not in elsizes:
722            return
723        npyv_store_till = eval(intrin)
724        data = self._data()
725        data_rev = self._data(reverse=True)
726        vdata = self.load(data)
727        lanes = list(range(1, self.nlanes + 1))
728        lanes += [self.nlanes**2, self.nlanes**4]
729        for n in lanes:
730            data_till = data_rev.copy()
731            data_till[:n * scale] = data[:n * scale]
732            store_till = self._data(reverse=True)
733            npyv_store_till(store_till, n, vdata)
734            assert store_till == data_till
735
736    @pytest.mark.parametrize("intrin, elsizes, scale", [
737        ("self.loadn", (32, 64), 1),
738        ("self.loadn2", (32, 64), 2),
739    ])
740    def test_memory_noncont_load(self, intrin, elsizes, scale):
741        if self._scalar_size() not in elsizes:
742            return
743        npyv_loadn = eval(intrin)
744        for stride in range(-64, 64):
745            if stride < 0:
746                data = self._data(stride, -stride * self.nlanes)
747                data_stride = list(itertools.chain(
748                    *zip(*[data[-i::stride] for i in range(scale, 0, -1)])
749                ))
750            elif stride == 0:
751                data = self._data()
752                data_stride = data[0:scale] * (self.nlanes // scale)
753            else:
754                data = self._data(count=stride * self.nlanes)
755                data_stride = list(itertools.chain(
756                    *zip(*[data[i::stride] for i in range(scale)]))
757                )
758            data_stride = self.load(data_stride)  # cast unsigned
759            loadn = npyv_loadn(data, stride)
760            assert loadn == data_stride
761
762    @pytest.mark.parametrize("intrin, elsizes, scale, fill", [
763        ("self.loadn_tillz, self.loadn_till", (32, 64), 1, [0xffff]),
764        ("self.loadn2_tillz, self.loadn2_till", (32, 64), 2, [0xffff, 0x7fff]),
765    ])
766    def test_memory_noncont_partial_load(self, intrin, elsizes, scale, fill):
767        if self._scalar_size() not in elsizes:
768            return
769        npyv_loadn_tillz, npyv_loadn_till = eval(intrin)
770        lanes = list(range(1, self.nlanes + 1))
771        lanes += [self.nlanes**2, self.nlanes**4]
772        for stride in range(-64, 64):
773            if stride < 0:
774                data = self._data(stride, -stride * self.nlanes)
775                data_stride = list(itertools.chain(
776                    *zip(*[data[-i::stride] for i in range(scale, 0, -1)])
777                ))
778            elif stride == 0:
779                data = self._data()
780                data_stride = data[0:scale] * (self.nlanes // scale)
781            else:
782                data = self._data(count=stride * self.nlanes)
783                data_stride = list(itertools.chain(
784                    *zip(*[data[i::stride] for i in range(scale)])
785                ))
786            data_stride = list(self.load(data_stride))  # cast unsigned
787            for n in lanes:
788                nscale = n * scale
789                llanes = self.nlanes - nscale
790                data_stride_till = (
791                    data_stride[:nscale] + fill * (llanes // scale)
792                )
793                loadn_till = npyv_loadn_till(data, stride, n, *fill)
794                assert loadn_till == data_stride_till
795                data_stride_tillz = data_stride[:nscale] + [0] * llanes
796                loadn_tillz = npyv_loadn_tillz(data, stride, n)
797                assert loadn_tillz == data_stride_tillz
798
799    @pytest.mark.parametrize("intrin, elsizes, scale", [
800        ("self.storen", (32, 64), 1),
801        ("self.storen2", (32, 64), 2),
802    ])
803    def test_memory_noncont_store(self, intrin, elsizes, scale):
804        if self._scalar_size() not in elsizes:
805            return
806        npyv_storen = eval(intrin)
807        data = self._data()
808        vdata = self.load(data)
809        hlanes = self.nlanes // scale
810        for stride in range(1, 64):
811            data_storen = [0xff] * stride * self.nlanes
812            for s in range(0, hlanes * stride, stride):
813                i = (s // stride) * scale
814                data_storen[s:s + scale] = data[i:i + scale]
815            storen = [0xff] * stride * self.nlanes
816            storen += [0x7f] * 64
817            npyv_storen(storen, stride, vdata)
818            assert storen[:-64] == data_storen
819            assert storen[-64:] == [0x7f] * 64  # detect overflow
820
821        for stride in range(-64, 0):
822            data_storen = [0xff] * -stride * self.nlanes
823            for s in range(0, hlanes * stride, stride):
824                i = (s // stride) * scale
825                data_storen[s - scale:s or None] = data[i:i + scale]
826            storen = [0x7f] * 64
827            storen += [0xff] * -stride * self.nlanes
828            npyv_storen(storen, stride, vdata)
829            assert storen[64:] == data_storen
830            assert storen[:64] == [0x7f] * 64  # detect overflow
831        # stride 0
832        data_storen = [0x7f] * self.nlanes
833        storen = data_storen.copy()
834        data_storen[0:scale] = data[-scale:]
835        npyv_storen(storen, 0, vdata)
836        assert storen == data_storen
837
838    @pytest.mark.parametrize("intrin, elsizes, scale", [
839        ("self.storen_till", (32, 64), 1),
840        ("self.storen2_till", (32, 64), 2),
841    ])
842    def test_memory_noncont_partial_store(self, intrin, elsizes, scale):
843        if self._scalar_size() not in elsizes:
844            return
845        npyv_storen_till = eval(intrin)
846        data = self._data()
847        vdata = self.load(data)
848        lanes = list(range(1, self.nlanes + 1))
849        lanes += [self.nlanes**2, self.nlanes**4]
850        hlanes = self.nlanes // scale
851        for stride in range(1, 64):
852            for n in lanes:
853                data_till = [0xff] * stride * self.nlanes
854                tdata = data[:n * scale] + [0xff] * (self.nlanes - n * scale)
855                for s in range(0, hlanes * stride, stride)[:n]:
856                    i = (s // stride) * scale
857                    data_till[s:s + scale] = tdata[i:i + scale]
858                storen_till = [0xff] * stride * self.nlanes
859                storen_till += [0x7f] * 64
860                npyv_storen_till(storen_till, stride, n, vdata)
861                assert storen_till[:-64] == data_till
862                assert storen_till[-64:] == [0x7f] * 64  # detect overflow
863
864        for stride in range(-64, 0):
865            for n in lanes:
866                data_till = [0xff] * -stride * self.nlanes
867                tdata = data[:n * scale] + [0xff] * (self.nlanes - n * scale)
868                for s in range(0, hlanes * stride, stride)[:n]:
869                    i = (s // stride) * scale
870                    data_till[s - scale:s or None] = tdata[i:i + scale]
871                storen_till = [0x7f] * 64
872                storen_till += [0xff] * -stride * self.nlanes
873                npyv_storen_till(storen_till, stride, n, vdata)
874                assert storen_till[64:] == data_till
875                assert storen_till[:64] == [0x7f] * 64  # detect overflow
876
877        # stride 0
878        for n in lanes:
879            data_till = [0x7f] * self.nlanes
880            storen_till = data_till.copy()
881            data_till[0:scale] = data[:n * scale][-scale:]
882            npyv_storen_till(storen_till, 0, n, vdata)
883            assert storen_till == data_till
884
885    @pytest.mark.parametrize("intrin, table_size, elsize", [
886        ("self.lut32", 32, 32),
887        ("self.lut16", 16, 64)
888    ])
889    def test_lut(self, intrin, table_size, elsize):
890        """
891        Test lookup table intrinsics:
892            npyv_lut32_##sfx
893            npyv_lut16_##sfx
894        """
895        if elsize != self._scalar_size():
896            return
897        intrin = eval(intrin)
898        idx_itrin = getattr(self.npyv, f"setall_u{elsize}")
899        table = range(table_size)
900        for i in table:
901            broadi = self.setall(i)
902            idx = idx_itrin(i)
903            lut = intrin(table, idx)
904            assert lut == broadi
905
906    def test_misc(self):
907        broadcast_zero = self.zero()
908        assert broadcast_zero == [0] * self.nlanes
909        for i in range(1, 10):
910            broadcasti = self.setall(i)
911            assert broadcasti == [i] * self.nlanes
912
913        data_a, data_b = self._data(), self._data(reverse=True)
914        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
915
916        # py level of npyv_set_* don't support ignoring the extra specified lanes or
917        # fill non-specified lanes with zero.
918        vset = self.set(*data_a)
919        assert vset == data_a
920        # py level of npyv_setf_* don't support ignoring the extra specified lanes or
921        # fill non-specified lanes with the specified scalar.
922        vsetf = self.setf(10, *data_a)
923        assert vsetf == data_a
924
925        # We're testing the sanity of _simd's type-vector,
926        # reinterpret* intrinsics itself are tested via compiler
927        # during the build of _simd module
928        sfxes = ["u8", "s8", "u16", "s16", "u32", "s32", "u64", "s64"]
929        if self.npyv.simd_f64:
930            sfxes.append("f64")
931        if self.npyv.simd_f32:
932            sfxes.append("f32")
933        for sfx in sfxes:
934            vec_name = getattr(self, "reinterpret_" + sfx)(vdata_a).__name__
935            assert vec_name == "npyv_" + sfx
936
937        # select & mask operations
938        select_a = self.select(self.cmpeq(self.zero(), self.zero()), vdata_a, vdata_b)
939        assert select_a == data_a
940        select_b = self.select(self.cmpneq(self.zero(), self.zero()), vdata_a, vdata_b)
941        assert select_b == data_b
942
943        # test extract elements
944        assert self.extract0(vdata_b) == vdata_b[0]
945
946        # cleanup intrinsic is only used with AVX for
947        # zeroing registers to avoid the AVX-SSE transition penalty,
948        # so nothing to test here
949        self.npyv.cleanup()
950
951    def test_reorder(self):
952        data_a, data_b = self._data(), self._data(reverse=True)
953        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
954        # lower half part
955        data_a_lo = data_a[:self.nlanes // 2]
956        data_b_lo = data_b[:self.nlanes // 2]
957        # higher half part
958        data_a_hi = data_a[self.nlanes // 2:]
959        data_b_hi = data_b[self.nlanes // 2:]
960        # combine two lower parts
961        combinel = self.combinel(vdata_a, vdata_b)
962        assert combinel == data_a_lo + data_b_lo
963        # combine two higher parts
964        combineh = self.combineh(vdata_a, vdata_b)
965        assert combineh == data_a_hi + data_b_hi
966        # combine x2
967        combine = self.combine(vdata_a, vdata_b)
968        assert combine == (data_a_lo + data_b_lo, data_a_hi + data_b_hi)
969
970        # zip(interleave)
971        data_zipl = self.load([
972            v for p in zip(data_a_lo, data_b_lo) for v in p
973        ])
974        data_ziph = self.load([
975            v for p in zip(data_a_hi, data_b_hi) for v in p
976        ])
977        vzip = self.zip(vdata_a, vdata_b)
978        assert vzip == (data_zipl, data_ziph)
979        vzip = [0] * self.nlanes * 2
980        self._x2("store")(vzip, (vdata_a, vdata_b))
981        assert vzip == list(data_zipl) + list(data_ziph)
982
983        # unzip(deinterleave)
984        unzip = self.unzip(data_zipl, data_ziph)
985        assert unzip == (data_a, data_b)
986        unzip = self._x2("load")(list(data_zipl) + list(data_ziph))
987        assert unzip == (data_a, data_b)
988
989    def test_reorder_rev64(self):
990        # Reverse elements of each 64-bit lane
991        ssize = self._scalar_size()
992        if ssize == 64:
993            return
994        data_rev64 = [
995            y for x in range(0, self.nlanes, 64 // ssize)
996              for y in reversed(range(x, x + 64 // ssize))
997        ]
998        rev64 = self.rev64(self.load(range(self.nlanes)))
999        assert rev64 == data_rev64
1000
1001    def test_reorder_permi128(self):
1002        """
1003        Test permuting elements for each 128-bit lane.
1004        npyv_permi128_##sfx
1005        """
1006        ssize = self._scalar_size()
1007        if ssize < 32:
1008            return
1009        data = self.load(self._data())
1010        permn = 128 // ssize
1011        permd = permn - 1
1012        nlane128 = self.nlanes // permn
1013        shfl = [0, 1] if ssize == 64 else [0, 2, 4, 6]
1014        for i in range(permn):
1015            indices = [(i >> shf) & permd for shf in shfl]
1016            vperm = self.permi128(data, *indices)
1017            data_vperm = [
1018                data[j + (e & -permn)]
1019                for e, j in enumerate(indices * nlane128)
1020            ]
1021            assert vperm == data_vperm
1022
1023    @pytest.mark.parametrize('func, intrin', [
1024        (operator.lt, "cmplt"),
1025        (operator.le, "cmple"),
1026        (operator.gt, "cmpgt"),
1027        (operator.ge, "cmpge"),
1028        (operator.eq, "cmpeq")
1029    ])
1030    def test_operators_comparison(self, func, intrin):
1031        if self._is_fp():
1032            data_a = self._data()
1033        else:
1034            data_a = self._data(self._int_max() - self.nlanes)
1035        data_b = self._data(self._int_min(), reverse=True)
1036        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
1037        intrin = getattr(self, intrin)
1038
1039        mask_true = self._true_mask()
1040
1041        def to_bool(vector):
1042            return [lane == mask_true for lane in vector]
1043
1044        data_cmp = [func(a, b) for a, b in zip(data_a, data_b)]
1045        cmp = to_bool(intrin(vdata_a, vdata_b))
1046        assert cmp == data_cmp
1047
1048    def test_operators_logical(self):
1049        if self._is_fp():
1050            data_a = self._data()
1051        else:
1052            data_a = self._data(self._int_max() - self.nlanes)
1053        data_b = self._data(self._int_min(), reverse=True)
1054        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
1055
1056        if self._is_fp():
1057            data_cast_a = self._to_unsigned(vdata_a)
1058            data_cast_b = self._to_unsigned(vdata_b)
1059            cast, cast_data = self._to_unsigned, self._to_unsigned
1060        else:
1061            data_cast_a, data_cast_b = data_a, data_b
1062            cast, cast_data = lambda a: a, self.load
1063
1064        data_xor = cast_data([a ^ b for a, b in zip(data_cast_a, data_cast_b)])
1065        vxor = cast(self.xor(vdata_a, vdata_b))
1066        assert vxor == data_xor
1067
1068        data_or = cast_data([a | b for a, b in zip(data_cast_a, data_cast_b)])
1069        vor = cast(getattr(self, "or")(vdata_a, vdata_b))
1070        assert vor == data_or
1071
1072        data_and = cast_data([a & b for a, b in zip(data_cast_a, data_cast_b)])
1073        vand = cast(getattr(self, "and")(vdata_a, vdata_b))
1074        assert vand == data_and
1075
1076        data_not = cast_data([~a for a in data_cast_a])
1077        vnot = cast(getattr(self, "not")(vdata_a))
1078        assert vnot == data_not
1079
1080        if self.sfx not in ("u8"):
1081            return
1082        data_andc = [a & ~b for a, b in zip(data_cast_a, data_cast_b)]
1083        vandc = cast(self.andc(vdata_a, vdata_b))
1084        assert vandc == data_andc
1085
1086    @pytest.mark.parametrize("intrin", ["any", "all"])
1087    @pytest.mark.parametrize("data", (
1088        [1, 2, 3, 4],
1089        [-1, -2, -3, -4],
1090        [0, 1, 2, 3, 4],
1091        [0x7f, 0x7fff, 0x7fffffff, 0x7fffffffffffffff],
1092        [0, -1, -2, -3, 4],
1093        [0],
1094        [1],
1095        [-1]
1096    ))
1097    def test_operators_crosstest(self, intrin, data):
1098        """
1099        Test intrinsics:
1100            npyv_any_##SFX
1101            npyv_all_##SFX
1102        """
1103        data_a = self.load(data * self.nlanes)
1104        func = eval(intrin)
1105        intrin = getattr(self, intrin)
1106        desired = func(data_a)
1107        simd = intrin(data_a)
1108        assert not not simd == desired
1109
1110    def test_conversion_boolean(self):
1111        bsfx = "b" + self.sfx[1:]
1112        to_boolean = getattr(self.npyv, f"cvt_{bsfx}_{self.sfx}")
1113        from_boolean = getattr(self.npyv, f"cvt_{self.sfx}_{bsfx}")
1114
1115        false_vb = to_boolean(self.setall(0))
1116        true_vb = self.cmpeq(self.setall(0), self.setall(0))
1117        assert false_vb != true_vb
1118
1119        false_vsfx = from_boolean(false_vb)
1120        true_vsfx = from_boolean(true_vb)
1121        assert false_vsfx != true_vsfx
1122
1123    def test_conversion_expand(self):
1124        """
1125        Test expand intrinsics:
1126            npyv_expand_u16_u8
1127            npyv_expand_u32_u16
1128        """
1129        if self.sfx not in ("u8", "u16"):
1130            return
1131        totype = self.sfx[0] + str(int(self.sfx[1:]) * 2)
1132        expand = getattr(self.npyv, f"expand_{totype}_{self.sfx}")
1133        # close enough from the edge to detect any deviation
1134        data = self._data(self._int_max() - self.nlanes)
1135        vdata = self.load(data)
1136        edata = expand(vdata)
1137        # lower half part
1138        data_lo = data[:self.nlanes // 2]
1139        # higher half part
1140        data_hi = data[self.nlanes // 2:]
1141        assert edata == (data_lo, data_hi)
1142
1143    def test_arithmetic_subadd(self):
1144        if self._is_fp():
1145            data_a = self._data()
1146        else:
1147            data_a = self._data(self._int_max() - self.nlanes)
1148        data_b = self._data(self._int_min(), reverse=True)
1149        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
1150
1151        # non-saturated
1152        data_add = self.load([a + b for a, b in zip(data_a, data_b)])  # load to cast
1153        add = self.add(vdata_a, vdata_b)
1154        assert add == data_add
1155        data_sub = self.load([a - b for a, b in zip(data_a, data_b)])
1156        sub = self.sub(vdata_a, vdata_b)
1157        assert sub == data_sub
1158
1159    def test_arithmetic_mul(self):
1160        if self.sfx in ("u64", "s64"):
1161            return
1162
1163        if self._is_fp():
1164            data_a = self._data()
1165        else:
1166            data_a = self._data(self._int_max() - self.nlanes)
1167        data_b = self._data(self._int_min(), reverse=True)
1168        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
1169
1170        data_mul = self.load([a * b for a, b in zip(data_a, data_b)])
1171        mul = self.mul(vdata_a, vdata_b)
1172        assert mul == data_mul
1173
1174    def test_arithmetic_div(self):
1175        if not self._is_fp():
1176            return
1177
1178        data_a, data_b = self._data(), self._data(reverse=True)
1179        vdata_a, vdata_b = self.load(data_a), self.load(data_b)
1180
1181        # load to truncate f64 to precision of f32
1182        data_div = self.load([a / b for a, b in zip(data_a, data_b)])
1183        div = self.div(vdata_a, vdata_b)
1184        assert div == data_div
1185
1186    def test_arithmetic_intdiv(self):
1187        """
1188        Test integer division intrinsics:
1189            npyv_divisor_##sfx
1190            npyv_divc_##sfx
1191        """
1192        if self._is_fp():
1193            return
1194
1195        int_min = self._int_min()
1196
1197        def trunc_div(a, d):
1198            """
1199            Divide towards zero works with large integers > 2^53,
1200            and wrap around overflow similar to what C does.

Showing the first 1,200 of 1346 lines. Download the file for the rest.

codekingpro/portable-devtools · Team Ai