Function parallelExecutor

BitSliceIndexing/bsi.go:156–195

(parallelism int, t *task, e action,
	foundSet *roaring.Bitmap)

type action func(t *task, batch []uint32, resultsChan chan *roaring.Bitmap, wg *sync.WaitGroup)

func parallelExecutor(parallelism int, t *task, e action,
	foundSet *roaring.Bitmap) *roaring.Bitmap {

	var n int = parallelism
	if n == 0 {
		n = runtime.NumCPU()
	}

	resultsChan := make(chan *roaring.Bitmap, n)

	card := foundSet.GetCardinality()
	x := card / uint64(n)

	remainder := card - (x * uint64(n))
	var batch []uint32
	var wg sync.WaitGroup
	iter := foundSet.ManyIterator()
	for i := 0; i < n; i++ {
		if i == n-1 {
			batch = make([]uint32, x+remainder)
		} else {
			batch = make([]uint32, x)
		}
		iter.NextMany(batch)
		wg.Add(1)
		go e(t, batch, resultsChan, &wg)
	}

	wg.Wait()

	close(resultsChan)

	ba := make([]*roaring.Bitmap, 0)
	for bm := range resultsChan {
		ba = append(ba, bm)
	}

	return roaring.ParOr(0, ba...)

}

type bsiAction func(input *BSI, batch []uint32, resultsChan chan *BSI, wg *sync.WaitGroup)