<commit_before>// A simple multiplexing protocol.
package frames

import (
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"time"
)

// Error returned when we've run out of channels.
var ChannelsExhausted = errors.New("channels exhausted")

func (f *frameConnection) connHandler() {
	defer f.c.Close()
	go f.readLoop()
	f.writeLoop()
}

type newconn struct {
	c net.Conn
	e error
}

type frameConnection struct {
	c           net.Conn
	channels    map[uint16]*frameChannel
	newConns    chan newconn
	egress      chan *FramePacket
	closeMarker chan bool
	lastChid    uint16
}

func (f *frameConnection) nextId() (uint16, error) {
	f.lastChid++
	for i := 0; i < 0xffff; i++ {
		if _, taken := f.channels[f.lastChid]; !taken {
			return f.lastChid, nil
		}
		f.lastChid++
	}
	return 0, ChannelsExhausted
}

func (f *frameConnection) Accept() (net.Conn, error) {
	c, ok := <-f.newConns
	if !ok {
		return nil, io.EOF
	}
	return c.c, c.e
}

func (f *frameConnection) Close() error {
	select {
	case <-f.closeMarker:
		return errors.New("already closed")
	default:
	}

	for _, c := range f.channels {
		c.Close()
	}
	close(f.closeMarker)
	return f.c.Close()
}

func (f *frameConnection) Addr() net.Addr {
	return f.c.LocalAddr()
}

func (f *frameConnection) openChannel(pkt *FramePacket) {
	chid, err := f.nextId()
	response := &FramePacket{
		Cmd:     pkt.Cmd,
		Status:  FrameSuccess,
		Channel: chid,
	}
	nc := newconn{}
	if err == nil {
		f.channels[chid] = &frameChannel{
			conn:        f,
			channel:     chid,
			incoming:    make(chan []byte, 1024),
			current:     nil,
			closeMarker: make(chan bool),
		}
		nc.c = f.channels[chid]
	} else {
		response.Status = FrameError
		nc.e = err
	}
	f.egress <- response
	f.newConns <- nc
}

func (f *frameConnection) closeChannel(pkt *FramePacket) {
	ch := f.channels[pkt.Channel]
	if ch == nil {
		log.Printf("Closing a closed channel: %v", pkt)
		return
	}
	ch.Close()
	delete(f.channels, pkt.Channel)
}

func (f *frameConnection) gotData(pkt *FramePacket) {
	ch := f.channels[pkt.Channel]
	if ch == nil {
		log.Printf("Write to nonexistent channel on %v %v",
			f.c.RemoteAddr(), pkt)
		return
	}
	ch.incoming <- pkt.Data
}

func (f *frameConnection) readLoop() {
	defer f.Close()
	for {
		hdr := make([]byte, minPktLen)
		_, err := io.ReadFull(f.c, hdr)
		if err != nil {
			log.Printf("Channel read error: %v", err)
			return
		}
		pkt := PacketFromHeader(hdr)
		_, err = io.ReadFull(f.c, pkt.Data)
		if err != nil {
			log.Printf("Channel read error: %v", err)
			return
		}

		switch pkt.Cmd {<commit_msg>Don't log EOF when reading a channel header.<commit_after>// A simple multiplexing protocol.
package frames

import (
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"time"
)

// Error returned when we've run out of channels.
var ChannelsExhausted = errors.New("channels exhausted")

func (f *frameConnection) connHandler() {
	defer f.c.Close()
	go f.readLoop()
	f.writeLoop()
}

type newconn struct {
	c net.Conn
	e error
}

type frameConnection struct {
	c           net.Conn
	channels    map[uint16]*frameChannel
	newConns    chan newconn
	egress      chan *FramePacket
	closeMarker chan bool
	lastChid    uint16
}

func (f *frameConnection) nextId() (uint16, error) {
	f.lastChid++
	for i := 0; i < 0xffff; i++ {
		if _, taken := f.channels[f.lastChid]; !taken {
			return f.lastChid, nil
		}
		f.lastChid++
	}
	return 0, ChannelsExhausted
}

func (f *frameConnection) Accept() (net.Conn, error) {
	c, ok := <-f.newConns
	if !ok {
		return nil, io.EOF
	}
	return c.c, c.e
}

func (f *frameConnection) Close() error {
	select {
	case <-f.closeMarker:
		return errors.New("already closed")
	default:
	}

	for _, c := range f.channels {
		c.Close()
	}
	close(f.closeMarker)
	return f.c.Close()
}

func (f *frameConnection) Addr() net.Addr {
	return f.c.LocalAddr()
}

func (f *frameConnection) openChannel(pkt *FramePacket) {
	chid, err := f.nextId()
	response := &FramePacket{
		Cmd:     pkt.Cmd,
		Status:  FrameSuccess,
		Channel: chid,
	}
	nc := newconn{}
	if err == nil {
		f.channels[chid] = &frameChannel{
			conn:        f,
			channel:     chid,
			incoming:    make(chan []byte, 1024),
			current:     nil,
			closeMarker: make(chan bool),
		}
		nc.c = f.channels[chid]
	} else {
		response.Status = FrameError
		nc.e = err
	}
	f.egress <- response
	f.newConns <- nc
}

func (f *frameConnection) closeChannel(pkt *FramePacket) {
	ch := f.channels[pkt.Channel]
	if ch == nil {
		log.Printf("Closing a closed channel: %v", pkt)
		return
	}
	ch.Close()
	delete(f.channels, pkt.Channel)
}

func (f *frameConnection) gotData(pkt *FramePacket) {
	ch := f.channels[pkt.Channel]
	if ch == nil {
		log.Printf("Write to nonexistent channel on %v %v",
			f.c.RemoteAddr(), pkt)
		return
	}
	ch.incoming <- pkt.Data
}

func (f *frameConnection) readLoop() {
	defer f.Close()
	for {
		hdr := make([]byte, minPktLen)
		_, err := io.ReadFull(f.c, hdr)
		if err != nil {
			if err != io.EOF {
				log.Printf("Channel header read error: %v", err)
			}
			return
		}
		pkt := PacketFromHeader(hdr)
		_, err = io.ReadFull(f.c, pkt.Data)
		if err != nil {
			log.Printf("Channel data read error: %v", err)
			return
		}

		switch pkt.Cmd {
{'fraction_non_alphanumeric': 0.14650667698858139, 'fraction_numerical': 0.005031933423650087, 'mean_word_length': 3.1711057304277643, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 3, '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': '6378666', 'n_tokens_mistral': 2126, 'n_tokens_neox': 1995, 'n_words': 522}