3
0
mirror of https://github.com/ergochat/ergo.git synced 2024-11-22 11:59:40 +01:00
ergo/irc/socket.go

245 lines
6.2 KiB
Go
Raw Normal View History

// Copyright (c) 2012-2014 Jeremy Latt
2017-03-27 14:15:02 +02:00
// Copyright (c) 2016-2017 Daniel Oaks <daniel@danieloaks.net>
// released under the MIT license
2014-02-14 04:37:16 +01:00
package irc
import (
2018-03-16 00:11:29 +01:00
"errors"
"io"
"sync"
2019-04-15 17:13:13 +02:00
2021-05-25 06:34:38 +02:00
"github.com/ergochat/ergo/irc/utils"
2014-02-14 04:37:16 +01:00
)
var (
2020-03-22 14:45:33 +01:00
errSendQExceeded = errors.New("SendQ exceeded")
sendQExceededMessage = []byte("\r\nERROR :SendQ Exceeded\r\n")
)
2016-06-15 13:21:45 +02:00
// Socket represents an IRC socket.
2014-02-14 04:37:16 +01:00
type Socket struct {
2018-03-16 00:11:29 +01:00
sync.Mutex
2020-05-05 04:29:10 +02:00
conn IRCConn
2018-03-18 02:32:12 +01:00
maxSendQBytes int
2017-04-18 14:26:01 +02:00
// this is a trylock enforcing that only one goroutine can write to `conn` at a time
2019-04-15 17:13:13 +02:00
writerSemaphore utils.Semaphore
2018-03-16 00:11:29 +01:00
2018-04-26 21:32:32 +02:00
buffers [][]byte
totalLength int
2018-03-16 00:11:29 +01:00
closed bool
sendQExceeded bool
finalData []byte // what to send when we die
finalized bool
2014-02-14 04:37:16 +01:00
}
2016-06-15 13:21:45 +02:00
// NewSocket returns a new Socket.
2020-05-05 04:29:10 +02:00
func NewSocket(conn IRCConn, maxSendQBytes int) *Socket {
result := Socket{
conn: conn,
maxSendQBytes: maxSendQBytes,
writerSemaphore: utils.NewSemaphore(1),
2014-02-14 04:37:16 +01:00
}
return &result
2014-02-14 04:37:16 +01:00
}
2016-06-15 13:21:45 +02:00
// Close stops a Socket from being able to send/receive any more data.
2014-02-14 04:37:16 +01:00
func (socket *Socket) Close() {
2018-03-16 23:16:04 +01:00
socket.Lock()
socket.closed = true
socket.Unlock()
2018-04-16 01:05:22 +02:00
socket.wakeWriter()
2014-02-14 04:37:16 +01:00
}
2016-06-15 13:21:45 +02:00
// Read returns a single IRC line from a Socket.
func (socket *Socket) Read() (string, error) {
2020-06-23 09:18:49 +02:00
// immediately fail if Close() has been called, even if there's
// still data in a bufio.Reader or websocket buffer:
2017-04-18 14:26:01 +02:00
if socket.IsClosed() {
2016-06-15 13:21:45 +02:00
return "", io.EOF
}
2020-05-05 04:29:10 +02:00
lineBytes, err := socket.conn.ReadLine()
2018-03-18 02:32:12 +01:00
line := string(lineBytes)
2014-02-14 04:37:16 +01:00
2016-06-15 13:21:45 +02:00
if err == io.EOF {
socket.Close()
}
2020-06-23 09:18:49 +02:00
return line, err
2016-06-15 13:21:45 +02:00
}
2014-02-20 20:15:42 +01:00
// Write sends the given string out of Socket. Requirements:
// 1. MUST NOT block for macroscopic amounts of time
// 2. MUST NOT reorder messages
// 3. MUST provide mutual exclusion for socket.conn.Write
// 4. SHOULD NOT tie up additional goroutines, beyond the one blocked on socket.conn.Write
2018-04-26 21:32:32 +02:00
func (socket *Socket) Write(data []byte) (err error) {
if len(data) == 0 {
return
}
2018-03-16 00:11:29 +01:00
socket.Lock()
if socket.closed {
err = io.EOF
} else {
2018-04-26 21:32:32 +02:00
prospectiveLen := socket.totalLength + len(data)
if prospectiveLen > socket.maxSendQBytes {
socket.sendQExceeded = true
socket.closed = true
2018-04-26 21:32:32 +02:00
err = errSendQExceeded
} else {
socket.buffers = append(socket.buffers, data)
socket.totalLength = prospectiveLen
}
2018-03-16 00:11:29 +01:00
}
socket.Unlock()
2016-06-15 13:21:45 +02:00
2018-04-16 01:05:22 +02:00
socket.wakeWriter()
2018-03-16 23:16:04 +01:00
return
}
// BlockingWrite sends the given string out of Socket. Requirements:
// 1. MUST block until the message is sent
// 2. MUST bypass sendq (calls to BlockingWrite cannot, on their own, cause a sendq overflow)
// 3. MUST provide mutual exclusion for socket.conn.Write
// 4. MUST respect the same ordering guarantees as Write (i.e., if a call to Write that sends
// message m1 happens-before a call to BlockingWrite that sends message m2,
// m1 must be sent on the wire before m2
// Callers MUST be writing to the client's socket from the client's own goroutine;
// other callers must use the nonblocking Write call instead. Otherwise, a client
// with a slow/unreliable connection risks stalling the progress of the system as a whole.
func (socket *Socket) BlockingWrite(data []byte) (err error) {
if len(data) == 0 {
return
}
// after releasing the semaphore, we must check for fresh data, same as `send`
defer func() {
if socket.readyToWrite() {
socket.wakeWriter()
}
}()
// blocking acquire of the trylock
socket.writerSemaphore.Acquire()
defer socket.writerSemaphore.Release()
// first, flush any buffered data, to preserve the ordering guarantees
closed := socket.performWrite()
if closed {
return io.EOF
}
err = socket.conn.WriteLine(data)
if err != nil {
socket.finalize()
}
return
}
2018-04-16 01:05:22 +02:00
// wakeWriter starts the goroutine that actually performs the write, without blocking
func (socket *Socket) wakeWriter() {
2018-04-25 02:34:28 +02:00
if socket.writerSemaphore.TryAcquire() {
2018-04-16 01:05:22 +02:00
// acquired the trylock; send() will release it
go socket.send()
}
2018-04-25 02:34:28 +02:00
// else: do nothing, the holder will check for more data after releasing it
2018-04-16 01:05:22 +02:00
}
2017-04-18 14:26:01 +02:00
// SetFinalData sets the final data to send when the SocketWriter closes.
func (socket *Socket) SetFinalData(data []byte) {
2018-03-16 00:11:29 +01:00
socket.Lock()
defer socket.Unlock()
2017-04-18 14:26:01 +02:00
socket.finalData = data
}
// IsClosed returns whether the socket is closed.
func (socket *Socket) IsClosed() bool {
2018-03-16 00:11:29 +01:00
socket.Lock()
defer socket.Unlock()
2017-04-18 14:26:01 +02:00
return socket.closed
}
// is there data to write?
func (socket *Socket) readyToWrite() bool {
socket.Lock()
defer socket.Unlock()
// on the first time observing socket.closed, we still have to write socket.finalData
return !socket.finalized && (socket.totalLength > 0 || socket.closed)
}
// send actually writes messages to socket.Conn; it may block
func (socket *Socket) send() {
for {
2018-04-16 01:05:22 +02:00
// we are holding the trylock: actually do the write
socket.performWrite()
// surrender the trylock, avoiding a race where a write comes in after we've
// checked readyToWrite() and it returned false, but while we still hold the trylock:
2018-04-25 02:34:28 +02:00
socket.writerSemaphore.Release()
2018-04-16 01:05:22 +02:00
// check if more data came in while we held the trylock:
if !socket.readyToWrite() {
return
}
2018-04-25 02:34:28 +02:00
if !socket.writerSemaphore.TryAcquire() {
2018-04-16 01:05:22 +02:00
// failed to acquire; exit and wait for the holder to observe readyToWrite()
// after releasing it
return
}
2018-04-25 02:34:28 +02:00
// got the lock again, loop back around and write
}
}
// write the contents of the buffer, then see if we need to close
// returns whether we closed
func (socket *Socket) performWrite() (closed bool) {
// retrieve the buffered data, clear the buffer
socket.Lock()
2018-04-26 21:32:32 +02:00
buffers := socket.buffers
socket.buffers = nil
socket.totalLength = 0
closed = socket.closed
socket.Unlock()
var err error
if 0 < len(buffers) {
err = socket.conn.WriteLines(buffers)
}
closed = closed || err != nil
if closed {
socket.finalize()
}
return
}
2017-04-18 14:26:01 +02:00
// mark closed and send final data. you must be holding the semaphore to call this:
func (socket *Socket) finalize() {
2018-03-16 00:11:29 +01:00
// mark the socket closed (if someone hasn't already), then write error lines
socket.Lock()
socket.closed = true
finalized := socket.finalized
socket.finalized = true
2018-03-16 00:11:29 +01:00
finalData := socket.finalData
if socket.sendQExceeded {
finalData = sendQExceededMessage
2018-03-16 00:11:29 +01:00
}
socket.Unlock()
if finalized {
return
}
if len(finalData) != 0 {
socket.conn.WriteLine(finalData)
}
2017-04-18 14:26:01 +02:00
// close the connection
socket.conn.Close()
2014-02-14 04:37:16 +01:00
}