m2pool-core/internal/gbt/gbt.go

// gbt.go
package gbt

import (
	"encoding/json"
	"fmt"
	"log"

	"io/ioutil"
	"pool/internal/db"
	"sync/atomic"

	//"pool/internal/cache"
	"pool/internal/gbt/coin"
	"pool/internal/gbt/monero"
	monero_rpc "pool/internal/gbt/monero/rpc"
	"pool/internal/utility"

	"os"
	"os/signal"
	"syscall"

	"time"

	"github.com/btcsuite/btcd/rpcclient"
	"github.com/redis/go-redis/v9"
	"go.uber.org/zap"
)

var logg *zap.Logger
var GbtCtx coin.GbtContext

func InitConfig(config *coin.GbtConfig) {
	data, err := ioutil.ReadFile("gbt.conf")
	if err != nil {
		panic(err.Error())
	}
	if err = json.Unmarshal(data, &config); err != nil {
		panic(err.Error())
	}
}

func InitClient(gbt *coin.GbtContext) error {
	fmt.Println(gbt.Coin)
	switch gbt.Coin {
	case "monero":
		url := "http://" + gbt.Config.Rpc.Host
		rpcClient := monero_rpc.NewHttpClient(url, 10*time.Second)
		gbt.MoneroClinet = rpcClient
		gbt.MoneroAddr = gbt.Config.Rpc.User
	default:
		var config rpcclient.ConnConfig
		if gbt.Config.Rpc.Type == "testnet" {
			config.Host = gbt.Config.Rpc.Testnet
		} else {
			config.Host = gbt.Config.Rpc.Host
		}
		config.User = gbt.Config.Rpc.User
		config.Pass = gbt.Config.Rpc.Pass
		config.HTTPPostMode = true
		config.DisableTLS = true
		client, err := rpcclient.New(&config, nil)
		if err != nil {
			logg.Info("[gbt]", zap.String("rpcclient new ", err.Error()))
			return err
		}
		gbt.Client = client
		blockCount, err := client.GetBlockCount()
		if err != nil {
			logg.Info("[gbt]", zap.String("GetBlockCount ", err.Error()))
			return err
		}
		logg.Info("[gbt]", zap.Int64("Block count ", blockCount))

	}
	return nil
}

/*func GbtLivingHandler(gbt *coin.GbtContext) {
	timer := time.NewTimer(time.Duration(600) * time.Second)
	for {
		select {
		case aliving := <-gbt.AlivingChan:
			if !aliving {
				timer.Stop()
				//log.Println("gbt aliving", aliving)
				return
			} else {
				timer.Reset(time.Duration(600) * time.Second)
			}
		case <-time.After(time.Duration(600) * time.Second):
			gbt.ExitGbtChan <- true

			return
		}
	}
}*/

func GbtLivingHandler(gbt *coin.GbtContext) {
	var to_cnt int = 0
	for {
		flagAliving := atomic.LoadInt32(&(gbt.FlagAliving))
		flagExit := atomic.LoadInt32(&(gbt.FlagAlivingExit))

		if flagExit == 1 {
			logg.Error("[server]", zap.String("GbtLivingHandler exited", "exit"))
			break
		}

		if flagAliving == 0 {
			//if to_cnt > 240 {
			if to_cnt > gbt.Config.Rpc.Timeout*3/1000 {
				logg.Error("[server]", zap.String("GbtLivingHandler exited", "timer expired"))
				cmd := "killall gbt_" + gbt.Coin + " &"
				utility.ExecShellCmd(cmd)
				gbt.ExitGbtChan <- true
				break
			}
			to_cnt++
		} else {
			to_cnt = 0
			atomic.StoreInt32(&(gbt.FlagAliving), 0)
		}

		time.Sleep(time.Second)
	}
}

type coinobj struct {
	Coin  string
	Init  func(GbtCtx *coin.GbtContext, DbCtx *db.DbContext)
	Start func()
	Stop  func()
}

var coinobjs = []coinobj{
	// {Coin: "nexa", Init: nexa.Init, Start: nexa.Start, Stop: nexa.Stop},
	{Coin: "monero", Init: monero.Init, Start: monero.Start, Stop: monero.Stop},
	// {Coin: "sha3x", Init: sha3x.Init, Start: sha3x.Start, Stop: sha3x.Stop},
}

func register_signal(dbctx *db.DbContext) {
	signal_ch := make(chan os.Signal, 1)
	signal.Notify(signal_ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	go signal_handle(signal_ch, dbctx)
}

func signal_handle(signal_ch chan os.Signal, dbctx *db.DbContext) {
	for s := range signal_ch {
		switch s {
		case syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT:
			log.Println("stop!")
			dbctx.AppExit <- true
		default:
			//fmt.Println("other signal", s)
		}
	}
}

/*func LoadCache(gbt *coin.GbtContext) {
	val := cache.LoadPoolCache(gbt.RedisClient, gbt.Coin, "blocks")
	if val != nil {
		if intVal, ok := val.(int64); ok {
			gbt.Blocks = intVal
		}
	}

	val_f := cache.LoadPoolCache(gbt.RedisClient, gbt.Coin, "rewards")
	if val_f != nil {
		if fVal, ok := val_f.(float64); ok {
			gbt.Reward = fVal
		}
	}

	val_f = cache.LoadPoolCache(gbt.RedisClient, gbt.Coin, "fee")
	if val_f != nil {
		if fVal, ok := val_f.(float64); ok {
			gbt.Fee = fVal
		}
	}
}*/

func Start(Coin string, DbCtx *db.DbContext) {
	GbtCtx.DbCtx = DbCtx

	GbtCtx.Coin = Coin

	atomic.StoreInt32(&(GbtCtx.FlagAliving), 0)
	atomic.StoreInt32(&(GbtCtx.FlagAlivingExit), 0)

	InitConfig(&GbtCtx.Config)
	l, lr, err := utility.InitLogg(&(GbtCtx.Config.Zaplog), &(GbtCtx.Config.Logrotae), Coin, "gbt")
	logg = l

	//defer logg.Sync()

	GbtCtx.Log = l
	GbtCtx.LogR = lr

	GbtCtx.RedisClient = redis.NewClient(&redis.Options{
		Addr:     GbtCtx.Config.Redis.Addr,
		Password: GbtCtx.Config.Redis.Password,
		DB:       GbtCtx.Config.Redis.DB,
	})

	register_signal(DbCtx)

	GbtCtx.PubCh = utility.InitZmqPub(GbtCtx.Config.Zmq.Pub)

	GbtCtx.SubCh = utility.InitZmqSub(GbtCtx.Config.Zmq.Sub, "blk"+Coin)

	GbtCtx.MoneroNewBlockPubCh = utility.InitZmqPub(GbtCtx.Config.Rpc.ZmqSub)
	GbtCtx.MoneroNewBlockSubCh = utility.InitZmqSub(GbtCtx.Config.Rpc.ZmqSub, "hashblock")
	//GbtCtx.PushCh = utility.InitZmqPush(GbtCtx.Config.Profit.Push)

	for {
		err = InitClient(&GbtCtx)
		if err != nil {
			logg.Error("[gbt]", zap.String("InitClient", err.Error()))
			time.Sleep(time.Duration(5) * time.Second)
			continue
		}
		break
	}

	if len(GbtCtx.Config.Rpc.ZmqSub) > 0 {
		GbtCtx.NodeSubCh = utility.InitZmqSub(GbtCtx.Config.Rpc.ZmqSub, utility.BITCOIND_ZMQ_HASHBLOCK)
	}

	GbtCtx.Started = true

	GbtCtx.ExitGbtChan = make(chan bool, 256)

	//GbtCtx.AlivingChan = make(chan bool, 256)

	GbtCtx.MinerAddrs = db.GetAddressFromTable(DbCtx)
	GbtCtx.MinerAddrIndex = 0

	//LoadCache(&GbtCtx)

	for _, coinobj := range coinobjs {
		if coinobj.Coin == Coin {
			coinobj.Init(&GbtCtx, DbCtx)
			go coinobj.Start()
			break
		}
	}

	go GbtLivingHandler(&GbtCtx)

	<-DbCtx.AppExit
}

func Stop(Coin string) {

	GbtCtx.Started = false

	GbtCtx.ExitGbtChan <- true

	//GbtCtx.AlivingChan <- false
	atomic.StoreInt32(&(GbtCtx.FlagAlivingExit), 1)

	for _, coinobj := range coinobjs {
		if coinobj.Coin == Coin {
			coinobj.Stop()
			break
		}
	}

	//time.Sleep(1 * time.Second)

	defer close(GbtCtx.ExitGbtChan)

	//defer close(GbtCtx.AlivingChan)

	if GbtCtx.NodeSubCh != nil {
		defer GbtCtx.NodeSubCh.Destroy()
	}
	if GbtCtx.PubCh != nil {
		defer GbtCtx.PubCh.Destroy()
	}
	if GbtCtx.SubCh != nil {
		defer GbtCtx.SubCh.Destroy()
	}
	/*if GbtCtx.PushCh != nil {
		defer GbtCtx.PushCh.Destroy()
	}*/
	defer GbtCtx.RedisClient.Close()
	defer logg.Sync()
}