Changeset 227 in code for trunk

Timestamp:

Apr 6, 2020, 4:05:36 PM (5 years ago)

Author:

contact

Message:

Remove per-network ring buffer goroutines

Just dispatch from the user goroutine. This allows removes a lot of complexity.

Location:

trunk

Files:

• doc/architecture.md (modified) (1 diff)
• downstream.go (modified) (2 diffs)
• upstream.go (modified) (1 diff)
• user.go (modified) (2 diffs)

trunk/doc/architecture.md

@@ -32 +32 @@
 and downstream message handlers are called from this goroutine, thus they can
 safely access both upstream and downstream state.
-
-In addition to these goroutines, each downstream connection also has one
-goroutine per network to handle new upstream messages coming from the ring
-buffer.

trunk/downstream.go

@@ -232 +232 @@
 
         dc.conn.SendMessage(msg)
+}
+
+func (dc *downstreamConn) sendFromUpstream(msg *irc.Message, uc *upstreamConn) {
+        dc.lock.Lock()
+        _, ours := dc.ourMessages[msg]
+        delete(dc.ourMessages, msg)
+        dc.lock.Unlock()
+        if ours && !dc.getCap("echo-message") {
+                // The message comes from our connection, don't echo it
+                // back
+                return
+        }
+
+        msg = msg.Copy()
+        switch msg.Command {
+        case "PRIVMSG", "NOTICE":
+                msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
+                msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
+        default:
+                panic(fmt.Sprintf("unexpected %q message", msg.Command))
+        }
+
+        dc.SendMessage(msg)
 }
 
@@ -664 +687 @@
 
         dc.forEachNetwork(func(net *network) {
-                dc.runNetwork(net, sendHistory)
+                var seqPtr *uint64
+                if sendHistory {
+                        seq, ok := net.history[dc.clientName]
+                        if ok {
+                                seqPtr = &seq
+                        }
+                }
+
+                consumer, _ := net.ring.NewConsumer(seqPtr)
+
+                if _, ok := dc.ringConsumers[net]; ok {
+                        panic("network has been added twice")
+                }
+                dc.ringConsumers[net] = consumer
+
+                // TODO: this means all history is lost when trying to send it while the
+                // upstream is disconnected. We need to store history differently so that
+                // we don't need access to upstreamConn to forward it to a downstream
+                // client.
+                uc := net.upstream()
+                if uc == nil {
+                        dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
+                        return
+                }
+
+                for {
+                        msg := consumer.Peek()
+                        if msg == nil {
+                                break
+                        }
+
+                        dc.sendFromUpstream(msg, uc)
+                        consumer.Consume()
+                }
         })
 
         return nil
-}
-
-// runNetwork starts listening for messages coming from the network's ring
-// buffer.
-//
-// It panics if the network is not suitable for the downstream connection.
-func (dc *downstreamConn) runNetwork(net *network, loadHistory bool) {
-        if dc.network != nil && net != dc.network {
-                panic("network not suitable for downstream connection")
-        }
-
-        var seqPtr *uint64
-        if loadHistory {
-                seq, ok := net.history[dc.clientName]
-                if ok {
-                        seqPtr = &seq
-                }
-        }
-
-        consumer, ch := net.ring.NewConsumer(seqPtr)
-
-        if _, ok := dc.ringConsumers[net]; ok {
-                panic("network has been added twice")
-        }
-        dc.ringConsumers[net] = consumer
-
-        go func() {
-                for range ch {
-                        uc := net.upstream()
-                        if uc == nil {
-                                dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
-                                continue
-                        }
-
-                        for {
-                                msg := consumer.Peek()
-                                if msg == nil {
-                                        break
-                                }
-
-                                dc.lock.Lock()
-                                _, ours := dc.ourMessages[msg]
-                                delete(dc.ourMessages, msg)
-                                dc.lock.Unlock()
-                                if ours && !dc.getCap("echo-message") {
-                                        // The message comes from our connection, don't echo it
-                                        // back
-                                        consumer.Consume()
-                                        continue
-                                }
-
-                                msg = msg.Copy()
-                                switch msg.Command {
-                                case "PRIVMSG", "NOTICE":
-                                        msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
-                                        msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
-                                default:
-                                        panic(fmt.Sprintf("unexpected %q message", msg.Command))
-                                }
-
-                                dc.SendMessage(msg)
-                                consumer.Consume()
-                        }
-                }
-        }()
 }
 

trunk/upstream.go

@@ -1366 +1366 @@
 func (uc *upstreamConn) produce(msg *irc.Message) {
         uc.network.ring.Produce(msg)
+
+        uc.forEachDownstream(func(dc *downstreamConn) {
+                dc.sendFromUpstream(dc.ringConsumers[uc.network].Consume(), uc)
+        })
 }
 

trunk/user.go

@@ -352 +352 @@
         u.forEachDownstream(func(dc *downstreamConn) {
                 if dc.network == nil {
-                        dc.runNetwork(network, false)
+                        consumer, _ := network.ring.NewConsumer(nil)
+                        dc.ringConsumers[network] = consumer
                 }
         })
@@ -376 +377 @@
                                 dc.Close()
                         }
+                        delete(dc.ringConsumers, net)
                 })