From fa5d4be71838aafa80b4fad124af94f02330e8b9 Mon Sep 17 00:00:00 2001 From: Shivaram Lingamneni Date: Thu, 15 Mar 2018 19:11:29 -0400 Subject: [PATCH 1/5] refactor irc.Socket --- irc/socket.go | 175 +++++++++++++++++++------------------------------- 1 file changed, 67 insertions(+), 108 deletions(-) diff --git a/irc/socket.go b/irc/socket.go index e9122086..dceff9a6 100644 --- a/irc/socket.go +++ b/irc/socket.go @@ -9,6 +9,7 @@ import ( "crypto/sha256" "crypto/tls" "encoding/hex" + "errors" "io" "net" "strings" @@ -18,24 +19,25 @@ import ( var ( handshakeTimeout, _ = time.ParseDuration("5s") + errSendQExceeded = errors.New("SendQ exceeded") ) // Socket represents an IRC socket. type Socket struct { + sync.Mutex + conn net.Conn reader *bufio.Reader MaxSendQBytes uint64 - closed bool - closedMutex sync.Mutex - - finalData string // what to send when we die - finalDataMutex sync.Mutex - + // coordination system for asynchronous writes + buffer []byte lineToSendExists chan bool - linesToSend []string - linesToSendMutex sync.Mutex + + closed bool + sendQExceeded bool + finalData string // what to send when we die } // NewSocket returns a new Socket. @@ -44,21 +46,24 @@ func NewSocket(conn net.Conn, maxSendQBytes uint64) Socket { conn: conn, reader: bufio.NewReader(conn), MaxSendQBytes: maxSendQBytes, - lineToSendExists: make(chan bool), + lineToSendExists: make(chan bool, 1), } } // Close stops a Socket from being able to send/receive any more data. func (socket *Socket) Close() { - socket.closedMutex.Lock() - defer socket.closedMutex.Unlock() - if socket.closed { - return - } - socket.closed = true + alreadyClosed := func() bool { + socket.Lock() + defer socket.Unlock() + result := socket.closed + socket.closed = true + return result + }() - // force close loop to happen if it hasn't already - go socket.timedFillLineToSendExists(200 * time.Millisecond) + if !alreadyClosed { + // force close loop to happen if it hasn't already + socket.Write("") + } } // CertFP returns the fingerprint of the certificate provided by the client. @@ -114,124 +119,78 @@ func (socket *Socket) Read() (string, error) { } // Write sends the given string out of Socket. -func (socket *Socket) Write(data string) error { - if socket.IsClosed() { - return io.EOF +func (socket *Socket) Write(data string) (err error) { + socket.Lock() + defer socket.Unlock() + + if socket.closed { + err = io.EOF + } else if uint64(len(data)+len(socket.buffer)) > socket.MaxSendQBytes { + socket.sendQExceeded = true + err = errSendQExceeded + } else { + socket.buffer = append(socket.buffer, data...) } - socket.linesToSendMutex.Lock() - socket.linesToSend = append(socket.linesToSend, data) - socket.linesToSendMutex.Unlock() - - go socket.timedFillLineToSendExists(15 * time.Second) - - return nil -} - -// timedFillLineToSendExists either sends the note or times out. -func (socket *Socket) timedFillLineToSendExists(duration time.Duration) { - lineToSendTimeout := time.NewTimer(duration) - defer lineToSendTimeout.Stop() - select { - case socket.lineToSendExists <- true: - // passed data successfully - case <-lineToSendTimeout.C: - // timed out send + // this can generate a spurious wakeup, since we are racing against the channel read, + // but since we are holding the mutex, we are not racing against the other writes + // and therefore we cannot miss a wakeup or block + if len(socket.lineToSendExists) == 0 { + socket.lineToSendExists <- true } + + return } // SetFinalData sets the final data to send when the SocketWriter closes. func (socket *Socket) SetFinalData(data string) { - socket.finalDataMutex.Lock() + socket.Lock() + defer socket.Unlock() socket.finalData = data - socket.finalDataMutex.Unlock() } // IsClosed returns whether the socket is closed. func (socket *Socket) IsClosed() bool { - socket.closedMutex.Lock() - defer socket.closedMutex.Unlock() + socket.Lock() + defer socket.Unlock() return socket.closed } // RunSocketWriter starts writing messages to the outgoing socket. func (socket *Socket) RunSocketWriter() { - for { + localBuffer := make([]byte, 0) + shouldStop := false + for !shouldStop { // wait for new lines select { case <-socket.lineToSendExists: - socket.linesToSendMutex.Lock() + // retrieve the buffered data, clear the buffer + socket.Lock() + localBuffer = append(localBuffer, socket.buffer...) + socket.buffer = socket.buffer[:0] + socket.Unlock() - // check if we're closed - if socket.IsClosed() { - socket.linesToSendMutex.Unlock() - break - } + _, err := socket.conn.Write(localBuffer) + localBuffer = localBuffer[:0] - // check whether new lines actually exist or not - if len(socket.linesToSend) < 1 { - socket.linesToSendMutex.Unlock() - continue - } - - // check sendq - var sendQBytes uint64 - for _, line := range socket.linesToSend { - sendQBytes += uint64(len(line)) - if socket.MaxSendQBytes < sendQBytes { - // don't unlock mutex because this break is just to escape this for loop - break - } - } - if socket.MaxSendQBytes < sendQBytes { - socket.SetFinalData("\r\nERROR :SendQ Exceeded\r\n") - socket.linesToSendMutex.Unlock() - break - } - - // get all existing data - data := strings.Join(socket.linesToSend, "") - socket.linesToSend = []string{} - - socket.linesToSendMutex.Unlock() - - // write data - if 0 < len(data) { - _, err := socket.conn.Write([]byte(data)) - if err != nil { - break - } - } - } - if socket.IsClosed() { - // error out or we've been closed - break + socket.Lock() + shouldStop = (err != nil) || socket.closed || socket.sendQExceeded + socket.Unlock() } } - // force closure of socket - socket.closedMutex.Lock() - if !socket.closed { - socket.closed = true - } - socket.closedMutex.Unlock() - // write error lines - socket.finalDataMutex.Lock() - if 0 < len(socket.finalData) { - socket.conn.Write([]byte(socket.finalData)) + // mark the socket closed (if someone hasn't already), then write error lines + socket.Lock() + socket.closed = true + finalData := socket.finalData + if socket.sendQExceeded { + finalData = "\r\nERROR :SendQ Exceeded\r\n" + } + socket.Unlock() + if finalData != "" { + socket.conn.Write([]byte(finalData)) } - socket.finalDataMutex.Unlock() // close the connection socket.conn.Close() - - // empty the lineToSendExists channel - for 0 < len(socket.lineToSendExists) { - <-socket.lineToSendExists - } -} - -// WriteLine writes the given line out of Socket. -func (socket *Socket) WriteLine(line string) error { - return socket.Write(line + "\r\n") } From 0a432c9d99a4cdcb82654af2b3bb7c6ab9e1d5eb Mon Sep 17 00:00:00 2001 From: Shivaram Lingamneni Date: Fri, 16 Mar 2018 12:39:11 -0400 Subject: [PATCH 2/5] do an actual nonblocking send instead of the len() trick --- irc/socket.go | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/irc/socket.go b/irc/socket.go index dceff9a6..9d3282ab 100644 --- a/irc/socket.go +++ b/irc/socket.go @@ -121,8 +121,6 @@ func (socket *Socket) Read() (string, error) { // Write sends the given string out of Socket. func (socket *Socket) Write(data string) (err error) { socket.Lock() - defer socket.Unlock() - if socket.closed { err = io.EOF } else if uint64(len(data)+len(socket.buffer)) > socket.MaxSendQBytes { @@ -131,12 +129,13 @@ func (socket *Socket) Write(data string) (err error) { } else { socket.buffer = append(socket.buffer, data...) } + socket.Unlock() - // this can generate a spurious wakeup, since we are racing against the channel read, - // but since we are holding the mutex, we are not racing against the other writes - // and therefore we cannot miss a wakeup or block - if len(socket.lineToSendExists) == 0 { - socket.lineToSendExists <- true + // notify the consumer that data is available + select { + case socket.lineToSendExists <- true: + default: + // a notification is already in the queue, this is fine } return From 8fd144662721c10842db0ee56ef8b4f76370bee2 Mon Sep 17 00:00:00 2001 From: Shivaram Lingamneni Date: Fri, 16 Mar 2018 18:16:04 -0400 Subject: [PATCH 3/5] tweak: clean up Socket.Close() --- irc/socket.go | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/irc/socket.go b/irc/socket.go index 9d3282ab..aa178530 100644 --- a/irc/socket.go +++ b/irc/socket.go @@ -52,18 +52,11 @@ func NewSocket(conn net.Conn, maxSendQBytes uint64) Socket { // Close stops a Socket from being able to send/receive any more data. func (socket *Socket) Close() { - alreadyClosed := func() bool { - socket.Lock() - defer socket.Unlock() - result := socket.closed - socket.closed = true - return result - }() + socket.Lock() + socket.closed = true + socket.Unlock() - if !alreadyClosed { - // force close loop to happen if it hasn't already - socket.Write("") - } + socket.wakeWriter() } // CertFP returns the fingerprint of the certificate provided by the client. @@ -131,14 +124,17 @@ func (socket *Socket) Write(data string) (err error) { } socket.Unlock() - // notify the consumer that data is available + socket.wakeWriter() + return +} + +// wakeWriter wakes up the goroutine that actually performs the write, without blocking +func (socket *Socket) wakeWriter() { + // nonblocking send to the channel, no-op if it's full select { case socket.lineToSendExists <- true: default: - // a notification is already in the queue, this is fine } - - return } // SetFinalData sets the final data to send when the SocketWriter closes. From d1f5c59eef10937370fc82cf3eac039742e03cb8 Mon Sep 17 00:00:00 2001 From: Shivaram Lingamneni Date: Sat, 17 Mar 2018 21:32:12 -0400 Subject: [PATCH 4/5] fix #190 --- irc/client.go | 10 ++++++++-- irc/config.go | 5 +++-- irc/errors.go | 1 + irc/getters.go | 9 +++++++++ irc/server.go | 13 ++----------- irc/socket.go | 19 +++++++++++-------- 6 files changed, 34 insertions(+), 23 deletions(-) diff --git a/irc/client.go b/irc/client.go index 67b95a10..2cce97be 100644 --- a/irc/client.go +++ b/irc/client.go @@ -86,7 +86,9 @@ type Client struct { // NewClient returns a client with all the appropriate info setup. func NewClient(server *Server, conn net.Conn, isTLS bool) *Client { now := time.Now() - socket := NewSocket(conn, server.MaxSendQBytes) + limits := server.Limits() + fullLineLenLimit := limits.LineLen.Tags + limits.LineLen.Rest + socket := NewSocket(conn, fullLineLenLimit*2, server.MaxSendQBytes()) go socket.RunSocketWriter() client := &Client{ atime: now, @@ -230,7 +232,11 @@ func (client *Client) run() { line, err = client.socket.Read() if err != nil { - client.Quit("connection closed") + quitMessage := "connection closed" + if err == errReadQ { + quitMessage = "readQ exceeded" + } + client.Quit(quitMessage) break } diff --git a/irc/config.go b/irc/config.go index cacec5a4..f5b849e2 100644 --- a/irc/config.go +++ b/irc/config.go @@ -208,7 +208,7 @@ type Config struct { ProxyAllowedFrom []string `yaml:"proxy-allowed-from"` WebIRC []webircConfig `yaml:"webirc"` MaxSendQString string `yaml:"max-sendq"` - MaxSendQBytes uint64 + MaxSendQBytes int ConnectionLimiter connection_limits.LimiterConfig `yaml:"connection-limits"` ConnectionThrottler connection_limits.ThrottlerConfig `yaml:"connection-throttling"` } @@ -520,10 +520,11 @@ func LoadConfig(filename string) (config *Config, err error) { } } - config.Server.MaxSendQBytes, err = bytefmt.ToBytes(config.Server.MaxSendQString) + maxSendQBytes, err := bytefmt.ToBytes(config.Server.MaxSendQString) if err != nil { return nil, fmt.Errorf("Could not parse maximum SendQ size (make sure it only contains whole numbers): %s", err.Error()) } + config.Server.MaxSendQBytes = int(maxSendQBytes) // get language files config.Languages.Data = make(map[string]languages.LangData) diff --git a/irc/errors.go b/irc/errors.go index c6a5fe87..7909099b 100644 --- a/irc/errors.go +++ b/irc/errors.go @@ -40,6 +40,7 @@ var ( var ( errNoPeerCerts = errors.New("Client did not provide a certificate") errNotTLS = errors.New("Not a TLS connection") + errReadQ = errors.New("ReadQ Exceeded") ) // String Errors diff --git a/irc/getters.go b/irc/getters.go index 29e32361..56b97057 100644 --- a/irc/getters.go +++ b/irc/getters.go @@ -6,8 +6,17 @@ package irc import ( "github.com/oragono/oragono/irc/isupport" "github.com/oragono/oragono/irc/modes" + "sync/atomic" ) +func (server *Server) MaxSendQBytes() int { + return int(atomic.LoadUint64(&server.maxSendQBytes)) +} + +func (server *Server) SetMaxSendQBytes(m int) { + atomic.StoreUint64(&server.maxSendQBytes, uint64(m)) +} + func (server *Server) ISupport() *isupport.List { server.configurableStateMutex.RLock() defer server.configurableStateMutex.RUnlock() diff --git a/irc/server.go b/irc/server.go index 9e6e0a22..260396eb 100644 --- a/irc/server.go +++ b/irc/server.go @@ -109,7 +109,7 @@ type Server struct { limits Limits listeners map[string]*ListenerWrapper logger *logger.Manager - MaxSendQBytes uint64 + maxSendQBytes uint64 monitorManager *MonitorManager motdLines []string name string @@ -932,16 +932,7 @@ func (server *Server) applyConfig(config *Config, initial bool) error { server.configurableStateMutex.Unlock() // set new sendqueue size - if config.Server.MaxSendQBytes != server.MaxSendQBytes { - server.configurableStateMutex.Lock() - server.MaxSendQBytes = config.Server.MaxSendQBytes - server.configurableStateMutex.Unlock() - - // update on all clients - for _, sClient := range server.clients.AllClients() { - sClient.socket.MaxSendQBytes = config.Server.MaxSendQBytes - } - } + server.SetMaxSendQBytes(config.Server.MaxSendQBytes) // set RPL_ISUPPORT var newISupportReplies [][]string diff --git a/irc/socket.go b/irc/socket.go index aa178530..0f98a6f9 100644 --- a/irc/socket.go +++ b/irc/socket.go @@ -29,7 +29,7 @@ type Socket struct { conn net.Conn reader *bufio.Reader - MaxSendQBytes uint64 + maxSendQBytes int // coordination system for asynchronous writes buffer []byte @@ -41,11 +41,11 @@ type Socket struct { } // NewSocket returns a new Socket. -func NewSocket(conn net.Conn, maxSendQBytes uint64) Socket { +func NewSocket(conn net.Conn, maxReadQBytes int, maxSendQBytes int) Socket { return Socket{ conn: conn, - reader: bufio.NewReader(conn), - MaxSendQBytes: maxSendQBytes, + reader: bufio.NewReaderSize(conn, maxReadQBytes), + maxSendQBytes: maxSendQBytes, lineToSendExists: make(chan bool, 1), } } @@ -92,10 +92,13 @@ func (socket *Socket) Read() (string, error) { return "", io.EOF } - lineBytes, err := socket.reader.ReadBytes('\n') + lineBytes, isPrefix, err := socket.reader.ReadLine() + if isPrefix { + return "", errReadQ + } // convert bytes to string - line := string(lineBytes[:]) + line := string(lineBytes) // read last message properly (such as ERROR/QUIT/etc), just fail next reads/writes if err == io.EOF { @@ -108,7 +111,7 @@ func (socket *Socket) Read() (string, error) { return "", err } - return strings.TrimRight(line, "\r\n"), nil + return line, nil } // Write sends the given string out of Socket. @@ -116,7 +119,7 @@ func (socket *Socket) Write(data string) (err error) { socket.Lock() if socket.closed { err = io.EOF - } else if uint64(len(data)+len(socket.buffer)) > socket.MaxSendQBytes { + } else if len(data)+len(socket.buffer) > socket.maxSendQBytes { socket.sendQExceeded = true err = errSendQExceeded } else { From a8b952da7726357de2ec070c3d28681883f87c9f Mon Sep 17 00:00:00 2001 From: Shivaram Lingamneni Date: Mon, 19 Mar 2018 00:24:20 -0400 Subject: [PATCH 5/5] store maxSendQBytes in a uint32 to avoid alignment problems The sync.atomic documentation says: "On both ARM and x86-32, it is the caller's responsibility to arrange for 64-bit alignment of 64-bit words accessed atomically." --- irc/getters.go | 4 ++-- irc/server.go | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/irc/getters.go b/irc/getters.go index 56b97057..7168a329 100644 --- a/irc/getters.go +++ b/irc/getters.go @@ -10,11 +10,11 @@ import ( ) func (server *Server) MaxSendQBytes() int { - return int(atomic.LoadUint64(&server.maxSendQBytes)) + return int(atomic.LoadUint32(&server.maxSendQBytes)) } func (server *Server) SetMaxSendQBytes(m int) { - atomic.StoreUint64(&server.maxSendQBytes, uint64(m)) + atomic.StoreUint32(&server.maxSendQBytes, uint32(m)) } func (server *Server) ISupport() *isupport.List { diff --git a/irc/server.go b/irc/server.go index 260396eb..c96c6959 100644 --- a/irc/server.go +++ b/irc/server.go @@ -109,7 +109,7 @@ type Server struct { limits Limits listeners map[string]*ListenerWrapper logger *logger.Manager - maxSendQBytes uint64 + maxSendQBytes uint32 monitorManager *MonitorManager motdLines []string name string