<commit_before>package main

import (
	"errors"
	"flag"
	"labix.org/v2/mgo"
	"log"
	. "replay"
	"runtime"
	"sync/atomic"
	"time"
)

func panicOnError(err error) {
	if err != nil {
		panic(err)
	}
}

var (
	opsFilename string
	url         string
	workers     int
	maxOps      int
	numSkipOps     int
	socketTimeout	int64
	style		string
	sampleRate	float64
)

const (
	// Set one minute timeout on mongo socket connections (nanoseconds) by default
	DEFAULT_MGO_SOCKET_TIMEOUT=60000000000
)

func init() {
	flag.StringVar(&opsFilename, "ops_filename", "",
		"The file for the serialized ops, generated by the record scripts.")
	flag.StringVar(&url, "url", "",
		"The database server's url, in the format of <host>[:<port>]")
	flag.StringVar(&style, "style", "",
		"How to replay the the ops. You can choose: \n"+
			"	stress: repaly ops at fast as possible\n"+
			"	real: repaly ops in accordance to ops' timestamps")
	flag.IntVar(&workers, "workers", 10,
		"Number of workers that sends ops to database.")
	flag.IntVar(&maxOps, "maxOps", 0,
		"[Optional] Maximal amount of ops to be replayed from the "+
			"ops_filename file. By setting it to `0`, replayer will "+
			"replay all the ops.")
	flag.IntVar(&numSkipOps, "numSkipOps", 0,
		"[Optional] Skip first N ops. Useful for when the total ops in ops_filename" +
		" exceeds available memory and you're running in stress mode.")
	flag.Int64Var(&socketTimeout, "socketTimeout", DEFAULT_MGO_SOCKET_TIMEOUT, "Mongo socket timeout in nanoseconds. Defaults to 60 seconds.")
	flag.Float64Var(&sampleRate, "sample_rate", 0.0, "sample ops for latency")
}

func parseFlags() error {
	flag.Parse()
	if style != "stress" && style != "real" {
		return errors.New("Cannot recognize the style: " + style)
	}
	if workers <= 0 {
		return errors.New("`workers` should be a positive number")
	}
	if maxOps == 0 {
		maxOps = 4294967295
	}
	return nil
}

func main() {
	// Will enable system threads to make sure all cpus can be well utilized.
	runtime.GOMAXPROCS(100)
	err := parseFlags()
	panicOnError(err)

	// Prepare to dispatch ops
	var reader OpsReader
	var opsChan chan *Op
	if style == "stress" {
		err, reader = NewFileByLineOpsReader(opsFilename)
		panicOnError(err)
		if numSkipOps > 0 {
			err = reader.SkipOps(numSkipOps)
			panicOnError(err)
		}
		opsChan = NewBestEffortOpsDispatcher(reader, maxOps)
	} else {
		// TODO NewCyclicOpsReader: do we really want to make it cyclic?
		reader = NewCyclicOpsReader(func() OpsReader {
			err, reader := NewFileByLineOpsReader(opsFilename)
			panicOnError(err)
			return reader
		})
		if numSkipOps > 0 {
			err = reader.SkipOps(numSkipOps)
			panicOnError(err)
		}
		opsChan = NewByTimeOpsDispatcher(reader, maxOps)
	}

	latencyChan := make(chan Latency, workers)

	// Set up workers to do the job
	exit := make(chan int)
	opsExecuted := int64(0)
	fetch := func(id int, statsCollector IStatsCollector) {
		log.Printf("Worker #%d report for duty\n", id)
		
		session, err := mgo.Dial(url)		
		panicOnError(err)
		session.SetSocketTimeout(time.Duration(socketTimeout))
		
		defer session.Close()
		exec := OpsExecutorWithStats(session, statsCollector)
		for {
			op := <-opsChan
			if op == nil {
				break
			}
			exec.Execute(op)
			atomic.AddInt64(&opsExecuted, 1)
		}
		exit <- 1
		log.Printf("Worker #%d done!\n", id)
	}
	statsCollectorList := make([]*StatsCollector, workers)
	for i := 0; i < workers; i++ {
		statsCollectorList[i] = NewStatsCollector()
		statsCollectorList[i].SampleLatencies(sampleRate, latencyChan)
		go fetch(i, statsCollectorList[i])
	}

	// Periodically report execution status
	go func() {
		statsAnalyzer := NewStatsAnalyzer(statsCollectorList, &opsExecuted,
			latencyChan, int(sampleRate*float64(maxOps)))
		toFloat := func(nano int64) float64 {
			return float64(nano) / float64(1e6)
		}
		report := func() {
			status := statsAnalyzer.GetStatus()<commit_msg>Add optional verbose option to print op errors and other info for future troubleshooting.<commit_after>package main

import (
	"errors"
	"flag"
	"labix.org/v2/mgo"
	"log"
	. "replay"
	"runtime"
	"sync/atomic"
	"time"
)

func panicOnError(err error) {
	if err != nil {
		panic(err)
	}
}

var (
	opsFilename string
	url         string
	workers     int
	maxOps      int
	numSkipOps     int
	socketTimeout	int64
	style		string
	sampleRate	float64
	verbose		bool
)

const (
	// Set one minute timeout on mongo socket connections (nanoseconds) by default
	DEFAULT_MGO_SOCKET_TIMEOUT=60000000000
)

func init() {
	flag.StringVar(&opsFilename, "ops_filename", "",
		"The file for the serialized ops, generated by the record scripts.")
	flag.StringVar(&url, "url", "",
		"The database server's url, in the format of <host>[:<port>]")
	flag.StringVar(&style, "style", "",
		"How to replay the the ops. You can choose: \n"+
			"	stress: repaly ops at fast as possible\n"+
			"	real: repaly ops in accordance to ops' timestamps")
	flag.IntVar(&workers, "workers", 10,
		"Number of workers that sends ops to database.")
	flag.IntVar(&maxOps, "maxOps", 0,
		"[Optional] Maximal amount of ops to be replayed from the "+
			"ops_filename file. By setting it to `0`, replayer will "+
			"replay all the ops.")
	flag.IntVar(&numSkipOps, "numSkipOps", 0,
		"[Optional] Skip first N ops. Useful for when the total ops in ops_filename" +
		" exceeds available memory and you're running in stress mode.")
	flag.Int64Var(&socketTimeout, "socketTimeout", DEFAULT_MGO_SOCKET_TIMEOUT, "Mongo socket timeout in nanoseconds. Defaults to 60 seconds.")
	flag.Float64Var(&sampleRate, "sample_rate", 0.0, "sample ops for latency")	
	// TODO: define a real error logger
	flag.BoolVar(&verbose, "verbose", false, "[Optional] Print op errors and other verbose information to stdout.")
}

func parseFlags() error {
	flag.Parse()
	if style != "stress" && style != "real" {
		return errors.New("Cannot recognize the style: " + style)
	}
	if workers <= 0 {
		return errors.New("`workers` should be a positive number")
	}
	if maxOps == 0 {
		maxOps = 4294967295
	}
	return nil
}

func main() {
	// Will enable system threads to make sure all cpus can be well utilized.
	runtime.GOMAXPROCS(100)
	err := parseFlags()
	panicOnError(err)

	// Prepare to dispatch ops
	var reader OpsReader
	var opsChan chan *Op
	if style == "stress" {
		err, reader = NewFileByLineOpsReader(opsFilename)
		panicOnError(err)
		if numSkipOps > 0 {
			err = reader.SkipOps(numSkipOps)
			panicOnError(err)
		}
		opsChan = NewBestEffortOpsDispatcher(reader, maxOps)
	} else {
		// TODO NewCyclicOpsReader: do we really want to make it cyclic?
		reader = NewCyclicOpsReader(func() OpsReader {
			err, reader := NewFileByLineOpsReader(opsFilename)
			panicOnError(err)
			return reader
		})
		if numSkipOps > 0 {
			err = reader.SkipOps(numSkipOps)
			panicOnError(err)
		}
		opsChan = NewByTimeOpsDispatcher(reader, maxOps)
	}

	latencyChan := make(chan Latency, workers)

	// Set up workers to do the job
	exit := make(chan int)
	opsExecuted := int64(0)
	fetch := func(id int, statsCollector IStatsCollector) {
		log.Printf("Worker #%d report for duty\n", id)
		
		session, err := mgo.Dial(url)		
		panicOnError(err)
		session.SetSocketTimeout(time.Duration(socketTimeout))
		
		defer session.Close()
		exec := OpsExecutorWithStats(session, statsCollector)
		for {
			op := <-opsChan
			if op == nil {
				break
			}
			err := exec.Execute(op)
			
			if verbose == true && err != nil {
				log.Println(err.Error())
			}
			atomic.AddInt64(&opsExecuted, 1)
		}
		exit <- 1
		log.Printf("Worker #%d done!\n", id)
	}
	statsCollectorList := make([]*StatsCollector, workers)
	for i := 0; i < workers; i++ {
		statsCollectorList[i] = NewStatsCollector()
		statsCollectorList[i].SampleLatencies(sampleRate, latencyChan)
		go fetch(i, statsCollectorList[i])
	}

	// Periodically report execution status
	go func() {
		statsAnalyzer := NewStatsAnalyzer(statsCollectorList, &opsExecuted,
			latencyChan, int(sampleRate*float64(maxOps)))
		toFloat := func(nano int64) float64 {
			return float64(nano) / float64(1e6)
		}
		report := func() {
			status := statsAnalyzer.GetStatus()
{'fraction_non_alphanumeric': 0.11926265000618583, 'fraction_numerical': 0.01633057033279723, 'mean_word_length': 3.993205682520074, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2084989', 'n_tokens_mistral': 3067, 'n_tokens_neox': 2736, 'n_words': 800}