Skip to content

Commit 0ec177b

Browse files
CMGSCMGS
committed
use itogami pool
1 parent 5e006b9 commit 0ec177b

25 files changed

Lines changed: 43 additions & 72 deletions

‎agent.go‎

Lines changed: 4 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"math/rand"
77
"os"
88
"os/signal"
9-
"runtime"
109
"sync"
1110
"syscall"
1211
"time"
@@ -57,12 +56,6 @@ func serve(c *cli.Context) error {
5756
utils.WritePid(config.PidFile)
5857
defer os.Remove(config.PidFile)
5958

60-
if err := utils.NewPool(config.MaxConcurrency); err != nil {
61-
log.Error(err)
62-
return err
63-
}
64-
defer utils.Pool.Release()
65-
6659
ctx, cancel := context.WithCancel(c.Context)
6760
defer cancel()
6861

@@ -78,7 +71,7 @@ func serve(c *cli.Context) error {
7871
if err != nil {
7972
return err
8073
}
81-
_ = utils.Pool.Submit(func() {
74+
utils.Pool.Submit(func() {
8275
defer wg.Done()
8376
if err := workloadsManager.Run(ctx); err != nil {
8477
log.Errorf("[agent] workload manager err: %v, exiting", err)
@@ -90,7 +83,7 @@ func serve(c *cli.Context) error {
9083
if err != nil {
9184
return err
9285
}
93-
_ = utils.Pool.Submit(func() {
86+
utils.Pool.Submit(func() {
9487
defer wg.Done()
9588
if err := nodeManager.Run(ctx); err != nil {
9689
log.Errorf("[agent] node manager err: %v, exiting", err)
@@ -99,9 +92,9 @@ func serve(c *cli.Context) error {
9992
})
10093

10194
apiHandler := api.NewHandler(config, workloadsManager)
102-
_ = utils.Pool.Submit(apiHandler.Serve)
95+
utils.Pool.Submit(apiHandler.Serve)
10396

104-
_ = utils.Pool.Submit(func() {
97+
utils.Pool.Submit(func() {
10598
select {
10699
case <-ctx.Done():
107100
log.Info("[agent] Agent exiting")
@@ -248,12 +241,6 @@ func main() {
248241
Value: false,
249242
Usage: "will only check containers belong to this node if set",
250243
},
251-
&cli.IntFlag{
252-
Name: "max-concurrency",
253-
Value: runtime.NumCPU() * 100,
254-
Usage: "max concurrency for goroutine pool",
255-
EnvVars: []string{"ERU_MAX_CONCURRENCY"},
256-
},
257244
},
258245
Action: serve,
259246
}

‎agent.yaml.sample‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,6 @@ core:
1616
# The default value of this option is 60.
1717
heartbeat_interval: 120
1818

19-
# max_concurrency defines the goroutine pool max size
20-
# need set a big number
21-
# The default value is 100X number of the CPU.
22-
max_concurrency: 1000
23-
2419
# check_only_mine defines check agent itself or not
2520
# The default value is false
2621
check_only_mine: false

‎go.mod‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,12 @@ go 1.19
55
require (
66
github.com/CMGS/statsd v0.0.0-20160223095033-48c421b3c1ab
77
github.com/alphadose/haxmap v1.1.0
8+
github.com/alphadose/itogami v0.4.1-0.20221016160904-c25d0a36bfe7
89
github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40
910
github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
1011
github.com/docker/docker v20.10.18+incompatible
1112
github.com/docker/go-units v0.5.0
1213
github.com/jinzhu/configor v1.2.1
13-
github.com/panjf2000/ants/v2 v2.5.0
1414
github.com/patrickmn/go-cache v2.1.0+incompatible
1515
github.com/projecteru2/core v0.0.0-20221011033901-ff9e2848a9e1
1616
github.com/projecteru2/libyavirt v0.0.0-20220621042712-95cdc6363b1c
@@ -47,6 +47,7 @@ require (
4747
github.com/moby/term v0.0.0-20220808134915-39b0c02b01ae // indirect
4848
github.com/opencontainers/go-digest v1.0.0 // indirect
4949
github.com/opencontainers/image-spec v1.1.0-rc2 // indirect
50+
github.com/panjf2000/ants/v2 v2.5.0 // indirect
5051
github.com/pkg/errors v0.9.1 // indirect
5152
github.com/pmezard/go-difflib v1.0.0 // indirect
5253
github.com/prometheus/client_model v0.2.0 // indirect

‎go.sum‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
5252
github.com/alexcesaro/statsd v2.0.0+incompatible h1:HG17k1Qk8V1F4UOoq6tx+IUoAbOcI5PHzzEUGeDD72w=
5353
github.com/alphadose/haxmap v1.1.0 h1:M7Dxdr+civMQkWCgDpoktNpLofDBz7XzdS3rF3Y6r4U=
5454
github.com/alphadose/haxmap v1.1.0/go.mod h1:Pq2IXbl9/ytYHfrIAd7rIVtZQ2ezdIhZfvdqOizDeWY=
55+
github.com/alphadose/itogami v0.4.1-0.20221016160904-c25d0a36bfe7 h1:YaYmhNYMp53C10C+Fzd4OMuwUjO7R9e/F17OOCHcjqQ=
56+
github.com/alphadose/itogami v0.4.1-0.20221016160904-c25d0a36bfe7/go.mod h1:djI1vuSef+44Ik61DOZcd04m7eyyY0kqcf7LmH98vC4=
5557
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
5658
github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
5759
github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=

‎logs/writer.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ func NewWriter(ctx context.Context, addr string, stdout bool) (writer *Writer, e
7575
log.Infof("[writer] create writer for %s success", addr)
7676
}
7777

78-
_ = utils.Pool.Submit(func() { writer.keepalive(ctx) })
78+
utils.Pool.Submit(func() { writer.keepalive(ctx) })
7979
return writer, nil
8080
}
8181

‎logs/writer_test.go‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,12 @@ import (
77
"time"
88

99
"github.com/projecteru2/agent/types"
10-
"github.com/projecteru2/agent/utils"
1110

1211
"github.com/stretchr/testify/assert"
1312
)
1413

1514
func TestNewWriterWithUDP(t *testing.T) {
1615
ctx, cancel := context.WithCancel(context.Background())
17-
utils.NewPool(1000)
1816
defer cancel()
1917
// udp writer
2018
addr := "udp://127.0.0.1:23456"

‎manager/node/heartbeat.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,15 @@ func (m *Manager) heartbeat(ctx context.Context) {
1515
if m.config.HeartbeatInterval <= 0 {
1616
return
1717
}
18-
_ = utils.Pool.Submit(func() { m.nodeStatusReport(ctx) })
18+
utils.Pool.Submit(func() { m.nodeStatusReport(ctx) })
1919

2020
tick := time.NewTicker(time.Duration(m.config.HeartbeatInterval) * time.Second)
2121
defer tick.Stop()
2222

2323
for {
2424
select {
2525
case <-tick.C:
26-
_ = utils.Pool.Submit(func() { m.nodeStatusReport(ctx) })
26+
utils.Pool.Submit(func() { m.nodeStatusReport(ctx) })
2727
case <-ctx.Done():
2828
return
2929
}

‎manager/node/heartbeat_test.go‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@ import (
77

88
runtimemocks "github.com/projecteru2/agent/runtime/mocks"
99
storemocks "github.com/projecteru2/agent/store/mocks"
10-
"github.com/projecteru2/agent/utils"
1110

1211
"github.com/stretchr/testify/assert"
1312
)
@@ -35,7 +34,6 @@ func TestNodeStatusReport(t *testing.T) {
3534
func TestHeartbeat(t *testing.T) {
3635
ctx, cancel := context.WithCancel(context.Background())
3736
defer cancel()
38-
utils.NewPool(1000)
3937
manager := newMockNodeManager(t)
4038
store := manager.store.(*storemocks.MockStore)
4139

‎manager/node/manager.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ func NewManager(ctx context.Context, config *types.Config) (*Manager, error) {
7676
// Run runs a node manager
7777
func (m *Manager) Run(ctx context.Context) error {
7878
log.Info("[NodeManager] start node status heartbeat")
79-
_ = utils.Pool.Submit(func() { m.heartbeat(ctx) })
79+
utils.Pool.Submit(func() { m.heartbeat(ctx) })
8080

8181
<-ctx.Done()
8282
log.Info("[NodeManager] exiting")

‎manager/workload/attach.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ func (m *Manager) attach(ctx context.Context, ID string) {
5757
log.Infof("[attach] attach %s workload %s success", workloadName, ID)
5858

5959
// attach metrics
60-
_ = utils.Pool.Submit(func() { m.runtimeClient.CollectWorkloadMetrics(ctx, ID) })
60+
utils.Pool.Submit(func() { m.runtimeClient.CollectWorkloadMetrics(ctx, ID) })
6161

6262
extra, err := m.runtimeClient.LogFieldsExtra(ctx, ID)
6363
if err != nil {
@@ -101,6 +101,6 @@ func (m *Manager) attach(ctx context.Context, ID string) {
101101
}
102102
wg.Add(2)
103103
defer wg.Wait()
104-
_ = utils.Pool.Submit(func() { pump("stdout", outr) })
105-
_ = utils.Pool.Submit(func() { pump("stderr", errr) })
104+
utils.Pool.Submit(func() { pump("stdout", outr) })
105+
utils.Pool.Submit(func() { pump("stderr", errr) })
106106
}

0 commit comments

Comments
 (0)