Merge branch 'master' into rbf_config

This commit is contained in:
jaten-molecula 2020-10-14 19:57:05 -05:00 committed by GitHub
commit 577600d17b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 360 additions and 231 deletions

View file

@ -62,6 +62,11 @@ jobs:
- checkout-plus
- run: go mod tidy
- run: git diff --exit-code -- go.mod go.sum
check-changelog-label:
executor:
name: golang
steps:
- run: curl https://moleculacorp:$GITHUB_PERSONAL_ACCESS_TOKEN@api.github.com/repos/molecula/pilosa/pulls/$CIRCLE_PR_NUMBER | jq "[.labels[] | .name | startswith(\"changelog\")] | any" -e
test-build-arm:
executor:
name: golang
@ -178,6 +183,13 @@ workflows:
- go-mod-tidy:
requires:
- setup
- check-changelog-label:
requires:
- setup
# the following should make this only run on pull requests
filters:
branches:
only: /^pull\/.*$/
- test-build-arm:
requires:
- setup

105
api.go
View file

@ -25,8 +25,6 @@ import (
"io/ioutil"
"math"
"net/url"
"os"
"path"
"sort"
"strconv"
"strings"
@ -793,71 +791,64 @@ func (api *API) Node() *Node {
return &node
}
// Usage gets the disk usage per index
func (api *API) Usage() (map[string]int64, int64, error) {
indexSizes := make(map[string]int64)
var totalSize int64
dirName, err := expandDirName(api.server.dataDir)
if err != nil {
return indexSizes, totalSize, errors.Wrap(err, "expanding data directory")
}
dir, err := os.Open(dirName)
if err != nil {
return indexSizes, totalSize, errors.Wrap(err, "opening data directory")
}
defer dir.Close()
files, err := dir.Readdir(-1)
if err != nil {
return indexSizes, totalSize, errors.Wrap(err, "reading data directory")
}
for _, file := range files {
if !file.IsDir() {
continue
}
if api.holder.Txf().IsTxDatabasePath(file.Name()) {
continue
}
fullName := path.Join(dirName, file.Name())
indexSizes[file.Name()], err = diskUsage(fullName)
if err != nil {
break
}
totalSize += indexSizes[file.Name()]
}
return indexSizes, totalSize, nil
// NodeUsage represents all usage measurements for one node.
type NodeUsage struct {
Disk DiskUsage `json:"bytesOnDisk"`
}
func diskUsage(fname string) (int64, error) {
var size int64
// DiskUsage represents the storage space used on disk by one node.
type DiskUsage struct {
Capacity uint64 `json:"capacity,omitempty"`
TotalUse int64 `json:"totalInUse"`
Indexes map[string]int64 `json:"indexes"`
}
dir, err := os.Open(fname)
// Usage gets the disk usage per index, in a map[nodeID]NodeUsage
func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.Usage")
defer span.Finish()
nodeUsages := make(map[string]NodeUsage)
indexSizes, err := api.holder.Txf().IndexSizes()
if err != nil {
return 0, errors.Wrap(err, "opening data subdirectory")
return nil, errors.Wrap(err, "getting index usage")
}
defer dir.Close()
files, err := dir.Readdir(-1)
if err != nil {
return 0, errors.Wrap(err, "reading data subdirectory")
var totalSize int64
for _, s := range indexSizes {
totalSize += s
}
for _, file := range files {
if file.IsDir() {
sz, err := diskUsage(path.Join(fname, file.Name()))
if err != nil {
return 0, err
capacity, err := api.server.systemInfo.DiskCapacity(api.holder.path)
if err != nil {
api.server.logger.Printf("couldn't read disk capacity: %s", err)
}
// Insert into result.
nodeUsage := NodeUsage{
Disk: DiskUsage{
Capacity: capacity,
TotalUse: totalSize,
Indexes: indexSizes,
},
}
nodeUsages[api.server.nodeID] = nodeUsage
// Collect size on disk from remote nodes
if !remote {
nodes := api.cluster.Nodes()
for _, node := range nodes {
if node.ID == api.server.nodeID {
continue
}
size += sz
} else {
size += file.Size()
nodeUsage, err := api.server.defaultClient.GetNodeUsage(ctx, &node.URI)
if err != nil {
return nil, errors.Wrapf(err, "collecting disk usage from %s", node.URI)
}
nodeUsages[node.ID] = nodeUsage[node.ID]
}
}
return size, nil
return nodeUsages, nil
}
// RecalculateCaches forces all TopN caches to be updated.

View file

@ -81,6 +81,8 @@ type InternalClient interface {
FinishTransaction(ctx context.Context, id string) (*Transaction, error)
Transactions(ctx context.Context) (map[string]*Transaction, error)
GetTransaction(ctx context.Context, id string) (*Transaction, error)
GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error)
}
//===============
@ -227,3 +229,7 @@ func (n nopInternalClient) Transactions(ctx context.Context) (map[string]*Transa
func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (*Transaction, error) {
return nil, nil
}
func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error) {
return nil, nil
}

View file

@ -275,6 +275,7 @@ type SystemInfo interface {
CPUCores() (physical int, logical int, err error)
CPUMHz() (int, error)
CPUArch() string
DiskCapacity(string) (uint64, error)
}
// newNopSystemInfo creates a no-op implementation of SystemInfo.
@ -345,3 +346,8 @@ func (n *nopSystemInfo) CPUMHz() (int, error) {
func (n *nopSystemInfo) CPUCores() (physical, logical int, err error) {
return 0, 0, nil
}
// DiskCapacity returns the disk capacity
func (n *nopSystemInfo) DiskCapacity(path string) (uint64, error) {
return 0, nil
}

2
go.mod
View file

@ -14,7 +14,7 @@ require (
github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361
github.com/dustin/go-humanize v1.0.0
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
github.com/glycerine/lmdb-go v1.9.32
github.com/glycerine/lmdb-go v1.9.34
github.com/go-ole/go-ole v1.2.4 // indirect
github.com/gogo/protobuf v1.2.1
github.com/golang/protobuf v1.3.3

5
go.sum
View file

@ -64,8 +64,8 @@ github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 h1:gclg6gY70GLy
github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31/go.mod h1:Ogl1Tioa0aV7gstGFO7KhffUsb9M4ydbEbbxpcEDc24=
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311 h1:AAXH0ZvYIHHqU06ASy0H2tYAkAGrQlZvEy2QZrrtt4E=
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311/go.mod h1:B72P/ZM99sNiCmaQJflpmMAF5LsDzStpLdWzn0+Vr2Y=
github.com/glycerine/lmdb-go v1.9.32 h1:thLnzCykFcmn2rACYnwpR4ovYauLNKaAuk+xj7YMbS0=
github.com/glycerine/lmdb-go v1.9.32/go.mod h1:DrPeeTGooMg6B7cjNSP14perptTJzzdBy5YoosthrRs=
github.com/glycerine/lmdb-go v1.9.34 h1:0lymJjpdelYnIMcNzsKROfIaApt99zhaHtjDJTHjGkE=
github.com/glycerine/lmdb-go v1.9.34/go.mod h1:DrPeeTGooMg6B7cjNSP14perptTJzzdBy5YoosthrRs=
github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
@ -328,6 +328,7 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e h1:aZzprAO9/8oim3qStq3wc1Xuxx4QmAGriC4VU4ojemQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

View file

@ -21,6 +21,7 @@ import (
"github.com/pilosa/pilosa/v2"
"github.com/shirou/gopsutil/cpu"
"github.com/shirou/gopsutil/disk"
"github.com/shirou/gopsutil/host"
"github.com/shirou/gopsutil/mem"
)
@ -243,6 +244,16 @@ func (s *systemInfo) CPUCores() (physical, logical int, err error) {
return s.cpuPhysicalCores, s.cpuLogicalCores, nil
}
// DiskCapacity returns the disk capacity.
func (s *systemInfo) DiskCapacity(path string) (uint64, error) {
diskInfo, err := disk.Usage(path)
if err != nil {
return 0, err
}
return diskInfo.Total, nil
}
// NewSystemInfo is a constructor for the gopsutil implementation of SystemInfo.
func NewSystemInfo() *systemInfo {
return &systemInfo{}

View file

@ -1247,6 +1247,37 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
return tkresp.Keys, nil
}
// GetNodeUsage retrieves the size-on-disk information for the specified node.
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pilosa.URI) (map[string]pilosa.NodeUsage, error) {
u := uri.Path("/ui/usage?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, err
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading")
}
nodeUsages := make(map[string]pilosa.NodeUsage) // map of size 1
if err := json.Unmarshal(body, &nodeUsages); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
}
return nodeUsages, nil
}
func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions")
defer span.Finish()

View file

@ -665,34 +665,25 @@ func (h *Handler) handleGetUsage(w http.ResponseWriter, r *http.Request) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
usageIndexes, usageTotal, err := h.api.Usage()
q := r.URL.Query()
remoteStr := q.Get("remote")
var remote bool
if remoteStr == "true" {
remote = true
}
nodeUsages, err := h.api.Usage(r.Context(), remote)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
disk := diskUsage{
Total: usageTotal,
Indexes: usageIndexes,
}
usage := getUsageResponse{
Disk: disk,
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(usage); err != nil {
if err := json.NewEncoder(w).Encode(nodeUsages); err != nil {
h.logger.Printf("write status response error: %s", err)
}
}
type getUsageResponse struct {
Disk diskUsage `json:"bytesOnDisk"`
}
type diskUsage struct {
Total int64 `json:"total"`
Indexes map[string]int64 `json:"indexes"`
}
// handleGetStatus handles GET /status requests.
func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
if !validHeaderAcceptJSON(r.Header) {

View file

@ -1180,7 +1180,7 @@ func (c *Cursor) merge(key uint64, data *roaring.Container) (bool, error) {
if err != nil {
return false, errors.Wrap(err, "cursor.merge")
}
container = roaring.NewContainerBitmap(cell.BitN, d)
container = roaring.NewContainerBitmap(-1, d)
case ContainerTypeRLE:
d := toInterval16(cell.Data)
container = roaring.NewContainerRun(d)
@ -1267,7 +1267,7 @@ func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
if err != nil {
return false, errors.Wrap(err, "cursor.difference")
}
container = roaring.NewContainerBitmap(cell.N, d)
container = roaring.NewContainerBitmap(-1, d)
case ContainerTypeRLE:
d := toInterval16(cell.Data)
container = roaring.NewContainerRun(d)

View file

@ -172,12 +172,16 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
cloneMaybe = make([]uint64, len(bm))
copy(cloneMaybe, bm)
}
c = roaring.NewContainerBitmap(l.N, cloneMaybe)
c = roaring.NewContainerBitmap(-1, cloneMaybe)
case ContainerTypeBitmap:
c = roaring.NewContainerBitmap(l.N, toArray64(cpMaybe))
c = roaring.NewContainerBitmap(-1, toArray64(cpMaybe))
case ContainerTypeRLE:
c = roaring.NewContainerRun(toInterval16(cpMaybe))
}
// Note: If the "roaringparanoia" build tag isn't set, this
// should be optimized away entirely. Otherwise it's moderately
// expensive.
c.CheckN()
c.SetMapped(mapped)
return c
}

View file

@ -440,7 +440,7 @@ func benchmarkGetSeq(b *testing.B, n int) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for j := 0; j < n; j++ {
r.Get(uint64(j))
_, _ = r.Get(uint64(j))
}
}
b.StopTimer()
@ -518,7 +518,7 @@ func benchmarkGetRnd(b *testing.B, n int) {
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for _, v := range a {
r.Get(uint64(v))
_, _ = r.Get(uint64(v))
}
}
b.StopTimer()

View file

@ -146,6 +146,9 @@ func NewContainerBitmap(n int, bitmap []uint64) *Container {
c.bitmapRepair()
} else {
c.setN(int32(n))
if roaringParanoia {
c.CheckN()
}
}
return c
}
@ -164,6 +167,9 @@ func NewContainerBitmapN(bitmap []uint64, n int32) *Container {
} else {
c.setBitmap(bitmap)
}
if roaringParanoia {
c.CheckN()
}
return c
}
@ -219,6 +225,9 @@ func NewContainerRunCopy(set []Interval16) *Container {
func NewContainerRunN(set []Interval16, n int32) *Container {
c := &Container{typeID: ContainerRun, n: n}
c.setRuns(set)
if roaringParanoia {
c.CheckN()
}
return c
}
@ -624,61 +633,6 @@ func (c *Container) setRunsMaybeCopy(runs []Interval16, doCopy bool) {
c.pointer, c.len, c.cap = &runs[0].Start, int32(len(runs)), int32(cap(runs))
}
// UpdateOrMake updates the container, yielding a new container if necessary.
func (c *Container) UpdateOrMake(typ byte, n int32, mapped bool) *Container {
if c == nil {
switch typ {
case ContainerRun:
c = NewContainerRunN(nil, n)
case ContainerBitmap:
c = NewContainerBitmapN(nil, n)
default:
c = NewContainerArrayN(nil, n)
}
c.flags |= flagMapped
return c
}
// ensure that we are allowed to modify this container
c = c.Thaw()
c.typeID = typ
c.n = n
// note: this probably shouldn't be happening, the decision should be getting
// made when we specify the storage.
c.setMapped(mapped)
// we don't know that any existing slice is usable, so let's ditch it
switch c.typeID {
case ContainerArray:
c.pointer, c.len, c.cap = &c.data[0], 0, stashedArraySize
case ContainerRun:
c.pointer, c.len, c.cap = &c.data[0], 0, stashedRunSize
default:
c.pointer, c.len, c.cap = nil, 0, 0
}
return c
}
// Update updates the container if possible. It is an error to
// call Update on a frozen container.
func (c *Container) Update(typ byte, n int32, mapped bool) {
if c == nil || c.frozen() {
panic("cannot Update a nil or frozen container")
}
c.typeID = typ
c.n = n
// note: this probably shouldn't be happening, the decision should be getting
// made when we specify the storage.
c.setMapped(mapped)
// we don't know that any existing slice is usable, so let's ditch it
switch c.typeID {
case ContainerArray:
c.pointer, c.len, c.cap = nil, 0, 0
case ContainerRun:
c.pointer, c.len, c.cap = nil, 0, 0
default:
c.pointer, c.len, c.cap = nil, 0, 0
}
}
// isArray returns true if the container is an array container.
func (c *Container) isArray() bool {
if c == nil {

View file

@ -188,7 +188,7 @@ func (btc *bTreeContainers) Repair() {
// (new-container, write). If write is true, the container is used to
// replace the given container.
func (btc *bTreeContainers) Update(key uint64, fn func(*Container, bool) (*Container, bool)) {
btc.tree.Put(key, fn)
_, _ = btc.tree.Put(key, fn)
btc.lastKey = ^uint64(0)
btc.lastContainer = nil
}

View file

@ -1568,7 +1568,7 @@ func (b *Bitmap) Shift(n int) (*Bitmap, error) {
}
o, carry := shift(ci)
if lastCarry {
o.add(0)
o, _ = o.add(0)
}
if o.N() > 0 {
output.Containers.Put(ki, o)
@ -3465,8 +3465,11 @@ func (c *Container) bitmapToArray() *Container {
}
// arrayToBitmap converts from array format to bitmap format.
func (c *Container) arrayToBitmap() *Container {
func (c *Container) arrayToBitmap() (out *Container) {
statsHit("arrayToBitmap")
if roaringParanoia {
defer func() { out.CheckN() }()
}
if c == nil {
if roaringParanoia {
panic("nil container for arrayToBitmap")
@ -3498,8 +3501,11 @@ func (c *Container) arrayToBitmap() *Container {
}
// runToBitmap converts from RLE format to bitmap format.
func (c *Container) runToBitmap() *Container {
func (c *Container) runToBitmap() (out *Container) {
statsHit("runToBitmap")
if roaringParanoia {
defer func() { c.CheckN() }()
}
if c == nil {
if roaringParanoia {
panic("nil container for runToBitmap")
@ -3725,6 +3731,9 @@ func (c *Container) runToArray() *Container {
// Clone returns a copy of c.
func (c *Container) Clone() (out *Container) {
if roaringParanoia {
defer func() { out.CheckN() }()
}
statsHit("Container/Clone")
if c == nil {
return nil
@ -3735,8 +3744,9 @@ func (c *Container) Clone() (out *Container) {
out = NewContainerArrayCopy(c.array())
case ContainerBitmap:
statsHit("Container/Clone/Bitmap")
other := NewContainerBitmapN(nil, c.N())
other := NewContainerBitmapN(nil, 0)
copy(other.bitmap(), c.bitmap())
other.n = c.n
out = other
case ContainerRun:
statsHit("Container/Clone/Run")
@ -4098,7 +4108,10 @@ func intersectionCountBitmapBitmap(a, b *Container) (n int32) {
return int32(popcountAndSlice(a.bitmap(), b.bitmap()))
}
func intersect(a, b *Container) *Container {
func intersect(a, b *Container) (c *Container) {
if roaringParanoia {
defer func() { c.CheckN() }()
}
if a.N() == MaxContainerVal+1 {
return b.Freeze()
}
@ -4320,7 +4333,10 @@ func intersectBitmapBitmap(a, b *Container) *Container {
return output
}
func union(a, b *Container) *Container {
func union(a, b *Container) (c *Container) {
if roaringParanoia {
defer func() { c.CheckN() }()
}
if a.N() == MaxContainerVal+1 || b.N() == MaxContainerVal+1 {
return fullContainer
}
@ -4543,7 +4559,7 @@ func unionRunRun(a, b *Container) *Container {
}
output.setN(n)
if len(output.runs()) > runMaxSize {
output.runToBitmap()
output = output.runToBitmap()
}
return output
}
@ -5029,7 +5045,10 @@ func appendInterval16At(a []Interval16, val Interval16, off int) ([]Interval16,
return a, off
}
func difference(a, b *Container) *Container {
func difference(a, b *Container) (c *Container) {
if roaringParanoia {
defer func() { c.CheckN() }()
}
if a.N() == 0 || b.N() == MaxContainerVal+1 {
return nil
}
@ -5072,14 +5091,14 @@ func differenceArrayArray(a, b *Container) *Container {
for i, j := 0, 0; i < na; {
va := aa[i]
if j >= nb {
output.add(va)
output, _ = output.add(va)
i++
continue
}
vb := ab[j]
if va < vb {
output.add(va)
output, _ = output.add(va)
i++
} else if va > vb {
j++
@ -5386,7 +5405,10 @@ func differenceBitmapBitmap(a, b *Container) *Container {
return output
}
func xor(a, b *Container) *Container {
func xor(a, b *Container) (c *Container) {
if roaringParanoia {
defer func() { c.CheckN() }()
}
if a.N() == 0 {
return b.Freeze()
}
@ -6482,14 +6504,15 @@ func (b *Bitmap) DifferenceInPlace(others ...*Bitmap) {
if targetKey == iKey {
// note: a nil container is valid, and has N == 0.
if iContainer.N() != 0 {
if curContainer.frozen() {
curContainer = curContainer.Clone()
b.Containers.Put(targetKey, curContainer)
}
curContainer.differenceInPlace(iContainer)
// Note: This Thaw() may be unnecessary, but some of the
// differenceInPlace code may be assuming the container is
// always writable.
curContainer = curContainer.Thaw().differenceInPlace(iContainer)
if curContainer.N() == 0 {
removeContainerKeys = append(removeContainerKeys, targetKey)
break
} else {
b.Containers.Put(targetKey, curContainer)
}
}
iIter.hasNext = iIter.iter.Next()
@ -6504,43 +6527,44 @@ func (b *Bitmap) DifferenceInPlace(others ...*Bitmap) {
target.Containers.Repair()
}
func (c *Container) differenceInPlace(other *Container) {
func (c *Container) differenceInPlace(other *Container) *Container {
if other == nil {
return
return c
}
if other.isArray() {
if c.isArray() {
differenceArrayArrayInPlace(c, other)
return differenceArrayArrayInPlace(c, other)
} else if c.isBitmap() {
differenceBitmapArrayInPlace(c, other)
return differenceBitmapArrayInPlace(c, other)
} else if c.isRun() {
differenceRunArrayInPlace(c, other)
return differenceRunArrayInPlace(c, other)
}
} else if other.isBitmap() {
if c.isArray() {
differenceArrayBitmapInPlace(c, other)
return differenceArrayBitmapInPlace(c, other)
} else if c.isBitmap() {
differenceBitmapBitmapInPlace(c, other)
return differenceBitmapBitmapInPlace(c, other)
} else if c.isRun() {
differenceRunBitmapInPlace(c, other)
return differenceRunBitmapInPlace(c, other)
}
} else if other.isRun() {
if c.isArray() {
differenceArrayRunInPlace(c, other)
return differenceArrayRunInPlace(c, other)
} else if c.isBitmap() {
differenceBitmapRunInPlace(c, other)
return differenceBitmapRunInPlace(c, other)
} else if c.isRun() {
differenceRunRunInPlace(c, other)
return differenceRunRunInPlace(c, other)
}
}
return c
}
func differenceArrayArrayInPlace(c, other *Container) {
func differenceArrayArrayInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/ArrayArray")
aa, ab := c.array(), other.array()
na, nb := len(aa), len(ab)
if na == 0 || nb == 0 {
return
return c
}
n := 0
for i, j := 0, 0; i < na; {
@ -6565,15 +6589,16 @@ func differenceArrayArrayInPlace(c, other *Container) {
}
aa = aa[:n]
c.setArray(aa)
return c
}
func differenceArrayBitmapInPlace(c, other *Container) {
func differenceArrayBitmapInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/ArrayBitmap")
aa := c.array()
n := 0
bitmap := other.bitmap()
if len(aa) == 0 || len(bitmap) == 0 {
return
return c
}
for _, va := range aa {
bmidx := va / 64
@ -6588,16 +6613,17 @@ func differenceArrayBitmapInPlace(c, other *Container) {
}
aa = aa[:n]
c.setArray(aa)
return c
}
func differenceArrayRunInPlace(c, other *Container) {
func differenceArrayRunInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/ArrayRun")
i := 0 // array index
j := 0 // run index
aa, rb := c.array(), other.runs()
if len(aa) == 0 || len(rb) == 0 {
return
return c
}
n := 0
@ -6632,14 +6658,15 @@ func differenceArrayRunInPlace(c, other *Container) {
}
aa = aa[:n]
c.setArray(aa)
return c
}
func differenceBitmapArrayInPlace(c, other *Container) {
func differenceBitmapArrayInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/BitmapArray")
bitmap := c.bitmap()
ab := other.array()
if len(bitmap) == 0 || len(ab) == 0 {
return
return c
}
n := c.N()
@ -6651,18 +6678,19 @@ func differenceBitmapArrayInPlace(c, other *Container) {
}
c.setN(n)
if n < ArrayMaxSize {
c.bitmapToArray() // With This Work
c = c.bitmapToArray() // With This Work
}
return c
}
func differenceBitmapBitmapInPlace(c, other *Container) {
func differenceBitmapBitmapInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/BitmapBitmap")
// local variables added to prevent BCE checks in loop
// see https://go101.org/article/bounds-check-elimination.html
a := c.bitmap()
b := other.bitmap()
if len(a) == 0 || len(b) == 0 {
return
return c
}
var (
@ -6677,25 +6705,27 @@ func differenceBitmapBitmapInPlace(c, other *Container) {
}
c.setN(n)
if n < ArrayMaxSize {
c.bitmapToArray() // Will this work?
c = c.bitmapToArray()
}
return c
}
func differenceBitmapRunInPlace(c, other *Container) {
func differenceBitmapRunInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/BitmapRun")
if len(c.bitmap()) == 0 {
return
return c
}
for _, run := range other.runs() {
c.bitmapZeroRange(uint64(run.Start), uint64(run.Last)+1)
}
return c
}
func differenceRunArrayInPlace(c, other *Container) {
func differenceRunArrayInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/RunArray")
ra, ab := c.runs(), other.array()
if len(ra) == 0 || len(ab) == 0 {
return
return c
}
runs := make([]Interval16, 0, len(ra))
bidx := 0
@ -6745,14 +6775,14 @@ RUNLOOP:
for _, run := range runs {
c.n += int32(run.Last-run.Start) + 1
}
c.optimize()
return c.optimize()
}
func differenceRunBitmapInPlace(c, other *Container) {
func differenceRunBitmapInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/RunBitmap")
ra := c.runs()
if len(ra) == 0 || len(other.bitmap()) == 0 {
return
return c
}
// If a is full, difference is the flip of b.
if len(ra) > 0 && ra[0].Start == 0 && ra[0].Last == 65535 {
@ -6765,7 +6795,7 @@ func differenceRunBitmapInPlace(c, other *Container) {
c.setMapped(false)
c.setBitmap(bitmap)
c.setN(c.count())
return
return c
}
runs := make([]Interval16, 0, len(ra))
for _, inputRun := range ra {
@ -6811,18 +6841,19 @@ func differenceRunBitmapInPlace(c, other *Container) {
c.n += int32(run.Last-run.Start) + 1
}
if c.N() < ArrayMaxSize && int32(len(runs)) > c.N()/2 {
c.runToArray()
c = c.runToArray()
} else if len(runs) > runMaxSize {
c.runToBitmap()
c = c.runToBitmap()
}
return c
}
func differenceRunRunInPlace(c, other *Container) {
func differenceRunRunInPlace(c, other *Container) *Container {
statsHit("differenceInPlace/RunRun")
ra, rb := c.runs(), other.runs()
if len(ra) == 0 || len(rb) == 0 {
return
return c
}
apos := 0 // current a-run index
bpos := 0 // current b-run index
@ -6882,6 +6913,7 @@ func differenceRunRunInPlace(c, other *Container) {
for _, run := range runs {
c.n += int32(run.Last-run.Start) + 1
}
return c
}
//RBF exports to be reconsidered as we progress

View file

@ -135,19 +135,19 @@ func TestRunCountRange(t *testing.T) {
if cnt != 0 {
t.Fatalf("should get 0 from empty container, but got: %v", cnt)
}
c.add(5)
c.add(6)
c.add(7)
c, _ = c.add(5)
c, _ = c.add(6)
c, _ = c.add(7)
cnt = RunCountRange(c.runs(), 2, 9)
if cnt != 3 {
t.Fatalf("should get 3 from interval within range, but got: %v", cnt)
}
c.add(8)
c.add(9)
c.add(10)
c.add(11)
c, _ = c.add(8)
c, _ = c.add(9)
c, _ = c.add(10)
c, _ = c.add(11)
cnt = RunCountRange(c.runs(), 4, 8)
if cnt != 3 {
@ -199,17 +199,17 @@ func TestRunCountRange(t *testing.T) {
t.Fatalf("should get 6 from interval equal to range, but got: %v", cnt)
}
c.add(17)
c.add(19)
c.add(18)
c, _ = c.add(17)
c, _ = c.add(19)
c, _ = c.add(18)
cnt = RunCountRange(c.runs(), 1, 22)
if cnt != 10 {
t.Fatalf("should get 10 from multiple ranges in interval, but got: %v", cnt)
}
c.add(13)
c.add(14)
c, _ = c.add(13)
c, _ = c.add(14)
cnt = RunCountRange(c.runs(), 6, 18)
if cnt != 9 {
@ -227,17 +227,17 @@ func TestRunContains(t *testing.T) {
if c.runContains(5) {
t.Fatalf("empty run container should not contain 5")
}
c.add(5)
c, _ = c.add(5)
if !c.runContains(5) {
t.Fatalf("run container with 5 should contain 5")
}
c.add(6)
c.add(7)
c, _ = c.add(6)
c, _ = c.add(7)
c.add(9)
c.add(10)
c.add(11)
c, _ = c.add(9)
c, _ = c.add(10)
c, _ = c.add(11)
if !c.runContains(10) {
t.Fatalf("run container with 10 in second run should contain 10")
@ -282,7 +282,7 @@ func TestIntersectionCountArrayBitmap3(t *testing.T) {
if res.N() != res.count() || res.N() != MaxContainerVal+1 {
t.Fatalf("test #2 intersectCountBitmapRun fail orig: %v new: %v exp: %v", res.N(), res.count(), MaxContainerVal+1)
}
b.bitmapToRun(0)
b = b.bitmapToRun(0)
res = intersectRunRun(a, b)
n := intersectionCountRunRun(a, b)
if res.N() != res.count() || res.N() != MaxContainerVal+1 || res.N() != int32(n) {
@ -613,7 +613,7 @@ func TestIntersectBitmapRunBitmap(t *testing.T) {
b.setN(4097)
ret := intersectBitmapRun(a, b)
if ret.isArray() {
ret.arrayToBitmap()
ret = ret.arrayToBitmap()
}
if !reflect.DeepEqual(ret.bitmap(), exp) {
t.Fatalf("test #%v expected %v, but got %v", i, exp, ret.bitmap())
@ -710,9 +710,9 @@ func TestUnionMixed(t *testing.T) {
res := union(tt.c1, tt.c2)
// convert to array for comparison
if res.isBitmap() {
res.bitmapToArray()
res = res.bitmapToArray()
} else if res.isRun() {
res.runToArray()
res = res.runToArray()
}
if !reflect.DeepEqual(res.array(), tt.exp) {
t.Fatalf("test %s expected %v, but got %v", tt.name, tt.exp, res.array())
@ -1304,11 +1304,11 @@ func TestBitmapToRun(t *testing.T) {
for i, test := range tests {
a := NewContainerBitmap(-1, test.bitmap)
x := a.bitmap()
a.bitmapToRun(0)
a = a.bitmapToRun(0)
if !reflect.DeepEqual(a.runs(), test.exp) {
t.Fatalf("test #%v expected %v, but got %v", i, test.exp, a.runs())
}
a.runToBitmap()
a = a.runToBitmap()
if !reflect.DeepEqual(a.bitmap(), x) {
t.Fatalf("test #%v expected %v, but got %v", i, a.bitmap(), x)
}
@ -1633,7 +1633,7 @@ func MakeBitmap(start []uint64) []uint64 {
func MakeLastBitSet() []uint64 {
obj := NewFileBitmap(65535)
c := obj.container(0)
c.arrayToBitmap()
c = c.arrayToBitmap()
return c.bitmap()
}
@ -1895,9 +1895,10 @@ func TestWriteReadArray(t *testing.T) {
func TestWriteReadBitmap(t *testing.T) {
// create bitmap containing > 4096 bits
cb := NewContainerBitmapN(nil, 129*32)
cb := NewContainerBitmapN(nil, 0)
for i := 0; i < 129; i++ {
cb.bitmap()[i] = 0x5555555555555555
cb.n += 32
}
bb := NewFileBitmap()
bb.Containers.Put(0, cb)
@ -1918,9 +1919,10 @@ func TestWriteReadBitmap(t *testing.T) {
func TestWriteReadFullBitmap(t *testing.T) {
// create bitmap containing > 4096 bits
cb := NewContainerBitmapN(nil, 65536)
cb := NewContainerBitmapN(nil, 0)
for i := 0; i < bitmapN; i++ {
cb.bitmap()[i] = 0xffffffffffffffff
cb.n += 64
}
bb := NewFileBitmap()
bb.Containers.Put(0, cb)
@ -2924,8 +2926,7 @@ func unionInPlaceWrapper(a, b *Container) *Container {
func differenceInPlaceWrapper(a, b *Container) *Container {
a = a.Clone()
// this should probably return its new value, but currently does not
a.differenceInPlace(b)
a = a.differenceInPlace(b)
return a
}
@ -3859,7 +3860,7 @@ func BenchmarkUnionBitmapBitmapInPlace(b *testing.B) {
b1 := newTestBitmapContainer()
b2 := newTestBitmapContainer()
for n := 0; n < b.N; n++ {
unionBitmapBitmapInPlace(b1, b2)
b1 = unionBitmapBitmapInPlace(b1, b2)
}
}
@ -4356,7 +4357,7 @@ func BenchmarkUnionRunRunInPlace(bm *testing.B) {
brun := doContainer(ContainerRun, br.fn())
abmp := arun.runToBitmap()
unionBitmapRunInPlace(abmp, brun)
_ = unionBitmapRunInPlace(abmp, brun)
}
})
@ -4365,7 +4366,7 @@ func BenchmarkUnionRunRunInPlace(bm *testing.B) {
arun := doContainer(ContainerRun, ar.fn())
brun := doContainer(ContainerRun, br.fn())
unionRunRunInPlace(arun, brun)
_ = unionRunRunInPlace(arun, brun)
}
})
}

View file

@ -17,3 +17,10 @@
package roaring
const roaringParanoia = false
// CheckN verifies that a container's cached count is correct, but
// there are two versions; this is the one which doesn't actually
// do anything, because the check is expensive. Which one you get is
// controlled by the roaringparanoia build tag.
func (c *Container) CheckN() {
}

View file

@ -16,4 +16,19 @@
package roaring
import "fmt"
const roaringParanoia = true
// CheckN verifies that the container's cached count is correct. Note
// that this has two definitions, depending on the presence of the
// roaringparanoia build tag.
func (c *Container) CheckN() {
if c == nil {
return
}
count := c.count()
if count != c.n {
panic(fmt.Sprintf("CheckN (%p): n %d, count %d", c, c.n, count))
}
}

View file

@ -60,7 +60,7 @@ func (b *Bitmap) UnmarshalBinary(data []byte) (err error) {
}
newC.setMapped(true)
if !b.preferMapping {
newC.unmapOrClone()
newC = newC.unmapOrClone()
}
b.Containers.Put(itrKey, newC)
itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
@ -152,7 +152,7 @@ func InspectBinary(data []byte, mapped bool, info *BitmapInfo) (b *Bitmap, mappe
}
newC.setMapped(true)
if !mapped {
newC.unmapOrClone()
newC = newC.unmapOrClone()
}
newC.flags |= flagPristine
if newC.flags&flagMapped != 0 {

View file

@ -385,13 +385,18 @@ func TestHandler_Endpoints(t *testing.T) {
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/ui/usage", nil))
if w.Code != gohttp.StatusOK {
fmt.Printf("%+v\n", w.Body)
t.Fatalf("unexpected status code: %d", w.Code)
}
ret := mustJSONDecode(t, w.Body)
usage := ret["bytesOnDisk"].(map[string]interface{})
indexes := usage["indexes"].(map[string]interface{})
if len(indexes) != 2 {
t.Fatalf("wrong length index size list: %#v", indexes)
nodeUsages := make(map[string]pilosa.NodeUsage)
if err := json.Unmarshal(w.Body.Bytes(), &nodeUsages); err != nil {
t.Fatalf("unmarshal")
}
for _, nodeUsage := range nodeUsages {
if len(nodeUsage.Disk.Indexes) != 2 {
t.Fatalf("wrong length index size list: %#v", nodeUsage.Disk.Indexes)
}
}
})

View file

@ -18,6 +18,7 @@ import (
"fmt"
"io"
"os"
"path"
"path/filepath"
"strconv"
"strings"
@ -564,6 +565,67 @@ func (f *TxFactory) DumpAll() {
f.dbPerShard.DumpAll()
}
func (f *TxFactory) IndexSizes() (index2bytes map[string]int64, err error) {
// Open storage directory.
index2bytes = make(map[string]int64)
dirName, err := expandDirName(f.holder.path)
if err != nil {
return index2bytes, errors.Wrap(err, "expanding data directory")
}
idxs := f.holder.Indexes()
for _, idx := range idxs {
index := idx.name
fullName := path.Join(dirName, index)
roaringAndMeta, err := directoryUsage(fullName)
if err != nil {
return index2bytes, errors.Wrap(err, "getting disk usage for roaring and meta")
}
fullName = index + ".index.txstores@@@"
rbfOrLmdb, err := directoryUsage(fullName)
if err != nil {
return index2bytes, errors.Wrap(err, "getting disk usage for backend")
}
index2bytes[index] = roaringAndMeta + rbfOrLmdb
}
return index2bytes, nil
}
func directoryUsage(fname string) (int64, error) {
if !DirExists(fname) {
return 0, nil
}
var size int64
dir, err := os.Open(fname)
if err != nil {
return 0, errors.Wrap(err, "opening data subdirectory")
}
defer dir.Close()
files, err := dir.Readdir(-1)
if err != nil {
return 0, errors.Wrap(err, "reading data subdirectory")
}
for _, file := range files {
if file.IsDir() {
sz, err := directoryUsage(path.Join(fname, file.Name()))
if err != nil {
return 0, err
}
size += sz
} else {
size += file.Size()
}
}
return size, nil
}
func (f *TxFactory) CloseIndex(idx *Index) error {
// under roaring and all the new databases, this is a no-op.
//idx.Dump("CloseIndex")