ringbuffer

ringbuffer.go

package ringbuffer_safe

import (
	"log"
	"os"
	"os/signal"
	"runtime"
	"sync/atomic"
	"time"
)

const free = int32(0)  // 0 check, state is free, ringbuffer can be accessed
const busy = int32(42) // 42 (or any number), state is busy, ringbuffer cannot be accessed

// ringbuffer underlying array length
const ringBufferLength = 1024

// fixed example entry to read/write
const TXT = "help-me!"

// entry type of ring buffer
type entry struct {
	prio int
	data string
}

func Lock(state *int32) {
	for !atomic.CompareAndSwapInt32(state, free, busy) { // any other non-zero value, 42 is just an example
		runtime.Gosched() // poke the scheduler
	}
}

func Unlock(state *int32) {
	atomic.StoreInt32(state, free) // once atomic always atomic
}

// ring buffer counting the number of rw-operations
type RingBuffer struct {
	writelock  *int32 // 0 means free
	readlock   *int32
	buffer     [ringBufferLength]entry
	writeIndex int
	readIndex  int
	counter    int
}

func (r *RingBuffer) read() (b []entry, ok bool) {
	b = make([]entry, 0, 2)
	// flag signaling if read value was valid
	for r.readIndex != r.writeIndex {
		ok = true
		b = append(b, r.buffer[r.readIndex])
		r.readIndex++
		if r.readIndex >= ringBufferLength {
			r.readIndex %= ringBufferLength
		}
		r.counter++
	}
	return b, ok
}

func (r *RingBuffer) write(b []entry) {
	for i := range b {
		if ((r.readIndex - 1) % ringBufferLength) == r.writeIndex {
			break
		}
		r.buffer[r.writeIndex] = b[i]
		r.writeIndex += 1
		if r.writeIndex >= ringBufferLength {
			r.writeIndex %= ringBufferLength
		}
	}
}

func writeRingBuffer(r *RingBuffer) {
	for {
		// only enter here, as long as the writelock value does not equal the free value
		// once the values are the same, in an atomic operation also set it to busy
		Lock(r.writelock)
		r.write([]entry{entry{prio: 1, data: TXT}})
		Unlock(r.writelock)
	}
}

func readRingBuffer(r *RingBuffer) {
	for {
		Lock(r.readlock)
		r.read()
		Unlock(r.readlock)
	}
}

func NewRingbuffer() *RingBuffer {
	readlock := int32(0)
	writelock := int32(0)
	return &RingBuffer{
		writelock: &writelock,
		readlock:  &readlock,
	}
}

func Run() int {
	rb := NewRingbuffer()
	// simulate external call to write to ringbuffer
	go writeRingBuffer(rb)
	// go writeRingBuffer(rb)
	// go writeRingBuffer(rb)
	// simulate external call to read from ringbuffer
	go readRingBuffer(rb)
	// go readRingBuffer(rb)
	// go readRingBuffer(rb)

	// Right way to stop the server using a SHUTDOWN HOOK
	// Create a channel to receive OS signals
	c := make(chan os.Signal)

	// Relay os.Interrupt to our channel (os.Interrupt = CTRL+C)
	// Ignore other incoming signals
	signal.Notify(c, os.Interrupt)

	go func() {
		time.Sleep(time.Second * 1)
		c <- os.Interrupt
	}()

	// Block main routine until a signal is received
	// As long as user doesn't press CTRL+C a message
	// is not passed and our main routine keeps running
	<-c
	log.Println("count", rb.counter)
	return rb.counter
}