<commit_before>package async

import (
	"context"
	"fmt"
	"sync"
	"time"
)

var (
	input    chan *cmd
	output   chan *batchCmd
	groupDic map[string]*group
	gLocker  sync.RWMutex
)

type CacheProvider interface {
	Put(interface{}, interface{}) error
	Get(interface{}) (interface{}, error)
}

func init() {
	groupDic = make(map[string]*group)
	input = make(chan *cmd, 128)
	output = make(chan *batchCmd, 128)
	go runloop()
}

type group struct {
	name      string
	batchSize int
	method    func(...interface{}) (map[interface{}]interface{}, error)
	cmdDic    map[interface{}][]*cmd
	keyDic    map[interface{}]bool
	cache     CacheProvider
}

type cmd struct {
	group    string
	key      interface{}
	callback chan interface{}
}

type batchCmd struct {
	group  string
	keys   []interface{}
	err    error
	result map[interface{}]interface{}
}

func RegisterGroup(
	name string,
	batchSize int,
	method func(...interface{}) (map[interface{}]interface{}, error),
	cache CacheProvider) error {
	gLocker.Lock()
	defer gLocker.Unlock()
	_, ok := groupDic[name]
	if ok {
		return fmt.Errorf("duplicate group:%s", name)
	}
	groupDic[name] = &group{
		name:      name,
		batchSize: batchSize,
		method:    method,
		cmdDic:    make(map[interface{}][]*cmd),
		keyDic:    make(map[interface{}]bool),
		cache:     cache,
	}
	return nil
}

func doing(ctx context.Context, b *batchCmd, method func(...interface{}) (map[interface{}]interface{}, error)) {
	res, err := Safety(func() (interface{}, error) {
		return method(b.keys...)
	})
	if err != nil {
		b.err = err
	} else {
		b.result = res.(map[interface{}]interface{})
	}
	output <- b
}

func runloop() {
	ctx := context.Background()
	timer := time.NewTicker(10 * time.Millisecond)
	for {
		select {
		case c := <-input:
			{
				gLocker.RLocker()
				defer gLocker.RUnlock()
				g, ok := groupDic[c.group]
				if false == ok {
					c.callback <- fmt.Errorf("invalid group:%s", c.group)
					continue
				}
				// cache provider
				if g.cache != nil {
					res, err := g.cache.Get(c.key)
					if res != nil && err == nil {
						c.callback <- res
						continue
					}
				}
				target, ok := g.cmdDic[c.key]
				if ok {
					g.cmdDic[c.key] = append(target, c)
				} else {
					g.cmdDic[c.key] = []*cmd{c}
					g.keyDic[c.key] = true
				}

				if len(g.keyDic) >= g.batchSize {
					b := &batchCmd{group: g.name}
					b.keys = make([]interface{}, len(g.keyDic))
					index := 0
					for k, _ := range g.keyDic {
						b.keys[index] = k
						index = index + 1
					}
					g.keyDic = make(map[interface{}]bool)
					go doing(ctx, b, g.method)
				}
			}
		case b := <-output:
			{
				gLocker.RLocker()
				defer gLocker.RUnlock()
				g, _ := groupDic[b.group]
				if b.err != nil {
					for _, key := range b.keys {
						target, ok := g.cmdDic[key]
						if ok {
							for _, c := range target {
								c.callback <- b.err
								close(c.callback)
							}
						}
						delete(g.cmdDic, key)
					}
				} else {
					for _, key := range b.keys {
						res, ok := b.result[key]
						if false == ok {
							res = fmt.Errorf("invalid key:%s", key)
						} else {
							// cache provider
							if g.cache != nil {
								g.cache.Put(key, res)
							}
						}
						target, ok := g.cmdDic[key]
						if ok {
							for _, c := range target {<commit_msg>add cache<commit_after>package async

import (
	"context"
	"fmt"
	"sync"
	"time"
)

var (
	input    chan *cmd
	output   chan *batchCmd
	groupDic map[string]*group
	gLocker  sync.RWMutex
)

type CacheProvider interface {
	Put(string, interface{}, interface{}) error
	Get(string, interface{}) (interface{}, error)
}

func init() {
	groupDic = make(map[string]*group)
	input = make(chan *cmd, 128)
	output = make(chan *batchCmd, 128)
	go runloop()
}

type group struct {
	name      string
	batchSize int
	method    func(...interface{}) (map[interface{}]interface{}, error)
	cmdDic    map[interface{}][]*cmd
	keyDic    map[interface{}]bool
	cache     CacheProvider
}

type cmd struct {
	group    string
	key      interface{}
	callback chan interface{}
}

type batchCmd struct {
	group  string
	keys   []interface{}
	err    error
	result map[interface{}]interface{}
}

func RegisterGroup(
	name string,
	batchSize int,
	method func(...interface{}) (map[interface{}]interface{}, error),
	cache CacheProvider) error {
	gLocker.Lock()
	defer gLocker.Unlock()
	_, ok := groupDic[name]
	if ok {
		return fmt.Errorf("duplicate group:%s", name)
	}
	groupDic[name] = &group{
		name:      name,
		batchSize: batchSize,
		method:    method,
		cmdDic:    make(map[interface{}][]*cmd),
		keyDic:    make(map[interface{}]bool),
		cache:     cache,
	}
	return nil
}

func doing(ctx context.Context, b *batchCmd, method func(...interface{}) (map[interface{}]interface{}, error)) {
	res, err := Safety(func() (interface{}, error) {
		return method(b.keys...)
	})
	if err != nil {
		b.err = err
	} else {
		b.result = res.(map[interface{}]interface{})
	}
	output <- b
}

func runloop() {
	ctx := context.Background()
	timer := time.NewTicker(10 * time.Millisecond)
	for {
		select {
		case c := <-input:
			{
				gLocker.RLocker()
				defer gLocker.RUnlock()
				g, ok := groupDic[c.group]
				if false == ok {
					c.callback <- fmt.Errorf("invalid group:%s", c.group)
					continue
				}
				// cache provider
				if g.cache != nil {
					res, err := g.cache.Get(g.name, c.key)
					if res != nil && err == nil {
						c.callback <- res
						continue
					}
				}
				target, ok := g.cmdDic[c.key]
				if ok {
					g.cmdDic[c.key] = append(target, c)
				} else {
					g.cmdDic[c.key] = []*cmd{c}
					g.keyDic[c.key] = true
				}

				if len(g.keyDic) >= g.batchSize {
					b := &batchCmd{group: g.name}
					b.keys = make([]interface{}, len(g.keyDic))
					index := 0
					for k, _ := range g.keyDic {
						b.keys[index] = k
						index = index + 1
					}
					g.keyDic = make(map[interface{}]bool)
					go doing(ctx, b, g.method)
				}
			}
		case b := <-output:
			{
				gLocker.RLocker()
				defer gLocker.RUnlock()
				g, _ := groupDic[b.group]
				if b.err != nil {
					for _, key := range b.keys {
						target, ok := g.cmdDic[key]
						if ok {
							for _, c := range target {
								c.callback <- b.err
								close(c.callback)
							}
						}
						delete(g.cmdDic, key)
					}
				} else {
					for _, key := range b.keys {
						res, ok := b.result[key]
						if false == ok {
							res = fmt.Errorf("invalid key:%s", key)
						} else {
							// cache provider
							if g.cache != nil {
								g.cache.Put(g.name, key, res)
							}
						}
						target, ok := g.cmdDic[key]
						if ok {
							for _, c := range target {
{'fraction_non_alphanumeric': 0.15751357875678937, 'fraction_numerical': 0.0030175015087507543, 'mean_word_length': 2.431159420289855, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3118359', 'n_tokens_mistral': 3002, 'n_tokens_neox': 2405, 'n_words': 579}