<commit_before>package flow

import (
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
)

type Flow struct {
	errChan  chan error
	stopChan chan struct{}
	ref      *int32
	wg       sync.WaitGroup
	Parent   *Flow
	Children []*Flow
	stoped   int32

	mutex sync.Mutex
}

func New(n int) *Flow {
	f := &Flow{
		errChan:  make(chan error, 1),
		stopChan: make(chan struct{}),
		ref:      new(int32),
	}
	f.Add(n)
	return f
}

func (f *Flow) Error(err error) {
	f.errChan <- err
}

func (f *Flow) Fork(n int) *Flow {
	f2 := New(n)
	f2.Parent = f
	f.Children = append(f.Children, f2)
	f.Add(1) // for f2
	return f2
}

func (f *Flow) StopAll() {
	flow := f
	for flow.Parent != nil {
		flow = flow.Parent
	}
	flow.Stop()
}

func (f *Flow) Close() {
	f.Stop()
	f.wait()
}

func (f *Flow) Stop() {
	if !atomic.CompareAndSwapInt32(&f.stoped, 0, 1) {
		return
	}

	close(f.stopChan)
	for _, cf := range f.Children {
		cf.Stop()
	}
}

func (f *Flow) IsClosed() bool {
	select {
	case <-f.stopChan:
		return true
	default:
		return false
	}
}

func (f *Flow) IsClose() chan struct{} {
	return f.stopChan
}

func (f *Flow) Add(n int) {
	atomic.AddInt32(f.ref, int32(n))
	f.wg.Add(n)
}

func (f *Flow) Done() {
	f.wg.Done()
	if atomic.AddInt32(f.ref, -1) == 0 {
		f.Stop()
	}
}

func (f *Flow) wait() {<commit_msg>add CloseOrWait<commit_after>package flow

import (
	"os"
	"os/signal"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
)

type Flow struct {
	errChan  chan error
	stopChan chan struct{}
	ref      *int32
	wg       sync.WaitGroup
	Parent   *Flow
	Children []*Flow
	stoped   int32

	mutex sync.Mutex
}

func New(n int) *Flow {
	f := &Flow{
		errChan:  make(chan error, 1),
		stopChan: make(chan struct{}),
		ref:      new(int32),
	}
	f.Add(n)
	return f
}

const (
	F_CLOSED  = true
	F_TIMEOUT = false
)

func (f *Flow) CloseOrWait(duration time.Duration) bool {
	select {
	case <-time.After(duration):
		return F_TIMEOUT
	case <-f.IsClose():
		return F_CLOSED
	}
}

func (f *Flow) Error(err error) {
	f.errChan <- err
}

func (f *Flow) Fork(n int) *Flow {
	f2 := New(n)
	f2.Parent = f
	f.Children = append(f.Children, f2)
	f.Add(1) // for f2
	return f2
}

func (f *Flow) StopAll() {
	flow := f
	for flow.Parent != nil {
		flow = flow.Parent
	}
	flow.Stop()
}

func (f *Flow) Close() {
	f.Stop()
	f.wait()
}

func (f *Flow) Stop() {
	if !atomic.CompareAndSwapInt32(&f.stoped, 0, 1) {
		return
	}

	close(f.stopChan)
	for _, cf := range f.Children {
		cf.Stop()
	}
}

func (f *Flow) IsClosed() bool {
	return atomic.LoadInt32(&f.stoped) == 1
}

func (f *Flow) IsClose() chan struct{} {
	return f.stopChan
}

func (f *Flow) Add(n int) {
	atomic.AddInt32(f.ref, int32(n))
	f.wg.Add(n)
}

func (f *Flow) Done() {
	f.wg.Done()
	if atomic.AddInt32(f.ref, -1) == 0 {
		f.Stop()
	}
}

func (f *Flow) wait() {
{'fraction_non_alphanumeric': 0.17054819061268361, 'fraction_numerical': 0.01898960945897528, 'mean_word_length': 3.005738880918221, 'pattern_counts': {'":': 0, '<': 8, '<?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': '1926344', 'n_tokens_mistral': 1321, 'n_tokens_neox': 1240, 'n_words': 313}