<commit_before>package core

import (
	"log"

	"github.com/nikhan/go-fetch"
)

// NewBlock creates a new block from a spec
func NewBlock(s Spec) *Block {
	var in []Route
	var out []Output

	for _, v := range s.Inputs {

		q, _ := fetch.Parse(".")
		in = append(in, Route{
			Name: v.Name,
			Path: q,
			C:    make(chan Message),
		})
	}

	for _, v := range s.Outputs {
		out = append(out, Output{
			Name:        v.Name,
			Connections: make(map[Connection]struct{}),
		})
	}

	return &Block{
		state: BlockState{
			make(MessageMap),
			make(MessageMap),
			make(Manifest),
			false,
		},
		routing: BlockRouting{
			Inputs:        in,
			Outputs:       out,
			InterruptChan: make(chan Interrupt),
			Shared: SharedStore{
				Type: s.Shared,
			},
		},
		kernel: s.Kernel,
	}
}

func (b *Block) process() Interrupt {
	if b.state.Processed == true {
		return nil
	}

	if b.routing.Shared.Type != NONE && b.routing.Shared.Store == nil {
		select {
		case f := <-b.routing.InterruptChan:
			return f
		}
	}

	if b.routing.Shared.Type != NONE {
		b.routing.Shared.Store.Lock()
	}

	interrupt := b.kernel(b.state.inputValues,
		b.state.outputValues,
		b.routing.Shared.Store,
		b.routing.InterruptChan)

	if interrupt != nil {
		if b.routing.Shared.Type != NONE {
			b.routing.Shared.Store.Unlock()
		}
		return interrupt
	}

	if b.routing.Shared.Type != NONE {
		b.routing.Shared.Store.Unlock()
	}

	b.state.Processed = true

	return nil
}

// suture: the main routine the block runs
func (b *Block) Serve() {
	for {
		var interrupt Interrupt

		b.routing.RLock()
		for {
			interrupt = b.receive()
			if interrupt != nil {
				break
			}

			interrupt = b.process()
			if interrupt != nil {
				break
			}

			interrupt = b.broadcast()
			if interrupt != nil {
				break
			}

			b.crank()
		}
		b.routing.RUnlock()
		b.routing.Lock()
		if ok := interrupt(); !ok {
			return
		}
		b.routing.Unlock()
	}
}

// todo: proper getter/setters of route properties
// 	GetRouteValue, GetRoutePath, GetRouteChan

// Input returns the specfied Route
func (b *Block) Input(id RouteID) Route {
	b.routing.RLock()
	defer b.routing.RUnlock()
	return b.routing.Inputs[id]
}

func (b *Block) Store(s Store) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Shared.Store = s
		return true
	}
}

// RouteValue sets the route to always be the specified value
func (b *Block) RouteValue(id RouteID, v Message) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Inputs[id].Value = &v
		return true
	}
}

// RoutePath sets a Route's Path to the supplied go-fetch Query
func (b *Block) RoutePath(id RouteID, p *fetch.Query) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Inputs[id].Path = p
		b.routing.Inputs[id].Value = nil
		return true
	}
}

// Connect connects a Route, specified by ID, to a connection
func (b *Block) Connect(id RouteID, c Connection) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Outputs[id].Connections[c] = struct{}{}
		return true
	}
}

// Disconnect removes a connection from a Route
func (b *Block) Disconnect(id RouteID, c Connection) {
	b.routing.InterruptChan <- func() bool {
		delete(b.routing.Outputs[id].Connections, c)
		return true
	}
}

// suture: stop the block
func (b *Block) Stop() {
	b.routing.InterruptChan <- func() bool {
		return false
	}
}

// wait and listen for all kernel inputs to be filled.
func (b *Block) receive() Interrupt {
	var err error
	for id, input := range b.routing.Inputs {
		//if we have already received a value on this input, skip.
		if _, ok := b.state.inputValues[RouteID(id)]; ok {
			continue
		}

		// if there is a value set for this input, place value on
		// buffer and set it in map.
		if input.Value != nil {
			b.state.inputValues[RouteID(id)] = *input.Value
			continue
		}

		select {
		case m := <-input.C:
			b.state.inputValues[RouteID(id)], err = fetch.Run(input.Path, m)
			if err != nil {
				log.Fatal(err)
			}
		case f := <-b.routing.InterruptChan:
			return f
		}
	}
	return nil<commit_msg>comments & org<commit_after>package core

import (
	"log"

	"github.com/nikhan/go-fetch"
)

// NewBlock creates a new block from a spec
func NewBlock(s Spec) *Block {
	var in []Route
	var out []Output

	for _, v := range s.Inputs {

		q, _ := fetch.Parse(".")
		in = append(in, Route{
			Name: v.Name,
			Path: q,
			C:    make(chan Message),
		})
	}

	for _, v := range s.Outputs {
		out = append(out, Output{
			Name:        v.Name,
			Connections: make(map[Connection]struct{}),
		})
	}

	return &Block{
		state: BlockState{
			make(MessageMap),
			make(MessageMap),
			make(Manifest),
			false,
		},
		routing: BlockRouting{
			Inputs:        in,
			Outputs:       out,
			InterruptChan: make(chan Interrupt),
			Shared: SharedStore{
				Type: s.Shared,
			},
		},
		kernel: s.Kernel,
	}
}

// suture: the main routine the block runs
func (b *Block) Serve() {
	for {
		var interrupt Interrupt

		b.routing.RLock()
		for {
			interrupt = b.receive()
			if interrupt != nil {
				break
			}

			interrupt = b.process()
			if interrupt != nil {
				break
			}

			interrupt = b.broadcast()
			if interrupt != nil {
				break
			}

			b.crank()
		}
		b.routing.RUnlock()
		b.routing.Lock()
		if ok := interrupt(); !ok {
			return
		}
		b.routing.Unlock()
	}
}

// todo: proper getter/setters of route properties
// 	GetRouteValue, GetRoutePath, GetRouteChan

// Input returns the specfied Route
func (b *Block) Input(id RouteID) Route {
	b.routing.RLock()
	defer b.routing.RUnlock()
	return b.routing.Inputs[id]
}

func (b *Block) Store(s Store) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Shared.Store = s
		return true
	}
}

// RouteValue sets the route to always be the specified value
func (b *Block) RouteValue(id RouteID, v Message) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Inputs[id].Value = &v
		return true
	}
}

// RoutePath sets a Route's Path to the supplied go-fetch Query
func (b *Block) RoutePath(id RouteID, p *fetch.Query) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Inputs[id].Path = p
		b.routing.Inputs[id].Value = nil
		return true
	}
}

// Connect connects a Route, specified by ID, to a connection
func (b *Block) Connect(id RouteID, c Connection) {
	b.routing.InterruptChan <- func() bool {
		b.routing.Outputs[id].Connections[c] = struct{}{}
		return true
	}
}

// Disconnect removes a connection from a Route
func (b *Block) Disconnect(id RouteID, c Connection) {
	b.routing.InterruptChan <- func() bool {
		delete(b.routing.Outputs[id].Connections, c)
		return true
	}
}

// suture: stop the block
func (b *Block) Stop() {
	b.routing.InterruptChan <- func() bool {
		return false
	}
}

// wait and listen for all kernel inputs to be filled.
func (b *Block) receive() Interrupt {
	var err error
	for id, input := range b.routing.Inputs {
		//if we have already received a value on this input, skip.
		if _, ok := b.state.inputValues[RouteID(id)]; ok {
			continue
		}

		// if there is a value set for this input, place value on
		// buffer and set it in map.
		if input.Value != nil {
			b.state.inputValues[RouteID(id)] = *input.Value
			continue
		}

		select {
		case m := <-input.C:
			b.state.inputValues[RouteID(id)], err = fetch.Run(input.Path, m)
			if err != nil {
				log.Fatal(err)
			}
		case f := <-b.routing.InterruptChan:
			return f
		}
	}
	return nil
}

// run kernel on inputs, produce outputs
func (b *Block) process() Interrupt {
	if b.state.Processed == true {
		return nil
	}

	// if this kernel relies on an external shared state then we need to
	// block until an interrupt connects us to a shared external state.
	if b.routing.Shared.Type != NONE && b.routing.Shared.Store == nil {
		select {
		case f := <-b.routing.InterruptChan:
			return f
		}
	}

	// we should only be able to get here if
	// - we don't need an shared state
	// - we have an external shared state and it has been attached
	if b.routing.Shared.Type != NONE {
		b.routing.Shared.Store.Lock()
	}

	// run the kernel
	interrupt := b.kernel(b.state.inputValues,
		b.state.outputValues,
		b.routing.Shared.Store,
		b.routing.InterruptChan)

	if interrupt != nil {
		if b.routing.Shared.Type != NONE {
			b.routing.Shared.Store.Unlock()
		}
		return interrupt
	}

	if b.routing.Shared.Type != NONE {
		b.routing.Shared.Store.Unlock()
	}

	b.state.Processed = true

	return nil
{'fraction_non_alphanumeric': 0.1345830329247777, 'fraction_numerical': 0.0, 'mean_word_length': 2.8965355805243447, 'pattern_counts': {'":': 0, '<': 21, '<?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': '1490803', 'n_tokens_mistral': 3280, 'n_tokens_neox': 2980, 'n_words': 841}