<commit_before>// Share server connections between different clients.

package main

import (
	"sync"
	"time"
)

// Maximum number of connections to a server.
const maxServerConnCnt = 20

// Store each server's connections in separate channels, getting
// connections for different servers can be done in parallel.
type ConnPool struct {
	idleConn map[string]chan *serverConn
	muxConn  chan *serverConn // connections support multiplexing
	sync.RWMutex
}

var connPool = &ConnPool{
	idleConn: map[string]chan *serverConn{},
	muxConn:  make(chan *serverConn, maxServerConnCnt*2),
}

func init() {
	// make sure hostPort here won't match any actual hostPort
	go closeStaleServerConn(connPool.muxConn, "muxConn")
}

func getConnFromChan(ch chan *serverConn) (sv *serverConn) {
	for {
		select {
		case sv = <-ch:
			if sv.mayBeClosed() {
				sv.Close()
				continue
			}
			return sv
		default:
			return nil
		}
	}
}

func putConnToChan(sv *serverConn, ch chan *serverConn, chname string) {
	select {
	case ch <- sv:
		debug.Printf("connPool channel %s: put conn\n", chname)
		return
	default:
		// Simply close the connection if can't put into channel immediately.
		// A better solution would remove old connections from the channel and
		// add the new one. But's it's more complicated and this should happen
		// rarely.
		debug.Printf("connPool channel %s: full", chname)
		sv.Close()
	}
}

func (cp *ConnPool) Get(hostPort string, asDirect bool) (sv *serverConn) {
	// Get from site specific connection first.
	// Direct connection are all site specific, so must use site specific
	// first to avoid using parent proxy for direct sites.
	cp.RLock()
	ch := cp.idleConn[hostPort]
	cp.RUnlock()

	if ch != nil {
		sv = getConnFromChan(ch)
	}
	if sv != nil {
		debug.Printf("connPool %s: get conn\n", hostPort)
		return sv
	}

	// All mulplexing connections are for blocked sites,
	// so for direct sites we should stop here.
	if asDirect {
		return nil
	}

	sv = getConnFromChan(cp.muxConn)
	if bool(debug) && sv != nil {
		debug.Println("connPool mux: get conn", hostPort)
	}
	return sv
}

func (cp *ConnPool) Put(sv *serverConn) {
	// Multiplexing connections.
	switch sv.Conn.(type) {
	case httpConn, cowConn:
		putConnToChan(sv, cp.muxConn, "muxConn")
		return
	}

	// Site specific connections.
	cp.RLock()
	ch := cp.idleConn[sv.hostPort]
	cp.RUnlock()

	if ch == nil {
		debug.Printf("connPool %s: new channel\n", sv.hostPort)
		ch = make(chan *serverConn, maxServerConnCnt)
		ch <- sv
		cp.Lock()
		cp.idleConn[sv.hostPort] = ch<commit_msg>Don't return mux conn from pool unless alwayUseProxy.<commit_after>// Share server connections between different clients.

package main

import (
	"sync"
	"time"
)

// Maximum number of connections to a server.
const maxServerConnCnt = 20

// Store each server's connections in separate channels, getting
// connections for different servers can be done in parallel.
type ConnPool struct {
	idleConn map[string]chan *serverConn
	muxConn  chan *serverConn // connections support multiplexing
	sync.RWMutex
}

var connPool = &ConnPool{
	idleConn: map[string]chan *serverConn{},
	muxConn:  make(chan *serverConn, maxServerConnCnt*2),
}

func init() {
	// make sure hostPort here won't match any actual hostPort
	go closeStaleServerConn(connPool.muxConn, "muxConn")
}

func getConnFromChan(ch chan *serverConn) (sv *serverConn) {
	for {
		select {
		case sv = <-ch:
			if sv.mayBeClosed() {
				sv.Close()
				continue
			}
			return sv
		default:
			return nil
		}
	}
}

func putConnToChan(sv *serverConn, ch chan *serverConn, chname string) {
	select {
	case ch <- sv:
		debug.Printf("connPool channel %s: put conn\n", chname)
		return
	default:
		// Simply close the connection if can't put into channel immediately.
		// A better solution would remove old connections from the channel and
		// add the new one. But's it's more complicated and this should happen
		// rarely.
		debug.Printf("connPool channel %s: full", chname)
		sv.Close()
	}
}

func (cp *ConnPool) Get(hostPort string, asDirect bool) (sv *serverConn) {
	// Get from site specific connection first.
	// Direct connection are all site specific, so must use site specific
	// first to avoid using parent proxy for direct sites.
	cp.RLock()
	ch := cp.idleConn[hostPort]
	cp.RUnlock()

	if ch != nil {
		sv = getConnFromChan(ch)
	}
	if sv != nil {
		debug.Printf("connPool %s: get conn\n", hostPort)
		return sv
	}

	// All mulplexing connections are for blocked sites,
	// so for direct sites we should stop here.
	if asDirect && !config.AlwaysProxy {
		return nil
	}

	sv = getConnFromChan(cp.muxConn)
	if bool(debug) && sv != nil {
		debug.Println("connPool mux: get conn", hostPort)
	}
	return sv
}

func (cp *ConnPool) Put(sv *serverConn) {
	// Multiplexing connections.
	switch sv.Conn.(type) {
	case httpConn, cowConn:
		putConnToChan(sv, cp.muxConn, "muxConn")
		return
	}

	// Site specific connections.
	cp.RLock()
	ch := cp.idleConn[sv.hostPort]
	cp.RUnlock()

	if ch == nil {
		debug.Printf("connPool %s: new channel\n", sv.hostPort)
		ch = make(chan *serverConn, maxServerConnCnt)
		ch <- sv
		cp.Lock()
		cp.idleConn[sv.hostPort] = ch
{'fraction_non_alphanumeric': 0.11048652839697616, 'fraction_numerical': 0.0011630160883892228, 'mean_word_length': 3.6486486486486487, 'pattern_counts': {'":': 0, '<': 9, '<?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': '3891883', 'n_tokens_mistral': 1831, 'n_tokens_neox': 1718, 'n_words': 571}