<commit_before>package tcp

import (
	"errors"
	"math"
	"network/ipv4"
	"sync"
	"time"

	"github.com/hsheth2/logs"
	"github.com/hsheth2/notifiers"
)

type TCB struct {
	read             chan *packet        // input
	writer           ipv4.Writer         // output
	ipAddress        *ipv4.Address       // destination ip address
	srcIP            *ipv4.Address       // src ip address
	lport, rport     uint16              // ports
	seqNum           uint32              // seq number (SND.NXT)
	ackNum           uint32              // ack number (RCV.NXT)
	seqAckMutex      *sync.RWMutex       // protects the seqNum and ackNum
	state            fsmState            // from the FSM
	timeWaitRestart  chan bool           // signals when the time_wait timer should restart
	stateUpdate      *sync.Cond          // signals when the state is changed
	kind             tcbParentType       // type (server or client)
	serverParent     *Server             // the parent server
	curWindow        uint16              // the current window size
	windowMutex      *sync.RWMutex       // protects access to the window size
	sendBuffer       []byte              // a buffer of bytes that need to be sent
	urgSendBuffer    []byte              // buffer of urgent data TODO urg data later
	sendBufferUpdate *sync.Cond          // notifies of send buffer updates
	stopSending      bool                // if the send function is allowed
	sendFinished     *notifiers.Notifier // broadcast when done sending
	recvBuffer       []byte              // bytes to received but not yet pushed
	pushBuffer       []byte              // bytes to push to client
	pushSignal       *sync.Cond          // signals upon push
	resendDelay      time.Duration       // the delay before resending
	iss              uint32              // the initial snd seq number
	irs              uint32              // the initial rcv seq number
	recentAckNum     uint32              // the last ack received (also SND.UNA)
	recentAckUpdate  *notifiers.Notifier // signals changes in recentAckNum
	maxSegSize       uint16              // MSS (MTU)
}

func newTCB(local, remote uint16, dstIP *ipv4.Address, read chan *packet, write ipv4.Writer, kind tcbParentType) (*TCB, error) {
	//ch logs.Trace.Println("New_TCB")

	seq, err := genRandSeqNum()
	if err != nil {
		logs.Error.Fatal(err)
		return nil, err
	}

	c := &TCB{
		lport:            local,
		rport:            remote,
		ipAddress:        dstIP,
		srcIP:            ipv4.GlobalRoutingTable.Query(dstIP),
		read:             read,
		writer:           write,
		seqNum:           seq,
		ackNum:           uint32(0), // Always 0 at start
		seqAckMutex:      &sync.RWMutex{},
		state:            fsmClosed,
		stateUpdate:      sync.NewCond(&sync.Mutex{}),
		timeWaitRestart:  make(chan bool, 1),
		kind:             kind,
		serverParent:     nil,
		curWindow:        43690, // TODO calc using http://ithitman.blogspot.com/2013/02/understanding-tcp-window-window-scaling.html
		windowMutex:      &sync.RWMutex{},
		sendBufferUpdate: sync.NewCond(&sync.Mutex{}),
		stopSending:      false,
		sendFinished:     notifiers.NewNotifier(),
		pushSignal:       sync.NewCond(&sync.Mutex{}),
		resendDelay:      250 * time.Millisecond,
		iss:              seq,
		irs:              0,
		recentAckNum:     0,
		recentAckUpdate:  notifiers.NewNotifier(),
		maxSegSize:       ipv4.IPMTU - basicHeaderSize,
	}
	////ch logs.Trace.Println("Starting the packet dealer")

	go c.packetSender()
	go c.packetDealer()

	return c, nil
}

func (c *TCB) Send(data []byte) error { // a blocking send call
	c.sendBufferUpdate.L.Lock()
	defer c.sendBufferUpdate.L.Unlock()
	if c.stopSending {
		return errors.New("Sending is not allowed anymore")
	}

	c.sendBuffer = append(c.sendBuffer, data...)
	c.sendBufferUpdate.Broadcast()
	return nil
}

func (c *TCB) Recv(num uint64) ([]byte, error) { // blocking recv call TODO add timeout
	// TODO check if conn is closed
	c.pushSignal.L.Lock()
	defer c.pushSignal.L.Unlock()
	for {
		//ch logs.Trace.Println(c.Hash(), "Attempting to read off of pushBuffer")
		//ch logs.Trace.Println(c.Hash(), "Amt of data on pushBuffer:", len(c.pushBuffer))
		amt := min(num, uint64(len(c.pushBuffer)))
		if amt != 0 {
			data := c.pushBuffer[:amt]
			c.pushBuffer = c.pushBuffer[amt:]
			return data, nil
		}
		if c.getState() == fsmClosed || c.getState() == fsmLastAck || c.getState() == fsmCloseWait {
			return nil, errors.New("Cannot reveive, losed")
		}
		//ch logs.Trace.Println(c.Hash(), "Waiting for push signal")
		c.pushSignal.Wait() // wait for a push
	}
	//return nil, errors.New("Read failed")
}

func (c *TCB) Close() error {
	//ch logs.Trace.Println(c.Hash(), "Closing TCB with lport:", c.lport)

	// block all future sends
	c.sendBufferUpdate.L.Lock()<commit_msg>Close check in TCB Receive<commit_after>package tcp

import (
	"errors"
	"math"
	"network/ipv4"
	"sync"
	"time"

	"github.com/hsheth2/logs"
	"github.com/hsheth2/notifiers"
)

type TCB struct {
	read             chan *packet        // input
	writer           ipv4.Writer         // output
	ipAddress        *ipv4.Address       // destination ip address
	srcIP            *ipv4.Address       // src ip address
	lport, rport     uint16              // ports
	seqNum           uint32              // seq number (SND.NXT)
	ackNum           uint32              // ack number (RCV.NXT)
	seqAckMutex      *sync.RWMutex       // protects the seqNum and ackNum
	state            fsmState            // from the FSM
	timeWaitRestart  chan bool           // signals when the time_wait timer should restart
	stateUpdate      *sync.Cond          // signals when the state is changed
	kind             tcbParentType       // type (server or client)
	serverParent     *Server             // the parent server
	curWindow        uint16              // the current window size
	windowMutex      *sync.RWMutex       // protects access to the window size
	sendBuffer       []byte              // a buffer of bytes that need to be sent
	urgSendBuffer    []byte              // buffer of urgent data TODO urg data later
	sendBufferUpdate *sync.Cond          // notifies of send buffer updates
	stopSending      bool                // if the send function is allowed
	sendFinished     *notifiers.Notifier // broadcast when done sending
	recvBuffer       []byte              // bytes to received but not yet pushed
	pushBuffer       []byte              // bytes to push to client
	pushSignal       *sync.Cond          // signals upon push
	resendDelay      time.Duration       // the delay before resending
	iss              uint32              // the initial snd seq number
	irs              uint32              // the initial rcv seq number
	recentAckNum     uint32              // the last ack received (also SND.UNA)
	recentAckUpdate  *notifiers.Notifier // signals changes in recentAckNum
	maxSegSize       uint16              // MSS (MTU)
}

func newTCB(local, remote uint16, dstIP *ipv4.Address, read chan *packet, write ipv4.Writer, kind tcbParentType) (*TCB, error) {
	//ch logs.Trace.Println("New_TCB")

	seq, err := genRandSeqNum()
	if err != nil {
		logs.Error.Fatal(err)
		return nil, err
	}

	c := &TCB{
		lport:            local,
		rport:            remote,
		ipAddress:        dstIP,
		srcIP:            ipv4.GlobalRoutingTable.Query(dstIP),
		read:             read,
		writer:           write,
		seqNum:           seq,
		ackNum:           uint32(0), // Always 0 at start
		seqAckMutex:      &sync.RWMutex{},
		state:            fsmClosed,
		stateUpdate:      sync.NewCond(&sync.Mutex{}),
		timeWaitRestart:  make(chan bool, 1),
		kind:             kind,
		serverParent:     nil,
		curWindow:        43690, // TODO calc using http://ithitman.blogspot.com/2013/02/understanding-tcp-window-window-scaling.html
		windowMutex:      &sync.RWMutex{},
		sendBufferUpdate: sync.NewCond(&sync.Mutex{}),
		stopSending:      false,
		sendFinished:     notifiers.NewNotifier(),
		pushSignal:       sync.NewCond(&sync.Mutex{}),
		resendDelay:      250 * time.Millisecond,
		iss:              seq,
		irs:              0,
		recentAckNum:     0,
		recentAckUpdate:  notifiers.NewNotifier(),
		maxSegSize:       ipv4.IPMTU - basicHeaderSize,
	}
	////ch logs.Trace.Println("Starting the packet dealer")

	go c.packetSender()
	go c.packetDealer()

	return c, nil
}

func (c *TCB) Send(data []byte) error { // a blocking send call
	c.sendBufferUpdate.L.Lock()
	defer c.sendBufferUpdate.L.Unlock()
	if c.stopSending {
		return errors.New("Sending is not allowed anymore")
	}

	c.sendBuffer = append(c.sendBuffer, data...)
	c.sendBufferUpdate.Broadcast()
	return nil
}

func (c *TCB) Recv(num uint64) ([]byte, error) { // blocking recv call TODO add timeout
	c.pushSignal.L.Lock()
	defer c.pushSignal.L.Unlock()
	for {
		//ch logs.Trace.Println(c.Hash(), "Attempting to read off of pushBuffer")
		//ch logs.Trace.Println(c.Hash(), "Amt of data on pushBuffer:", len(c.pushBuffer))
		amt := min(num, uint64(len(c.pushBuffer)))
		if amt != 0 {
			data := c.pushBuffer[:amt]
			c.pushBuffer = c.pushBuffer[amt:]
			return data, nil
		}
		if c.getState() == fsmClosed || c.getState() == fsmLastAck || c.getStat() == fsmCloseWait {
			return nil, errors.New("Cannot reveive, losed")
		}
		//ch logs.Trace.Println(c.Hash(), "Waiting for push signal")
		c.pushSignal.Wait() // wait for a push
	}
	//return nil, errors.New("Read failed")
}

func (c *TCB) Close() error {
	//ch logs.Trace.Println(c.Hash(), "Closing TCB with lport:", c.lport)

	// block all future sends
	c.sendBufferUpdate.L.Lock()
{'fraction_non_alphanumeric': 0.10800041941910454, 'fraction_numerical': 0.011324315822585718, 'mean_word_length': 2.140599275600922, 'pattern_counts': {'":': 0, '<': 3, '<?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': '1393632', 'n_tokens_mistral': 3162, 'n_tokens_neox': 3017, 'n_words': 853}