<commit_before>package ringbuffer

import (
	"fmt"
	"sync"
	"sync/atomic"
)

type RingBuffer struct {
	readIndex  int64      //读序号
	writeIndex int64      //写序号
	buf        []*[]byte  //环形buffer指针数组
	bufSize    int64      //初始化环形buffer指针数组大小
	mask       int64      //初始化环形buffer指针数组大小
	pcond      *sync.Cond //生产者
	ccond      *sync.Cond //消费者
}

func powerOfTwo64(n int64) bool {
	return n != 0 && (n&(n-1)) == 0
}

/**
初始化ringbuffer
参数bufferSize：初始化环形buffer指针数组大小
*/
func NewRingBuffer(size int64) (*RingBuffer, error) {
	if !powerOfTwo64(size) {
		return nil, fmt.Errorf("This size is not able to used")
	}
	buffer := RingBuffer{
		readIndex:  int64(0),
		writeIndex: int64(0),
		buf:        make([]*[]byte, size),
		bufSize:    size,
		mask:       size - int64(1),
		pcond:      sync.NewCond(new(sync.Mutex)),
		ccond:      sync.NewCond(new(sync.Mutex)),
	}
	for i := 0; i < size; i++ {
		buffer.buf[i] = nil
	}
	return &buffer, nil
}

/**
获取当前读序号
*/
func (this *RingBuffer) GetCurrentReadIndex() uint64 {
	return atomic.LoadInt64(&this.readIndex)
}

/**
获取当前写序号
*/
func (this *RingBuffer) GetCurrentWriteIndex() uint64 {
	return atomic.LoadInt64(&this.writeIndex)
}

/**
读取ringbuffer指定的buffer指针，返回该指针并清空ringbuffer该位置存在的指针内容，以及将读序号加1
*/
func (this *RingBuffer) ReadBuffer() (p *[]byte, ok bool) {
	this.ccond.L.Lock()
	defer func() {
		this.pcond.Broadcast()
		this.ccond.L.Unlock()
	}()
	ok = false
	p = nil
	readIndex := this.GetCurrentReadIndex()
	writeIndex := this.GetCurrentWriteIndex()
	for readIndex >= writeIndex {
		writeIndex = this.GetCurrentWriteIndex()
		this.ccond.Wait()
		continue
	}
	index := readIndex & this.mask //替代求模
	p = this.buf[index]
	this.buf[index] = nil
	atomic.AddInt64(&this.readIndex, int64(1))
	if p != nil {
		ok = true
	}
	return p, ok
}

/**
写入ringbuffer指针，以及将写序号加1
*/
func (this *RingBuffer) WriteBuffer(in *[]byte) (ok bool) {
	this.pcond.L.Lock()
	defer func() {
		this.ccond.Broadcast()
		this.pcond.L.Unlock()
	}()
	ok = false
	readIndex := this.GetCurrentReadIndex()
	writeIndex := this.GetCurrentWriteIndex()
	for writeIndex-readIndex >= this.bufSize {
		readIndex = this.GetCurrentReadIndex()
		this.pcond.Wait()
		continue
	}
	index := writeIndex & this.mask //替代求模
	if this.buf[index] == nil {
		this.buf[index] = in
		atomic.AddInt64(&this.writeIndex, int64(1))
		ok = true
	}
	return ok
}
<commit_msg>修改bug<commit_after>package ringbuffer

import (
	"fmt"
	"sync"
	"sync/atomic"
)

type RingBuffer struct {
	readIndex  int64      //读序号
	writeIndex int64      //写序号
	buf        []*[]byte  //环形buffer指针数组
	bufSize    int64      //初始化环形buffer指针数组大小
	mask       int64      //初始化环形buffer指针数组大小
	pcond      *sync.Cond //生产者
	ccond      *sync.Cond //消费者
}

func powerOfTwo64(n int64) bool {
	return n != 0 && (n&(n-1)) == 0
}

/**
初始化ringbuffer
参数bufferSize：初始化环形buffer指针数组大小
*/
func NewRingBuffer(size int64) (*RingBuffer, error) {
	if !powerOfTwo64(size) {
		return nil, fmt.Errorf("This size is not able to used")
	}
	buffer := RingBuffer{
		readIndex:  int64(0),
		writeIndex: int64(0),
		buf:        make([]*[]byte, size),
		bufSize:    size,
		mask:       size - int64(1),
		pcond:      sync.NewCond(new(sync.Mutex)),
		ccond:      sync.NewCond(new(sync.Mutex)),
	}
	for i := 0; i < size; i++ {
		buffer.buf[i] = nil
	}
	return &buffer, nil
}

/**
获取当前读序号
*/
func (this *RingBuffer) GetCurrentReadIndex() int64 {
	return atomic.LoadInt64(&this.readIndex)
}

/**
获取当前写序号
*/
func (this *RingBuffer) GetCurrentWriteIndex() int64 {
	return atomic.LoadInt64(&this.writeIndex)
}

/**
读取ringbuffer指定的buffer指针，返回该指针并清空ringbuffer该位置存在的指针内容，以及将读序号加1
*/
func (this *RingBuffer) ReadBuffer() (p *[]byte, ok bool) {
	this.ccond.L.Lock()
	defer func() {
		this.pcond.Broadcast()
		this.ccond.L.Unlock()
	}()
	ok = false
	p = nil
	readIndex := this.GetCurrentReadIndex()
	writeIndex := this.GetCurrentWriteIndex()
	for readIndex >= writeIndex {
		writeIndex = this.GetCurrentWriteIndex()
		this.ccond.Wait()
		continue
	}
	index := readIndex & this.mask //替代求模
	p = this.buf[index]
	this.buf[index] = nil
	atomic.AddInt64(&this.readIndex, int64(1))
	if p != nil {
		ok = true
	}
	return p, ok
}

/**
写入ringbuffer指针，以及将写序号加1
*/
func (this *RingBuffer) WriteBuffer(in *[]byte) (ok bool) {
	this.pcond.L.Lock()
	defer func() {
		this.ccond.Broadcast()
		this.pcond.L.Unlock()
	}()
	ok = false
	readIndex := this.GetCurrentReadIndex()
	writeIndex := this.GetCurrentWriteIndex()
	for writeIndex-readIndex >= this.bufSize {
		readIndex = this.GetCurrentReadIndex()
		this.pcond.Wait()
		continue
	}
	index := writeIndex & this.mask //替代求模
	if this.buf[index] == nil {
		this.buf[index] = in
		atomic.AddInt64(&this.writeIndex, int64(1))
		ok = true
	}
	return ok
}

{'fraction_non_alphanumeric': 0.1504047720494248, 'fraction_numerical': 0.020877716233489562, 'mean_word_length': 3.50143815915628, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '779818', 'n_tokens_mistral': 2029, 'n_tokens_neox': 2019, 'n_words': 380}