版本号修改为1.2.3

build/build.sh

@@ -54,10 +54,11 @@ function build() {
 	cp -R "$ROOT"/www "$DIST"/
 	cp -R "$ROOT"/pages "$DIST"/
 
-	# we support TOA on linux/amd64 only
+	# we support TOA on linux only
-	if [ "$OS" == "linux" ] && [ "$ARCH" == "amd64" ]
+	if [ "$OS" == "linux" ] && [ -f "${ROOT}/edge-toa/edge-toa-${ARCH}" ]
 	then
-		cp -R "$ROOT"/edge-toa "$DIST"
+		mkdir "$DIST/edge-toa"
+		cp "${ROOT}/edge-toa/edge-toa-${ARCH}" "$DIST/edge-toa/edge-toa"
 	fi
 
 	echo "building ..."

internal/caches/storage_file.go

@@ -198,6 +198,9 @@ func (this *FileStorage) UpdatePolicy(newPolicy *serverconfigs.HTTPCachePolicy)
 	if newPolicy.PersistenceAutoPurgeInterval != this.policy.PersistenceAutoPurgeInterval {
 		this.initPurgeTicker()
 	}
+
+	// reset ignored keys
+	this.ignoreKeys.Reset()
 }
 
 // Init 初始化
@@ -410,14 +413,14 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
 	}
 
 	// 是否已忽略
-	if this.ignoreKeys.Has(key) {
+	if maxSize > 0 && this.ignoreKeys.Has(types.String(maxSize)+"$"+key) {
 		return nil, ErrEntityTooLarge
 	}
 
 	// 先尝试内存缓存
 	// 我们限定仅小文件优先存在内存中
 	var maxMemorySize = FileToMemoryMaxSize
-	if maxSize > maxMemorySize {
+	if maxSize > 0 && maxSize < maxMemorySize {
 		maxMemorySize = maxSize
 	}
 	var memoryStorage = this.memoryStorage
@@ -642,7 +645,7 @@ func (this *FileStorage) openWriter(key string, expiredAt int64, status int, hea
 			sharedWritingFileKeyLocker.Unlock()
 		}), nil
 	} else {
-		return NewFileWriter(this, writer, key, expiredAt, metaHeaderSize, metaBodySize, -1, func() {
+		return NewFileWriter(this, writer, key, expiredAt, metaHeaderSize, metaBodySize, maxSize, func() {
 			sharedWritingFileKeyLocker.Lock()
 			delete(sharedWritingFileKeyMap, key)
 			if len(sharedWritingFileKeyMap) == 0 {
@@ -914,8 +917,8 @@ func (this *FileStorage) TotalMemorySize() int64 {
 }
 
 // IgnoreKey 忽略某个Key，即不缓存某个Key
-func (this *FileStorage) IgnoreKey(key string) {
+func (this *FileStorage) IgnoreKey(key string, maxSize int64) {
-	this.ignoreKeys.Push(key)
+	this.ignoreKeys.Push(types.String(maxSize) + "$" + key)
 }
 
 // CanSendfile 是否支持Sendfile

internal/caches/storage_file_test.go

@@ -42,7 +42,9 @@ func TestFileStorage_Init(t *testing.T) {
 
 	time.Sleep(2 * time.Second)
 	storage.purgeLoop()
-	t.Log(storage.list.(*FileList).total, "entries left")
+	t.Log(storage.list.(*FileList).Stat(func(hash string) bool {
+		return true
+	}))
 }
 
 func TestFileStorage_OpenWriter(t *testing.T) {

internal/caches/storage_interface.go

@@ -54,7 +54,7 @@ type StorageInterface interface {
 	AddToList(item *Item)
 
 	// IgnoreKey 忽略某个Key，即不缓存某个Key
-	IgnoreKey(key string)
+	IgnoreKey(key string, maxSize int64)
 
 	// CanSendfile 是否支持Sendfile
 	CanSendfile() bool

internal/caches/storage_memory.go

@@ -158,7 +158,7 @@ func (this *MemoryStorage) OpenReader(key string, useStale bool, isPartial bool)
 
 // OpenWriter 打开缓存写入器等待写入
 func (this *MemoryStorage) OpenWriter(key string, expiredAt int64, status int, headerSize int, bodySize int64, maxSize int64, isPartial bool) (Writer, error) {
-	if this.ignoreKeys.Has(key) {
+	if maxSize > 0 && this.ignoreKeys.Has(types.String(maxSize)+"$"+key) {
 		return nil, ErrEntityTooLarge
 	}
 
@@ -362,6 +362,9 @@ func (this *MemoryStorage) UpdatePolicy(newPolicy *serverconfigs.HTTPCachePolicy
 	if newPolicy.CapacityBytes() == 0 {
 		_ = this.CleanAll()
 	}
+
+	// reset ignored keys
+	this.ignoreKeys.Reset()
 }
 
 // CanUpdatePolicy 检查策略是否可以更新
@@ -392,8 +395,8 @@ func (this *MemoryStorage) TotalMemorySize() int64 {
 }
 
 // IgnoreKey 忽略某个Key，即不缓存某个Key
-func (this *MemoryStorage) IgnoreKey(key string) {
+func (this *MemoryStorage) IgnoreKey(key string, maxSize int64) {
-	this.ignoreKeys.Push(key)
+	this.ignoreKeys.Push(types.String(maxSize) + "$" + key)
 }
 
 // CanSendfile 是否支持Sendfile

internal/caches/writer_file.go

@@ -79,7 +79,7 @@ func (this *FileWriter) Write(data []byte) (n int, err error) {
 		err = ErrEntityTooLarge
 
 		if this.storage != nil {
-			this.storage.IgnoreKey(this.key)
+			this.storage.IgnoreKey(this.key, this.maxSize)
 		}
 	}
 

internal/caches/writer_memory.go

@@ -59,7 +59,7 @@ func (this *MemoryWriter) Write(data []byte) (n int, err error) {
 	// 检查尺寸
 	if this.maxSize > 0 && this.bodySize > this.maxSize {
 		err = ErrEntityTooLarge
-		this.storage.IgnoreKey(this.key)
+		this.storage.IgnoreKey(this.key, this.maxSize)
 		return len(data), err
 	}
 

internal/const/const.go

@@ -1,7 +1,7 @@
 package teaconst
 
 const (
-	Version = "1.2.1"
+	Version = "1.2.3"
 
 	ProductName = "Edge Node"
 	ProcessName = "edge-node"

internal/nodes/client_conn.go

@@ -129,9 +129,8 @@ func (this *ClientConn) Read(b []byte) (n int, err error) {
 	// 检测是否为超时错误
 	var isTimeout = err != nil && os.IsTimeout(err)
 	var isHandshakeError = isTimeout && !this.hasRead
-	if isTimeout {
+
-		_ = this.SetLinger(0)
+	if err != nil {
-	} else {
 		_ = this.SetLinger(nodeconfigs.DefaultTCPLinger)
 	}
 

internal/nodes/http_client_pool.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
 	"github.com/TeaOSLab/EdgeNode/internal/goman"
-	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/pires/go-proxyproto"
 	"golang.org/x/net/http2"
 	"net"
@@ -132,14 +131,8 @@ func (this *HTTPClientPool) Client(req *HTTPRequest,
 	var transport = &HTTPClientTransport{
 		Transport: &http.Transport{
 			DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
-				// 支持TOA的连接
-				conn, err := this.handleTOA(req, ctx, network, originAddr, connectionTimeout)
-				if conn != nil || err != nil {
-					return conn, err
-				}
-
 				// 普通的连接
-				conn, err = (&net.Dialer{
+				conn, err := (&net.Dialer{
 					Timeout:   connectionTimeout,
 					KeepAlive: 1 * time.Minute,
 				}).DialContext(ctx, network, originAddr)
@@ -215,38 +208,6 @@ func (this *HTTPClientPool) cleanClients() {
 	}
 }
 
-// 支持TOA
-func (this *HTTPClientPool) handleTOA(req *HTTPRequest, ctx context.Context, network string, originAddr string, connectionTimeout time.Duration) (net.Conn, error) {
-	// TODO 每个服务读取自身所属集群的TOA设置
-	var toaConfig = sharedTOAManager.Config()
-	if toaConfig != nil && toaConfig.IsOn {
-		var retries = 3
-		for i := 1; i <= retries; i++ {
-			var port = int(toaConfig.RandLocalPort())
-			// TODO 思考是否支持X-Real-IP/X-Forwarded-IP
-			err := sharedTOAManager.SendMsg("add:" + strconv.Itoa(port) + ":" + req.requestRemoteAddr(true))
-			if err != nil {
-				remotelogs.Error("TOA", "add failed: "+err.Error())
-			} else {
-				dialer := net.Dialer{
-					Timeout:   connectionTimeout,
-					KeepAlive: 1 * time.Minute,
-					LocalAddr: &net.TCPAddr{
-						Port: port,
-					},
-				}
-				conn, err := dialer.DialContext(ctx, network, originAddr)
-				// TODO 需要在合适的时机删除TOA记录
-				if err == nil || i == retries {
-					return conn, err
-				}
-			}
-		}
-	}
-
-	return nil, nil
-}
-
 // 支持PROXY Protocol
 func (this *HTTPClientPool) handlePROXYProtocol(conn net.Conn, req *HTTPRequest, proxyProtocol *serverconfigs.ProxyProtocolConfig) error {
 	if proxyProtocol != nil && proxyProtocol.IsOn && (proxyProtocol.Version == serverconfigs.ProxyProtocolVersion1 || proxyProtocol.Version == serverconfigs.ProxyProtocolVersion2) {

internal/nodes/http_request_cache.go

@@ -38,8 +38,13 @@ func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
 
 	// 添加 X-Cache Header
 	var addStatusHeader = this.web.Cache.AddStatusHeader
+	var cacheBypassDescription = ""
 	if addStatusHeader {
 		defer func() {
+			if len(cacheBypassDescription) > 0 {
+				this.writer.Header().Set("X-Cache", cacheBypassDescription)
+				return
+			}
 			var cacheStatus = this.varMapping["cache.status"]
 			if cacheStatus != "HIT" {
 				this.writer.Header().Set("X-Cache", cacheStatus)
@@ -94,6 +99,7 @@ func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
 	// 校验请求
 	if !this.cacheRef.MatchRequest(this.RawReq) {
 		this.cacheRef = nil
+		cacheBypassDescription = "BYPASS, not match"
 		return
 	}
 
@@ -106,6 +112,7 @@ func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
 	if this.cacheRef.EnableRequestCachePragma {
 		if this.RawReq.Header.Get("Cache-Control") == "no-cache" || this.RawReq.Header.Get("Pragma") == "no-cache" {
 			this.cacheRef = nil
+			cacheBypassDescription = "BYPASS, Cache-Control or Pragma"
 			return
 		}
 	}
@@ -119,6 +126,7 @@ func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
 	var key = this.Format(this.cacheRef.Key)
 	if len(key) == 0 {
 		this.cacheRef = nil
+		cacheBypassDescription = "BYPASS, empty key"
 		return
 	}
 	var method = this.Method()
@@ -134,6 +142,7 @@ func (this *HTTPRequest) doCacheRead(useStale bool) (shouldStop bool) {
 	var storage = caches.SharedManager.FindStorageWithPolicy(cachePolicy.Id)
 	if storage == nil {
 		this.cacheRef = nil
+		cacheBypassDescription = "BYPASS, no policy found"
 		return
 	}
 	this.writer.cacheStorage = storage

internal/nodes/http_request_user_agent.go

@@ -7,7 +7,7 @@ import (
 )
 
 func (this *HTTPRequest) doCheckUserAgent() (shouldStop bool) {
-	if this.web.UserAgent == nil {
+	if this.web.UserAgent == nil || !this.web.UserAgent.IsOn {
 		return
 	}
 

internal/nodes/node_ext.go

@@ -15,3 +15,7 @@ func (this *Node) reloadIPLibrary() {
 func (this *Node) notifyPlusChange() error {
 	return nil
 }
+
+func (this *Node) execTOAChangedTask() error {
+	return nil
+}

internal/nodes/node_tasks.go

@@ -90,6 +90,8 @@ func (this *Node) execTask(rpcClient *rpc.RPCClient, task *pb.NodeTask) error {
 		err = this.execUpdatingServersTask(rpcClient)
 	case "plusChanged":
 		err = this.notifyPlusChange()
+	case "toaChanged":
+		err = this.execTOAChangedTask()
 	default:
 		remotelogs.Error("NODE", "task '"+types.String(task.Id)+"', type '"+task.Type+"' has not been handled")
 	}

internal/nodes/toa_manager.go

@@ -1,117 +1,31 @@
+// Copyright 2023 GoEdge CDN goedge.cdn@gmail.com. All rights reserved. Official site: https://goedge.cn .
+//go:build !plus
+
 package nodes
 
-import (
+import "github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
-	"errors"
-	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
-	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
-	"github.com/TeaOSLab/EdgeNode/internal/events"
-	"github.com/TeaOSLab/EdgeNode/internal/goman"
-	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
-	executils "github.com/TeaOSLab/EdgeNode/internal/utils/exec"
-	"github.com/iwind/TeaGo/Tea"
-	"net"
-	"os"
-	"strings"
-	"time"
-)
 
 var sharedTOAManager = NewTOAManager()
 
-func init() {
-	if !teaconst.IsMain {
-		return
-	}
-
-	events.On(events.EventReload, func() {
-		err := sharedTOAManager.Run(sharedNodeConfig.TOA)
-		if err != nil {
-			remotelogs.Error("TOA", err.Error())
-		}
-	})
-}
-
 type TOAManager struct {
-	config *nodeconfigs.TOAConfig
-	pid    int
-	conn   net.Conn
 }
 
 func NewTOAManager() *TOAManager {
 	return &TOAManager{}
 }
 
-func (this *TOAManager) Run(config *nodeconfigs.TOAConfig) error {
+func (this *TOAManager) Apply(config *nodeconfigs.TOAConfig) error {
-	this.config = config
-
-	if this.pid > 0 {
-		remotelogs.Println("TOA", "stopping ...")
-		err := this.Quit()
-		if err != nil {
-			remotelogs.Error("TOA", "quit error: "+err.Error())
-		}
-		if this.conn != nil {
-			_ = this.conn.Close()
-		}
-		this.conn = nil
-		this.pid = 0
-	}
-
-	if !config.IsOn {
-		return nil
-	}
-
-	binPath := Tea.Root + "/edge-toa/edge-toa" // TODO 可以做成配置
-	_, err := os.Stat(binPath)
-	if err != nil {
-		return err
-	}
-	remotelogs.Println("TOA", "starting ...")
-	remotelogs.Println("TOA", "args: "+strings.Join(config.AsArgs(), " "))
-	cmd := executils.NewCmd(binPath, config.AsArgs()...)
-	err = cmd.Start()
-	if err != nil {
-		return err
-	}
-	var process = cmd.Process()
-	if process == nil {
-		return errors.New("start failed")
-	}
-	this.pid = process.Pid
-
-	goman.New(func() {
-		_ = cmd.Wait()
-	})
-
 	return nil
 }
 
 func (this *TOAManager) Config() *nodeconfigs.TOAConfig {
-	return this.config
+	return nil
 }
 
 func (this *TOAManager) Quit() error {
-	return this.SendMsg("quit:0")
+	return nil
 }
 
 func (this *TOAManager) SendMsg(msg string) error {
-	if this.config == nil {
+	return nil
-		return nil
-	}
-
-	if this.conn != nil {
-		_, err := this.conn.Write([]byte(msg + "\n"))
-		if err != nil {
-			_ = this.conn.Close()
-			this.conn = nil
-		}
-		return err
-	}
-
-	conn, err := net.DialTimeout("unix", this.config.SockFile(), 1*time.Second)
-	if err != nil {
-		return err
-	}
-	this.conn = conn
-	_, err = this.conn.Write([]byte(msg + "\n"))
-	return err
 }

internal/nodes/toa_manager_test.go

@@ -1,17 +0,0 @@
-package nodes
-
-import (
-	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
-	"testing"
-)
-
-func TestTOAManager_Run(t *testing.T) {
-	manager := NewTOAManager()
-	err := manager.Run(&nodeconfigs.TOAConfig{
-		IsOn: true,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	t.Log("ok")
-}

internal/utils/counters/counter_test.go

@@ -74,6 +74,20 @@ func TestCounter_GC2(t *testing.T) {
 	}
 }
 
+func TestCounterMemory(t *testing.T) {
+	var stat = &runtime.MemStats{}
+	runtime.ReadMemStats(stat)
+
+	var counter = counters.NewCounter()
+	for i := 0; i < 1e5; i++ {
+		counter.Increase(uint64(i), rands.Int(10, 300))
+	}
+
+	var stat1 = &runtime.MemStats{}
+	runtime.ReadMemStats(stat1)
+	t.Log((stat1.TotalAlloc-stat.TotalAlloc)/(1<<20), "MB")
+}
+
 func BenchmarkCounter_Increase(b *testing.B) {
 	runtime.GOMAXPROCS(4)
 

internal/utils/readers/reader_closer_tee.go

@@ -25,7 +25,7 @@ func (this *TeeReaderCloser) Read(p []byte) (n int, err error) {
 	n, err = this.r.Read(p)
 	if n > 0 {
 		_, wErr := this.w.Write(p[:n])
-		if err == nil && wErr != nil {
+		if (err == nil || err == io.EOF) && wErr != nil {
 			err = wErr
 		}
 	}

internal/utils/sets/set_fixed.go

@@ -13,8 +13,8 @@ type FixedSet struct {
 	maxSize int
 	locker  sync.RWMutex
 
-	m    map[interface{}]zero.Zero
+	m    map[any]zero.Zero
-	keys []interface{}
+	keys []any
 }
 
 func NewFixedSet(maxSize int) *FixedSet {
@@ -23,11 +23,11 @@ func NewFixedSet(maxSize int) *FixedSet {
 	}
 	return &FixedSet{
 		maxSize: maxSize,
-		m:       map[interface{}]zero.Zero{},
+		m:       map[any]zero.Zero{},
 	}
 }
 
-func (this *FixedSet) Push(item interface{}) {
+func (this *FixedSet) Push(item any) {
 	this.locker.Lock()
 	_, ok := this.m[item]
 	if !ok {
@@ -44,7 +44,7 @@ func (this *FixedSet) Push(item interface{}) {
 	this.locker.Unlock()
 }
 
-func (this *FixedSet) Has(item interface{}) bool {
+func (this *FixedSet) Has(item any) bool {
 	this.locker.RLock()
 	defer this.locker.RUnlock()
 
@@ -60,7 +60,7 @@ func (this *FixedSet) Size() int {
 
 func (this *FixedSet) Reset() {
 	this.locker.Lock()
-	this.m = map[interface{}]zero.Zero{}
+	this.m = map[any]zero.Zero{}
 	this.keys = nil
 	this.locker.Unlock()
 }