Changeset 103 in code

Timestamp:

Mar 16, 2020, 11:44:59 AM (5 years ago)

Author:

contact

Message:

Per-user dispatcher goroutine

This allows message handlers to read upstream/downstream connection
information without causing any race condition.

References: ​https://todo.sr.ht/~emersion/soju/1

Location:

trunk

Files:

• downstream.go (modified) (3 diffs)
• server.go (modified) (1 diff)
• upstream.go (modified) (2 diffs)
• user.go (modified) (5 diffs)

trunk/downstream.go

@@ -192 +192 @@
 }
 
-func (dc *downstreamConn) readMessages() error {
+func (dc *downstreamConn) readMessages(ch chan<- downstreamIncomingMessage) error {
         dc.logger.Printf("new connection")
 
@@ -207 +207 @@
                 }
 
-                err = dc.handleMessage(msg)
-                if ircErr, ok := err.(ircError); ok {
-                        ircErr.Message.Prefix = dc.srv.prefix()
-                        dc.SendMessage(ircErr.Message)
-                } else if err != nil {
-                        return fmt.Errorf("failed to handle IRC command %q: %v", msg.Command, err)
-                }
-
-                if dc.isClosed() {
-                        return nil
-                }
+                ch <- downstreamIncomingMessage{msg, dc}
         }
 
@@ -485 +475 @@
 }
 
+func (dc *downstreamConn) runUntilRegistered() error {
+        for !dc.registered {
+                msg, err := dc.irc.ReadMessage()
+                if err == io.EOF {
+                        break
+                } else if err != nil {
+                        return fmt.Errorf("failed to read IRC command: %v", err)
+                }
+
+                err = dc.handleMessage(msg)
+                if ircErr, ok := err.(ircError); ok {
+                        ircErr.Message.Prefix = dc.srv.prefix()
+                        dc.SendMessage(ircErr.Message)
+                } else if err != nil {
+                        return fmt.Errorf("failed to handle IRC command %q: %v", msg, err)
+                }
+        }
+
+        return nil
+}
+
 func (dc *downstreamConn) handleMessageRegistered(msg *irc.Message) error {
         switch msg.Command {

trunk/server.go

@@ -115 +115 @@
                         s.lock.Unlock()
 
-                        if err := dc.readMessages(); err != nil {
-                                dc.logger.Printf("failed to handle messages: %v", err)
+                        if err := dc.runUntilRegistered(); err != nil {
+                                dc.logger.Print(err)
+                        } else {
+                                if err := dc.readMessages(dc.user.downstreamIncoming); err != nil {
+                                        dc.logger.Print(err)
+                                }
                         }
                         dc.Close()

trunk/upstream.go

@@ -660 +660 @@
 }
 
-func (uc *upstreamConn) readMessages() error {
+func (uc *upstreamConn) readMessages(ch chan<- upstreamIncomingMessage) error {
         for {
                 msg, err := uc.irc.ReadMessage()
@@ -673 +673 @@
                 }
 
-                if err := uc.handleMessage(msg); err != nil {
-                        uc.logger.Printf("failed to handle message %q: %v", msg, err)
-                }
+                ch <- upstreamIncomingMessage{msg, uc}
         }
 

trunk/user.go

@@ -4 +4 @@
         "sync"
         "time"
+
+        "gopkg.in/irc.v3"
 )
+
+type upstreamIncomingMessage struct {
+        msg *irc.Message
+        uc  *upstreamConn
+}
+
+type downstreamIncomingMessage struct {
+        msg *irc.Message
+        dc  *downstreamConn
+}
 
 type network struct {
@@ -41 +53 @@
                 net.user.lock.Unlock()
 
-                if err := uc.readMessages(); err != nil {
+                if err := uc.readMessages(net.user.upstreamIncoming); err != nil {
                         uc.logger.Printf("failed to handle messages: %v", err)
                 }
@@ -56 +68 @@
         srv *Server
 
+        upstreamIncoming   chan upstreamIncomingMessage
+        downstreamIncoming chan downstreamIncomingMessage
+
         lock            sync.Mutex
         networks        []*network
@@ -63 +78 @@
 func newUser(srv *Server, record *User) *user {
         return &user{
-                User: *record,
-                srv:  srv,
+                User:               *record,
+                srv:                srv,
+                upstreamIncoming:   make(chan upstreamIncomingMessage, 64),
+                downstreamIncoming: make(chan downstreamIncomingMessage, 64),
         }
 }
@@ -120 +137 @@
         }
         u.lock.Unlock()
+
+        for {
+                select {
+                case upstreamMsg := <-u.upstreamIncoming:
+                        msg, uc := upstreamMsg.msg, upstreamMsg.uc
+                        if err := uc.handleMessage(msg); err != nil {
+                                uc.logger.Printf("failed to handle message %q: %v", msg, err)
+                        }
+                case downstreamMsg := <-u.downstreamIncoming:
+                        msg, dc := downstreamMsg.msg, downstreamMsg.dc
+                        err := dc.handleMessage(msg)
+                        if ircErr, ok := err.(ircError); ok {
+                                ircErr.Message.Prefix = dc.srv.prefix()
+                                dc.SendMessage(ircErr.Message)
+                        } else if err != nil {
+                                dc.logger.Printf("failed to handle message %q: %v", msg, err)
+                                dc.Close()
+                        }
+                }
+        }
 }