| 251 | } |
| 252 | |
| 253 | func (s *SFlow) dynWorkers() { |
| 254 | var load, nSeq, newWorkers, workers, n int |
| 255 | |
| 256 | tick := time.Tick(120 * time.Second) |
| 257 | |
| 258 | for { |
| 259 | <-tick |
| 260 | load = 0 |
| 261 | |
| 262 | for n = 0; n < 30; n++ { |
| 263 | time.Sleep(1 * time.Second) |
| 264 | load += len(sFlowUDPCh) |
| 265 | } |
| 266 | |
| 267 | if load > 15 { |
| 268 | |
| 269 | switch { |
| 270 | case load > 300: |
| 271 | newWorkers = 100 |
| 272 | case load > 200: |
| 273 | newWorkers = 60 |
| 274 | case load > 100: |
| 275 | newWorkers = 40 |
| 276 | default: |
| 277 | newWorkers = 30 |
| 278 | } |
| 279 | |
| 280 | workers = int(atomic.LoadInt32(&s.stats.Workers)) |
| 281 | if workers+newWorkers > maxWorkers { |
| 282 | logger.Println("sflow :: max out workers") |
| 283 | continue |
| 284 | } |
| 285 | |
| 286 | for n = 0; n < newWorkers; n++ { |
| 287 | go func() { |
| 288 | atomic.AddInt32(&s.stats.Workers, 1) |
| 289 | wQuit := make(chan struct{}) |
| 290 | s.pool <- wQuit |
| 291 | s.sFlowWorker(wQuit) |
| 292 | }() |
| 293 | } |
| 294 | |
| 295 | } |
| 296 | |
| 297 | if load == 0 { |
| 298 | nSeq++ |
| 299 | } else { |
| 300 | nSeq = 0 |
| 301 | continue |
| 302 | } |
| 303 | |
| 304 | if nSeq > 15 { |
| 305 | for n = 0; n < 10; n++ { |
| 306 | if len(s.pool) > s.workers { |
| 307 | atomic.AddInt32(&s.stats.Workers, -1) |
| 308 | wQuit := <-s.pool |
| 309 | close(wQuit) |
| 310 | } |