Changeset 188 in code

Timestamp:

Mar 31, 2020, 4:16:54 PM (5 years ago)

Author:

contact

Message:

Consume ring messages outside of writer goroutine

This fixes out-of-order JOIN and PRIVMSG messages.

Closes: ​https://todo.sr.ht/~emersion/soju/36

File:

• trunk/downstream.go (modified) (4 diffs)

trunk/downstream.go

@@ -52 +52 @@
 }}
 
-type ringMessage struct {
-        consumer     *RingConsumer
-        upstreamConn *upstreamConn
-}
-
 type downstreamConn struct {
-        id           uint64
-        net          net.Conn
-        irc          *irc.Conn
-        srv          *Server
-        logger       Logger
-        outgoing     chan *irc.Message
-        ringMessages chan ringMessage
-        closed       chan struct{}
+        id       uint64
+        net      net.Conn
+        irc      *irc.Conn
+        srv      *Server
+        logger   Logger
+        outgoing chan *irc.Message
+        closed   chan struct{}
 
         registered  bool
@@ -90 +84 @@
 func newDownstreamConn(srv *Server, netConn net.Conn, id uint64) *downstreamConn {
         dc := &downstreamConn{
-                id:           id,
-                net:          netConn,
-                irc:          irc.NewConn(netConn),
-                srv:          srv,
-                logger:       &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
-                outgoing:     make(chan *irc.Message, 64),
-                ringMessages: make(chan ringMessage),
-                closed:       make(chan struct{}),
-                caps:         make(map[string]bool),
-                ourMessages:  make(map[*irc.Message]struct{}),
+                id:          id,
+                net:         netConn,
+                irc:         irc.NewConn(netConn),
+                srv:         srv,
+                logger:      &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
+                outgoing:    make(chan *irc.Message, 64),
+                closed:      make(chan struct{}),
+                caps:        make(map[string]bool),
+                ourMessages: make(map[*irc.Message]struct{}),
         }
         dc.hostname = netConn.RemoteAddr().String()
@@ -258 +251 @@
                         }
                         err = dc.irc.WriteMessage(msg)
-                case ringMessage := <-dc.ringMessages:
-                        consumer, uc := ringMessage.consumer, ringMessage.upstreamConn
-                        for {
-                                msg := consumer.Peek()
-                                if msg == nil {
-                                        break
-                                }
-
-                                dc.lock.Lock()
-                                _, ours := dc.ourMessages[msg]
-                                delete(dc.ourMessages, msg)
-                                dc.lock.Unlock()
-                                if ours {
-                                        // The message comes from our connection, don't echo it
-                                        // back
-                                        consumer.Consume()
-                                        continue
-                                }
-
-                                msg = msg.Copy()
-                                switch msg.Command {
-                                case "PRIVMSG":
-                                        msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
-                                        msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
-                                default:
-                                        panic("expected to consume a PRIVMSG message")
-                                }
-                                if dc.srv.Debug {
-                                        dc.logger.Printf("sent: %v", msg)
-                                }
-                                err = dc.irc.WriteMessage(msg)
-                                if err != nil {
-                                        break
-                                }
-                                consumer.Consume()
-                        }
                 case <-dc.closed:
                         closed = true
@@ -775 +732 @@
                                         break
                                 }
-                                dc.ringMessages <- ringMessage{consumer, uc}
+
+                                for {
+                                        msg := consumer.Peek()
+                                        if msg == nil {
+                                                break
+                                        }
+
+                                        dc.lock.Lock()
+                                        _, ours := dc.ourMessages[msg]
+                                        delete(dc.ourMessages, msg)
+                                        dc.lock.Unlock()
+                                        if ours {
+                                                // The message comes from our connection, don't echo it
+                                                // back
+                                                consumer.Consume()
+                                                continue
+                                        }
+
+                                        msg = msg.Copy()
+                                        switch msg.Command {
+                                        case "PRIVMSG":
+                                                msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
+                                                msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
+                                        default:
+                                                panic("expected to consume a PRIVMSG message")
+                                        }
+
+                                        dc.SendMessage(msg)
+                                        consumer.Consume()
+                                }
                         case <-dc.closed:
                                 closed = true