<commit_before>package sync

import (
	"errors"
	"io"
	"sync"
)

// ErrStopped is returned by ThreadGroup methods if Stop has already been
// called.
var ErrStopped = errors.New("ThreadGroup already stopped")

// ThreadGroup is a sync.WaitGroup with additional functionality for
// facilitating clean shutdown. Namely, it provides a StopChan method for
// notifying callers when shutdown occurrs. Another key difference is that a
// ThreadGroup is only intended be used once; as such, its Add and Stop
// methods return errors if Stop has already been called.
//
// During shutdown, it is common to close resources such as net.Conns and
// net.Listeners. Typically, this would require spawning a goroutine to wait
// on the ThreadGroup's StopChan and then close the resource. To make this
// more convenient, ThreadGroup provides a RegisterCloser method. io.Closers
// registered in this manner will be automatically closed when Stop is called.
type ThreadGroup struct {
	stopChan chan struct{}
	chanOnce sync.Once
	mu       sync.Mutex
	wg       sync.WaitGroup
	closers  []io.Closer
}

// StopChan provides read-only access to the ThreadGroup's stopChan. Callers
// should select on StopChan in order to interrupt long-running reads (such as
// time.After).
func (tg *ThreadGroup) StopChan() <-chan struct{} {
	// Initialize tg.stopChan if it is nil; this makes an unitialized
	// ThreadGroup valid. (Otherwise, a NewThreadGroup function would be
	// necessary.)
	tg.chanOnce.Do(func() { tg.stopChan = make(chan struct{}) })
	return tg.stopChan
}

// IsStopped returns true if Stop has been called.
func (tg *ThreadGroup) IsStopped() bool {
	select {
	case <-tg.StopChan():
		return true
	default:
		return false
	}
}

// Add increments the ThreadGroup counter.
func (tg *ThreadGroup) Add() error {
	tg.mu.Lock()
	defer tg.mu.Unlock()
	if tg.IsStopped() {
		return ErrStopped
	}
	tg.wg.Add(1)
	return nil
}

// Done decrements the ThreadGroup counter.
func (tg *ThreadGroup) Done() {
	tg.wg.Done()
}

// Stop closes the ThreadGroup's stopChan, closes members of the closer set,
// and blocks until the counter is zero.
func (tg *ThreadGroup) Stop() error {
	tg.mu.Lock()
	if tg.IsStopped() {
		tg.mu.Unlock()
		return ErrStopped
	}
	close(tg.stopChan)
	for _, c := range tg.closers {
		c.Close()
	}
	tg.mu.Unlock()
	tg.wg.Wait()
	return nil
}

// RegisterCloser adds an io.Closer to the ThreadGroups closer set. Members of
// the set will be closed when Stop is called. Note that this means the errors
// returned by Close are not recoverable.
func (tg *ThreadGroup) RegisterCloser(c io.Closer) error {
	tg.mu.Lock()
	defer tg.mu.Unlock()
	if tg.IsStopped() {
		return ErrStopped
	}
	tg.closers = append(tg.closers, c)
	return nil
}
<commit_msg>clean up references to closers<commit_after>package sync

import (
	"errors"
	"io"
	"sync"
)

// ErrStopped is returned by ThreadGroup methods if Stop has already been
// called.
var ErrStopped = errors.New("ThreadGroup already stopped")

// ThreadGroup is a sync.WaitGroup with additional functionality for
// facilitating clean shutdown. Namely, it provides a StopChan method for
// notifying callers when shutdown occurrs. Another key difference is that a
// ThreadGroup is only intended be used once; as such, its Add and Stop
// methods return errors if Stop has already been called.
//
// During shutdown, it is common to close resources such as net.Conns and
// net.Listeners. Typically, this would require spawning a goroutine to wait
// on the ThreadGroup's StopChan and then close the resource. To make this
// more convenient, ThreadGroup provides a RegisterCloser method. io.Closers
// registered in this manner will be automatically closed when Stop is called.
type ThreadGroup struct {
	stopChan chan struct{}
	chanOnce sync.Once
	mu       sync.Mutex
	wg       sync.WaitGroup
	closers  []io.Closer
}

// StopChan provides read-only access to the ThreadGroup's stopChan. Callers
// should select on StopChan in order to interrupt long-running reads (such as
// time.After).
func (tg *ThreadGroup) StopChan() <-chan struct{} {
	// Initialize tg.stopChan if it is nil; this makes an unitialized
	// ThreadGroup valid. (Otherwise, a NewThreadGroup function would be
	// necessary.)
	tg.chanOnce.Do(func() { tg.stopChan = make(chan struct{}) })
	return tg.stopChan
}

// IsStopped returns true if Stop has been called.
func (tg *ThreadGroup) IsStopped() bool {
	select {
	case <-tg.StopChan():
		return true
	default:
		return false
	}
}

// Add increments the ThreadGroup counter.
func (tg *ThreadGroup) Add() error {
	tg.mu.Lock()
	defer tg.mu.Unlock()
	if tg.IsStopped() {
		return ErrStopped
	}
	tg.wg.Add(1)
	return nil
}

// Done decrements the ThreadGroup counter.
func (tg *ThreadGroup) Done() {
	tg.wg.Done()
}

// Stop closes the ThreadGroup's stopChan, closes members of the closer set,
// and blocks until the counter is zero.
func (tg *ThreadGroup) Stop() error {
	tg.mu.Lock()
	if tg.IsStopped() {
		tg.mu.Unlock()
		return ErrStopped
	}
	close(tg.stopChan)
	for _, c := range tg.closers {
		c.Close()
	}
	tg.closers = nil
	tg.mu.Unlock()
	tg.wg.Wait()
	return nil
}

// RegisterCloser adds an io.Closer to the ThreadGroups closer set. Members of
// the set will be closed when Stop is called. Note that this means the errors
// returned by Close are not recoverable.
func (tg *ThreadGroup) RegisterCloser(c io.Closer) error {
	tg.mu.Lock()
	defer tg.mu.Unlock()
	if tg.IsStopped() {
		return ErrStopped
	}
	tg.closers = append(tg.closers, c)
	return nil
}

{'fraction_non_alphanumeric': 0.09809954751131222, 'fraction_numerical': 0.00036199095022624434, 'mean_word_length': 3.620401337792642, 'pattern_counts': {'":': 0, '<': 7, '<?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': '2315601', 'n_tokens_mistral': 1784, 'n_tokens_neox': 1705, 'n_words': 676}