socket: Move to a timing-out send method that reduces goroutines and ensures QUIT/ERROR are sent

This commit is contained in:
Daniel Oaks 2017-04-18 20:29:00 +10:00
parent 067f982da4
commit f7a4f5d956
2 changed files with 35 additions and 22 deletions

View File

@ -436,9 +436,13 @@ func (client *Client) ChangeNickname(nickname string) error {
// Quit sends the given quit message to the client (but does not destroy them). // Quit sends the given quit message to the client (but does not destroy them).
func (client *Client) Quit(message string) { func (client *Client) Quit(message string) {
if !client.quitMessageSent { if !client.quitMessageSent {
client.Send(nil, client.nickMaskString, "QUIT", message) quitMsg := ircmsg.MakeMessage(nil, client.nickMaskString, "QUIT", message)
client.Send(nil, "", "ERROR", message) quitLine, _ := quitMsg.Line()
client.socket.Write("\r\n")
errorMsg := ircmsg.MakeMessage(nil, "", "ERROR", message)
errorLine, _ := errorMsg.Line()
client.socket.FinalData = quitLine + errorLine
client.quitMessageSent = true client.quitMessageSent = true
} }
} }

View File

@ -31,6 +31,7 @@ type Socket struct {
reader *bufio.Reader reader *bufio.Reader
MaxSendQBytes uint64 MaxSendQBytes uint64
FinalData string // what to send when we die
lineToSendExists chan bool lineToSendExists chan bool
linesToSend []string linesToSend []string
@ -54,8 +55,8 @@ func (socket *Socket) Close() {
} }
socket.Closed = true socket.Closed = true
// force close loop to happen // force close loop to happen if it hasn't already
go socket.fillLineToSendExists(true) go socket.timedFillLineToSendExists(200 * time.Millisecond)
} }
// CertFP returns the fingerprint of the certificate provided by the client. // CertFP returns the fingerprint of the certificate provided by the client.
@ -119,15 +120,21 @@ func (socket *Socket) Write(data string) error {
socket.linesToSendMutex.Lock() socket.linesToSendMutex.Lock()
socket.linesToSend = append(socket.linesToSend, data) socket.linesToSend = append(socket.linesToSend, data)
socket.linesToSendMutex.Unlock() socket.linesToSendMutex.Unlock()
go socket.fillLineToSendExists(false)
if !socket.Closed {
go socket.timedFillLineToSendExists(15 * time.Second)
}
return nil return nil
} }
// fillLineToSendExists only exists because you can't goroutine single statements. // timedFillLineToSendExists either sends the note or times out.
func (socket *Socket) fillLineToSendExists(force bool) { func (socket *Socket) timedFillLineToSendExists(duration time.Duration) {
if force || !socket.Closed { select {
socket.lineToSendExists <- true case socket.lineToSendExists <- true:
// passed data successfully
case <-time.After(duration):
// timed out send
} }
} }
@ -142,12 +149,13 @@ func (socket *Socket) RunSocketWriter() {
// check if we're closed // check if we're closed
if socket.Closed { if socket.Closed {
socket.linesToSendMutex.Unlock()
break break
} }
// check number of lines to send // check whether new lines actually exist or not
if len(socket.linesToSend) < 1 { if len(socket.linesToSend) < 1 {
fmt.Println("No line to send found on socket writer") socket.linesToSendMutex.Unlock()
continue continue
} }
@ -156,22 +164,19 @@ func (socket *Socket) RunSocketWriter() {
for _, line := range socket.linesToSend { for _, line := range socket.linesToSend {
sendQBytes += uint64(len(line)) sendQBytes += uint64(len(line))
if socket.MaxSendQBytes < sendQBytes { if socket.MaxSendQBytes < sendQBytes {
socket.linesToSendMutex.Unlock()
break break
} }
} }
if socket.MaxSendQBytes < sendQBytes { if socket.MaxSendQBytes < sendQBytes {
socket.conn.Write([]byte("\r\nERROR :SendQ Exceeded\r\n")) socket.FinalData = "\r\nERROR :SendQ Exceeded\r\n"
fmt.Println("SendQ exceeded, disconnected client") socket.linesToSendMutex.Unlock()
break break
} }
// get data // get all existing data
data := socket.linesToSend[0] data := strings.Join(socket.linesToSend, "")
if len(socket.linesToSend) > 1 { socket.linesToSend = []string{}
socket.linesToSend = socket.linesToSend[1:]
} else {
socket.linesToSend = []string{}
}
socket.linesToSendMutex.Unlock() socket.linesToSendMutex.Unlock()
@ -190,10 +195,14 @@ func (socket *Socket) RunSocketWriter() {
break break
} }
} }
socket.conn.Close()
if !socket.Closed { if !socket.Closed {
socket.Closed = true socket.Closed = true
} }
// write error lines
if 0 < len(socket.FinalData) {
socket.conn.Write([]byte(socket.FinalData))
}
socket.conn.Close()
// empty the lineToSendExists channel // empty the lineToSendExists channel
for 0 < len(socket.lineToSendExists) { for 0 < len(socket.lineToSendExists) {