<commit_before>package main

import (
	"errors"
	"flag"
	"fmt"
	"github.com/Symantec/Dominator/dom/herd"
	"github.com/Symantec/Dominator/dom/mdbd"
	"github.com/Symantec/Dominator/lib/constants"
	"github.com/Symantec/Dominator/lib/logbuf"
	"github.com/Symantec/Dominator/lib/mdb"
	"github.com/Symantec/tricorder/go/tricorder"
	"log"
	"os"
	"os/user"
	"path"
	"runtime"
	"strconv"
	"syscall"
	"time"
)

var (
	caFile = flag.String("CAfile", "/etc/ssl/CA.pem",
		"Name of file containing the root of trust")
	certDir = flag.String("certDir", "/etc/ssl/Dominator",
		"Name of file (relative to certDir) containing the SSL certificate")
	certFile = flag.String("certFile", "cert.pem",
		"Name of file (relative to stateDir) containing the SSL certificate")
	debug = flag.Bool("debug", false,
		"If true, show debugging output")
	fdLimit = flag.Uint64("fdLimit", getFdLimit(),
		"Maximum number of open file descriptors (this limits concurrent connection attempts)")
	imageServerHostname = flag.String("imageServerHostname", "localhost",
		"Hostname of image server")
	imageServerPortNum = flag.Uint("imageServerPortNum",
		constants.ImageServerPortNumber,
		"Port number of image server")
	keyFile = flag.String("keyFile", "key.pem",
		"Name of file (relative to certDir) containing the SSL certificate")
	logbufLines = flag.Uint("logbufLines", 1024,
		"Number of lines to store in the log buffer")
	mdbFile = flag.String("mdbFile", "mdb",
		"File to read MDB data from, relative to stateDir (default format is JSON)")
	minInterval = flag.Uint("minInterval", 1,
		"Minimum interval between loops (in seconds)")
	portNum = flag.Uint("portNum", constants.DomPortNumber,
		"Port number to allocate and listen on for HTTP/RPC")
	stateDir = flag.String("stateDir", "/var/lib/Dominator",
		"Name of dominator state directory.")
	username = flag.String("username", "",
		"If running as root, username to switch to.")
)

func showMdb(mdb *mdb.Mdb) {
	fmt.Println()
	mdb.DebugWrite(os.Stdout)
	fmt.Println()
}

func getFdLimit() uint64 {
	var rlim syscall.Rlimit
	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rlim); err != nil {
		panic(err)
	}
	return rlim.Max
}

func setUser(username string) error {
	if username == "" {
		return errors.New("-username argument missing")
	}
	newUser, err := user.Lookup(username)
	if err != nil {
		return err
	}
	uid, err := strconv.Atoi(newUser.Uid)
	if err != nil {
		return err
	}
	gid, err := strconv.Atoi(newUser.Gid)
	if err != nil {
		return err
	}
	if uid == 0 {
		return errors.New("Do not run the Dominator as root")
		os.Exit(1)
	}
	if err := syscall.Setresgid(gid, gid, gid); err != nil {
		return err
	}
	return syscall.Setresuid(uid, uid, uid)
}

func pathJoin(first, second string) string {
	if path.IsAbs(second) {
		return path.Clean(second)
	}
	return path.Join(first, second)
}

func main() {
	flag.Parse()
	tricorder.RegisterFlags()
	setupTls(*caFile,
		pathJoin(*certDir, *certFile), pathJoin(*certDir, *keyFile))
	rlim := syscall.Rlimit{*fdLimit, *fdLimit}
	if err := syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rlim); err != nil {
		fmt.Fprintf(os.Stderr, "Cannot set FD limit\t%s\n", err)
		os.Exit(1)
	}
	if os.Geteuid() == 0 {
		if err := setUser(*username); err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(1)
		}
	}
	fi, err := os.Lstat(*stateDir)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Cannot stat: %s\t%s\n", *stateDir, err)
		os.Exit(1)
	}
	if !fi.IsDir() {
		fmt.Fprintf(os.Stderr, "%s is not a directory\n", *stateDir)
		os.Exit(1)
	}
	interval := time.Duration(*minInterval) * time.Second
	circularBuffer := logbuf.New(*logbufLines)
	logger := log.New(circularBuffer, "", log.LstdFlags)
	mdbChannel := mdbd.StartMdbDaemon(path.Join(*stateDir, *mdbFile), logger)
	herd := herd.NewHerd(fmt.Sprintf("%s:%d", *imageServerHostname,
		*imageServerPortNum), logger)
	herd.AddHtmlWriter(circularBuffer)
	if err = herd.StartServer(*portNum, true); err != nil {
		fmt.Fprintf(os.Stderr, "Unable to create http server\t%s\n", err)
		os.Exit(1)
	}
	nextCycleStopTime := time.Now().Add(interval)
	haveMdb := false
	for {
		select {
		case mdb := <-mdbChannel:
			herd.MdbUpdate(mdb)
			haveMdb = true
			if *debug {
				showMdb(mdb)
			}
			runtime.GC() // An opportune time to take out the garbage.
		default:
			// Do work.
			if herd.PollNextSub() {
				if *debug {
					fmt.Print(".")
				}
				var sleepTime time.Duration
				if haveMdb {
					sleepTime = nextCycleStopTime.Sub(time.Now())
				} else {
					sleepTime = time.Millisecond * 100
				}
				time.Sleep(sleepTime)
				nextCycleStopTime = time.Now().Add(interval)
				if sleepTime < 0 { // There was no time to rest.
					runtime.GC() // An opportune time to take out the garbage.
				}
			}
		}
	}
}<commit_msg>Eliminate dominator delay in processing first MDB.<commit_after>package main

import (
	"errors"
	"flag"
	"fmt"
	"github.com/Symantec/Dominator/dom/herd"
	"github.com/Symantec/Dominator/dom/mdbd"
	"github.com/Symantec/Dominator/lib/constants"
	"github.com/Symantec/Dominator/lib/logbuf"
	"github.com/Symantec/Dominator/lib/mdb"
	"github.com/Symantec/tricorder/go/tricorder"
	"log"
	"os"
	"os/user"
	"path"
	"runtime"
	"strconv"
	"syscall"
	"time"
)

var (
	caFile = flag.String("CAfile", "/etc/ssl/CA.pem",
		"Name of file containing the root of trust")
	certDir = flag.String("certDir", "/etc/ssl/Dominator",
		"Name of file (relative to certDir) containing the SSL certificate")
	certFile = flag.String("certFile", "cert.pem",
		"Name of file (relative to stateDir) containing the SSL certificate")
	debug = flag.Bool("debug", false,
		"If true, show debugging output")
	fdLimit = flag.Uint64("fdLimit", getFdLimit(),
		"Maximum number of open file descriptors (this limits concurrent connection attempts)")
	imageServerHostname = flag.String("imageServerHostname", "localhost",
		"Hostname of image server")
	imageServerPortNum = flag.Uint("imageServerPortNum",
		constants.ImageServerPortNumber,
		"Port number of image server")
	keyFile = flag.String("keyFile", "key.pem",
		"Name of file (relative to certDir) containing the SSL certificate")
	logbufLines = flag.Uint("logbufLines", 1024,
		"Number of lines to store in the log buffer")
	mdbFile = flag.String("mdbFile", "mdb",
		"File to read MDB data from, relative to stateDir (default format is JSON)")
	minInterval = flag.Uint("minInterval", 1,
		"Minimum interval between loops (in seconds)")
	portNum = flag.Uint("portNum", constants.DomPortNumber,
		"Port number to allocate and listen on for HTTP/RPC")
	stateDir = flag.String("stateDir", "/var/lib/Dominator",
		"Name of dominator state directory.")
	username = flag.String("username", "",
		"If running as root, username to switch to.")
)

func showMdb(mdb *mdb.Mdb) {
	fmt.Println()
	mdb.DebugWrite(os.Stdout)
	fmt.Println()
}

func getFdLimit() uint64 {
	var rlim syscall.Rlimit
	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rlim); err != nil {
		panic(err)
	}
	return rlim.Max
}

func setUser(username string) error {
	if username == "" {
		return errors.New("-username argument missing")
	}
	newUser, err := user.Lookup(username)
	if err != nil {
		return err
	}
	uid, err := strconv.Atoi(newUser.Uid)
	if err != nil {
		return err
	}
	gid, err := strconv.Atoi(newUser.Gid)
	if err != nil {
		return err
	}
	if uid == 0 {
		return errors.New("Do not run the Dominator as root")
		os.Exit(1)
	}
	if err := syscall.Setresgid(gid, gid, gid); err != nil {
		return err
	}
	return syscall.Setresuid(uid, uid, uid)
}

func pathJoin(first, second string) string {
	if path.IsAbs(second) {
		return path.Clean(second)
	}
	return path.Join(first, second)
}

func main() {
	flag.Parse()
	tricorder.RegisterFlags()
	setupTls(*caFile,
		pathJoin(*certDir, *certFile), pathJoin(*certDir, *keyFile))
	rlim := syscall.Rlimit{*fdLimit, *fdLimit}
	if err := syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rlim); err != nil {
		fmt.Fprintf(os.Stderr, "Cannot set FD limit\t%s\n", err)
		os.Exit(1)
	}
	if os.Geteuid() == 0 {
		if err := setUser(*username); err != nil {
			fmt.Fprintln(os.Stderr, err)
			os.Exit(1)
		}
	}
	fi, err := os.Lstat(*stateDir)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Cannot stat: %s\t%s\n", *stateDir, err)
		os.Exit(1)
	}
	if !fi.IsDir() {
		fmt.Fprintf(os.Stderr, "%s is not a directory\n", *stateDir)
		os.Exit(1)
	}
	interval := time.Duration(*minInterval) * time.Second
	circularBuffer := logbuf.New(*logbufLines)
	logger := log.New(circularBuffer, "", log.LstdFlags)
	mdbChannel := mdbd.StartMdbDaemon(path.Join(*stateDir, *mdbFile), logger)
	herd := herd.NewHerd(fmt.Sprintf("%s:%d", *imageServerHostname,
		*imageServerPortNum), logger)
	herd.AddHtmlWriter(circularBuffer)
	if err = herd.StartServer(*portNum, true); err != nil {
		fmt.Fprintf(os.Stderr, "Unable to create http server\t%s\n", err)
		os.Exit(1)
	}
	scanTokenChannel := make(chan bool, 1)
	scanTokenChannel <- true
	nextCycleStopTime := time.Now().Add(interval)
	for {
		select {
		case mdb := <-mdbChannel:
			herd.MdbUpdate(mdb)
			if *debug {
				showMdb(mdb)
			}
			runtime.GC() // An opportune time to take out the garbage.
		case <-scanTokenChannel:
			// Scan one sub.
			if herd.PollNextSub() { // We've reached the end of a scan cycle.
				if *debug {
					fmt.Print(".")
				}
				go func(sleepDuration time.Duration) {
					if sleepDuration < 0 { // There was no time to rest.
						runtime.GC()
					} else {
						time.Sleep(sleepDuration)
					}
					scanTokenChannel <- true
				}(nextCycleStopTime.Sub(time.Now()))
				nextCycleStopTime = time.Now().Add(interval)
			} else {
				scanTokenChannel <- true
			}
		}
	}
}
{'fraction_non_alphanumeric': 0.14716626656925164, 'fraction_numerical': 0.004174929548063876, 'mean_word_length': 3.9802494802494803, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '6920275', 'n_tokens_mistral': 3819, 'n_tokens_neox': 3526, 'n_words': 857}