<commit_before>package main

import (
	"flag"
	"io/ioutil"
	"log"
	"os"
	"runtime/pprof"
	"strconv"
	"sync"
	"time"

	"github.com/nabeken/delayd"
	"github.com/nabeken/delayd/testutil"
)

var (
	flagNumMsg    = flag.Int("n", 10, "Specify a number of messages to send")
	flagNumServer = flag.Int("s", 3, "Specify a number of servers")
	flagDuration  = flag.Duration("d", 1*time.Second, "Specify a delay. Default is 1s.")
	flagConfig    = flag.String("c", "/etc/delayd.toml", "Specify a config. Default is /etc/delayd.toml")
	flagProfile   = flag.String("p", "", "Specify a output file for cpu profiling.")
)

// mps returns message per seconds.
func mps(n int, d time.Duration) float64 {
	return float64(n) / d.Seconds()
}

func generateMessages(n int, d time.Duration) []testutil.Message {
	msgs := make([]testutil.Message, n)
	for i := 0; i < n; i++ {
		msgs[i] = testutil.Message{
			Value: strconv.FormatInt(int64(i+1), 10),
			Delay: int64(d / time.Millisecond),
		}
	}
	return msgs
}

func main() {
	flag.Parse()

	config, err := delayd.LoadConfig(*flagConfig)
	if err != nil {
		log.Fatal(err)
	}

	// Use stdout instead of file
	config.LogDir = ""

	if *flagProfile != "" {
		f, err := os.Create(*flagProfile)
		if err != nil {
			delayd.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	client, err := testutil.AMQPClientFunc(config, ioutil.Discard)
	if err != nil {
		log.Fatal(err)
	}

	os.Setenv("DELAYD_CONSUL_AGENT_SERVICE_ADDRESS_TWEAK", "127.0.0.1")
	defer os.Setenv("DELAYD_CONSUL_AGENT_SERVICE_ADDRESS_TWEAK", "")
	raftServers, err := testutil.RaftServers(*flagNumServer, testutil.AMQPServerFunc)(config)
	if err != nil {
		log.Fatal(err)
	}

	for _, s := range raftServers {
		go s.Run()
		defer s.Stop()
	}

	// Send messages to delayd exchange
	msgs := generateMessages(*flagNumMsg, *flagDuration)
	n := len(msgs)

	start := time.Now()
	if err := client.SendMessages(msgs); err != nil {
		log.Println(err)
		return
	}
	sent := time.Now()

	durationSent := sent.Sub(start)
	delayd.Infof("sent %d messages for %s, %f msg/s", n, durationSent, mps(n, durationSent))

	var wg sync.WaitGroup
	wg.Add(len(msgs))
	go func() {
		for _ = range client.RecvLoop() {
			wg.Done()
		}
	}()

	// Wait for messages to be processed
	wg.Wait()
	done := time.Now()

	durationProcessed := done.Sub(sent)
	durationAll := done.Sub(start)

	delayd.Infof("processed %d messages for %s, %f msg/s", n, durationProcessed, mps(n, durationProcessed))
	delayd.Infof("total: processed %d messages for %s, %f msg/s", n, durationAll, mps(n, durationAll))

	// shutdown consumer
	client.Close()
}<commit_msg>delayd-bench: Do not panic on dups<commit_after>package main

import (
	"flag"
	"io/ioutil"
	"log"
	"os"
	"runtime/pprof"
	"strconv"
	"sync"
	"time"

	"github.com/nabeken/delayd"
	"github.com/nabeken/delayd/testutil"
)

var (
	flagNumMsg    = flag.Int("n", 10, "Specify a number of messages to send")
	flagNumServer = flag.Int("s", 3, "Specify a number of servers")
	flagDuration  = flag.Duration("d", 1*time.Second, "Specify a delay. Default is 1s.")
	flagConfig    = flag.String("c", "/etc/delayd.toml", "Specify a config. Default is /etc/delayd.toml")
	flagProfile   = flag.String("p", "", "Specify a output file for cpu profiling.")
)

// mps returns message per seconds.
func mps(n int, d time.Duration) float64 {
	return float64(n) / d.Seconds()
}

func generateMessages(n int, d time.Duration) []testutil.Message {
	msgs := make([]testutil.Message, n)
	for i := 0; i < n; i++ {
		msgs[i] = testutil.Message{
			Value: strconv.FormatInt(int64(i+1), 10),
			Delay: int64(d / time.Millisecond),
		}
	}
	return msgs
}

func main() {
	flag.Parse()

	config, err := delayd.LoadConfig(*flagConfig)
	if err != nil {
		log.Fatal(err)
	}

	// Use stdout instead of file
	config.LogDir = ""

	if *flagProfile != "" {
		f, err := os.Create(*flagProfile)
		if err != nil {
			delayd.Fatal(err)
		}
		pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	client, err := testutil.AMQPClientFunc(config, ioutil.Discard)
	if err != nil {
		log.Fatal(err)
	}

	os.Setenv("DELAYD_CONSUL_AGENT_SERVICE_ADDRESS_TWEAK", "127.0.0.1")
	defer os.Setenv("DELAYD_CONSUL_AGENT_SERVICE_ADDRESS_TWEAK", "")
	raftServers, err := testutil.RaftServers(*flagNumServer, testutil.AMQPServerFunc)(config)
	if err != nil {
		log.Fatal(err)
	}

	for _, s := range raftServers {
		go s.Run()
		defer s.Stop()
	}

	// Send messages to delayd exchange
	msgs := generateMessages(*flagNumMsg, *flagDuration)
	n := len(msgs)

	start := time.Now()
	if err := client.SendMessages(msgs); err != nil {
		log.Println(err)
		return
	}
	sent := time.Now()

	durationSent := sent.Sub(start)
	delayd.Infof("sent %d messages for %s, %f msg/s", n, durationSent, mps(n, durationSent))

	var wg sync.WaitGroup
	wg.Add(n)
	acked := 0
	go func() {
		for _ = range client.RecvLoop() {
			acked++
			if acked > 10000 {
				continue
			}
			wg.Done()
		}
	}()

	// Wait for messages to be processed
	wg.Wait()
	done := time.Now()

	durationProcessed := done.Sub(sent)
	durationAll := done.Sub(start)

	delayd.Infof("processed %d messages for %s, %f msg/s", n, durationProcessed, mps(n, durationProcessed))
	delayd.Infof("total: processed %d messages for %s, %f msg/s", n, durationAll, mps(n, durationAll))

	// shutdown consumer
	client.Close()

	if acked > n {
		delayd.Infof("%d messages are duplicated", acked - n)
	}
}
{'fraction_non_alphanumeric': 0.15795328142380421, 'fraction_numerical': 0.0096403411197627, 'mean_word_length': 3.842908438061041, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '3591643', 'n_tokens_mistral': 2225, 'n_tokens_neox': 2087, 'n_words': 495}