// TODO need to run this and prometheus through some benchmarks for allocations
501570
func extractCollectdDimensions(doit bool, metricName string) (string, map[string]string) {
	dimensions := make(map[string]string)
	if !doit {
		return metricName, dimensions
	}
	index := strings.Index(metricName, "..")
	var toAddDims map[string]string
	for {
		metricName, toAddDims = collectd.GetDimensionsFromName(&metricName)
		if len(toAddDims) == 0 {
			// we may have put two dots next to each other extracting this stuff, see
			// if there were two beforehand and if not, replace them if they occur with
			// a single dot
			// TODO could do better here if wavefront users prefixed all metrics with
			// collectd, we could know how they're constructed and could make them be
			// almost as good as coming from collectd.
			// could even be crazy and put the types.db file in here and provide
			// accurate metric types... overkill?
			if index == -1 {
				metricName = strings.Replace(metricName, "..", ".", -1)
			}

			return metricName, dimensions
		}
		for k, v := range toAddDims {
			dimensions[k] = v
		}
	}
}
324059
func ParallelDoWithLimit(jobs []func() interface{}, nThreads int) FutureGroup {
	if nThreads > len(jobs) {
		nThreads = len(jobs)
	}
	tp :=	NewPool(nThreads, len(jobs))
	tp.Start()
	defer func() {
		tp.Shutdown()
		tp.Wait()
	}()
	result := make([]*Future, len(jobs))
	for idx, nj := range(jobs) {
		result[idx] = tp.Submit(nj)
	}
	return FutureGroup{result}
}
552601
func benchmark32s(n int) {
	//var hashes = make(Uint32Slice, n)
	const nbytes = 16384 // fix
	bs := make([]byte, nbytes, nbytes)
	bs = bs[:]
	//fp20 := make([]byte, 20, 20)
	Hf2 = "nullhash"
	for _, ksiz := range keySizes {
		if ksiz == 512 {
			n = n / 10
		}
		bs = bs[:ksiz]
		//hashes = make(Uint32Slice, n, n)
		//hashes = hashes[:]
		//fmt.Printf("ksiz=%d, len(bs)=%d\n", ksiz, len(bs))
		pn := hrff.Int64{int64(n), ""}
		ps := hrff.Int64{int64(n * ksiz), "B"}
		//fmt.Printf("benchmark32s: gen n=%d, n=%h, keySize=%d, size=%h\n", n, pn, ksiz, ps)
		if false {
			_ = gomap.Hash64(bs, 0)
			_ = aeshash.Hash(bs, 0)
		}
		start, stop := time.Now(), time.Now()
		switch ksiz {
		case 4:
			start = time.Now()
			for i := 0; i < n; i++ {
				_ = aeshash.Hash(bs, 0)
				// sha1160.Reset()
				// sha1160.Write(bs)
				// fp20 = fp20[0:0]
				// fp20 = sha1160.Sum(fp20)
				// _ = uint64(fp20[0])<<56 | uint64(fp20[1])<<48 | uint64(fp20[2])<<40 | uint64(fp20[3])<<32 |
				// 	uint64(fp20[4])<<24 | uint64(fp20[5])<<16 | uint64(fp20[6])<<8  | uint64(fp20[7])<<0
				//k224.Reset()
				//k224.Write(bs)
				//_ = k224.Sum(nil)
				//bs[0], bs[1], bs[2], bs[3] = byte(i), byte(i>>8), byte(i>>16), byte(i>>24)
				//_, _ = jenkins.Jenkins364(bs, 0, 0, 0)
				//_ = jenkins.Hash232(bs, 0)
				//_, _ = jenkins.Jenkins364(bs, 0, 0, 0)
				//_ = gomap.Hash64(bs, 0)
				//_ =  nhf64.Hash64S(bs, 0) // chnage below too
				//nh.Reset()
				//nh.Write(bs)
				//_ = nh.Sum64()
				//Hashf(bs, 0)
				//_ = siphash.Hash(0, 0, bs)
				//hashes[i] = h
				//fmt.Printf("i=%d, 0x%08x, h=0x%08x\n", i, i, h)
				//hashes[i] = h
				//fmt.Printf("i=%d, 0x%08x, h=0x%08x\n", i, i, h)
			}
			stop = time.Now()
		default:
			start = time.Now()
			for i := 0; i < n; i++ {
				// sha1160.Reset()
				// sha1160.Write(bs)
				// fp20 = fp20[0:0]
				// fp20 = sha1160.Sum(fp20)
				// _ = uint64(fp20[0])<<56 | uint64(fp20[1])<<48 | uint64(fp20[2])<<40 | uint64(fp20[3])<<32 |
				// 	uint64(fp20[4])<<24 | uint64(fp20[5])<<16 | uint64(fp20[6])<<8  | uint64(fp20[7])<<0
				_ = aeshash.Hash(bs, 0)
				//k224.Reset()
				//k224.Write(bs)
				//_ = k224.Sum(nil)
				//bs[0], bs[1], bs[2], bs[3], bs[4], bs[5], bs[6], bs[7] = byte(i), byte(i>>8), byte(i>>16), byte(i>>24), byte(i>>32), byte(i>>40), byte(i>>48), byte(i>>56)
				//_, _ = jenkins.Jenkins364(bs, 0, 0, 0)
				//_ = aeshash.Hash(bs, 0)
				//_ = gomap.Hash64(bs, 0)
				//_ = siphash.Hash(0, 0, bs)
				//_ =  nhf64.Hash64S(bs, 0)
				//nh.Reset()
				//nh.Write(bs)
				//_ = nh.Sum64()
				//Hashf(bs, 0)
			}
			stop = time.Now()
		}
		d := tdiff(start, stop)
		hsec := hrff.Float64{(float64(n) / d.Seconds()), "hashes/sec"}
		bsec := hrff.Float64{(float64(n) * float64(ksiz) / d.Seconds()), "B/sec"}
		//fmt.Printf("benchmark32s: %h\n", hsec)
		//fmt.Printf("benchmark32s: %.2h\n\n", bsec)
		fmt.Printf("\tksize=%d, n=%h, size=%h, %h, %.1h, time=%v\n", ksiz, pn, ps, hsec, bsec, d)
	}
	return

	fmt.Printf("benchmark32s: sort n=%d\n", n)
	//hashes.Sort()
	/*
		for i := 0; i < n; i++ {
			fmt.Printf("i=%d, 0x%08x, h=0x%08x\n", i, i, hashes[i])
		}
	*/
	fmt.Printf("benchmark32s: dup check n=%d\n", n)
	//dups, mrun := checkForDups32(hashes)
	//fmt.Printf("benchmark32: dups=%d, mrun=%d\n", dups, mrun)
}
328014
func (s *SystemScheduler) computeJobAllocs() error {
	// Lookup the allocations by JobID
	ws := memdb.NewWatchSet()
	allocs, err := s.state.AllocsByJob(ws, s.eval.Namespace, s.eval.JobID, true)
	if err != nil {
		return fmt.Errorf("failed to get allocs for job '%s': %v",
			s.eval.JobID, err)
	}

	// Determine the tainted nodes containing job allocs
	tainted, err := taintedNodes(s.state, allocs)
	if err != nil {
		return fmt.Errorf("failed to get tainted nodes for job '%s': %v",
			s.eval.JobID, err)
	}

	// Update the allocations which are in pending/running state on tainted
	// nodes to lost
	updateNonTerminalAllocsToLost(s.plan, tainted, allocs)

	// Filter out the allocations in a terminal state
	allocs, terminalAllocs := structs.FilterTerminalAllocs(allocs)

	// Diff the required and existing allocations
	diff := diffSystemAllocs(s.job, s.nodes, tainted, allocs, terminalAllocs)
	s.logger.Debug("reconciled current state with desired state",
		"place", len(diff.place), "update", len(diff.update),
		"migrate", len(diff.migrate), "stop", len(diff.stop),
		"ignore", len(diff.ignore), "lost", len(diff.lost))

	// Add all the allocs to stop
	for _, e := range diff.stop {
		s.plan.AppendStoppedAlloc(e.Alloc, allocNotNeeded, "")
	}

	// Add all the allocs to migrate
	for _, e := range diff.migrate {
		s.plan.AppendStoppedAlloc(e.Alloc, allocNodeTainted, "")
	}

	// Lost allocations should be transitioned to desired status stop and client
	// status lost.
	for _, e := range diff.lost {
		s.plan.AppendStoppedAlloc(e.Alloc, allocLost, structs.AllocClientStatusLost)
	}

	// Attempt to do the upgrades in place
	destructiveUpdates, inplaceUpdates := inplaceUpdate(s.ctx, s.eval, s.job, s.stack, diff.update)
	diff.update = destructiveUpdates

	if s.eval.AnnotatePlan {
		s.plan.Annotations = &structs.PlanAnnotations{
			DesiredTGUpdates: desiredUpdates(diff, inplaceUpdates, destructiveUpdates),
		}
	}

	// Check if a rolling upgrade strategy is being used
	limit := len(diff.update)
	if !s.job.Stopped() && s.job.Update.Rolling() {
		limit = s.job.Update.MaxParallel
	}

	// Treat non in-place updates as an eviction and new placement.
	s.limitReached = evictAndPlace(s.ctx, diff, diff.update, allocUpdating, &limit)

	// Nothing remaining to do if placement is not required
	if len(diff.place) == 0 {
		if !s.job.Stopped() {
			for _, tg := range s.job.TaskGroups {
				s.queuedAllocs[tg.Name] = 0
			}
		}
		return nil
	}

	// Record the number of allocations that needs to be placed per Task Group
	for _, allocTuple := range diff.place {
		s.queuedAllocs[allocTuple.TaskGroup.Name] += 1
	}

	// Compute the placements
	return s.computePlacements(diff.place)
}
522398
func measureDownstreamBandwidth_SingleRun(
	bucket s3.Bucket,
	keys []string,
	parallelism uint) (bytesPerSecond float64, err error) {
	// Time the whole process.
	timeBefore := time.Now()

	// Start several workers.
	errs := make([]error, int(parallelism))
	done := make(chan bool)

	var totalLoaded uint64
	for i := 0; i < int(parallelism); i++ {
		go func(i int) {
			// Load data from one of the keys.
			data, err := bucket.GetObject(keys[i%len(keys)])
			if err != nil {
				errs[i] = err
			}

			atomic.AddUint64(&totalLoaded, uint64(len(data)))
			done <- true
		}(i)
	}

	// Wait for all of the workers.
	for i := 0; i < int(parallelism); i++ {
		<-done
	}

	elapsed := time.Since(timeBefore)

	// Did any of the workers fail?
	for i := 0; i < int(parallelism); i++ {
		if errs[i] != nil {
			err = errs[i]
			return
		}
	}

	// Estimate the bandwidth.
	bytesTransferred := float64(totalLoaded)
	secondsElapsed := float64(elapsed) / float64(time.Second)

	bytesPerSecond = bytesTransferred / secondsElapsed
	return
}
952761
func benchmark(c *redis.Client, testname string, command string, params ...interface{}) {
	fmt.Printf("===== %s =====\n", testname)
	start := time.Now()

	for i := 0; i < *requests; i++ {
		c.Cmd(command, params...)
	}

	duration := time.Now().Sub(start)
	rps := float64(*requests) / duration.Seconds()
	fmt.Println("Requests per second:", rps)
	fmt.Printf("Duration: %v\n\n", duration.Seconds())
}
790520
func (s *StateStore) upsertAllocsImpl(index uint64, allocs []*structs.Allocation, txn *memdb.Txn) error {
	// Handle the allocations
	jobs := make(map[structs.NamespacedID]string, 1)
	for _, alloc := range allocs {
		existing, err := txn.First("allocs", "id", alloc.ID)
		if err != nil {
			return fmt.Errorf("alloc lookup failed: %v", err)
		}
		exist, _ := existing.(*structs.Allocation)

		if exist == nil {
			alloc.CreateIndex = index
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index
			if alloc.DeploymentStatus != nil {
				alloc.DeploymentStatus.ModifyIndex = index
			}

			// Issue https://github.com/hashicorp/nomad/issues/2583 uncovered
			// the a race between a forced garbage collection and the scheduler
			// marking an allocation as terminal. The issue is that the
			// allocation from the scheduler has its job normalized and the FSM
			// will only denormalize if the allocation is not terminal.  However
			// if the allocation is garbage collected, that will result in a
			// allocation being upserted for the first time without a job
			// attached. By returning an error here, it will cause the FSM to
			// error, causing the plan_apply to error and thus causing the
			// evaluation to be failed. This will force an index refresh that
			// should solve this issue.
			if alloc.Job == nil {
				return fmt.Errorf("attempting to upsert allocation %q without a job", alloc.ID)
			}
		} else {
			alloc.CreateIndex = exist.CreateIndex
			alloc.ModifyIndex = index
			alloc.AllocModifyIndex = index

			// Keep the clients task states
			alloc.TaskStates = exist.TaskStates

			// If the scheduler is marking this allocation as lost we do not
			// want to reuse the status of the existing allocation.
			if alloc.ClientStatus != structs.AllocClientStatusLost {
				alloc.ClientStatus = exist.ClientStatus
				alloc.ClientDescription = exist.ClientDescription
			}

			// The job has been denormalized so re-attach the original job
			if alloc.Job == nil {
				alloc.Job = exist.Job
			}
		}

		// COMPAT 0.7: Upgrade old objects that do not have namespaces
		if alloc.Namespace == "" {
			alloc.Namespace = structs.DefaultNamespace
		}

		// OPTIMIZATION:
		// These should be given a map of new to old allocation and the updates
		// should be one on all changes. The current implementation causes O(n)
		// lookups/copies/insertions rather than O(1)
		if err := s.updateDeploymentWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating deployment: %v", err)
		}

		if err := s.updateSummaryWithAlloc(index, alloc, exist, txn); err != nil {
			return fmt.Errorf("error updating job summary: %v", err)
		}

		if err := s.updateEntWithAlloc(index, alloc, exist, txn); err != nil {
			return err
		}

		if err := txn.Insert("allocs", alloc); err != nil {
			return fmt.Errorf("alloc insert failed: %v", err)
		}

		if alloc.PreviousAllocation != "" {
			prevAlloc, err := txn.First("allocs", "id", alloc.PreviousAllocation)
			if err != nil {
				return fmt.Errorf("alloc lookup failed: %v", err)
			}
			existingPrevAlloc, _ := prevAlloc.(*structs.Allocation)
			if existingPrevAlloc != nil {
				prevAllocCopy := existingPrevAlloc.Copy()
				prevAllocCopy.NextAllocation = alloc.ID
				prevAllocCopy.ModifyIndex = index
				if err := txn.Insert("allocs", prevAllocCopy); err != nil {
					return fmt.Errorf("alloc insert failed: %v", err)
				}
			}
		}

		// If the allocation is running, force the job to running status.
		forceStatus := ""
		if !alloc.TerminalStatus() {
			forceStatus = structs.JobStatusRunning
		}

		tuple := structs.NamespacedID{
			ID:        alloc.JobID,
			Namespace: alloc.Namespace,
		}
		jobs[tuple] = forceStatus
	}

	// Update the indexes
	if err := txn.Insert("index", &IndexEntry{"allocs", index}); err != nil {
		return fmt.Errorf("index update failed: %v", err)
	}

	// Set the job's status
	if err := s.setJobStatuses(index, txn, jobs, false); err != nil {
		return fmt.Errorf("setting job status failed: %v", err)
	}

	return nil
}
158767
func (ar *allocRunner) Alloc() *structs.Allocation {
	ar.allocLock.RLock()
	defer ar.allocLock.RUnlock()
	return ar.alloc
}
1363488
func (m *fileMetrics) PrometheusCollectors() []prometheus.Collector {
	return []prometheus.Collector{
		m.DiskSize,
		m.Files,
	}
}
746173
func (vm *Machine) FeedrateMultiplier(feedMultiplier float64) {
	for idx := range vm.Positions {
		vm.Positions[idx].State.Feedrate *= feedMultiplier
	}
}
77144
func main() {
	runtime.GOMAXPROCS(8)
	server := redismq.NewServer("localhost", "6379", "", 9, "9999")
	server.Start()
	testQueue := redismq.CreateBufferedQueue("localhost", "6379", "", 9, "example", 200)
	err := testQueue.Start()
	if err != nil {
		panic(err)
	}
	go write(testQueue)

	go read(testQueue, "1")
	go read(testQueue, "2")
	go read(testQueue, "3")
	go read(testQueue, "4")
	go read(testQueue, "5")
	go read(testQueue, "6")
	go read(testQueue, "7")
	go read(testQueue, "8")

	select {}
}
130002
func (collector *volumeStatsCollector) Collect(ch chan<- prometheus.Metric) {
	podStats, err := collector.statsProvider.ListPodStats()
	if err != nil {
		return
	}
	addGauge := func(desc *prometheus.Desc, pvcRef *stats.PVCReference, v float64, lv ...string) {
		lv = append([]string{pvcRef.Namespace, pvcRef.Name}, lv...)
		metric, err := prometheus.NewConstMetric(desc, prometheus.GaugeValue, v, lv...)
		if err != nil {
			klog.Warningf("Failed to generate metric: %v", err)
			return
		}
		ch <- metric
	}
	allPVCs := sets.String{}
	for _, podStat := range podStats {
		if podStat.VolumeStats == nil {
			continue
		}
		for _, volumeStat := range podStat.VolumeStats {
			pvcRef := volumeStat.PVCRef
			if pvcRef == nil {
				// ignore if no PVC reference
				continue
			}
			pvcUniqStr := pvcRef.Namespace + "/" + pvcRef.Name
			if allPVCs.Has(pvcUniqStr) {
				// ignore if already collected
				continue
			}
			addGauge(volumeStatsCapacityBytesDesc, pvcRef, float64(*volumeStat.CapacityBytes))
			addGauge(volumeStatsAvailableBytesDesc, pvcRef, float64(*volumeStat.AvailableBytes))
			addGauge(volumeStatsUsedBytesDesc, pvcRef, float64(*volumeStat.UsedBytes))
			addGauge(volumeStatsInodesDesc, pvcRef, float64(*volumeStat.Inodes))
			addGauge(volumeStatsInodesFreeDesc, pvcRef, float64(*volumeStat.InodesFree))
			addGauge(volumeStatsInodesUsedDesc, pvcRef, float64(*volumeStat.InodesUsed))
			allPVCs.Insert(pvcUniqStr)
		}
	}
}
582878
func (e *Evaluations) Allocations(evalID string, q *QueryOptions) ([]*AllocationListStub, *QueryMeta, error) {
	var resp []*AllocationListStub
	qm, err := e.client.query("/v1/evaluation/"+evalID+"/allocations", &resp, q)
	if err != nil {
		return nil, nil, err
	}
	sort.Sort(AllocIndexSort(resp))
	return resp, qm, nil
}
27465
func (mod *PrometheusBee) Run(eventChan chan bees.Event) {

	// Counter vector registration

	mod.counter = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: mod.counterVecName,
			Help: "Collection of all counter variables being iterated by hives",
		},
		[]string{"name"},
	)
	err := prometheus.Register(mod.counter)
	if err != nil {
		mod.LogErrorf("Error registering counter vector: %v", err)
		panic("Unable to start Prometheus due to counter initialization failure")
	}

	// Gauge vector registration

	mod.gauge = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: mod.gaugeVecName,
			Help: "Collection of all gauge variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.gauge)
	if err != nil {
		mod.LogErrorf("Error registering gauge vector: %v", err)
		panic("Unable to start Prometheus due to gauge initialization failure")
	}

	// Histogram vector registration

	mod.histogram = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name: mod.histogramVecName,
			Help: "Collection of all histogram variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.histogram)
	if err != nil {
		mod.LogErrorf("Error registering histogram vector: %v", err)
		panic("Unable to start Prometheus due to histogram initialization failure")
	}

	// Summary vector registration

	mod.summary = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name: mod.summaryVecName,
			Help: "Collection of all summary variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.summary)
	if err != nil {
		mod.LogErrorf("Error registering summary vector: %v", err)
		panic("Unable to start Prometheus due to summary initialization failure")
	}

	// Now, to serve everything up:

	go func() {
		http.Handle("/metrics", promhttp.Handler())
		err := http.ListenAndServe(":2112", nil)
		if err != nil {
			mod.LogErrorf("Error running prometheus metric handler: %v", err)
		}
	}()

	select {
	case <-mod.SigChan:
		return
	}
}
1262896
func (s *sticky) AllCacheMisses() uint64 {
	var total uint64
	for _, cache := range s.metricCache {
		total += cache.allCacheMisses()
	}
	return total
}
520491
func NewProcessStatCollector() (Collector, error) {
	fs, err := procfs.NewFS(*procPath)
	if err != nil {
		return nil, fmt.Errorf("failed to open procfs: %v", err)
	}
	subsystem := "processes"
	return &processCollector{
		fs: fs,
		threadAlloc: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, subsystem, "threads"),
			"Allocated threads in system",
			nil, nil,
		),
		threadLimit: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, subsystem, "max_threads"),
			"Limit of threads in the system",
			nil, nil,
		),
		procsState: prometheus.NewDesc(
			prometheus.BuildFQName(namespace, subsystem, "state"),
			"Number of processes in each state.",
			[]string{"state"}, nil,
		),
		pidUsed: prometheus.NewDesc(prometheus.BuildFQName(namespace, subsystem, "pids"),
			"Number of PIDs", nil, nil,
		),
		pidMax: prometheus.NewDesc(prometheus.BuildFQName(namespace, subsystem, "max_processes"),
			"Number of max PIDs limit", nil, nil,
		),
	}, nil
}
315626
func Benchmark2(benchSize int, benchCleanup bool) {
	docs := make([]int, 0, benchSize*2+1000)
	wp := new(sync.WaitGroup)
	numThreads := runtime.GOMAXPROCS(-1)
	// There are goroutines doing document operations: insert, read, query, update, delete
	wp.Add(5 * numThreads)
	// And one more changing schema and stuff
	wp.Add(1)

	// Prepare a collection with two indexes
	tmp := "/tmp/tiedot_bench2"
	os.RemoveAll(tmp)
	if benchCleanup {
		defer os.RemoveAll(tmp)
	}
	benchdb, col := mkTmpDBAndCol(tmp, "tmp")
	defer benchdb.Close()
	col.Index([]string{"nested", "nested", "str"})
	col.Index([]string{"nested", "nested", "int"})
	col.Index([]string{"nested", "nested", "float"})
	col.Index([]string{"strs"})
	col.Index([]string{"ints"})
	col.Index([]string{"floats"})

	// Insert 1000 documents to make a start
	for j := 0; j < 1000; j++ {
		if newID, err := col.Insert(sampleDoc(benchSize)); err == nil {
			docs = append(docs, newID)
		} else {
			fmt.Println("Insert error", err)
		}
	}
	start := float64(time.Now().UTC().UnixNano())

	// Insert benchSize * 2 documents
	for i := 0; i < numThreads; i++ {
		go func(i int) {
			fmt.Printf("Insert thread %d starting\n", i)
			defer wp.Done()
			for j := 0; j < benchSize/numThreads*2; j++ {
				if newID, err := col.Insert(sampleDoc(benchSize)); err == nil {
					docs = append(docs, newID)
				} else {
					fmt.Println("Insert error", err)
				}
			}
			fmt.Printf("Insert thread %d completed\n", i)
		}(i)
	}

	// Read benchSize * 2 documents
	var readCount int64
	for i := 0; i < numThreads; i++ {
		go func(i int) {
			fmt.Printf("Read thread %d starting\n", i)
			defer wp.Done()
			for j := 0; j < benchSize/numThreads*2; j++ {
				if _, err := col.Read(docs[rand.Intn(len(docs))]); err == nil {
					atomic.AddInt64(&readCount, 1)
				}
			}
			fmt.Printf("Read thread %d completed\n", i)
		}(i)
	}

	// Query benchSize times (lookup on two attributes)
	for i := 0; i < numThreads; i++ {
		go func(i int) {
			fmt.Printf("Query thread %d starting\n", i)
			defer wp.Done()
			var err error
			for j := 0; j < benchSize/numThreads; j++ {
				result := make(map[int]struct{})
				if err = db.EvalQuery(sampleQuery(benchSize), col, &result); err != nil {
					fmt.Println("Query error", err)
				}
			}
			fmt.Printf("Query thread %d completed\n", i)
		}(i)
	}

	// Update benchSize documents
	var updateCount int64
	for i := 0; i < numThreads; i++ {
		go func(i int) {
			fmt.Printf("Update thread %d starting\n", i)
			defer wp.Done()
			for j := 0; j < benchSize/numThreads; j++ {
				if err := col.Update(docs[rand.Intn(len(docs))], sampleDoc(benchSize)); err == nil {
					atomic.AddInt64(&updateCount, 1)
				}
			}
			fmt.Printf("Update thread %d completed\n", i)
		}(i)
	}

	// Delete benchSize documents
	var delCount int64
	for i := 0; i < numThreads; i++ {
		go func(i int) {
			fmt.Printf("Delete thread %d starting\n", i)
			defer wp.Done()
			for j := 0; j < benchSize/numThreads; j++ {
				if err := col.Delete(docs[rand.Intn(len(docs))]); err == nil {
					atomic.AddInt64(&delCount, 1)
				}
			}
			fmt.Printf("Delete thread %d completed\n", i)
		}(i)
	}

	// This one does a bunch of schema-changing stuff, testing the engine while document operations are busy
	go func() {
		time.Sleep(2 * time.Second)
		if err := benchdb.Create("foo"); err != nil {
			panic(err)
		} else if err := benchdb.Rename("foo", "bar"); err != nil {
			panic(err)
		} else if err := benchdb.Truncate("bar"); err != nil {
			panic(err)
		} else if err := benchdb.Scrub("bar"); err != nil {
			panic(err)
		} else if benchdb.Use("bar") == nil {
			panic("Missing collection")
		}
		for _, colName := range benchdb.AllCols() {
			if colName != "bar" && colName != "tmp" {
				panic("Wrong collections in benchmark db")
			}
		}
		defer os.RemoveAll("/tmp/tiedot_bench2_dump")
		if err := benchdb.Dump("/tmp/tiedot_bench2_dump"); err != nil {
			panic(err)
		} else if err := benchdb.Drop("bar"); err != nil {
			panic(err)
		}
		defer wp.Done()
	}()

	// Wait for all goroutines to finish, then print summary
	wp.Wait()
	end := float64(time.Now().UTC().UnixNano())
	fmt.Printf("Total operations %d: %d ns/iter, %d iter/sec\n", benchSize*7, int((end-start)/float64(benchSize)/7), int(1000000000/((end-start)/float64(benchSize)/7)))
	fmt.Printf("Read %d documents\n", readCount)
	fmt.Printf("Updated %d documents\n", updateCount)
	fmt.Printf("Deleted %d documents\n", delCount)
}