<commit_before>/*
Package inet handles connecting to an irc server and reading and writing to
the connection
*/
package inet

import (
	"bytes"
	"io"
	"log"
	"net"
	"sync"
	"time"
)

const (
	// bufferSize is the size of the buffer to be allocated for writes
	bufferSize = 16348
	// nBufferedWrites is how many writes can succeed before blocking
	nBufferedWrites = 25
	// defaultTimeScale is the default scale of the sleeps and timeouts.
	defaultTimeScale = time.Second
)

var (
	// pong allows replies from pong to write directly without waiting on sleeps
	pong = []byte("PONG")
)

// Format strings for errors and logging output
const (
	fmtDiscarded          = "(%v) <- (DISCARDED) %s\n"
	fmtWrite              = "(%v) <- %s\n"
	fmtWriteErr           = "(%v) <- (%v) %s\n"
	fmtRead               = "(%v) -> %s\n"
	fmtErrSiphonReadError = "inet: (%v) read socket error (%s)\n"
	fmtErrPumpReadError   = "inet: (%v) write socket error (%s)\n"
	fmtErrSiphonClosed    = "inet: (%v) siphon closed (%s)\n"
	fmtErrPumpClosed      = "inet: (%v) pump closed (%s)\n"
	errMsgShutdown        = "Shut Down"
)

// IrcClient represents a connection to an irc server. It uses a queueing system
// to throttle writes to the server. And it implements ReadWriteCloser interface
type IrcClient struct {
	isShutdown        bool
	isShutdownProtect sync.RWMutex

	conn        net.Conn
	siphonchan  chan []byte
	pumpchan    chan []byte
	pumpservice chan chan []byte
	killpump    chan int
	killsiphon  chan int
	queue       Queue

	// The name of the connection for logging
	name string

	// write throttling
	nThrottled int
	lastwrite  time.Time
	scale      time.Duration
	burst      int
	timeout    time.Duration
	step       time.Duration

	// buffering for io.Reader interface
	readbuf []byte
	pos     int
}

// CreateIrcClient initializes the required fields in the IrcClient
func CreateIrcClient(conn net.Conn, name string) *IrcClient {
	return &IrcClient{
		name:        name,
		conn:        conn,
		siphonchan:  make(chan []byte),
		pumpchan:    make(chan []byte),
		pumpservice: make(chan chan []byte),
		lastwrite:   time.Time{},
		scale:       defaultTimeScale,
	}
}

// CreateIrcClientFloodProtect creates an irc client and sets up the variables
// required for flood protection. The timeout and step variables are in seconds.
func CreateIrcClientFloodProtect(conn net.Conn, name string, burst, timeout,
	step int, scale time.Duration) *IrcClient {

	c := CreateIrcClient(conn, name)
	if scale != 0 {
		c.scale = scale
	}
	c.burst = burst
	c.timeout = time.Duration(timeout) * c.scale
	c.step = time.Duration(step) * c.scale
	return c
}

// SpawnWorkers creates two goroutines, one that is constantly reading using
// Siphon, and one that is constantly working on eliminating the write queue by
// writing. Also sets up the instances kill channels.
func (c *IrcClient) SpawnWorkers(pump, siphon bool) {
	c.isShutdownProtect.Lock()
	defer c.isShutdownProtect.Unlock()
	c.isShutdown = false

	if pump {
		c.killpump = make(chan int)
		go c.pump()
	}
	if siphon {
		c.killsiphon = make(chan int)
		go c.siphon()
	}
}

// calcSleepTime calculates the sleep time required by the flood protection
// given a time of write.
func (c *IrcClient) calcSleepTime(t time.Time) time.Duration {
	if c.timeout == 0 || c.step == 0 {
		return 0
	}

	dur := t.Sub(c.lastwrite)
	if dur < 0 {
		dur = 0
	}

	if dur > c.timeout {
		c.nThrottled = 1
	} else {
		if c.nThrottled >= c.burst {
			return c.step
		}
		c.nThrottled++
	}

	return 0
}

// pump enqueues the messages given to Write and writes them to the connection.
// It also sleeps a don't-get-glined amount of time between writes.
func (c *IrcClient) pump() {
	var err error
	var sleeper <-chan time.Time
	defer close(c.pumpservice)

	for err == nil {
		select {
		case c.pumpservice <- c.pumpchan:
			message := <-c.pumpchan
			if len(message) == 0 {
				break
			}

			if bytes.HasPrefix(message, pong) {
				if err = c.writeMessage(message); err != nil {
					break
				}
			} else if sleeper == nil {
				sleepTime := c.calcSleepTime(time.Now())
				if sleepTime == 0 {
					if err = c.writeMessage(message); err != nil {
						break
					}
				} else {
					c.queue.Enqueue(message)
					sleeper = time.After(sleepTime)
				}
			} else {
				c.queue.Enqueue(message)
			}
		case <-sleeper:
			if err = c.writeMessage(c.queue.Dequeue()); err != nil {
				break
			}
			if c.queue.length > 0 {
				sleepTime := c.calcSleepTime(time.Now())
				sleeper = time.After(sleepTime)
			} else {
				sleeper = nil
			}
		case <-c.killpump:
			log.Printf(fmtErrPumpClosed, c.name, errMsgShutdown)
			return
		}
	}

	<-c.killpump
}

// writeMessage writes a byte array out to the socket, sets the last write time.
func (c *IrcClient) writeMessage(msg []byte) error {
	var n int
	var err error
	for written := 0; written < len(msg); written += n {<commit_msg>Adding timer to send ping requests to the server every 30 seconds<commit_after>/*
Package inet handles connecting to an irc server and reading and writing to
the connection
*/
package inet

import (
	"bytes"
	"io"
	"log"
	"net"
	"sync"
	"time"
)

const (
	// bufferSize is the size of the buffer to be allocated for writes
	bufferSize = 16348
	// nBufferedWrites is how many writes can succeed before blocking
	nBufferedWrites = 25
	// defaultTimeScale is the default scale of the sleeps and timeouts.
	defaultTimeScale = time.Second
)

var (
	// pong allows replies from pong to write directly without waiting on sleeps
	pong = []byte("PONG")
)

// Format strings for errors and logging output
const (
	fmtDiscarded          = "(%v) <- (DISCARDED) %s\n"
	fmtWrite              = "(%v) <- %s\n"
	fmtWriteErr           = "(%v) <- (%v) %s\n"
	fmtRead               = "(%v) -> %s\n"
	fmtErrSiphonReadError = "inet: (%v) read socket error (%s)\n"
	fmtErrPumpReadError   = "inet: (%v) write socket error (%s)\n"
	fmtErrSiphonClosed    = "inet: (%v) siphon closed (%s)\n"
	fmtErrPumpClosed      = "inet: (%v) pump closed (%s)\n"
	errMsgShutdown        = "Shut Down"
)

// IrcClient represents a connection to an irc server. It uses a queueing system
// to throttle writes to the server. And it implements ReadWriteCloser interface
type IrcClient struct {
	isShutdown        bool
	isShutdownProtect sync.RWMutex

	conn        net.Conn
	siphonchan  chan []byte
	pumpchan    chan []byte
	pumpservice chan chan []byte
	killpump    chan int
	killsiphon  chan int
	queue       Queue

	// The name of the connection for logging
	name string

	// write throttling
	nThrottled int
	lastwrite  time.Time
	scale      time.Duration
	burst      int
	timeout    time.Duration
	step       time.Duration

	// buffering for io.Reader interface
	readbuf []byte
	pos     int
}

// CreateIrcClient initializes the required fields in the IrcClient
func CreateIrcClient(conn net.Conn, name string) *IrcClient {
	return &IrcClient{
		name:        name,
		conn:        conn,
		siphonchan:  make(chan []byte),
		pumpchan:    make(chan []byte),
		pumpservice: make(chan chan []byte),
		lastwrite:   time.Time{},
		scale:       defaultTimeScale,
	}
}

// CreateIrcClientFloodProtect creates an irc client and sets up the variables
// required for flood protection. The timeout and step variables are in seconds.
func CreateIrcClientFloodProtect(conn net.Conn, name string, burst, timeout,
	step int, scale time.Duration) *IrcClient {

	c := CreateIrcClient(conn, name)
	if scale != 0 {
		c.scale = scale
	}
	c.burst = burst
	c.timeout = time.Duration(timeout) * c.scale
	c.step = time.Duration(step) * c.scale
	return c
}

// SpawnWorkers creates two goroutines, one that is constantly reading using
// Siphon, and one that is constantly working on eliminating the write queue by
// writing. Also sets up the instances kill channels.
func (c *IrcClient) SpawnWorkers(pump, siphon bool) {
	c.isShutdownProtect.Lock()
	defer c.isShutdownProtect.Unlock()
	c.isShutdown = false

	if pump {
		c.killpump = make(chan int)
		go c.pump()
	}
	if siphon {
		c.killsiphon = make(chan int)
		go c.siphon()
	}
}

// calcSleepTime calculates the sleep time required by the flood protection
// given a time of write.
func (c *IrcClient) calcSleepTime(t time.Time) time.Duration {
	if c.timeout == 0 || c.step == 0 {
		return 0
	}

	dur := t.Sub(c.lastwrite)
	if dur < 0 {
		dur = 0
	}

	if dur > c.timeout {
		c.nThrottled = 1
	} else {
		if c.nThrottled >= c.burst {
			return c.step
		}
		c.nThrottled++
	}

	return 0
}

// pump enqueues the messages given to Write and writes them to the connection.
// It also sleeps a don't-get-glined amount of time between writes.
func (c *IrcClient) pump() {
	var err error
	var sleeper <-chan time.Time
	var pinger <- chan time.Time
	pinger = time.After(30 * time.Second)
	defer close(c.pumpservice)

	for err == nil {
		select {
		case c.pumpservice <- c.pumpchan:
			message := <-c.pumpchan
			if len(message) == 0 {
				break
			}

			if bytes.HasPrefix(message, pong) {
				if err = c.writeMessage(message); err != nil {
					break
				}
			} else if sleeper == nil {
				sleepTime := c.calcSleepTime(time.Now())
				if sleepTime == 0 {
					if err = c.writeMessage(message); err != nil {
						break
					}
				} else {
					c.queue.Enqueue(message)
					sleeper = time.After(sleepTime)
				}
			} else {
				c.queue.Enqueue(message)
			}
		case <-sleeper:
			if err = c.writeMessage(c.queue.Dequeue()); err != nil {
				break
			}
			if c.queue.length > 0 {
				sleepTime := c.calcSleepTime(time.Now())
				sleeper = time.After(sleepTime)
			} else {
				sleeper = nil
			}
		case <- pinger:
			if err = c.writeMessage([]byte("PING :ghostcheck\r\n")); err != nil {
				break
			}
			pinger = time.After(30 * time.Second)
		case <-c.killpump:
			log.Printf(fmtErrPumpClosed, c.name, errMsgShutdown)
			return
		}
	}

	<-c.killpump
}

// writeMessage writes a byte array out to the socket, sets the last write time.
func (c *IrcClient) writeMessage(msg []byte) error {
	var n int
	var err error
	for written := 0; written < len(msg); written += n {
{'fraction_non_alphanumeric': 0.10833998403830807, 'fraction_numerical': 0.004389465283320032, 'mean_word_length': 2.9375490966221522, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2587568', 'n_tokens_mistral': 3815, 'n_tokens_neox': 3438, 'n_words': 1108}