remove goroutine for db

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

README.md

@@ -1,5 +1,5 @@
 # Micro
-![Coverage](https://img.shields.io/badge/Coverage-33.6%25-yellow)
+![Coverage](https://img.shields.io/badge/Coverage-34.1%25-yellow)
 [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0)
 [![Doc](https://img.shields.io/badge/go.dev-reference-007d9c?logo=go&logoColor=white&style=flat-square)](https://pkg.go.dev/go.unistack.org/micro/v4?tab=overview)
 [![Status](https://git.unistack.org/unistack-org/micro/actions/workflows/job_tests.yml/badge.svg?branch=v4)](https://git.unistack.org/unistack-org/micro/actions?query=workflow%3Abuild+branch%3Av4+event%3Apush)

client/noop.go

@@ -15,6 +15,11 @@ import (
 	"go.unistack.org/micro/v4/tracer"
 )
 
+// DefaultCodecs will be used to encode/decode data
+var DefaultCodecs = map[string]codec.Codec{
+	"application/octet-stream": codec.NewCodec(),
+}
+
 type noopClient struct {
 	funcCall   FuncCall
 	funcStream FuncStream

client/options.go

@@ -161,7 +161,7 @@ func NewOptions(opts ...Option) Options {
 	options := Options{
 		Context:     context.Background(),
 		ContentType: DefaultContentType,
-		Codecs:      make(map[string]codec.Codec),
+		Codecs:      DefaultCodecs,
 		CallOptions: CallOptions{
 			Context:        context.Background(),
 			Backoff:        DefaultBackoff,

logger/options.go

@@ -52,12 +52,6 @@ type Options struct {
 	AddStacktrace bool
 	// DedupKeys deduplicate keys in log output
 	DedupKeys bool
-	// FatalFinalizers runs in order in [logger.Fatal] method
-	FatalFinalizers []func(context.Context)
-}
-
-var DefaultFatalFinalizer = func(ctx context.Context) {
-	os.Exit(1)
 }
 
 // NewOptions creates new options struct
@@ -71,7 +65,6 @@ func NewOptions(opts ...Option) Options {
 		AddSource:        true,
 		TimeFunc:         time.Now,
 		Meter:            meter.DefaultMeter,
-		FatalFinalizers:  []func(context.Context){DefaultFatalFinalizer},
 	}
 
 	WithMicroKeys()(&options)
@@ -83,13 +76,6 @@ func NewOptions(opts ...Option) Options {
 	return options
 }
 
-// WithFatalFinalizers set logger.Fatal finalizers
-func WithFatalFinalizers(fncs ...func(context.Context)) Option {
-	return func(o *Options) {
-		o.FatalFinalizers = fncs
-	}
-}
-
 // WithContextAttrFuncs appends default funcs for the context attrs filler
 func WithContextAttrFuncs(fncs ...ContextAttrFunc) Option {
 	return func(o *Options) {

logger/slog/slog.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"io"
 	"log/slog"
+	"os"
 	"reflect"
 	"regexp"
 	"runtime"
 	"strconv"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/semconv"
@@ -229,12 +231,11 @@ func (s *slogLogger) Error(ctx context.Context, msg string, attrs ...interface{}
 
 func (s *slogLogger) Fatal(ctx context.Context, msg string, attrs ...interface{}) {
 	s.printLog(ctx, logger.FatalLevel, msg, attrs...)
-	for _, fn := range s.opts.FatalFinalizers {
-		fn(ctx)
-	}
 	if closer, ok := s.opts.Out.(io.Closer); ok {
 		closer.Close()
 	}
+	time.Sleep(1 * time.Second)
+	os.Exit(1)
 }
 
 func (s *slogLogger) Warn(ctx context.Context, msg string, attrs ...interface{}) {

logger/slog/slog_test.go

@@ -469,25 +469,3 @@ func Test_WithContextAttrFunc(t *testing.T) {
 
 	// t.Logf("xxx %s", buf.Bytes())
 }
-
-func TestFatalFinalizers(t *testing.T) {
-	ctx := context.TODO()
-	buf := bytes.NewBuffer(nil)
-	l := NewLogger(
-		logger.WithLevel(logger.TraceLevel),
-		logger.WithOutput(buf),
-	)
-	if err := l.Init(
-		logger.WithFatalFinalizers(func(ctx context.Context) {
-			l.Info(ctx, "fatal finalizer")
-		})); err != nil {
-		t.Fatal(err)
-	}
-	l.Fatal(ctx, "info_msg1")
-	if !bytes.Contains(buf.Bytes(), []byte("fatal finalizer")) {
-		t.Fatalf("logger dont have fatal message, buf %s", buf.Bytes())
-	}
-	if !bytes.Contains(buf.Bytes(), []byte("info_msg1")) {
-		t.Fatalf("logger dont have info_msg1 message, buf %s", buf.Bytes())
-	}
-}

meter/meter.go

@@ -49,11 +49,9 @@ type Meter interface {
 	Set(opts ...Option) Meter
 	// Histogram get or create histogram
 	Histogram(name string, labels ...string) Histogram
-	// HistogramExt get or create histogram with specified quantiles
-	HistogramExt(name string, quantiles []float64, labels ...string) Histogram
 	// Summary get or create summary
 	Summary(name string, labels ...string) Summary
-	// SummaryExt get or create summary with specified quantiles and window time
+	// SummaryExt get or create summary with spcified quantiles and window time
 	SummaryExt(name string, window time.Duration, quantiles []float64, labels ...string) Summary
 	// Write writes metrics to io.Writer
 	Write(w io.Writer, opts ...Option) error
@@ -84,11 +82,7 @@ type FloatCounter interface {
 
 // Gauge is a float64 gauge
 type Gauge interface {
-	Add(float64)
 	Get() float64
-	Set(float64)
-	Dec()
-	Inc()
 }
 
 // Histogram is a histogram for non-negative values with automatically created buckets

meter/noop.go

@@ -70,11 +70,6 @@ func (r *noopMeter) Histogram(_ string, labels ...string) Histogram {
 	return &noopHistogram{labels: labels}
 }
 
-// HistogramExt implements the Meter interface
-func (r *noopMeter) HistogramExt(_ string, quantiles []float64, labels ...string) Histogram {
-	return &noopHistogram{labels: labels}
-}
-
 // Set implements the Meter interface
 func (r *noopMeter) Set(opts ...Option) Meter {
 	m := &noopMeter{opts: r.opts}
@@ -141,18 +136,6 @@ type noopGauge struct {
 	labels []string
 }
 
-func (r *noopGauge) Add(float64) {
-}
-
-func (r *noopGauge) Set(float64) {
-}
-
-func (r *noopGauge) Inc() {
-}
-
-func (r *noopGauge) Dec() {
-}
-
 func (r *noopGauge) Get() float64 {
 	return 0
 }

meter/options.go

@@ -4,8 +4,6 @@ import (
 	"context"
 )
 
-var DefaultQuantiles = []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10}
-
 // Option powers the configuration for metrics implementations:
 type Option func(*Options)
 
@@ -25,8 +23,6 @@ type Options struct {
 	WriteProcessMetrics bool
 	// WriteFDMetrics flag to write fd metrics
 	WriteFDMetrics bool
-	// Quantiles specifies buckets for histogram
-	Quantiles []float64
 }
 
 // NewOptions prepares a set of options:
@@ -65,12 +61,14 @@ func Address(value string) Option {
 	}
 }
 
-// Quantiles defines the desired spread of statistics for histogram metrics:
+/*
-func Quantiles(quantiles []float64) Option {
+// TimingObjectives defines the desired spread of statistics for histogram / timing metrics:
+func TimingObjectives(value map[float64]float64) Option {
 	return func(o *Options) {
-		o.Quantiles = quantiles
+		o.TimingObjectives = value
 	}
 }
+*/
 
 // Labels add the meter labels
 func Labels(ls ...string) Option {

server/noop.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"time"
 
+	"go.unistack.org/micro/v4/codec"
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/register"
 	maddr "go.unistack.org/micro/v4/util/addr"
@@ -13,6 +14,11 @@ import (
 	"go.unistack.org/micro/v4/util/rand"
 )
 
+// DefaultCodecs will be used to encode/decode
+var DefaultCodecs = map[string]codec.Codec{
+	"application/octet-stream": codec.NewCodec(),
+}
+
 type rpcHandler struct {
 	opts    HandlerOptions
 	handler interface{}

util/xpool/pool.go

@@ -6,18 +6,18 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"go.unistack.org/micro/v4/meter"
 	"go.unistack.org/micro/v4/semconv"
 )
 
-func unregisterMetrics(size int) {
+var (
-	meter.DefaultMeter.Unregister(semconv.PoolGetTotal, "capacity", strconv.Itoa(size))
+	pools   = make([]Statser, 0)
-	meter.DefaultMeter.Unregister(semconv.PoolPutTotal, "capacity", strconv.Itoa(size))
+	poolsMu sync.Mutex
-	meter.DefaultMeter.Unregister(semconv.PoolMisTotal, "capacity", strconv.Itoa(size))
+)
-	meter.DefaultMeter.Unregister(semconv.PoolRetTotal, "capacity", strconv.Itoa(size))
-}
 
+// Stats struct
 type Stats struct {
 	Get uint64
 	Put uint64
@@ -25,13 +25,41 @@ type Stats struct {
 	Ret uint64
 }
 
+// Statser provides buffer pool stats
+type Statser interface {
+	Stats() Stats
+	Cap() int
+}
+
+func init() {
+	go newStatsMeter()
+}
+
+func newStatsMeter() {
+	ticker := time.NewTicker(meter.DefaultMeterStatsInterval)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		poolsMu.Lock()
+		for _, st := range pools {
+			stats := st.Stats()
+			meter.DefaultMeter.Counter(semconv.PoolGetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Get)
+			meter.DefaultMeter.Counter(semconv.PoolPutTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Put)
+			meter.DefaultMeter.Counter(semconv.PoolMisTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Mis)
+			meter.DefaultMeter.Counter(semconv.PoolRetTotal, "capacity", strconv.Itoa(st.Cap())).Set(stats.Ret)
+		}
+		poolsMu.Unlock()
+	}
+}
+
+var (
+	_ Statser = (*BytePool)(nil)
+	_ Statser = (*BytesPool)(nil)
+	_ Statser = (*StringsPool)(nil)
+)
+
 type Pool[T any] struct {
-	p   *sync.Pool
+	p *sync.Pool
-	get *atomic.Uint64
-	put *atomic.Uint64
-	mis *atomic.Uint64
-	ret *atomic.Uint64
-	c   int
 }
 
 func (p Pool[T]) Put(t T) {
@@ -42,82 +70,37 @@ func (p Pool[T]) Get() T {
 	return p.p.Get().(T)
 }
 
-func NewPool[T any](fn func() T, size int) Pool[T] {
+func NewPool[T any](fn func() T) Pool[T] {
-	p := Pool[T]{
+	return Pool[T]{
-		c:   size,
+		p: &sync.Pool{
-		get: &atomic.Uint64{},
+			New: func() interface{} {
-		put: &atomic.Uint64{},
+				return fn()
-		mis: &atomic.Uint64{},
+			},
-		ret: &atomic.Uint64{},
-	}
-
-	p.p = &sync.Pool{
-		New: func() interface{} {
-			p.mis.Add(1)
-			return fn()
 		},
 	}
-
-	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
-		return float64(p.get.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
-		return float64(p.put.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
-		return float64(p.mis.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
-		return float64(p.ret.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	return p
 }
 
 type BytePool struct {
 	p   *sync.Pool
-	get *atomic.Uint64
+	get uint64
-	put *atomic.Uint64
+	put uint64
-	mis *atomic.Uint64
+	mis uint64
-	ret *atomic.Uint64
+	ret uint64
 	c   int
 }
 
 func NewBytePool(size int) *BytePool {
-	p := &BytePool{
+	p := &BytePool{c: size}
-		c:   size,
-		get: &atomic.Uint64{},
-		put: &atomic.Uint64{},
-		mis: &atomic.Uint64{},
-		ret: &atomic.Uint64{},
-	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			p.mis.Add(1)
+			atomic.AddUint64(&p.mis, 1)
 			b := make([]byte, 0, size)
 			return &b
 		},
 	}
-
+	poolsMu.Lock()
-	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
+	pools = append(pools, p)
-		return float64(p.get.Load())
+	poolsMu.Unlock()
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
-		return float64(p.put.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
-		return float64(p.mis.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
-		return float64(p.ret.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
 	return p
 }
 
@@ -127,73 +110,49 @@ func (p *BytePool) Cap() int {
 
 func (p *BytePool) Stats() Stats {
 	return Stats{
-		Put: p.put.Load(),
+		Put: atomic.LoadUint64(&p.put),
-		Get: p.get.Load(),
+		Get: atomic.LoadUint64(&p.get),
-		Mis: p.mis.Load(),
+		Mis: atomic.LoadUint64(&p.mis),
-		Ret: p.ret.Load(),
+		Ret: atomic.LoadUint64(&p.ret),
 	}
 }
 
 func (p *BytePool) Get() *[]byte {
-	p.get.Add(1)
+	atomic.AddUint64(&p.get, 1)
 	return p.p.Get().(*[]byte)
 }
 
 func (p *BytePool) Put(b *[]byte) {
-	p.put.Add(1)
+	atomic.AddUint64(&p.put, 1)
 	if cap(*b) > p.c {
-		p.ret.Add(1)
+		atomic.AddUint64(&p.ret, 1)
 		return
 	}
 	*b = (*b)[:0]
 	p.p.Put(b)
 }
 
-func (p *BytePool) Close() {
-	unregisterMetrics(p.c)
-}
-
 type BytesPool struct {
 	p   *sync.Pool
-	get *atomic.Uint64
+	get uint64
-	put *atomic.Uint64
+	put uint64
-	mis *atomic.Uint64
+	mis uint64
-	ret *atomic.Uint64
+	ret uint64
 	c   int
 }
 
 func NewBytesPool(size int) *BytesPool {
-	p := &BytesPool{
+	p := &BytesPool{c: size}
-		c:   size,
-		get: &atomic.Uint64{},
-		put: &atomic.Uint64{},
-		mis: &atomic.Uint64{},
-		ret: &atomic.Uint64{},
-	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			p.mis.Add(1)
+			atomic.AddUint64(&p.mis, 1)
 			b := bytes.NewBuffer(make([]byte, 0, size))
 			return b
 		},
 	}
-
+	poolsMu.Lock()
-	meter.DefaultMeter.Gauge(semconv.PoolGetTotal, func() float64 {
+	pools = append(pools, p)
-		return float64(p.get.Load())
+	poolsMu.Unlock()
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolPutTotal, func() float64 {
-		return float64(p.put.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolMisTotal, func() float64 {
-		return float64(p.mis.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
-	meter.DefaultMeter.Gauge(semconv.PoolRetTotal, func() float64 {
-		return float64(p.ret.Load())
-	}, "capacity", strconv.Itoa(p.c))
-
 	return p
 }
 
@@ -203,10 +162,10 @@ func (p *BytesPool) Cap() int {
 
 func (p *BytesPool) Stats() Stats {
 	return Stats{
-		Put: p.put.Load(),
+		Put: atomic.LoadUint64(&p.put),
-		Get: p.get.Load(),
+		Get: atomic.LoadUint64(&p.get),
-		Mis: p.mis.Load(),
+		Mis: atomic.LoadUint64(&p.mis),
-		Ret: p.ret.Load(),
+		Ret: atomic.LoadUint64(&p.ret),
 	}
 }
 
@@ -215,43 +174,34 @@ func (p *BytesPool) Get() *bytes.Buffer {
 }
 
 func (p *BytesPool) Put(b *bytes.Buffer) {
-	p.put.Add(1)
 	if (*b).Cap() > p.c {
-		p.ret.Add(1)
+		atomic.AddUint64(&p.ret, 1)
 		return
 	}
 	b.Reset()
 	p.p.Put(b)
 }
 
-func (p *BytesPool) Close() {
-	unregisterMetrics(p.c)
-}
-
 type StringsPool struct {
 	p   *sync.Pool
-	get *atomic.Uint64
+	get uint64
-	put *atomic.Uint64
+	put uint64
-	mis *atomic.Uint64
+	mis uint64
-	ret *atomic.Uint64
+	ret uint64
 	c   int
 }
 
 func NewStringsPool(size int) *StringsPool {
-	p := &StringsPool{
+	p := &StringsPool{c: size}
-		c:   size,
-		get: &atomic.Uint64{},
-		put: &atomic.Uint64{},
-		mis: &atomic.Uint64{},
-		ret: &atomic.Uint64{},
-	}
 	p.p = &sync.Pool{
 		New: func() interface{} {
-			p.mis.Add(1)
+			atomic.AddUint64(&p.mis, 1)
 			return &strings.Builder{}
 		},
 	}
-
+	poolsMu.Lock()
+	pools = append(pools, p)
+	poolsMu.Unlock()
 	return p
 }
 
@@ -261,28 +211,24 @@ func (p *StringsPool) Cap() int {
 
 func (p *StringsPool) Stats() Stats {
 	return Stats{
-		Put: p.put.Load(),
+		Put: atomic.LoadUint64(&p.put),
-		Get: p.get.Load(),
+		Get: atomic.LoadUint64(&p.get),
-		Mis: p.mis.Load(),
+		Mis: atomic.LoadUint64(&p.mis),
-		Ret: p.ret.Load(),
+		Ret: atomic.LoadUint64(&p.ret),
 	}
 }
 
 func (p *StringsPool) Get() *strings.Builder {
-	p.get.Add(1)
+	atomic.AddUint64(&p.get, 1)
 	return p.p.Get().(*strings.Builder)
 }
 
 func (p *StringsPool) Put(b *strings.Builder) {
-	p.put.Add(1)
+	atomic.AddUint64(&p.put, 1)
 	if b.Cap() > p.c {
-		p.ret.Add(1)
+		atomic.AddUint64(&p.ret, 1)
 		return
 	}
 	b.Reset()
 	p.p.Put(b)
 }
-
-func (p *StringsPool) Close() {
-	unregisterMetrics(p.c)
-}