<commit_before>package fluentd_forwarder

import (
	"bytes"
	"github.com/ugorji/go/codec"
	logging "github.com/op/go-logging"
	"net"
	"reflect"
	"sync"
	"sync/atomic"
	"time"
	"io"
	"os"
	"math/rand"
	"unsafe"
)

var randSource = rand.NewSource(time.Now().UnixNano())

type ForwardOutput struct {
	logger            *logging.Logger
	codec             *codec.MsgpackHandle
	bind              string
	retryInterval     time.Duration
	connectionTimeout time.Duration
	writeTimeout      time.Duration
	enc               *codec.Encoder
	conn              net.Conn
	flushInterval     time.Duration
	wg                sync.WaitGroup
	journalGroup      JournalGroup
	journal           Journal
	emitterChan       chan FluentRecordSet
	spoolerShutdownChan chan struct{}
	isShuttingDown    unsafe.Pointer
}

func encodeRecordSet(encoder *codec.Encoder, recordSet FluentRecordSet) error {
	v := []interface{}{recordSet.Tag, recordSet.Records}
	err := encoder.Encode(v)
	if err != nil {
		return err
	}
	return err
}

func (output *ForwardOutput) ensureConnected() error {
	if output.conn == nil {
		output.logger.Notice("Connecting to %s...", output.bind)
		conn, err := net.DialTimeout("tcp", output.bind, output.connectionTimeout)
		if err != nil {
			output.logger.Error("Failed to connect to %s (reason: %s)", output.bind, err.Error())
			return err
		} else {
			output.conn = conn
		}
	}
	return nil
}

func (output *ForwardOutput) sendBuffer(buf []byte) error {
	for len(buf) > 0 {
		if atomic.LoadPointer(&output.isShuttingDown) != unsafe.Pointer(uintptr(0)) {
			break
		}
		err := output.ensureConnected()
		if err != nil {
			output.logger.Info("Will be retried in %s", output.retryInterval.String())
			time.Sleep(output.retryInterval)
			continue
		}
		startTime := time.Now()
		if output.writeTimeout == 0 {
			output.conn.SetWriteDeadline(time.Time {})
		} else {
			output.conn.SetWriteDeadline(startTime.Add(output.writeTimeout))
		}
		n, err := output.conn.Write(buf)
		buf = buf[n:]
		if err != nil {
			output.logger.Error("Failed to flush buffer (reason: %s, left: %d bytes)", err.Error(), len(buf))
			err_, ok := err.(net.Error)
			if !ok || (!err_.Timeout() && !err_.Temporary()) {
				return err
			}
		}
		if n > 0 {
			elapsed := time.Now().Sub(startTime)
			output.logger.Info("Forwarded %d bytes in %f seconds (%d bytes left)\n", n, elapsed.Seconds(), len(buf))
		}
	}
	return nil
}

func (output *ForwardOutput) spawnSpooler() {
	output.logger.Notice("Spawning spooler")
	output.wg.Add(1)
	go func() {
		ticker := time.NewTicker(output.flushInterval)
		defer func () {
			ticker.Stop()
			output.journal.Dispose()<commit_msg>Fix error recovery scheme of the emitter.<commit_after>package fluentd_forwarder

import (
	"bytes"
	"github.com/ugorji/go/codec"
	logging "github.com/op/go-logging"
	"net"
	"reflect"
	"sync"
	"sync/atomic"
	"time"
	"io"
	"os"
	"math/rand"
	"unsafe"
)

var randSource = rand.NewSource(time.Now().UnixNano())

type ForwardOutput struct {
	logger            *logging.Logger
	codec             *codec.MsgpackHandle
	bind              string
	retryInterval     time.Duration
	connectionTimeout time.Duration
	writeTimeout      time.Duration
	enc               *codec.Encoder
	conn              net.Conn
	flushInterval     time.Duration
	wg                sync.WaitGroup
	journalGroup      JournalGroup
	journal           Journal
	emitterChan       chan FluentRecordSet
	spoolerShutdownChan chan struct{}
	isShuttingDown    unsafe.Pointer
}

func encodeRecordSet(encoder *codec.Encoder, recordSet FluentRecordSet) error {
	v := []interface{}{recordSet.Tag, recordSet.Records}
	err := encoder.Encode(v)
	if err != nil {
		return err
	}
	return err
}

func (output *ForwardOutput) ensureConnected() error {
	if output.conn == nil {
		output.logger.Notice("Connecting to %s...", output.bind)
		conn, err := net.DialTimeout("tcp", output.bind, output.connectionTimeout)
		if err != nil {
			output.logger.Error("Failed to connect to %s (reason: %s)", output.bind, err.Error())
			return err
		} else {
			output.conn = conn
		}
	}
	return nil
}

func (output *ForwardOutput) sendBuffer(buf []byte) error {
	for len(buf) > 0 {
		if atomic.LoadPointer(&output.isShuttingDown) != unsafe.Pointer(uintptr(0)) {
			break
		}
		err := output.ensureConnected()
		if err != nil {
			output.logger.Info("Will be retried in %s", output.retryInterval.String())
			time.Sleep(output.retryInterval)
			continue
		}
		startTime := time.Now()
		if output.writeTimeout == 0 {
			output.conn.SetWriteDeadline(time.Time {})
		} else {
			output.conn.SetWriteDeadline(startTime.Add(output.writeTimeout))
		}
		n, err := output.conn.Write(buf)
		buf = buf[n:]
		if err != nil {
			output.logger.Error("Failed to flush buffer (reason: %s, left: %d bytes)", err.Error(), len(buf))
			err_, ok := err.(net.Error)
			if !ok || (!err_.Timeout() && !err_.Temporary()) {
				output.conn.Close()
				output.conn = nil
				continue
			}
		}
		if n > 0 {
			elapsed := time.Now().Sub(startTime)
			output.logger.Info("Forwarded %d bytes in %f seconds (%d bytes left)\n", n, elapsed.Seconds(), len(buf))
		}
	}
	return nil
}

func (output *ForwardOutput) spawnSpooler() {
	output.logger.Notice("Spawning spooler")
	output.wg.Add(1)
	go func() {
		ticker := time.NewTicker(output.flushInterval)
		defer func () {
			ticker.Stop()
			output.journal.Dispose()
{'fraction_non_alphanumeric': 0.1353116834639791, 'fraction_numerical': 0.001866368047779022, 'mean_word_length': 3.2872, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '6105531', 'n_tokens_mistral': 1958, 'n_tokens_neox': 1794, 'n_words': 415}