Changeset 57 in code for trunk


Ignore:
Timestamp:
Feb 17, 2020, 2:46:29 PM (5 years ago)
Author:
contact
Message:

Fix issues related to Ring

  • RingConsumer is now used directly in the goroutine responsible for writing downstream messages. This allows the ring buffer not to be consumed on write error.
  • RingConsumer now has a channel attached. This allows PRIVMSG messages to always use RingConsumer, instead of also directly pushing messages to all downstream connections.
  • Multiple clients with the same history name are now supported.
  • Ring is now protected by a mutex
Location:
trunk
Files:
4 edited

Legend:

Unmodified
Added
Removed

  • trunk/downstream.go

    r56 r57  
    4141
    4242type downstreamConn struct {
    43         net      net.Conn
    44         irc      *irc.Conn
    45         srv      *Server
    46         logger   Logger
    47         messages chan *irc.Message
     43        net       net.Conn
     44        irc       *irc.Conn
     45        srv       *Server
     46        logger    Logger
     47        messages  chan *irc.Message
     48        consumers chan *RingConsumer
     49        closed    chan struct{}
    4850
    4951        registered bool
    5052        user       *user
    51         closed     bool
    5253        nick       string
    5354        username   string
    … …  
    5758func newDownstreamConn(srv *Server, netConn net.Conn) *downstreamConn {
    5859        dc := &downstreamConn{
    59                 net:      netConn,
    60                 irc:      irc.NewConn(netConn),
    61                 srv:      srv,
    62                 logger:   &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
    63                 messages: make(chan *irc.Message, 64),
     60                net:       netConn,
     61                irc:       irc.NewConn(netConn),
     62                srv:       srv,
     63                logger:    &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
     64                messages:  make(chan *irc.Message, 64),
     65                consumers: make(chan *RingConsumer),
     66                closed:    make(chan struct{}),
    6467        }
    6568
    … …  
    8386                User: dc.username,
    8487                // TODO: fill the host?
     88        }
     89}
     90
     91func (dc *downstreamConn) isClosed() bool {
     92        select {
     93        case <-dc.closed:
     94                return true
     95        default:
     96                return false
    8597        }
    8698}
    … …  
    105117                }
    106118
    107                 if dc.closed {
     119                if dc.isClosed() {
    108120                        return nil
    109121                }
    … …  
    114126
    115127func (dc *downstreamConn) writeMessages() error {
    116         for msg := range dc.messages {
    117                 if err := dc.irc.WriteMessage(msg); err != nil {
    118                         return err
     128        for {
     129                var err error
     130                var closed bool
     131                select {
     132                case msg := <-dc.messages:
     133                        err = dc.irc.WriteMessage(msg)
     134                case consumer := <-dc.consumers:
     135                        for {
     136                                msg := consumer.Peek()
     137                                if msg == nil {
     138                                        break
     139                                }
     140                                err = dc.irc.WriteMessage(msg)
     141                                if err != nil {
     142                                        break
     143                                }
     144                                consumer.Consume()
     145                        }
     146                case <-dc.closed:
     147                        closed = true
     148                }
     149                if err != nil {
     150                        return err
     151                }
     152                if closed {
     153                        break
    119154                }
    120155        }
    … …  
    123158
    124159func (dc *downstreamConn) Close() error {
    125         if dc.closed {
     160        if dc.isClosed() {
    126161                return fmt.Errorf("downstream connection already closed")
    127162        }
    … …  
    135170                }
    136171                u.lock.Unlock()
    137 
    138                 // TODO: figure out a better way to advance the ring buffer consumer cursor
    139                 u.forEachUpstream(func(uc *upstreamConn) {
    140                         // TODO: let clients specify the ring buffer name in their username
    141                         uc.ring.Consumer("").Reset()
    142                 })
    143         }
    144 
    145         close(dc.messages)
    146         dc.closed = true
    147 
     172        }
     173
     174        close(dc.closed)
    148175        return nil
    149176}
    … …  
    212239
    213240        u.lock.Lock()
     241        firstDownstream := len(u.downstreamConns) == 0
    214242        u.downstreamConns = append(u.downstreamConns, dc)
    215243        u.lock.Unlock()
    … …  
    250278
    251279                // TODO: let clients specify the ring buffer name in their username
    252                 consumer := uc.ring.Consumer("")
    253                 for {
    254                         // TODO: these messages will get lost if the connection is closed
    255                         msg := consumer.Consume()
    256                         if msg == nil {
    257                                 break
    258                         }
    259                         dc.SendMessage(msg)
    260                 }
     280                historyName := ""
     281
     282                var seqPtr *uint64
     283                if firstDownstream {
     284                        seq, ok := uc.history[historyName]
     285                        if ok {
     286                                seqPtr = &seq
     287                        }
     288                }
     289
     290                consumer, ch := uc.ring.Consumer(seqPtr)
     291                go func() {
     292                        for {
     293                                var closed bool
     294                                select {
     295                                case <-ch:
     296                                        dc.consumers <- consumer
     297                                case <-dc.closed:
     298                                        closed = true
     299                                }
     300                                if closed {
     301                                        break
     302                                }
     303                        }
     304
     305                        seq := consumer.Close()
     306
     307                        dc.user.lock.Lock()
     308                        lastDownstream := len(dc.user.downstreamConns) == 0
     309                        dc.user.lock.Unlock()
     310
     311                        if lastDownstream {
     312                                uc.history[historyName] = seq
     313                        }
     314                }()
    261315        })
    262316

  • trunk/ring.go

    r51 r57  
    22
    33import (
     4        "sync"
     5
    46        "gopkg.in/irc.v3"
    57)
    … …  
    810// buffer size is fixed. The ring buffer is stored in memory.
    911type Ring struct {
    10         buffer   []*irc.Message
    11         cap, cur uint64
     12        buffer []*irc.Message
     13        cap    uint64
    1214
    13         consumers map[string]*RingConsumer
     15        lock      sync.Mutex
     16        cur       uint64
     17        consumers []*RingConsumer
    1418}
    1519
    1620func NewRing(capacity int) *Ring {
    1721        return &Ring{
    18                 buffer:    make([]*irc.Message, capacity),
    19                 cap:       uint64(capacity),
    20                 consumers: make(map[string]*RingConsumer),
     22                buffer: make([]*irc.Message, capacity),
     23                cap:    uint64(capacity),
    2124        }
    2225}
    2326
    2427func (r *Ring) Produce(msg *irc.Message) {
     28        r.lock.Lock()
     29        defer r.lock.Unlock()
     30
    2531        i := int(r.cur % r.cap)
    2632        r.buffer[i] = msg
    2733        r.cur++
     34
     35        for _, consumer := range r.consumers {
     36                select {
     37                case consumer.ch <- struct{}{}:
     38                        // This space is intentionally left blank
     39                default:
     40                        // The channel already has a pending item
     41                }
     42        }
    2843}
    2944
    30 func (r *Ring) Consumer(name string) *RingConsumer {
    31         consumer, ok := r.consumers[name]
    32         if ok {
    33                 return consumer
     45func (r *Ring) Consumer(seq *uint64) (*RingConsumer, <-chan struct{}) {
     46        consumer := &RingConsumer{
     47                ring: r,
     48                ch:   make(chan struct{}, 1),
    3449        }
    3550
    36         consumer = &RingConsumer{
    37                 ring: r,
    38                 cur:  r.cur,
     51        r.lock.Lock()
     52        if seq != nil {
     53                consumer.cur = *seq
     54        } else {
     55                consumer.cur = r.cur
    3956        }
    40         r.consumers[name] = consumer
    41         return consumer
     57        if consumer.diff() > 0 {
     58                consumer.ch <- struct{}{}
     59        }
     60        r.consumers = append(r.consumers, consumer)
     61        r.lock.Unlock()
     62
     63        return consumer, consumer.ch
    4264}
    4365
    4466type RingConsumer struct {
    45         ring *Ring
    46         cur  uint64
     67        ring   *Ring
     68        cur    uint64
     69        ch     chan struct{}
     70        closed bool
    4771}
    4872
    49 func (rc *RingConsumer) Diff() uint64 {
     73// diff returns the number of pending messages. It assumes the Ring is locked.
     74func (rc *RingConsumer) diff() uint64 {
    5075        if rc.cur > rc.ring.cur {
    5176                panic("jounce: consumer cursor greater than producer cursor")
    … …  
    5580
    5681func (rc *RingConsumer) Peek() *irc.Message {
    57         diff := rc.Diff()
     82        if rc.closed {
     83                panic("jounce: RingConsumer.Peek called after Close")
     84        }
     85
     86        rc.ring.lock.Lock()
     87        defer rc.ring.lock.Unlock()
     88
     89        diff := rc.diff()
    5890        if diff == 0 {
    5991                return nil
    … …  
    79111}
    80112
    81 func (rc *RingConsumer) Reset() {
    82         rc.cur = rc.ring.cur
     113func (rc *RingConsumer) Close() uint64 {
     114        rc.ring.lock.Lock()
     115        for i := range rc.ring.consumers {
     116                if rc.ring.consumers[i] == rc {
     117                        rc.ring.consumers = append(rc.ring.consumers[:i], rc.ring.consumers[i+1:]...)
     118                        break
     119                }
     120        }
     121        rc.ring.lock.Unlock()
     122
     123        close(rc.ch)
     124        rc.closed = true
     125        return rc.cur
    83126}

  • trunk/server.go

    r50 r57  
    3939        upstreamConns   []*upstreamConn
    4040        downstreamConns []*downstreamConn
     41}
     42
     43func newUser(srv *Server, username string) *user {
     44        return &user{
     45                username: username,
     46                srv:      srv,
     47        }
    4148}
    4249
    … …  
    117124func (s *Server) Run() {
    118125        // TODO: multi-user
    119         u := &user{username: "jounce", srv: s}
     126        u := newUser(s, "jounce")
    120127
    121128        s.lock.Lock()

  • trunk/upstream.go

    r55 r57  
    4545        modes      modeSet
    4646        channels   map[string]*upstreamChannel
     47        history    map[string]uint64
    4748}
    4849
    … …  
    6768                ring:     NewRing(u.srv.RingCap),
    6869                channels: make(map[string]*upstreamChannel),
     70                history:  make(map[string]uint64),
    6971        }
    7072
    … …  
    306308        case "PRIVMSG":
    307309                uc.ring.Produce(msg)
    308                 uc.user.forEachDownstream(func(dc *downstreamConn) {
    309                         dc.SendMessage(msg)
    310                 })
    311310        case irc.RPL_YOURHOST, irc.RPL_CREATED:
    312311                // Ignore
Note: See TracChangeset for help on using the changeset viewer.