<commit_before>package engine

import (
	"runtime/debug"
	"sync"

	log "github.com/funkygao/log4go"
)

var (
	_ PluginRunner = &iRunner{}
	_ InputRunner  = &iRunner{}
)

// Input is the input plugin.
type Input interface {
	Plugin

	Acker

	// Run starts the main loop of the Input plugin.
	Run(r InputRunner, h PluginHelper) (err error)

	// Stop is the callback which stops the Input plugin.
	Stop(InputRunner)
}

// InputRunner is a helper for Input plugin to access some context data.
type InputRunner interface {
	PluginRunner

	// InChan returns input channel from which Inputs can get fresh Packets.
	InChan() chan *Packet

	// Input returns the associated Input plugin object.
	Input() Input

	// Injects Packet into the Router's input channel for delivery
	// to all Filter and Output plugins with corresponding matcher.
	Inject(pack *Packet)

	// DeclareResource declares the current Input plugin is interested in the specified resource.
	DeclareResource(resources ...string) error

	// RebalanceChannel returns a channel that notifies client every reblance event.
	RebalanceChannel() chan struct{}

	// LeadingResources returns all the resources this InputRunner is leading.
	LeadingResources() []string
}

type iRunner struct {
	pRunnerBase

	inChan chan *Packet

	leadingResources []string
	rebalanceCh      chan struct{}
}

func newInputRunner(input Input, pluginCommons *pluginCommons) (r *iRunner) {
	return &iRunner{
		pRunnerBase: pRunnerBase{
			plugin:        input.(Plugin),
			pluginCommons: pluginCommons,
		},
		rebalanceCh: make(chan struct{}, 1),
	}
}

func (ir *iRunner) Inject(pack *Packet) {
	if len(pack.Ident) == 0 {
		pack.Ident = ir.Name()
	}

	pack.input = ir.Input()
	ir.engine.router.hub <- pack
}

func (ir *iRunner) InChan() chan *Packet {
	return ir.inChan
}

func (ir *iRunner) Input() Input {
	return ir.plugin.(Input)
}

func (ir *iRunner) DeclareResource(resources ...string) error {
	return ir.engine.DeclareResource(ir.Name(), resources)
}

func (ir *iRunner) LeadingResources() []string {
	return ir.leadingResources
}

func (ir *iRunner) feedResources(resources []string) {
	ir.leadingResources = resources
}

func (ir *iRunner) rebalance() {
	ir.rebalanceCh <- struct{}{}
}

func (ir *iRunner) RebalanceChannel() chan struct{} {
	return ir.rebalanceCh
}

func (ir *iRunner) start(e *Engine, wg *sync.WaitGroup) error {
	ir.engine = e
	ir.inChan = e.inputRecycleChans[ir.Name()]

	go ir.runMainloop(e, wg)
	return nil
}

func (ir *iRunner) runMainloop(e *Engine, wg *sync.WaitGroup) {
	defer func() {
		if err := recover(); err != nil {
			log.Critical("[%s] %v\n%s", ir.Name(), err, string(debug.Stack()))
		}

		wg.Done()
	}()

	globals := Globals()
	for {
		log.Info("Input[%s] started", ir.Name())
		if err := ir.Input().Run(ir, e); err == nil {
			log.Info("Input[%s] stopped", ir.Name())
		} else {
			log.Error("Input[%s] stopped: %v", ir.Name(), err)
		}

		if globals.Stopping {
			e.stopInputRunner(ir.Name())

			return
		}

		if restart, ok := ir.plugin.(Restarter); ok {
			if !restart.CleanupForRestart() {
				// when we found all Input stopped, shutdown engine
				e.stopInputRunner(ir.Name())

				return
			}
		}

		log.Trace("Input[%s] restarting", ir.Name())

		// Re-initialize our plugin with its wrapper
		iw := e.inputWrappers[ir.Name()]
		ir.plugin = iw.Create()
	}

}
<commit_msg>when cluster disabled, call cluster funcs should raise err<commit_after>package engine

import (
	"runtime/debug"
	"sync"

	log "github.com/funkygao/log4go"
)

var (
	_ PluginRunner = &iRunner{}
	_ InputRunner  = &iRunner{}
)

// Input is the input plugin.
type Input interface {
	Plugin

	Acker

	// Run starts the main loop of the Input plugin.
	Run(r InputRunner, h PluginHelper) (err error)

	// Stop is the callback which stops the Input plugin.
	Stop(InputRunner)
}

// InputRunner is a helper for Input plugin to access some context data.
type InputRunner interface {
	PluginRunner

	// InChan returns input channel from which Inputs can get fresh Packets.
	InChan() chan *Packet

	// Input returns the associated Input plugin object.
	Input() Input

	// Injects Packet into the Router's input channel for delivery
	// to all Filter and Output plugins with corresponding matcher.
	Inject(pack *Packet)

	// DeclareResource declares the current Input plugin is interested in the specified resource.
	DeclareResource(resources ...string) error

	// RebalanceChannel returns a channel that notifies client every reblance event.
	RebalanceChannel() chan struct{}

	// LeadingResources returns all the resources this InputRunner is leading.
	LeadingResources() []string
}

type iRunner struct {
	pRunnerBase

	inChan chan *Packet

	leadingResources []string
	rebalanceCh      chan struct{}
}

func newInputRunner(input Input, pluginCommons *pluginCommons) (r *iRunner) {
	return &iRunner{
		pRunnerBase: pRunnerBase{
			plugin:        input.(Plugin),
			pluginCommons: pluginCommons,
		},
		rebalanceCh: make(chan struct{}, 1),
	}
}

func (ir *iRunner) Inject(pack *Packet) {
	if len(pack.Ident) == 0 {
		pack.Ident = ir.Name()
	}

	pack.input = ir.Input()
	ir.engine.router.hub <- pack
}

func (ir *iRunner) InChan() chan *Packet {
	return ir.inChan
}

func (ir *iRunner) Input() Input {
	return ir.plugin.(Input)
}

func (ir *iRunner) DeclareResource(resources ...string) error {
	if !Globals().ClusterEnabled {
		return ErrClusterDisabled
	}

	return ir.engine.DeclareResource(ir.Name(), resources)
}

func (ir *iRunner) LeadingResources() []string {
	return ir.leadingResources
}

func (ir *iRunner) feedResources(resources []string) {
	ir.leadingResources = resources
}

func (ir *iRunner) rebalance() {
	ir.rebalanceCh <- struct{}{}
}

func (ir *iRunner) RebalanceChannel() chan struct{} {
	return ir.rebalanceCh
}

func (ir *iRunner) start(e *Engine, wg *sync.WaitGroup) error {
	ir.engine = e
	ir.inChan = e.inputRecycleChans[ir.Name()]

	go ir.runMainloop(e, wg)
	return nil
}

func (ir *iRunner) runMainloop(e *Engine, wg *sync.WaitGroup) {
	defer func() {
		if err := recover(); err != nil {
			log.Critical("[%s] %v\n%s", ir.Name(), err, string(debug.Stack()))
		}

		wg.Done()
	}()

	globals := Globals()
	for {
		log.Info("Input[%s] started", ir.Name())
		if err := ir.Input().Run(ir, e); err == nil {
			log.Info("Input[%s] stopped", ir.Name())
		} else {
			log.Error("Input[%s] stopped: %v", ir.Name(), err)
		}

		if globals.Stopping {
			e.stopInputRunner(ir.Name())

			return
		}

		if restart, ok := ir.plugin.(Restarter); ok {
			if !restart.CleanupForRestart() {
				// when we found all Input stopped, shutdown engine
				e.stopInputRunner(ir.Name())

				return
			}
		}

		log.Trace("Input[%s] restarting", ir.Name())

		// Re-initialize our plugin with its wrapper
		iw := e.inputWrappers[ir.Name()]
		ir.plugin = iw.Create()
	}

}

{'fraction_non_alphanumeric': 0.12986060161408658, 'fraction_numerical': 0.0008804108584005869, 'mean_word_length': 3.7898805340829234, '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': '7171313', 'n_tokens_mistral': 2447, 'n_tokens_neox': 2262, 'n_words': 680}