Merge pull request #54 from ncwsky/main

大量添加索引引发缓冲通道满造成服务崩溃

config.yaml

@@ -22,4 +22,7 @@ enableGzip: true
 timeout: 600
 
 # 分片数量
-shard: 10
+shard: 10
+
+# 分片缓冲数量
+bufferNum: 1000

core/initialize.go

@@ -24,6 +24,7 @@ func NewContainer(tokenizer *words.Tokenizer) *searcher.Container {
 		Tokenizer: tokenizer,
 		Shard:     global.CONFIG.Shard,
 		Timeout:   global.CONFIG.Timeout,
+		BufferNum: global.CONFIG.BufferNum,
 	}
 	if err := container.Init(); err != nil {
 		panic(err)

core/parser.go

@@ -4,10 +4,10 @@ import (
 	"flag"
 	"fmt"
 	"gofound/global"
-	"gopkg.in/yaml.v2"
-	"io/ioutil"
 	"os"
 	"runtime"
+
+	"gopkg.in/yaml.v2"
 )
 
 // Parser 解析器
@@ -31,6 +31,7 @@ func Parser() *global.Config {
 
 	var enableGzip = flag.Bool("enableGzip", true, "是否开启gzip压缩")
 	var timeout = flag.Int64("timeout", 10*60, "数据库超时关闭时间(秒)")
+	var bufferNum = flag.Int("bufferNum", 1000, "分片缓冲数量")
 
 	var configPath = flag.String("config", "", "配置文件路径，配置此项其他参数忽略")
 
@@ -40,7 +41,8 @@ func Parser() *global.Config {
 
 	if *configPath != "" {
 		//解析配置文件
-		file, err := ioutil.ReadFile(*configPath)
+		//file, err := ioutil.ReadFile(*configPath)
+		file, err := os.ReadFile(*configPath) //详情：https://github.com/golang/go/issues/42026
 		if err != nil {
 			panic(err)
 		}
@@ -60,6 +62,7 @@ func Parser() *global.Config {
 		Auth:        *auth,
 		EnableGzip:  *enableGzip,
 		Timeout:     *timeout,
+		BufferNum:   *bufferNum,
 	}
 
 	return config

global/config.go

@@ -12,4 +12,5 @@ type Config struct {
 	Auth        string `json:"auth"`        //认证
 	EnableGzip  bool   `yaml:"enableGzip"`  //是否开启gzip压缩
 	Timeout     int64  `json:"timeout"`     //超时时间
+	BufferNum   int    `yaml:"bufferNum"`   //分片缓冲数
 }

gofound.d

@@ -0,0 +1,107 @@
+#!/bin/bash
+# chkconfig: 2345 90 10
+# Description: Startup script for gofound on Debian. Place in /etc/init.d and
+# run 'update-rc.d -f gofound defaults', or use the appropriate command on your
+# distro. For CentOS/Redhat run: 'chkconfig --add gofound'
+
+### BEGIN INIT INFO
+#  
+# Provides:         gofound.d
+# Required-Start:   $local_fs  $remote_fs  
+# Required-Stop:    $local_fs  $remote_fs  
+# Default-Start:    2 3 4 5
+# Default-Stop:     0 1 6
+# Short-Description: starts gofound
+# Description:       This file should be used to gofound scripts to be placed in /etc/init.d.  
+#  
+### END INIT INFO 
+
+
+## 2345是默认启动级别，级别有0-6共7个级别。 90是启动优先级，10是停止优先级，优先级范围是0－100，数字越大，优先级越低。
+
+## Fill in name of program here.  
+PROG="gofound"
+PROG_PATH="/usr/local/bin" ## Not need, but sometimes helpful (if $PROG resides in /opt for example). 
+PROG_ARGS="--config=/gofound_path/config.yaml"
+PID_PATH="/var/run/"
+
+start() {  
+    if [ -e "$PID_PATH/$PROG.pid" ]; then  
+        ## Program is running, exit with error.  
+        echo "Error! $PROG_PATH/$PROG is currently running!" 1>&2
+        exit 1  
+    else  
+        ## Change from /dev/null to something like /var/log/$PROG if you want to save output.  
+        $PROG_PATH/$PROG $PROG_ARGS 2>&1 >/var/log/$PROG &  
+        #pid=`ps ax | grep -i '/usr/bin/frps' | grep -v 'grep' |  sed 's/^\([0-9]\{1,\}\).*/\1/g' | head -n 1`  
+        pid=`ps -ef | grep $PROG_PATH/$PROG | grep -v grep | awk '{print $2}'`
+        #echo $PROG_PATH/$PROG $PROG_ARGS 
+        echo "$PROG_PATH/$PROG($pid) started"  
+        echo $pid > "$PID_PATH/$PROG.pid"  
+    fi  
+}  
+
+stop() {  
+    echo "begin stop"  
+    if [ -e "$PID_PATH/$PROG.pid" ]; then  
+        ## Program is running, so stop it  
+        #pid=`ps ax | grep -i '/usr/bin/frps' | grep -v 'grep' | sed 's/^\([0-9]\{1,\}\).*/\1/g' | head -n 1`  
+        pid=`ps -ef | grep $PROG_PATH/$PROG | grep -v grep | awk '{print $2}'`
+        kill $pid  
+
+        rm -f  "$PID_PATH/$PROG.pid"  
+        echo "$PROG_PATH/$PROG($pid) stopped"
+    else  
+        ## Program is not running, exit with error.  
+        echo "Error! $PROG_PATH/$PROG not started!" 1>&2
+    fi  
+}
+
+status() {  
+    if [ -e "$PID_PATH/$PROG.pid" ]; then  
+        ## Program is running, so stop it  
+        #pid=`ps ax | grep -i '/usr/bin/frps' | grep -v 'grep' | sed 's/^\([0-9]\{1,\}\).*/\1/g' | head -n 1`  
+        pid=`ps -ef | grep $PROG_PATH/$PROG | grep -v grep | awk '{print $2}'`
+
+		if [ $pid ]; then
+			echo "$PROG_PATH/$PROG($pid) is running..."
+		else
+			echo "$PROG_PATH/$PROG dead but pid file exists" 1>&2
+		fi
+    else  
+        ## Program is not running, exit with error.  
+        echo "Error! $PROG_PATH/$PROG not started!" 1>&2
+    fi  
+}
+
+
+## Check to see if we are running as root first.  
+## Found at http://www.cyberciti.biz/tips/shell-root-user-check-script.html  
+if [ "$(id -u)" != "0" ]; then  
+    echo "This script must be run as root" 1>&2
+    exit 1  
+fi  
+
+case "$1" in  
+    start)  
+        start  
+        exit 0  
+    ;;  
+    stop)  
+        stop  
+        exit 0  
+    ;;  
+    reload|restart|force-reload)  
+        stop  
+        start  
+        exit 0  
+    ;;
+    status)
+		status
+        exit 0
+	;;
+    *)  
+        echo "Usage: $0 {start|stop|restart|status}" 1>&2
+        exit 1  
+    ;;  
+esac

searcher/container.go

@@ -1,9 +1,9 @@
 package searcher
 
 import (
+	"errors"
 	"fmt"
 	"gofound/searcher/words"
-	"io/ioutil"
 	"log"
 	"os"
 	"runtime"
@@ -17,14 +17,15 @@ type Container struct {
 	Tokenizer *words.Tokenizer   //分词器
 	Shard     int                //分片
 	Timeout   int64              //超时关闭数据库
+	BufferNum int                //分片缓冲数
 }
 
 func (c *Container) Init() error {
 
 	c.engines = make(map[string]*Engine)
 
 	//读取当前路径下的所有目录，就是数据库名称
-	dirs, err := ioutil.ReadDir(c.Dir)
+	dirs, err := os.ReadDir(c.Dir)
 	if err != nil {
 		if os.IsNotExist(err) {
 			//创建
@@ -55,6 +56,7 @@ func (c *Container) NewEngine(name string) *Engine {
 		Tokenizer:    c.Tokenizer,
 		Shard:        c.Shard,
 		Timeout:      c.Timeout,
+		BufferNum:    c.BufferNum,
 	}
 	option := engine.GetOptions()
 
@@ -116,6 +118,9 @@ func (c *Container) GetDocumentCount() int64 {
 
 // DropDataBase 删除数据库
 func (c *Container) DropDataBase(name string) error {
+	if _, ok := c.engines[name]; !ok {
+		return errors.New("数据库不存在")
+	}
 	err := c.engines[name].Drop()
 	if err != nil {
 		return err

searcher/engine.go

@@ -38,6 +38,7 @@ type Engine struct {
 
 	Shard   int   //分片数
 	Timeout int64 //超时时间,单位秒
+	BufferNum int   //分片缓冲数
 
 	documentCount int64 //文档总数量
 }
@@ -62,13 +63,14 @@ func (e *Engine) Init() {
 	//-1代表没有初始化
 	e.documentCount = -1
 	//log.Println("数据存储目录：", e.IndexPath)
+	log.Println("chain num:", e.Shard*e.BufferNum)
 
 	e.addDocumentWorkerChan = make([]chan *model.IndexDoc, e.Shard)
 	//初始化文件存储
 	for shard := 0; shard < e.Shard; shard++ {
 
 		//初始化chan
-		worker := make(chan *model.IndexDoc, 1000)
+		worker := make(chan *model.IndexDoc, e.BufferNum)
 		e.addDocumentWorkerChan[shard] = worker
 
 		//初始化chan
@@ -105,14 +107,26 @@ func (e *Engine) automaticGC() {
 		<-ticker.C
 		//定时GC
 		runtime.GC()
+		if e.IsDebug {
+			log.Println("waiting:", e.GetQueue())
+		}
 	}
 }
 
-func (e *Engine) IndexDocument(doc *model.IndexDoc) {
-	//根据ID来判断，使用多线程，提速
+func (e *Engine) IndexDocument(doc *model.IndexDoc) error {
 	//数量增加
 	e.documentCount++
 	e.addDocumentWorkerChan[e.getShard(doc.Id)] <- doc
+	return nil
+	/*
+		select {
+		case e.addDocumentWorkerChan[e.getShard(doc.Id)] <- doc:
+			e.documentCount++
+		default:
+			return errors.New("处理缓冲已满")
+		}
+		return nil
+	*/
 }
 
 // GetQueue 获取队列剩余
@@ -153,6 +167,9 @@ func (e *Engine) InitOption(option *Option) {
 	if e.Shard <= 0 {
 		e.Shard = 10
 	}
+	if e.BufferNum <= 0 {
+		e.BufferNum = 1000
+	}
 	//初始化其他的
 	e.Init()
 

searcher/utils/utils.go

@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"encoding/gob"
 	"io/fs"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -202,7 +201,7 @@ func ReleaseAssets(file fs.File, out string) {
 			}
 		}(outFile)
 
-		err = ioutil.WriteFile(out, buffer, os.ModePerm)
+		err = os.WriteFile(out, buffer, os.ModePerm)
 		if err != nil {
 			panic(err)
 		}