// runAll executes all the canned benchmarks and prints out the stats.
545076
func runAll(ctx context.Context, st storage.Store, chanSize, bulkSize int) int {
	//   - Add non existing triples.        (done)
	//   - Add triples that already exist.  (done)
	//   - Remove non existing triples.     (done)
	//   - Remove existing triples.         (done)
	//   - BQL tree walking from root.      (done)
	//   - BQL random graph hopping.        (done)
	//   - BQL sorting.                     (done)
	//   - BQL grouping.                    (done)
	//   - BQL counting.                    (bounded by sort and grouping)
	//   - BQL filter existent              (bounded by sort and grouping)
	//   - BQL filter non existent          (bounded by sort and grouping)
	fmt.Printf("DISCLAIMER: Running this benchmarks is expensive. Consider using a machine with at least 3G of RAM.\n\n")

	var out int
	// Add non existing triples.
	out += runBattery(ctx, st, "adding non existing tree triples", chanSize, bulkSize, batteries.AddTreeTriplesBenchmark)
	out += runBattery(ctx, st, "adding non existing graph triples", chanSize, bulkSize, batteries.AddGraphTriplesBenchmark)

	// Add existing triples.
	out += runBattery(ctx, st, "adding existing tree triples", chanSize, bulkSize, batteries.AddExistingTreeTriplesBenchmark)
	out += runBattery(ctx, st, "adding existing graph triples", chanSize, bulkSize, batteries.AddExistingGraphTriplesBenchmark)

	// Remove non existing triples.
	out += runBattery(ctx, st, "removing non existing tree triples", chanSize, bulkSize, batteries.RemoveTreeTriplesBenchmark)
	out += runBattery(ctx, st, "removing non existing graph triples", chanSize, bulkSize, batteries.RemoveGraphTriplesBenchmark)

	// Remove existing triples.
	out += runBattery(ctx, st, "removing existing tree triples", chanSize, bulkSize, batteries.RemoveExistingTreeTriplesBenchmark)
	out += runBattery(ctx, st, "removing existing graph triples", chanSize, bulkSize, batteries.RemoveExistingGraphTriplesBenchmark)

	// BQL graph walking.
	out += runBattery(ctx, st, "walking the tree graph with BQL", chanSize, bulkSize, batteries.BQLTreeGraphWalking)
	out += runBattery(ctx, st, "walking the random graph with BQL", chanSize, bulkSize, batteries.BQLRandomGraphWalking)
	return out
}
279722
func GetAllRunPolicies(lister buildlister.BuildLister, updater buildclient.BuildUpdater) []RunPolicy {
	return []RunPolicy{
		&ParallelPolicy{BuildLister: lister, BuildUpdater: updater},
		&SerialPolicy{BuildLister: lister, BuildUpdater: updater},
		&SerialLatestOnlyPolicy{BuildLister: lister, BuildUpdater: updater},
	}
}
1212241
func (c *Command) Runnable() bool {
	return c.Run != nil || c.RunE != nil
}
1289327
func (n *NodeDrainer) run(ctx context.Context) {
	for {
		select {
		case <-n.ctx.Done():
			return
		case nodes := <-n.deadlineNotifier.NextBatch():
			n.handleDeadlinedNodes(nodes)
		case req := <-n.jobWatcher.Drain():
			n.handleJobAllocDrain(req)
		case allocs := <-n.jobWatcher.Migrated():
			n.handleMigratedAllocs(allocs)
		}
	}
}
825175
func writeAll(d []byte, w io.Writer) (err error) {
    i := 0
    n := 0
    l := len(d)
    for i < l && err != nil {
        n, err = w.Write(d[i:])
        i += n
    }
    return
}
565625
func (runner *suiteRunner) runFunc(method *methodType, kind funcKind, testName string, logb *logger, dispatcher func(c *C)) *C {
	c := runner.forkCall(method, kind, testName, logb, dispatcher)
	<-c.done
	return c
}
1157947
func Ran(tempdir string) bool {
	p := path.Join(tempdir, ".ran")
	_, err := os.Stat(p)
	return err == nil || !os.IsNotExist(err)
}
536559
func (engine *Engine) DumpAll(w io.Writer, tp ...core.DbType) error {
	tables, err := engine.DBMetas()
	if err != nil {
		return err
	}
	return engine.DumpTables(tables, w, tp...)
}
254668
func (c *Client) runAllocs(update *allocUpdates) {
	// Get the existing allocs
	c.allocLock.RLock()
	existing := make(map[string]uint64, len(c.allocs))
	for id, ar := range c.allocs {
		existing[id] = ar.Alloc().AllocModifyIndex
	}
	c.allocLock.RUnlock()

	// Diff the existing and updated allocations
	diff := diffAllocs(existing, update)
	c.logger.Debug("allocation updates", "added", len(diff.added), "removed", len(diff.removed),
		"updated", len(diff.updated), "ignored", len(diff.ignore))

	errs := 0

	// Remove the old allocations
	for _, remove := range diff.removed {
		c.removeAlloc(remove)
	}

	// Update the existing allocations
	for _, update := range diff.updated {
		c.logger.Trace("updating alloc", "alloc_id", update.ID, "index", update.AllocModifyIndex)
		c.updateAlloc(update)
	}

	// Make room for new allocations before running
	if err := c.garbageCollector.MakeRoomFor(diff.added); err != nil {
		c.logger.Error("error making room for new allocations", "error", err)
		errs++
	}

	// Start the new allocations
	for _, add := range diff.added {
		migrateToken := update.migrateTokens[add.ID]
		if err := c.addAlloc(add, migrateToken); err != nil {
			c.logger.Error("error adding alloc", "error", err, "alloc_id", add.ID)
			errs++
			// We mark the alloc as failed and send an update to the server
			// We track the fact that creating an allocrunner failed so that we don't send updates again
			if add.ClientStatus != structs.AllocClientStatusFailed {
				c.handleInvalidAllocs(add, err)
			}
		}
	}

	// Trigger the GC once more now that new allocs are started that could
	// have caused thresholds to be exceeded
	c.garbageCollector.Trigger()
	c.logger.Debug("allocation updates applied", "added", len(diff.added), "removed", len(diff.removed),
		"updated", len(diff.updated), "ignored", len(diff.ignore), "errors", errs)
}
390693
func (f Func) Run(rootPath string, data Data) error {
	return chdir(rootPath, func() error {
		if f.Should != nil && !f.Should(data) {
			return nil
		}
		return f.Runner(rootPath, data)
	})
}
1107838
func (e *BackgroundTask) Run(m libkb.MetaContext) (err error) {
	defer m.Trace(e.Name(), func() error { return err })()

	// use a new background context with a saved cancel function
	var cancel func()
	m, cancel = m.BackgroundWithCancel()

	e.Lock()
	defer e.Unlock()

	e.shutdownFunc = cancel
	if e.shutdown {
		// Shutdown before started
		cancel()
		e.meta("early-shutdown")
		return nil
	}

	// start the loop and return
	go func() {
		err := e.loop(m)
		if err != nil {
			e.log(m, "loop error: %s", err)
		}
		cancel()
		e.meta("loop-exit")
	}()

	return nil
}
556229
func (c *asyncClient) Flushall() (stat FutureBool, err Error) {
	resp, err := c.conn.QueueRequest(&FLUSHALL, [][]byte{})
	if err == nil {
		stat = resp.future.(FutureBool)
	}

	return
}
30523
func ListAll(runConf *RunConf) []string {
	var names []string
	for _, suite := range allSuites {
		names = append(names, List(suite, runConf)...)
	}
	return names
}
679649
func (job *MrJob) Run() error {
	if job.NewMapperF == nil {
		return errorsp.NewWithStacks("MrJob: NewMapperF undefined!")
	}
	if job.NewReducerF == nil {
		return errorsp.NewWithStacks("MrJob: NewReducerF undefined!")
	}
	if job.Source == nil {
		return errorsp.NewWithStacks("MrJob: Source undefined!")
	}
	/*
	 * Map
	 */
	sorters := job.Sorter
	if sorters == nil {
		log.Println("Sorter not specified, using MemSorters...")
		sorters = NewMemSorters()
	}

	log.Println("Start mapping...")
	endss := make([][]chan error, 0, len(job.Source))
	totalPart := 0
	for i := range job.Source {
		partCount, err := job.Source[i].PartCount()
		if err != nil {
			return errorsp.WithStacksAndMessage(err, "part count of source %d", i)
		}
		ends := make([]chan error, 0, partCount)
		for part := 0; part < partCount; part++ {
			end := make(chan error, 1)
			ends = append(ends, end)
			go func(i, part, totalPart int, end chan error) {
				end <- func() error {
					c, err := sorters.NewPartCollector(totalPart)
					if err != nil {
						return errorsp.WithStacksAndMessage(err, "new part collector for source %d part %d", i, part)
					}
					mapper := job.NewMapperF(i, part)
					key, val := mapper.NewKey(), mapper.NewVal()
					iter, err := job.Source[i].Iterator(part)
					if err != nil {
						return errorsp.WithStacksAndMessage(err, "source %d part %d", i, part)
					}
					defer iter.Close()

					for {
						if err := iter.Next(key, val); err != nil {
							if errorsp.Cause(err) == io.EOF {
								break
							}
							if errorsp.Cause(err) == io.ErrUnexpectedEOF {
								log.Printf("Ignoring the error: %v", err)
								break
							}
							return errorsp.WithStacksAndMessage(err, "next in source %d part %d", i, part)
						}
						if err := mapper.Map(key, val, c); err != nil {
							return errorsp.WithStacksAndMessage(err, "mapping %v %v in source %d part %d", key, val, i, part)
						}
					}
					return mapper.MapEnd(c)
				}()
			}(i, part, totalPart, end)
			totalPart++
		}
		endss = append(endss, ends)
	}
	for _, ends := range endss {
		for _, end := range ends {
			err := <-end
			if err != nil {
				return err
			}
		}
	}
	if err := sorters.ClosePartCollectors(); err != nil {
		log.Printf("sorters.ClosePartCollectors(): %v", err)
	}
	log.Printf("Map ends, begin to reduce")

	var ends []chan error
	parts := sorters.ReduceParts()
	for _, part := range parts {
		end := make(chan error, 1)
		ends = append(ends, end)
		go func(part int, end chan error) {
			end <- func() error {
				it, err := sorters.NewReduceIterator(part)
				if err != nil {
					return errorsp.WithStacksAndMessage(err, "new reduce iterator for part %d", part)
				}
				cs := make([]sophie.Collector, 0, len(job.Dest))
				for _, dst := range job.Dest {
					c, err := dst.Collector(part)
					if err != nil {
						return err
					}
					defer c.Close()
					cs = append(cs, c)
				}
				reducer := job.NewReducerF(part)
				return it.Iterate(cs, reducer)
			}()
		}(part, end)
	}
	for _, end := range ends {
		if err := <-end; err != nil {
			return err
		}
	}
	log.Println("Reduce ends.")

	return nil
}
229548
func (ctx *BatchedContext) Run(errChan chan error) error {
	runtime.LockOSThread()
	defer runtime.UnlockOSThread()
	for {
		select {
		case <-ctx.workAvailable:
			ctx.DoWork()
			if err := ctx.Errors(); err != nil {
				if errChan == nil {
					return err
				}
				errChan <- err

			}
		case w := <-ctx.Work():
			ctx.ErrChan() <- w()
		}
	}
}
518960
func (j *Job) run() (result []reflect.Value, err error) {
	f := reflect.ValueOf(j.funcs[j.jobFunc])
	params := j.fparams[j.jobFunc]
	if len(params) != f.Type().NumIn() {
		err = errors.New("the number of param is not adapted")
		return
	}
	in := make([]reflect.Value, len(params))
	for k, param := range params {
		in[k] = reflect.ValueOf(param)
	}
	result = f.Call(in)
	j.lastRun = time.Now()
	j.scheduleNextRun()
	return
}
547225
func (cluster *Cluster) RunBatch(bat *Batch) ([]interface{}, error) {
    for i := range bat.batches {
	go doBatch(&bat.batches[i])
    }

    for i := range bat.batches {
	<-bat.batches[i].done
    }

    var replies []interface{}
    for _, i := range bat.index {
	if bat.batches[i].err != nil {
	    return nil, bat.batches[i].err
	}

	replies = append(replies, bat.batches[i].cmds[0].reply)
	bat.batches[i].cmds = bat.batches[i].cmds[1:]
    }

    return replies, nil
}