diff --git a/bitmapcontainer.go b/bitmapcontainer.go index 354af17e..87afef6d 100644 --- a/bitmapcontainer.go +++ b/bitmapcontainer.go @@ -1153,12 +1153,30 @@ func (bc *bitmapContainer) toArrayContainer() *arrayContainer { return ac } +// bitmapContainerSkipMaxCardinality is where scanning eight words at a time and +// skipping the empty groups stops paying for itself and compressing every word +// becomes cheaper. Measured crossover is just under 768 values. +const bitmapContainerSkipMaxCardinality = 768 + +// fillArray writes the values of the container into the given slice, whose +// length is the container's cardinality. func (bc *bitmapContainer) fillArray(container []uint16) { - // TODO: rewrite in assembly + if useVectorFill { + if len(container) < bitmapContainerSkipMaxCardinality { + fillArraySkipVector(bc.bitmap, container) + } else { + fillArrayVector(bc.bitmap, container) + } + return + } + fillArrayScalar(bc.bitmap, container) +} + +func fillArrayScalar(bitmap []uint64, container []uint16) { pos := 0 base := 0 - for k := 0; k < len(bc.bitmap); k++ { - bitset := bc.bitmap[k] + for k := 0; k < len(bitmap); k++ { + bitset := bitmap[k] for bitset != 0 { t := bitset & -bitset container[pos] = uint16((base + bits.OnesCount64(t-1))) diff --git a/fillarray_generic.go b/fillarray_generic.go new file mode 100644 index 00000000..8e90207d --- /dev/null +++ b/fillarray_generic.go @@ -0,0 +1,31 @@ +//go:build !amd64 || appengine +// +build !amd64 appengine + +package roaring + +// On these targets useVectorFill is a compile-time constant false, so the +// branches that would reach these functions are eliminated and none of them is +// ever executed. They exist so that the shared call sites in bitmapcontainer.go +// and util.go compile, and they delegate back to the scalar paths so that they +// remain correct if anything ever does call them. There is no recursion: the +// guard in each caller is false at compile time on these targets. + +func fillArrayVector(bitmap []uint64, container []uint16) { + fillArrayScalar(bitmap, container) +} + +func fillArraySkipVector(bitmap []uint64, container []uint16) { + fillArrayScalar(bitmap, container) +} + +func fillArrayANDVector(container []uint16, bitmap1, bitmap2 []uint64) { + fillArrayAND(container, bitmap1, bitmap2) +} + +func fillArrayANDNOTVector(container []uint16, bitmap1, bitmap2 []uint64) { + fillArrayANDNOT(container, bitmap1, bitmap2) +} + +func fillArrayXORVector(container []uint16, bitmap1, bitmap2 []uint64) { + fillArrayXOR(container, bitmap1, bitmap2) +} diff --git a/fillarray_test.go b/fillarray_test.go new file mode 100644 index 00000000..0a369abb --- /dev/null +++ b/fillarray_test.go @@ -0,0 +1,153 @@ +package roaring + +import ( + "math/bits" + "math/rand" + "testing" +) + +func fillArrayTestBitmap(cardinality int, seed int64) *bitmapContainer { + r := rand.New(rand.NewSource(seed)) + bc := newBitmapContainer() + for bc.cardinality < cardinality { + bc.iadd(uint16(r.Intn(65536))) + } + return bc +} + +func wordsToValues(words []uint64) []uint16 { + out := make([]uint16, 0, 65536) + for k, w := range words { + for w != 0 { + out = append(out, uint16(k*64+bits.TrailingZeros64(w))) + w &= w - 1 + } + } + return out +} + +func combineWords(a, b []uint64, op func(x, y uint64) uint64) []uint64 { + out := make([]uint64, len(a)) + for i := range a { + out[i] = op(a[i], b[i]) + } + return out +} + +// cardinalities straddle the skip/compress crossover and the 32-values-per- +// store and 64-values-per-word boundaries. +var fillArrayTestCardinalities = []int{0, 1, 31, 32, 33, 63, 64, 65, 511, 512, 513, 4096, 30000, 65535, 65536} + +func TestFillArrayMatchesScalar(t *testing.T) { + for _, cardinality := range fillArrayTestCardinalities { + bc := fillArrayTestBitmap(cardinality, int64(cardinality)+1) + + want := make([]uint16, cardinality) + fillArrayScalar(bc.bitmap, want) + got := make([]uint16, cardinality) + bc.fillArray(got) + for i := range want { + if got[i] != want[i] { + t.Fatalf("fillArray card=%d at %d: got %d want %d", cardinality, i, got[i], want[i]) + } + } + + other := fillArrayTestBitmap(cardinality, int64(cardinality)+99) + for _, tc := range []struct { + name string + op func(x, y uint64) uint64 + fn func(container []uint16, bitmap1, bitmap2 []uint64) + }{ + {"AND", func(x, y uint64) uint64 { return x & y }, fillArrayAND}, + {"ANDNOT", func(x, y uint64) uint64 { return x &^ y }, fillArrayANDNOT}, + {"XOR", func(x, y uint64) uint64 { return x ^ y }, fillArrayXOR}, + } { + combined := combineWords(bc.bitmap, other.bitmap, tc.op) + expected := wordsToValues(combined) + got := make([]uint16, len(expected)) + tc.fn(got, bc.bitmap, other.bitmap) + for i := range expected { + if got[i] != expected[i] { + t.Fatalf("fillArray%s card=%d at %d: got %d want %d", + tc.name, cardinality, i, got[i], expected[i]) + } + } + } + } +} + +func benchFillArray(b *testing.B, cardinality int) { + bc := fillArrayTestBitmap(cardinality, int64(cardinality)+1) + out := make([]uint16, cardinality) + for b.Loop() { + bc.fillArray(out) + } + if cardinality > 0 { + b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*cardinality), "ns/value") + } +} + +func BenchmarkFillArray(b *testing.B) { + for _, c := range []int{64, 256, 512, 1024, 2048, 4096} { + b.Run("card="+itoaFill(c), func(b *testing.B) { benchFillArray(b, c) }) + } +} + +func BenchmarkFillArrayAND(b *testing.B) { + // source containers sized so the AND lands at or below arrayDefaultMaxSize, + // the only regime in which fillArrayAND is reached + for _, src := range []int{4096, 8192, 12288, 16384} { + a := fillArrayTestBitmap(src, 11) + c := fillArrayTestBitmap(src, 22) + n := 0 + for i := range a.bitmap { + n += bits.OnesCount64(a.bitmap[i] & c.bitmap[i]) + } + out := make([]uint16, n) + b.Run("src="+itoaFill(src)+"/out="+itoaFill(n), func(b *testing.B) { + for b.Loop() { + fillArrayAND(out, a.bitmap, c.bitmap) + } + b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N*n), "ns/value") + }) + } +} + +func itoaFill(v int) string { + if v == 0 { + return "0" + } + var buf [12]byte + i := len(buf) + for v > 0 { + i-- + buf[i] = byte('0' + v%10) + v /= 10 + } + return string(buf[i:]) +} + +// End to end: bitmapContainer.and / andNot / xor when the result is small +// enough to become an array container, which is the only way the two-input +// fillArray helpers are reached. +func benchBitmapToArrayOp(b *testing.B, srcCardinality int, op func(x, y *bitmapContainer) container) { + x := fillArrayTestBitmap(srcCardinality, 11) + y := fillArrayTestBitmap(srcCardinality, 22) + for b.Loop() { + c := op(x, y) + if c == nil { + b.Fatal("nil container") + } + } +} + +func BenchmarkBitmapContainerToArrayOps(b *testing.B) { + for _, src := range []int{4096, 8192, 16384} { + b.Run("and/src="+itoaFill(src), func(b *testing.B) { + benchBitmapToArrayOp(b, src, func(x, y *bitmapContainer) container { return x.andBitmap(y) }) + }) + b.Run("andNot/src="+itoaFill(src), func(b *testing.B) { + benchBitmapToArrayOp(b, src, func(x, y *bitmapContainer) container { return x.andNotBitmap(y) }) + }) + } +} diff --git a/fillarray_vbmi2_amd64.go b/fillarray_vbmi2_amd64.go new file mode 100644 index 00000000..89bf0327 --- /dev/null +++ b/fillarray_vbmi2_amd64.go @@ -0,0 +1,22 @@ +//go:build amd64 && !appengine +// +build amd64,!appengine + +package roaring + +// Implemented in fillarray_vbmi2_amd64.s. Only called when useVectorFill (see +// fillbits_vbmi2_amd64.go) reports that the CPU can run them. + +//go:noescape +func fillArrayVector(bitmap []uint64, container []uint16) + +//go:noescape +func fillArraySkipVector(bitmap []uint64, container []uint16) + +//go:noescape +func fillArrayANDVector(container []uint16, bitmap1, bitmap2 []uint64) + +//go:noescape +func fillArrayANDNOTVector(container []uint16, bitmap1, bitmap2 []uint64) + +//go:noescape +func fillArrayXORVector(container []uint16, bitmap1, bitmap2 []uint64) diff --git a/fillarray_vbmi2_amd64.s b/fillarray_vbmi2_amd64.s new file mode 100644 index 00000000..bb79027c --- /dev/null +++ b/fillarray_vbmi2_amd64.s @@ -0,0 +1,203 @@ +//go:build amd64 && !appengine +// +build amd64,!appengine + +#include "textflag.h" + +// Decoding a bitmap container into a uint16 array container. +// +// These share the shape of fillbits_vbmi2_amd64.s: one VPCOMPRESSB per 64-bit +// word turns the whole word into 64 byte-sized bit positions, which are then +// widened 32 at a time with VPMOVZXBW and written with a masked store, so +// exactly popcount(word) uint16s are written and the output needs no slack. +// The compress-then-widen idea is taken from simdjson's bit_indexer::write +// (icelake kernel, Apache-2.0): https://github.com/simdjson/simdjson +// +// The AND, ANDNOT and XOR variants combine two containers word by word before +// emitting, which is what bitmapContainer.and/andNot/xor need when their result +// is small enough to become an array container. +// +// fillArraySkipVector additionally tests eight words at a time with VPTESTMQ +// and skips wholly empty groups. It wins only when the container is very +// sparse, where most 512-bit groups hold nothing at all. + +DATA fa_bitPositions<>+0x00(SB)/8, $0x0706050403020100 +DATA fa_bitPositions<>+0x08(SB)/8, $0x0f0e0d0c0b0a0908 +DATA fa_bitPositions<>+0x10(SB)/8, $0x1716151413121110 +DATA fa_bitPositions<>+0x18(SB)/8, $0x1f1e1d1c1b1a1918 +DATA fa_bitPositions<>+0x20(SB)/8, $0x2726252423222120 +DATA fa_bitPositions<>+0x28(SB)/8, $0x2f2e2d2c2b2a2928 +DATA fa_bitPositions<>+0x30(SB)/8, $0x3736353433323130 +DATA fa_bitPositions<>+0x38(SB)/8, $0x3f3e3d3c3b3a3938 +GLOBL fa_bitPositions<>(SB), RODATA|NOPTR, $64 + +// Emit the set bits of AX as uint16 values at (DI), advancing DI. +// Z0 = byte positions, Z1 = running base (16-bit lanes). Clobbers R10, R11, +// K1, K2, Z2, Z4, Y5. +#define EMIT16(skip, adv) \ + TESTQ AX, AX \ + JZ skip \ + KMOVQ AX, K1 \ + VPCOMPRESSB.Z Z0, K1, Z2 \ + POPCNTQ AX, R10 \ + MOVQ $-1, R11 \ + BZHIQ R10, R11, R11 \ + KMOVD R11, K2 \ + VPMOVZXBW Y2, Z4 \ + VPADDW Z1, Z4, Z4 \ + VMOVDQU16 Z4, K2, (DI) \ + CMPQ R10, $32 \ + JLE adv \ + SHRQ $32, R11 \ + KMOVD R11, K2 \ + VEXTRACTI64X4 $1, Z2, Y5 \ + VPMOVZXBW Y5, Z4 \ + VPADDW Z1, Z4, Z4 \ + VMOVDQU16 Z4, K2, 64(DI) \ +adv: \ + LEAQ (DI)(R10*2), DI \ +skip: \ + VPADDW Z3, Z1, Z1 + +#define SETUP16 \ + VMOVDQU64 fa_bitPositions<>(SB), Z0 \ + VPXORQ Z1, Z1, Z1 \ + MOVL $64, AX \ + VPBROADCASTW AX, Z3 + +// func fillArrayVector(bitmap []uint64, container []uint16) +TEXT ·fillArrayVector(SB), NOSPLIT, $0-48 + MOVQ bitmap_base+0(FP), SI + MOVQ bitmap_len+8(FP), CX + MOVQ container_base+24(FP), DI + SETUP16 + TESTQ CX, CX + JZ fa_done +fa_loop: + MOVQ (SI), AX + EMIT16(fa_s, fa_a) + ADDQ $8, SI + DECQ CX + JNZ fa_loop +fa_done: + VZEROUPPER + RET + +// func fillArrayANDVector(container []uint16, bitmap1, bitmap2 []uint64) +TEXT ·fillArrayANDVector(SB), NOSPLIT, $0-72 + MOVQ container_base+0(FP), DI + MOVQ bitmap1_base+24(FP), SI + MOVQ bitmap1_len+32(FP), CX + MOVQ bitmap2_base+48(FP), BX + SETUP16 + TESTQ CX, CX + JZ fand_done +fand_loop: + MOVQ (SI), AX + ANDQ (BX), AX + EMIT16(fand_s, fand_a) + ADDQ $8, SI + ADDQ $8, BX + DECQ CX + JNZ fand_loop +fand_done: + VZEROUPPER + RET + +// func fillArrayANDNOTVector(container []uint16, bitmap1, bitmap2 []uint64) +TEXT ·fillArrayANDNOTVector(SB), NOSPLIT, $0-72 + MOVQ container_base+0(FP), DI + MOVQ bitmap1_base+24(FP), SI + MOVQ bitmap1_len+32(FP), CX + MOVQ bitmap2_base+48(FP), BX + SETUP16 + TESTQ CX, CX + JZ fandn_done +fandn_loop: + MOVQ (BX), AX + NOTQ AX + ANDQ (SI), AX + EMIT16(fandn_s, fandn_a) + ADDQ $8, SI + ADDQ $8, BX + DECQ CX + JNZ fandn_loop +fandn_done: + VZEROUPPER + RET + +// func fillArrayXORVector(container []uint16, bitmap1, bitmap2 []uint64) +TEXT ·fillArrayXORVector(SB), NOSPLIT, $0-72 + MOVQ container_base+0(FP), DI + MOVQ bitmap1_base+24(FP), SI + MOVQ bitmap1_len+32(FP), CX + MOVQ bitmap2_base+48(FP), BX + SETUP16 + TESTQ CX, CX + JZ fxor_done +fxor_loop: + MOVQ (SI), AX + XORQ (BX), AX + EMIT16(fxor_s, fxor_a) + ADDQ $8, SI + ADDQ $8, BX + DECQ CX + JNZ fxor_loop +fxor_done: + VZEROUPPER + RET + +// func fillArraySkipVector(bitmap []uint64, container []uint16) +// +// Same as fillArrayVector but tests eight words at a time with VPTESTMQ and +// skips wholly empty groups. Wins when the container is very sparse. +TEXT ·fillArraySkipVector(SB), NOSPLIT, $0-48 + MOVQ bitmap_base+0(FP), SI + MOVQ bitmap_len+8(FP), CX + MOVQ container_base+24(FP), DI + VMOVDQU64 fa_bitPositions<>(SB), Z0 + SHRQ $3, CX + TESTQ CX, CX + JZ fs_done + XORQ DX, DX // word index of the group +fs_loop: + VMOVDQU64 (SI), Z6 + VPTESTMQ Z6, Z6, K3 + KMOVB K3, R9 + TESTQ R9, R9 + JZ fs_next +fs_word: + TZCNTQ R9, R12 + BLSRQ R9, R9 + MOVQ (SI)(R12*8), AX + KMOVQ AX, K1 + VPCOMPRESSB.Z Z0, K1, Z2 + POPCNTQ AX, R10 + MOVQ $-1, R11 + BZHIQ R10, R11, R11 + LEAQ (DX)(R12*1), R13 + SHLQ $6, R13 + VPBROADCASTW R13, Z1 + KMOVD R11, K2 + VPMOVZXBW Y2, Z4 + VPADDW Z1, Z4, Z4 + VMOVDQU16 Z4, K2, (DI) + CMPQ R10, $32 + JLE fs_adv + SHRQ $32, R11 + KMOVD R11, K2 + VEXTRACTI64X4 $1, Z2, Y5 + VPMOVZXBW Y5, Z4 + VPADDW Z1, Z4, Z4 + VMOVDQU16 Z4, K2, 64(DI) +fs_adv: + LEAQ (DI)(R10*2), DI + TESTQ R9, R9 + JNZ fs_word +fs_next: + ADDQ $64, SI + ADDQ $8, DX + DECQ CX + JNZ fs_loop +fs_done: + VZEROUPPER + RET diff --git a/util.go b/util.go index e727d6e0..e49ad29c 100644 --- a/util.go +++ b/util.go @@ -65,7 +65,10 @@ func fillArrayAND(container []uint16, bitmap1, bitmap2 []uint64) { if len(bitmap1) != len(bitmap2) { panic("array lengths don't match") } - // TODO: rewrite in assembly + if useVectorFill { + fillArrayANDVector(container, bitmap1, bitmap2) + return + } pos := 0 for k := range bitmap1 { bitset := bitmap1[k] & bitmap2[k] @@ -82,7 +85,10 @@ func fillArrayANDNOT(container []uint16, bitmap1, bitmap2 []uint64) { if len(bitmap1) != len(bitmap2) { panic("array lengths don't match") } - // TODO: rewrite in assembly + if useVectorFill { + fillArrayANDNOTVector(container, bitmap1, bitmap2) + return + } pos := 0 for k := range bitmap1 { bitset := bitmap1[k] &^ bitmap2[k] @@ -99,7 +105,10 @@ func fillArrayXOR(container []uint16, bitmap1, bitmap2 []uint64) { if len(bitmap1) != len(bitmap2) { panic("array lengths don't match") } - // TODO: rewrite in assembly + if useVectorFill { + fillArrayXORVector(container, bitmap1, bitmap2) + return + } pos := 0 for k := 0; k < len(bitmap1); k++ { bitset := bitmap1[k] ^ bitmap2[k]