Merge branch 'master' into impossible

.circleci/config.yml

@@ -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

api.go

@@ -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.

boltdb/translate.go

@@ -116,6 +116,15 @@ func NewTranslateStore(index, field string, partitionID, partitionN int) *Transl
 
 // Open opens the translate file.
 func (s *TranslateStore) Open() (err error) {
+
+	// add the path to the problem database if we panic handling it.
+	defer func() {
+		r := recover()
+		if r != nil {
+			panic(fmt.Sprintf("pilosa/boltdb/TranslateStore.Open(s.Path='%v') panic with '%v'", s.Path, r))
+		}
+	}()
+
 	if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil {
 		return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path))
 	} else if s.db, err = bolt.Open(s.Path, 0666, &bolt.Options{Timeout: 1 * time.Second}); err != nil {

client.go

@@ -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
+}

cmd/pilosa-chk/chk.go

@@ -79,7 +79,7 @@ func main() {
 	const checkKeys = false
 	const applyKeyRepairs = false
 	for _, idx := range holder.Indexes() {
-		asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID")
+		asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs, nil, "fake-nodeID", 10)
 		if err != nil {
 			log.Fatal(err)
 		}

cmd/pilosa-fsck/fsck.go

@@ -73,6 +73,8 @@ type FsckConfig struct {
 	ReplicaN         int    // -replicas
 	PilosaConfigPath string // -config
 
+	ParallelReaders int //  -readers
+
 	topo *pilosa.Topology
 }
 
@@ -85,6 +87,8 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
 
 	fs.IntVar(&cfg.ReplicaN, "replicas", 0, "(required) manually entered replicaN; the number of replicas maintained in the cluster. Must be the same as the [cluster] 'replicas = R' entry in the pilosa.conf file for the cluster.")
 
+	fs.IntVar(&cfg.ParallelReaders, "readers", 10, "how many parallel readers to use to scan at once. 0 means do everything possible in parallel. 1 means serialize everything through a single reader. Can be adjusted to control memory consumption.")
+
 	fs.StringVar(&cfg.PilosaConfigPath, "config", "", "(required: -replicas or -config, with -config preferred) path to the pilosa.conf for the cluster (e.g. /etc/pilosa.conf)")
 
 	fs.StringVar(&cfg.JustThisIndex, "index", "", "(optional) restrict to just this index. Otherwise we default to all indexes.")
@@ -104,6 +108,12 @@ func (cfg *FsckConfig) DefineFlags(fs *flag.FlagSet) {
   -index index_name
         (optional) restrict to just this index. Otherwise we default to all indexes.
 
+  -readers PR
+        how many parallel readers to use to scan at once. PR==0 means do everything 
+        possible in parallel. PR==1 means serialize everything through a single reader. 
+        Adjust PR to control memory consumption if needed. As a practical limit, setting 
+        PR > 10000 will have no effect. (default is 10).
+
   -q
         be very quiet during analysis and repair
 
@@ -633,7 +643,7 @@ func (cfg *FsckConfig) readOneDir(dir string) (idx2frag map[string]*pilosa.Index
 
 		//vv("calling idx.ComputeTranslatorSummary(verbose, checkKeys=%v, cfg.FixCol='%v')", checkKeys, cfg.FixCol)
 
-		asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID)
+		asum, err := idx.ComputeTranslatorSummary(verbose, checkKeys, cfg.FixCol, topo, nodeID, cfg.ParallelReaders)
 		if err != nil {
 			log.Fatal(err)
 		}

cmd/pilosa-fsck/fsck_test.go

@@ -201,8 +201,9 @@ func Test_Repair(t *testing.T) {
 		FixCol: false,
 		Quiet:  true,
 		//Verbose:  true,
-		ReplicaN: nReplicas,
-		Dirs:     dirs,
+		ReplicaN:        nReplicas,
+		Dirs:            dirs,
+		ParallelReaders: 5,
 	}
 	panicOn(cfg.ValidateConfig())
 

diagnostics.go

@@ -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
+}

go.mod

@@ -13,7 +13,8 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/dgraph-io/badger/v2 v2.0.1-rc1.0.20200709123515-8e896a7af361
 	github.com/dustin/go-humanize v1.0.0
-	github.com/glycerine/lmdb-go v1.9.32
+	github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
+	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

go.sum

@@ -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=

gopsutil/systeminfo.go

@@ -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{}

http/client.go

@@ -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()

http/handler.go

@@ -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) {

index.go

@@ -26,6 +26,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/glycerine/idem"
 	"github.com/gogo/protobuf/proto"
 	"github.com/pilosa/pilosa/v2/hash"
 	"github.com/pilosa/pilosa/v2/internal"
@@ -787,30 +788,74 @@ func (ats *AllTranslatorSummary) Sort() {
 }
 
 // sums is only guaranteed to be sorted by (index, PartitionID, field) iff err returns nil
-func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string) (ats *AllTranslatorSummary, err error) {
-	i.mu.RLock()
-	defer i.mu.RUnlock()
+func (idx *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs bool, topo *Topology, nodeID string, parallelReaders int) (ats *AllTranslatorSummary, err error) {
+	idx.mu.RLock()
+	defer idx.mu.RUnlock()
 
 	ats = &AllTranslatorSummary{}
 	var atsMu sync.Mutex
 
 	if verbose {
-		fmt.Printf("\n# index: %v\n# =================\n", i.name)
+		fmt.Printf("\n# index: %v\n# =================\n", idx.name)
 	}
 
-	var g errgroup.Group
+	jobQ := make(chan func() error, 10000)
+	var errmu sync.Mutex
 
-	for _, fld := range i.fields {
+	if parallelReaders < 1 {
+		// turn it up to 11
+		parallelReaders = 10000
+	}
+
+	halters := make([]*idem.Halter, parallelReaders)
+	for j := 0; j < parallelReaders; j++ {
+		h := idem.NewHalter()
+		halters[j] = h
+	}
+	for _, h := range halters {
+		go func(h *idem.Halter) {
+			defer h.MarkDone()
+			for {
+				select {
+				case <-h.ReqStop.Chan:
+					return
+				case f, ok := <-jobQ:
+					if !ok || f == nil {
+						// channel closed, finish up
+						return
+					}
+
+					err1 := f()
+					if err1 != nil {
+						errmu.Lock()
+						if err == nil {
+							err = err1
+						}
+						errmu.Unlock()
+						// an error occurred, tell everyone to stop
+						for _, h2 := range halters {
+							h2.RequestStop()
+						}
+						return
+					}
+				}
+			}
+		}(h)
+	}
+
+floop:
+	for _, fld := range idx.fields {
 		fld := fld
-		g.Go(func() error {
-			//vv("g.Go() on fld '%v'", fld.name)
+
+		fun := func() error {
+			//vv("ComputeTranslatorSummary() on fld '%v'", fld.name)
 			sum, err := fld.translateStore.ComputeTranslatorSummaryRows()
 			if err != nil {
 				return err
 			}
 			sum.Field = fld.name
-			sum.Index = i.Name()
-			sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, i.Name())))
+			sum.Index = idx.Name()
+			sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, fld.name, idx.Name())))
 			sum.IsColKey = false
 			if verbose {
 				fmt.Printf("# row blake3-%v keyN: %5v idN: %5v field: '%v'\n", sum.Checksum, sum.KeyCount, sum.IDCount, fld.name)
@@ -819,17 +864,26 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
 			ats.Sums = append(ats.Sums, sum)
 			atsMu.Unlock()
 			return nil
-		})
-	}
+		}
+
+		select {
+		case <-halters[0].ReqStop.Chan:
+			break floop
+		case jobQ <- fun:
+		}
+	} // end floop
+
 	if verbose {
 		fmt.Printf("# ====================\n")
 	}
 
-	for partitionID, store := range i.translateStores {
+tloop:
+	for partitionID, store := range idx.translateStores {
 		partitionID := partitionID
 		store := store
-		g.Go(func() error {
-			//vv("g.Go() running on store.Path = '%v'", store.GetStorePath())
+
+		fun2 := func() error {
+			//vv("ComputeTranslatorSummary() running on store.Path = '%v'", store.GetStorePath())
 			if checkKeys {
 				prim := topo.PrimaryNodeIndex(partitionID)
 				primID := topo.nodeIDs[prim]
@@ -864,7 +918,7 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
 			}
 			sum.IsColKey = true
 			sum.PartitionID = partitionID
-			sum.Index = i.Name()
+			sum.Index = idx.Name()
 			sum.StorePath = store.GetStorePath()
 			sum.NodeID = nodeID
 			sum.IsPrimary = topo.IsPrimary(nodeID, partitionID)
@@ -877,7 +931,7 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
 				}
 			}
 
-			sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, i.Name())))
+			sum.Checksum = hash.Blake3sum16([]byte(fmt.Sprintf("%v/%v/%v", sum.Checksum, partitionID, idx.Name())))
 			if verbose {
 				// This is not regular index logging. This is output of the pilosa-fsck tool.
 				// So it must be printing straight to stdout.
@@ -888,9 +942,20 @@ func (i *Index) ComputeTranslatorSummary(verbose, checkKeys, applyKeyRepairs boo
 			atsMu.Unlock()
 
 			return nil
-		})
+		}
+		select {
+		case <-halters[0].ReqStop.Chan:
+			break tloop
+		case jobQ <- fun2:
+		}
+	} // end tloop
+
+	close(jobQ) // tell the workers no more jobs.
+
+	// wait for everyone to finish
+	for _, h := range halters {
+		<-h.Done.Chan
 	}
-	err = g.Wait()
 	return ats, err
 }
 

roaring/btree_test.go

@@ -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()

roaring/containers_btree.go

@@ -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
 }

roaring/roaring.go

@@ -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)
@@ -4559,7 +4559,7 @@ func unionRunRun(a, b *Container) *Container {
 	}
 	output.setN(n)
 	if len(output.runs()) > runMaxSize {
-		output.runToBitmap()
+		output = output.runToBitmap()
 	}
 	return output
 }
@@ -5091,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++
@@ -6504,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()
@@ -6526,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; {
@@ -6587,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
@@ -6610,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
 
@@ -6654,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()
@@ -6673,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 (
@@ -6699,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
@@ -6767,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 {
@@ -6787,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 {
@@ -6833,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
@@ -6904,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

roaring/roaring_internal_test.go

@@ -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()
 }
 
@@ -2926,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
 }
 
@@ -3861,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)
 	}
 }
 
@@ -4358,7 +4357,7 @@ func BenchmarkUnionRunRunInPlace(bm *testing.B) {
 					brun := doContainer(ContainerRun, br.fn())
 
 					abmp := arun.runToBitmap()
-					unionBitmapRunInPlace(abmp, brun)
+					_ = unionBitmapRunInPlace(abmp, brun)
 				}
 			})
 
@@ -4367,7 +4366,7 @@ func BenchmarkUnionRunRunInPlace(bm *testing.B) {
 					arun := doContainer(ContainerRun, ar.fn())
 					brun := doContainer(ContainerRun, br.fn())
 
-					unionRunRunInPlace(arun, brun)
+					_ = unionRunRunInPlace(arun, brun)
 				}
 			})
 		}

roaring/unmarshal_binary.go

@@ -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 {

server/handler_test.go

@@ -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)
+			}
 		}
 	})
 

txfactory.go

@@ -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")