kern/v4/d2/kwg/kwg.go

// package kwg -- именованный ожидатель потоков ядра.
//
// Не позволяет завершиться ядру, если есть хоть один работающий поток
package kwg

import (
	"context"
	"time"

	"gitp78su.ipnodns.ru/svi/kern/v4/d0"
	"gitp78su.ipnodns.ru/svi/kern/v4/d1"
	"gitp78su.ipnodns.ru/svi/kern/v4/d2/kern_ent/stream_name"
)

// KernWg -- именованный ожидатель потоков ядра.
type KernWg struct {
	*d0.MSafeMutex
	owner      *d0.LOwner
	ctx        context.Context
	dictStream map[*stream_name.AStreamName]d0.Bool // Словарь имён потоков с признаком работы
	isWork     *d0.MSafeBool
	log        *d1.LogBuf
}

var (
	kernWg *KernWg // Глобальный объект
	block  = d0.MutSafeMutex()
)

// GetKernelWg -- возвращает новый именованный ожидатель потоков ядра.
func GetKernelWg(ctx context.Context) *KernWg {
	block.Lock()
	defer block.Unlock()
	if kernWg != nil {
		kernWg.log.Debug("GetKernelWg()")
		return kernWg
	}
	owner := d0.LetOwner(1, "isol_kern_wg")
	d0.If(ctx == nil).Hassert("GetKernelWg(): ctx==nil")
	paramLogBuf := &d1.LogBufParam{
		Owner_:owner,
		IsTerm_: d0.MutSafeBool(owner, true),
		Prefix_: d0.LetTxt(owner, "kernelWg"),
	}
	sf := &KernWg{
		MSafeMutex: d0.MutSafeMutex(),
		owner:      owner,
		ctx:        ctx,
		dictStream: map[*stream_name.AStreamName]d0.Bool{},
		isWork:     d0.MutSafeBool(owner, false),
		log:        d1.NewLogBuf(paramLogBuf),
	}
	sf.log.Debug("GetKernelWg(): run")
	go sf.close()
	sf.isWork.Set()
	kernWg = sf
	return kernWg
}

// Log -- возвращает лог ожидателя потоков.
func (sf *KernWg) Log() *d1.LogBuf {
	return sf.log
}

// Len -- возвращает размер списка ожидания потоков.
func (sf *KernWg) Len() d0.Num {
	sf.RLock()
	defer sf.RUnlock()
	return d0.Num(len(sf.dictStream))
}

// IsWork -- возвращает признак работы ядра.
func (sf *KernWg) IsWork() d0.Bool {
	return sf.isWork.IsOk()
}

// IsStop -- возвращает признак остановки ядра.
func (sf *KernWg) IsStop() d0.Bool {
	return !sf.isWork.IsOk()
}

// List -- возвращает список имён потоков на ожидании.
func (sf *KernWg) List() []*stream_name.AStreamName {
	sf.RLock()
	defer sf.RUnlock()
	lst := make([]*stream_name.AStreamName, 0, len(sf.dictStream))
	for name := range sf.dictStream {
		lst = append(lst, name)
	}
	return lst
}

// Done -- удаляет поток из ожидания.
func (sf *KernWg) Done(name *stream_name.AStreamName) {
	sf.Lock()
	defer sf.Unlock()
	delete(sf.dictStream, name)
	sf.log.Debug("Done(): stream(%v) done", name)
}

// Wait -- блокирующий вызов; возвращает управление, только когда все потоки завершили работу.
func (sf *KernWg) Wait() {
	for {
		time.Sleep(time.Millisecond * 1)
		if sf.isWork.IsNot() {
			break
		}
	}
	sf.log.Debug("Wait(): done")
}

// Add -- добавляет поток в ожидание.
func (sf *KernWg) Add(name *stream_name.AStreamName) {
	sf.Lock()
	defer sf.Unlock()
	sf.log.Debug("Add(): stream('%v') done", name)
	d0.If(sf.isWork.IsNot()).Hassert("Add(): stream=%v, work end", []any{name})
	d0.If(name.Get() == "").Hassert("Add(): name stream is empty")
	_, isOk := sf.dictStream[name]
	d0.If(d0.Bool(isOk)).Hassert("Add(): stream '%v' already exists", name)
	sf.dictStream[name] = true
}

// Ожидает окончания работы ожидателя групп.
func (sf *KernWg) close() {
	<-sf.ctx.Done()
	fnDone := func() bool {
		sf.Lock()
		defer sf.Unlock()
		return len(sf.dictStream) == 0
	}
	for {
		time.Sleep(time.Millisecond * 1)
		if fnDone() {
			break
		}
	}
	sf.Lock()
	defer sf.Unlock()
	if sf.isWork.IsNot() {
		return
	}
	sf.isWork.Reset()
	sf.log.Debug("close(): end")
}