cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package blobstore
16

17
type Instance struct {
18
	mq chan task
19
}
20

21
type task struct {
22
	op *rwSlice
23
	fn func(op *rwSlice)
24
}
25

26
func New(worker int, size int) Instance {
27
	mq := make(chan task, size)
28
	for i := 0; i < worker; i++ {
29
		go func() {
30
			for {
31
				task, ok := <-mq
32
				if !ok {
33
					break
34
				}
35
				task.fn(task.op)
36
			}
37
		}()
38
	}
39
	return Instance{mq}
40
}
41

42
func (r Instance) Execute(op *rwSlice, fn func(op *rwSlice)) {
43
	r.mq <- task{
44
		op: op,
45
		fn: fn,
46
	}
47
}
48

49
func (r Instance) Close() {
50
	close(r.mq)
51
}
52

53
type Executor struct {
54
	tokens chan int
55
}
56

57
func NewExecutor(maxConcurrency int) *Executor {
58
	exec := &Executor{
59
		tokens: make(chan int, maxConcurrency),
60
	}
61
	for i := 0; i < maxConcurrency; i++ {
62
		exec.tokens <- i
63
	}
64
	return exec
65
}
66

67
func (exec *Executor) Run(fn func()) {
68
	i := <-exec.tokens
69
	go func() {
70
		fn()
71
		exec.tokens <- i
72
	}()
73
}
74