sad/soju
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Remove locks from ring buffer

Browse Source 
Everything is now accessed from the user goroutine.
This commit is contained in:
Simon Ser 2020-04-06 18:18:50 +02:00
parent d541587701
commit 3bece53335
No known key found for this signature in database
GPG Key ID: 0FDE7BE0E88F5E48
1 changed files with 0 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
ring.go
View File

@ -2,7 +2,6 @@ package soju
import ( import (
"fmt" "fmt"
"sync"
"gopkg.in/irc.v3" "gopkg.in/irc.v3"
) )
@ -13,7 +12,6 @@ type Ring struct {
buffer []*irc.Message buffer []*irc.Message
cap uint64 cap uint64
lock sync.Mutex
cur uint64 cur uint64
consumers []*RingConsumer consumers []*RingConsumer
closed bool closed bool
@ -29,9 +27,6 @@ func NewRing(capacity int) *Ring {
// Produce appends a new message to the ring buffer. // Produce appends a new message to the ring buffer.
func (r *Ring) Produce(msg *irc.Message) { func (r *Ring) Produce(msg *irc.Message) {
r.lock.Lock()
defer r.lock.Unlock()
if r.closed { if r.closed {
panic("soju: Ring.Produce called after Close") panic("soju: Ring.Produce called after Close")
} }
@ -42,9 +37,6 @@ func (r *Ring) Produce(msg *irc.Message) {
} }
func (r *Ring) Close() { func (r *Ring) Close() {
r.lock.Lock()
defer r.lock.Unlock()
if r.closed { if r.closed {
panic("soju: Ring.Close called twice") panic("soju: Ring.Close called twice")
} }
@ -62,14 +54,12 @@ func (r *Ring) Close() {
func (r *Ring) NewConsumer(seq *uint64) *RingConsumer { func (r *Ring) NewConsumer(seq *uint64) *RingConsumer {
consumer := &RingConsumer{ring: r} consumer := &RingConsumer{ring: r}
r.lock.Lock()
if seq != nil { if seq != nil {
consumer.cur = *seq consumer.cur = *seq
} else { } else {
consumer.cur = r.cur consumer.cur = r.cur
} }
r.consumers = append(r.consumers, consumer) r.consumers = append(r.consumers, consumer)
r.lock.Unlock()
return consumer return consumer
} }
@ -96,9 +86,6 @@ func (rc *RingConsumer) Peek() *irc.Message {
panic("soju: RingConsumer.Peek called after Close") panic("soju: RingConsumer.Peek called after Close")
} }
rc.ring.lock.Lock()
defer rc.ring.lock.Unlock()
diff := rc.diff() diff := rc.diff()
if diff == 0 { if diff == 0 {
return nil return nil
@ -129,14 +116,12 @@ func (rc *RingConsumer) Consume() *irc.Message {
// current history sequence number is returned. It can be provided later as an // current history sequence number is returned. It can be provided later as an
// argument to Ring.NewConsumer to resume the message stream. // argument to Ring.NewConsumer to resume the message stream.
func (rc *RingConsumer) Close() uint64 { func (rc *RingConsumer) Close() uint64 {
rc.ring.lock.Lock()
for i := range rc.ring.consumers { for i := range rc.ring.consumers {
if rc.ring.consumers[i] == rc { if rc.ring.consumers[i] == rc {
rc.ring.consumers = append(rc.ring.consumers[:i], rc.ring.consumers[i+1:]...) rc.ring.consumers = append(rc.ring.consumers[:i], rc.ring.consumers[i+1:]...)
break break
} }
} }
rc.ring.lock.Unlock()
rc.closed = true rc.closed = true
return rc.cur return rc.cur