<commit_before>package v2

import (
	"fmt"
	"sync/atomic"
)

type scFull struct{}
type pcFull struct{}

type demuxer struct {
	input     chan Event
	scheduler *Routine
	processor *Routine
	fin       chan error
	stopped   chan struct{}
	running   *uint32
}

// TODO
// demuxer_test
// Termination process
// Logger
// Metrics
// Adhere to interface
func newDemuxer(scheduler *Routine, processor *Routine) *demuxer {
	return &demuxer{
		input:     make(chan Event, 10),
		scheduler: scheduler,
		processor: processor,
		stopped:   make(chan struct{}, 1),
		fin:       make(chan error, 1),
		running:   new(uint32),
	}
}

func (dm *demuxer) start() {
	starting := atomic.CompareAndSwapUint32(dm.running, uint32(0), uint32(1))
	if !starting {
		panic("Routine has already started")
	}
	fmt.Printf("demuxer: run\n")
	for {
		if !dm.isRunning() {
			break
		}
		select {
		case event, ok := <-dm.input:
			if !ok {
				fmt.Printf("demuxer: stopping\n")
				dm.terminate(fmt.Errorf("stopped"))
				dm.stopped <- struct{}{}
				return
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		case event, ok := <-dm.scheduler.next():
			if !ok {
				fmt.Printf("demuxer: scheduler output closed\n")
				continue
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		case event, ok := <-dm.processor.next():
			if !ok {
				fmt.Printf("demuxer: processor output closed\n")
				continue
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		}
	}
}

func (dm *demuxer) handle(event Event) (Events, error) {
	received := dm.scheduler.trySend(event)
	if !received {
		return Events{scFull{}}, nil // backpressure
	}

	received = dm.processor.trySend(event)
	if !received {
		return Events{pcFull{}}, nil // backpressure
	}

	return Events{}, nil
}

func (dm *demuxer) trySend(event Event) bool {
	if !dm.isRunning() {
		fmt.Println("dummuxer isn't running")
		return false
	}
	select {
	case dm.input <- event:
		return true
	default:
		fmt.Printf("demuxer channel was full\n")
		return false
	}
}

func (dm *demuxer) isRunning() bool {
	return atomic.LoadUint32(dm.running) == 1
}

func (dm *demuxer) stop() {
	if !dm.isRunning() {
		return
	}
	fmt.Printf("demuxer stop\n")
	close(dm.input)
	<-dm.stopped
}

func (dm *demuxer) terminate(reason error) {
	stopped := atomic.CompareAndSwapUint32(dm.running, uint32(1), uint32(0))
	if !stopped {
		panic("called terminate but already terminated")
	}
	dm.fin <- reason
}

func (dm *demuxer) final() chan error {
	return dm.fin
}
<commit_msg>demuxer cleanup<commit_after>package v2

import (
	"fmt"
	"sync/atomic"

	"github.com/tendermint/tendermint/libs/log"
)

type scFull struct{}
type pcFull struct{}

type demuxer struct {
	input     chan Event
	scheduler *Routine
	processor *Routine
	fin       chan error
	stopped   chan struct{}
	rdy       chan struct{}
	running   *uint32
	stopping  *uint32
	logger    log.Logger
}

func newDemuxer(scheduler *Routine, processor *Routine) *demuxer {
	return &demuxer{
		input:     make(chan Event, 10),
		scheduler: scheduler,
		processor: processor,
		stopped:   make(chan struct{}, 1),
		fin:       make(chan error, 1),
		rdy:       make(chan struct{}, 1),
		running:   new(uint32),
		stopping:  new(uint32),
		logger:    log.NewNopLogger(),
	}
}

func (dm *demuxer) setLogger(logger log.Logger) {
	dm.logger = logger
}

func (dm *demuxer) start() {
	starting := atomic.CompareAndSwapUint32(dm.running, uint32(0), uint32(1))
	if !starting {
		panic("Routine has already started")
	}
	dm.logger.Info("demuxer: run")
	for {
		if !dm.isRunning() {
			break
		}
		select {
		case event, ok := <-dm.input:
			if !ok {
				dm.logger.Info("demuxer: stopping")
				dm.terminate(fmt.Errorf("stopped"))
				dm.stopped <- struct{}{}
				return
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		case event, ok := <-dm.scheduler.next():
			if !ok {
				dm.logger.Info("demuxer: scheduler output closed")
				continue
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		case event, ok := <-dm.processor.next():
			if !ok {
				dm.logger.Info("demuxer: processor output closed")
				continue
			}
			oEvents, err := dm.handle(event)
			if err != nil {
				dm.terminate(err)
				return
			}
			for _, event := range oEvents {
				dm.input <- event
			}
		}
	}
}

func (dm *demuxer) handle(event Event) (Events, error) {
	received := dm.scheduler.trySend(event)
	if !received {
		return Events{scFull{}}, nil // backpressure
	}

	received = dm.processor.trySend(event)
	if !received {
		return Events{pcFull{}}, nil // backpressure
	}

	return Events{}, nil
}

func (dm *demuxer) trySend(event Event) bool {
	if !dm.isRunning() || dm.isStopping() {
		dm.logger.Info("dummuxer isn't running")
		return false
	}
	select {
	case dm.input <- event:
		return true
	default:
		dm.logger.Info("demuxer channel was full")
		return false
	}
}

func (dm *demuxer) isRunning() bool {
	return atomic.LoadUint32(dm.running) == 1
}

func (dm *demuxer) isStopping() bool {
	return atomic.LoadUint32(dm.stopping) == 1
}

func (dm *demuxer) ready() chan struct{} {
	return dm.rdy
}

func (dm *demuxer) stop() {
	if !dm.isRunning() {
		return
	}
	stopping := atomic.CompareAndSwapUint32(dm.stopping, uint32(0), uint32(1))
	if !stopping {
		panic("Demuxer has already stopped")
	}
	dm.logger.Info("demuxer stop")
	close(dm.input)
	<-dm.stopped
}

func (dm *demuxer) terminate(reason error) {
	stopped := atomic.CompareAndSwapUint32(dm.running, uint32(1), uint32(0))
	if !stopped {
		panic("called terminate but already terminated")
	}
	dm.fin <- reason
}

func (dm *demuxer) final() chan error {
	return dm.fin
}

{'fraction_non_alphanumeric': 0.14257588187038556, 'fraction_numerical': 0.011812961443806398, 'mean_word_length': 3.1497617426820965, 'pattern_counts': {'":': 0, '<': 23, '<?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': '1517149', 'n_tokens_mistral': 2591, 'n_tokens_neox': 2244, 'n_words': 558}