Function ParOr

roaring64/parallel64.go:15–134

ParOr computes the union (OR) of all provided bitmaps in parallel, where the parameter "parallelism" determines how many workers are to be used (if it is set to 0, a default number of workers is chosen)

(parallelism int, bitmaps ...*Bitmap)

// where the parameter "parallelism" determines how many workers are to be used
// (if it is set to 0, a default number of workers is chosen)
func ParOr(parallelism int, bitmaps ...*Bitmap) *Bitmap {
	var lKey uint32 = maxUint32
	var hKey uint32

	bitmapsFiltered := bitmaps[:0]
	for _, b := range bitmaps {
		if !b.IsEmpty() {
			bitmapsFiltered = append(bitmapsFiltered, b)
		}
	}
	bitmaps = bitmapsFiltered

	for _, b := range bitmaps {
		lKey = minOfUint32(lKey, b.highlowcontainer.keys[0])
		hKey = maxOfUint32(hKey, b.highlowcontainer.keys[b.highlowcontainer.size()-1])
	}

	if lKey == maxUint32 && hKey == 0 {
		return New()
	} else if len(bitmaps) == 1 {
		return bitmaps[0]
	}
	// The following might overflow and we do not want that!
	// as it might lead to a channel of size 0 later which,
	// on some systems, would block indefinitely.
	keyRange := uint64(hKey) - uint64(lKey) + 1
	if keyRange == 1 {
		// All bitmaps have the same key,
		// we can merge the 32-bit roaring bitmaps in parallel
		var bms32s = make([]*roaring.Bitmap, 0, len(bitmaps))
		for _, b := range bitmaps {
			bms32s = append(bms32s, b.highlowcontainer.containers...)
		}
		return roaring32AsRoaring64(roaring.ParOr(parallelism, bms32s...), lKey)
	}

	if parallelism == 0 {
		parallelism = defaultWorkerCount
	}
	// We cannot use int since int is 32-bit on 32-bit systems.
	var chunkSize int64
	var chunkCount int64
	if int64(parallelism)*4 > int64(keyRange) {
		chunkSize = 1
		chunkCount = int64(keyRange)
	} else {
		chunkCount = int64(parallelism) * 4
		chunkSize = (int64(keyRange) + chunkCount - 1) / chunkCount
	}

	if chunkCount*chunkSize < int64(keyRange) {
		// it's fine to panic to indicate an implementation error
		panic(fmt.Sprintf("invariant check failed: chunkCount * chunkSize < keyRange, %d * %d < %d", chunkCount, chunkSize, keyRange))
	}

	chunks := make([]*roaringArray64, chunkCount)

	chunkSpecChan := make(chan parChunkSpec, minOfInt(maxOfInt(64, 2*parallelism), int(chunkCount)))