<commit_before>package lfchan

import (
	"runtime"
	"sync/atomic"
)

type innerChan struct {
	q       []AtomicValue
	sendIdx uint32
	recvIdx uint32
	len     int32
	die     int32
}

// Chan is a lock free channel that supports concurrent channel operations.
type Chan struct {
	*innerChan
}

// New returns a new channel with the buffer set to 1
func New() Chan {
	return NewSize(1)
}

// NewSize creates a buffered channel, with minimum length of 1
func NewSize(sz int) Chan {
	if sz < 1 {
		panic("sz < 1")
	}
	return Chan{&innerChan{
		q:       make([]AtomicValue, sz),
		sendIdx: ^uint32(0),
		recvIdx: ^uint32(0),
	}}
}

// Send adds v to the buffer of the channel and returns true, if the channel is closed it returns false
func (ch Chan) Send(v interface{}, block bool) bool {
	if !block && ch.Len() == ch.Cap() {
		return false
	}
	ncpu, ln, cnt := uint32(runtime.NumCPU()), uint32(len(ch.q)), uint32(0)
	for !ch.Closed() {

		i := atomic.AddUint32(&ch.sendIdx, 1)
		if ch.q[i%ln].CompareAndSwap(nil, v) {
			atomic.AddInt32(&ch.len, 1)
			return true
		}
		if block {
			if i%(ncpu*100) == 0 {
				for i := uint32(0); i < ncpu; i++ {
					runtime.Gosched()
				}
			}
		} else if cnt++; cnt == ln {
			break
		}
		runtime.Gosched()
	}
	return false
}

// Recv blocks until a value is available and returns v, true, or if the channel is closed and
// the buffer is empty, it will return nil, false
func (ch Chan) Recv(block bool) (interface{}, bool) {
	if !block && ch.Len() == 0 { // fast path
		return nil, false
	}
	ncpu, ln, cnt := uint32(runtime.NumCPU()), uint32(len(ch.q)), uint32(0)
	for !ch.Closed() || ch.Len() > 0 {
		i := atomic.AddUint32(&ch.recvIdx, 1)
		if v := ch.q[i%ln].Swap(nil); v != nil {
			atomic.AddInt32(&ch.len, -1)
			return v, true
		}
		if block {
			if i%(ncpu*100) == 0 {
				for i := uint32(0); i < ncpu; i++ {
					runtime.Gosched()
				}
			}
		} else if cnt++; cnt == ln {
			break
		}
		runtime.Gosched()
	}
	return nil, false
}

// Close marks the channel as closed
func (ch Chan) Close() { atomic.StoreInt32(&ch.die, 1) }

// Closed returns true if the channel have been closed
func (ch Chan) Closed() bool { return atomic.LoadInt32(&ch.die) == 1 }

// Cap returns the size of the internal queue
func (ch Chan) Cap() int { return len(ch.q) }

// Len returns the number of elements queued
func (ch Chan) Len() int { return int(atomic.LoadInt32(&ch.len)) }

// SelectSend sends v to the first available channel, if block is true, it blocks until a channel a accepts the value.
// returns false if all channels were full and block is false.
func SelectSend(block bool, v interface{}, chans ...Chan) bool {
	for {
		for i := range chans {
			if ok := chans[i].Send(v, false); ok {
				return ok
			}
		}
		if !block {
			return false
		}
		runtime.Gosched()
	}
}

// SelectRecv returns the first available value from chans, if block is true, it blocks until a value is available.
// returns nil, false if all channels were empty and block is false.
func SelectRecv(block bool, chans ...Chan) (interface{}, bool) {
	for {
		for i := range chans {
			if v, ok := chans[i].Recv(false); ok {
				return v, ok
			}
		}
		if !block {
			return nil, false
		}
		runtime.Gosched()
	}
}<commit_msg>Send/RecvOnly impl<commit_after>package lfchan

import (
	"runtime"
	"sync/atomic"
)

type innerChan struct {
	q       []AtomicValue
	sendIdx uint32
	recvIdx uint32
	len     int32
	die     int32
}

// Chan is a lock free channel that supports concurrent channel operations.
type Chan struct {
	*innerChan
}

// New returns a new channel with the buffer set to 1
func New() Chan {
	return NewSize(1)
}

// NewSize creates a buffered channel, with minimum length of 1
func NewSize(sz int) Chan {
	if sz < 1 {
		panic("sz < 1")
	}
	return Chan{&innerChan{
		q:       make([]AtomicValue, sz),
		sendIdx: ^uint32(0),
		recvIdx: ^uint32(0),
	}}
}

// Send adds v to the buffer of the channel and returns true, if the channel is closed it returns false
func (ch Chan) Send(v interface{}, block bool) bool {
	if !block && ch.Len() == ch.Cap() {
		return false
	}
	ncpu, ln, cnt := uint32(runtime.NumCPU()), uint32(len(ch.q)), uint32(0)
	for !ch.Closed() {

		i := atomic.AddUint32(&ch.sendIdx, 1)
		if ch.q[i%ln].CompareAndSwap(nil, v) {
			atomic.AddInt32(&ch.len, 1)
			return true
		}
		if block {
			if i%(ncpu*100) == 0 {
				for i := uint32(0); i < ncpu; i++ {
					runtime.Gosched()
				}
			}
		} else if cnt++; cnt == ln {
			break
		}
		runtime.Gosched()
	}
	return false
}

// Recv blocks until a value is available and returns v, true, or if the channel is closed and
// the buffer is empty, it will return nil, false
func (ch Chan) Recv(block bool) (interface{}, bool) {
	if !block && ch.Len() == 0 { // fast path
		return nil, false
	}
	ncpu, ln, cnt := uint32(runtime.NumCPU()), uint32(len(ch.q)), uint32(0)
	for !ch.Closed() || ch.Len() > 0 {
		i := atomic.AddUint32(&ch.recvIdx, 1)
		if v := ch.q[i%ln].Swap(nil); v != nil {
			atomic.AddInt32(&ch.len, -1)
			return v, true
		}
		if block {
			if i%(ncpu*100) == 0 {
				for i := uint32(0); i < ncpu; i++ {
					runtime.Gosched()
				}
			}
		} else if cnt++; cnt == ln {
			break
		}
		runtime.Gosched()
	}
	return nil, false
}

// RecvOnly returns a receive-only channel.
func (ch Chan) RecvOnly() RecvOnly { return RecvOnly{ch} }

// RecvOnly returns a send-only channel.
func (ch Chan) SendOnly() SendOnly { return SendOnly{ch} }

// Close marks the channel as closed
func (ch Chan) Close() { atomic.StoreInt32(&ch.die, 1) }

// Closed returns true if the channel have been closed
func (ch Chan) Closed() bool { return atomic.LoadInt32(&ch.die) == 1 }

// Cap returns the size of the internal queue
func (ch Chan) Cap() int { return len(ch.q) }

// Len returns the number of elements queued
func (ch Chan) Len() int { return int(atomic.LoadInt32(&ch.len)) }

// SelectSend sends v to the first available channel, if block is true, it blocks until a channel a accepts the value.
// returns false if all channels were full and block is false.
func SelectSend(block bool, v interface{}, chans ...Chan) bool {
	for {
		for i := range chans {
			if ok := chans[i].Send(v, false); ok {
				return ok
			}
		}
		if !block {
			return false
		}
		runtime.Gosched()
	}
}

// SelectRecv returns the first available value from chans, if block is true, it blocks until a value is available.
// returns nil, false if all channels were empty and block is false.
func SelectRecv(block bool, chans ...Chan) (interface{}, bool) {
	for {
		for i := range chans {
			if v, ok := chans[i].Recv(false); ok {
				return v, ok
			}
		}
		if !block {
			return nil, false
		}
		runtime.Gosched()
	}
}

// RecvOnly is a channel only has receive operations.
type RecvOnly struct{ c Chan }

// Recv
func (ro RecvOnly) Recv(block bool) (interface{}, bool) { return ro.c.Recv(block) }

// SebdOnly is a channel only has send operations.
type SendOnly struct{ c Chan }

// Send
func (so SendOnly) Send(v interface{}, block bool) bool { return so.c.Send(v, block) }
{'fraction_non_alphanumeric': 0.13574274331246441, 'fraction_numerical': 0.019635742743312465, 'mean_word_length': 3.0466321243523318, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 5, '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': '2693590', 'n_tokens_mistral': 2775, 'n_tokens_neox': 2475, 'n_words': 893}