<commit_before>package main

import (
	"errors"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"

	"github.com/cyberdelia/heroku-go/v3"
	flag "github.com/ogier/pflag"
)

var ErrNoReleases = errors.New("No releases found")

type ApiPoller struct {
	Cl  *heroku.Service
	App string
	Dd  DynoDriver

	lastReleaseID string
}

type Processes struct {
	r     *Release
	forms []Formation

	dd        DynoDriver
	executors []*Executor
}

type Formation interface {
	Args() []string
	Quantity() int
	Type() string
}

type ApiFormation struct {
	h *heroku.Formation
}

func (f *ApiFormation) Args() []string {
	return []string{f.h.Command}
}
func (f *ApiFormation) Quantity() int {
	return f.h.Quantity
}

func (f *ApiFormation) Type() string {
	return f.h.Type
}

func (ap *ApiPoller) fetchLatest() (*heroku.Release, error) {
	releases, err := ap.Cl.ReleaseList(
		ap.App, &heroku.ListRange{Descending: true, Field: "version",
			Max: 1})
	if err != nil {
		return nil, err
	}

	if len(releases) < 1 {
		return nil, ErrNoReleases
	}

	return releases[0], nil
}

func (ap *ApiPoller) fillProcesses(rel *heroku.Release) (*Processes, error) {
	config, err := ap.Cl.ConfigVarInfo(ap.App)
	if err != nil {
		return nil, err
	}

	slug, err := ap.Cl.SlugInfo(ap.App, rel.Slug.ID)
	if err != nil {
		return nil, err
	}

	hForms, err := ap.Cl.FormationList(ap.App, &heroku.ListRange{})
	if err != nil {
		return nil, err
	}

	procs := Processes{
		r: &Release{
			appName: ap.App,
			config:  config,
			slugURL: slug.Blob.URL,
			version: rel.Version,
		},
		forms: make([]Formation, len(hForms), len(hForms)),
		dd:    ap.Dd,
	}

	for i, hForm := range hForms {
		procs.forms[i] = &ApiFormation{h: hForm}
	}

	return &procs, nil
}

func (ap *ApiPoller) pollOnce() (*Processes, error) {
	release, err := ap.fetchLatest()
	if err != nil {
		return nil, err
	}

	if release != nil && ap.lastReleaseID != release.ID {
		ap.lastReleaseID = release.ID

		log.Printf("New release %s detected", ap.lastReleaseID)
		return ap.fillProcesses(release)
	}

	return nil, nil
}

func (ap *ApiPoller) pollSynchronous(out chan<- *Processes) {
	for {
		procs, err := ap.pollOnce()
		if err != nil {
			log.Println("Could not fetch new release information:",
				err)
			goto wait
		}

		if procs != nil {
			out <- procs
		}

	wait:
		time.Sleep(10 * time.Second)
	}
}

// Listens for new releases by periodically polling the Heroku
// API. When a new release is detected it is sent to the returned
// channel.
func (ap *ApiPoller) poll() <-chan *Processes {
	out := make(chan *Processes)
	go ap.pollSynchronous(out)
	return out
}

func (p *Processes) start(command string, args []string, concurrency int) (
	err error) {
	err = p.dd.Build(p.r)
	if err != nil {
		log.Printf("hsup could not bake image for release %s: %s",
			p.r.Name(), err.Error())
		return err
	}

	if command == "start" {
		for _, form := range p.forms {
			log.Printf("formation quantity=%v type=%v\n",
				form.Quantity(), form.Type())

			for i := 0; i < getConcurrency(concurrency,
				form.Quantity()); i++ {
				executor := &Executor{
					args:        form.Args(),
					dynoDriver:  p.dd,
					processID:   strconv.Itoa(i + 1),
					processType: form.Type(),
					release:     p.r,
					complete:    make(chan struct{}),
					state:       Stopped,
					newInput:    make(chan DynoInput),
				}

				p.executors = append(p.executors, executor)
			}
		}
	} else if command == "run" {
		for i := 0; i < getConcurrency(concurrency, 1); i++ {
			executor := &Executor{
				args:        args,
				dynoDriver:  p.dd,
				processID:   strconv.Itoa(i + 1),
				processType: "run",
				release:     p.r,<commit_msg>Properly report passed concurrency, respecting default<commit_after>package main

import (
	"errors"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"

	"github.com/cyberdelia/heroku-go/v3"
	flag "github.com/ogier/pflag"
)

var ErrNoReleases = errors.New("No releases found")

type ApiPoller struct {
	Cl  *heroku.Service
	App string
	Dd  DynoDriver

	lastReleaseID string
}

type Processes struct {
	r     *Release
	forms []Formation

	dd        DynoDriver
	executors []*Executor
}

type Formation interface {
	Args() []string
	Quantity() int
	Type() string
}

type ApiFormation struct {
	h *heroku.Formation
}

func (f *ApiFormation) Args() []string {
	return []string{f.h.Command}
}
func (f *ApiFormation) Quantity() int {
	return f.h.Quantity
}

func (f *ApiFormation) Type() string {
	return f.h.Type
}

func (ap *ApiPoller) fetchLatest() (*heroku.Release, error) {
	releases, err := ap.Cl.ReleaseList(
		ap.App, &heroku.ListRange{Descending: true, Field: "version",
			Max: 1})
	if err != nil {
		return nil, err
	}

	if len(releases) < 1 {
		return nil, ErrNoReleases
	}

	return releases[0], nil
}

func (ap *ApiPoller) fillProcesses(rel *heroku.Release) (*Processes, error) {
	config, err := ap.Cl.ConfigVarInfo(ap.App)
	if err != nil {
		return nil, err
	}

	slug, err := ap.Cl.SlugInfo(ap.App, rel.Slug.ID)
	if err != nil {
		return nil, err
	}

	hForms, err := ap.Cl.FormationList(ap.App, &heroku.ListRange{})
	if err != nil {
		return nil, err
	}

	procs := Processes{
		r: &Release{
			appName: ap.App,
			config:  config,
			slugURL: slug.Blob.URL,
			version: rel.Version,
		},
		forms: make([]Formation, len(hForms), len(hForms)),
		dd:    ap.Dd,
	}

	for i, hForm := range hForms {
		procs.forms[i] = &ApiFormation{h: hForm}
	}

	return &procs, nil
}

func (ap *ApiPoller) pollOnce() (*Processes, error) {
	release, err := ap.fetchLatest()
	if err != nil {
		return nil, err
	}

	if release != nil && ap.lastReleaseID != release.ID {
		ap.lastReleaseID = release.ID

		log.Printf("New release %s detected", ap.lastReleaseID)
		return ap.fillProcesses(release)
	}

	return nil, nil
}

func (ap *ApiPoller) pollSynchronous(out chan<- *Processes) {
	for {
		procs, err := ap.pollOnce()
		if err != nil {
			log.Println("Could not fetch new release information:",
				err)
			goto wait
		}

		if procs != nil {
			out <- procs
		}

	wait:
		time.Sleep(10 * time.Second)
	}
}

// Listens for new releases by periodically polling the Heroku
// API. When a new release is detected it is sent to the returned
// channel.
func (ap *ApiPoller) poll() <-chan *Processes {
	out := make(chan *Processes)
	go ap.pollSynchronous(out)
	return out
}

func (p *Processes) start(command string, args []string, concurrency int) (
	err error) {
	err = p.dd.Build(p.r)
	if err != nil {
		log.Printf("hsup could not bake image for release %s: %s",
			p.r.Name(), err.Error())
		return err
	}

	if command == "start" {
		for _, form := range p.forms {
			conc := getConcurrency(concurrency, form.Quantity())
			log.Printf("formation quantity=%v type=%v\n",
				conc, form.Type())

			for i := 0; i < conc; i++ {
				executor := &Executor{
					args:        form.Args(),
					dynoDriver:  p.dd,
					processID:   strconv.Itoa(i + 1),
					processType: form.Type(),
					release:     p.r,
					complete:    make(chan struct{}),
					state:       Stopped,
					newInput:    make(chan DynoInput),
				}

				p.executors = append(p.executors, executor)
			}
		}
	} else if command == "run" {
		for i := 0; i < getConcurrency(concurrency, 1); i++ {
			executor := &Executor{
				args:        args,
				dynoDriver:  p.dd,
				processID:   strconv.Itoa(i + 1),
				processType: "run",
				release:     p.r,
{'fraction_non_alphanumeric': 0.1457654723127036, 'fraction_numerical': 0.002985884907709012, 'mean_word_length': 3.137563166760247, 'pattern_counts': {'":': 0, '<': 15, '<?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': '7667261', 'n_tokens_mistral': 3089, 'n_tokens_neox': 2765, 'n_words': 738}