Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Contribute to GitLab
Sign in / Register
Toggle navigation
G
getty
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
wei.xuan
getty
Commits
7886bc27
Unverified
Commit
7886bc27
authored
Jun 11, 2019
by
望哥
Committed by
GitHub
Jun 11, 2019
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #6 from divebomb/master
Add: task worker pool
parents
8b0e100a
58156632
Show whitespace changes
Inline
Side-by-side
Showing
7 changed files
with
301 additions
and
55 deletions
+301
-55
change_log.md
change_log.md
+4
-0
client.go
client.go
+1
-1
getty.go
getty.go
+3
-0
options.go
options.go
+55
-0
server.go
server.go
+0
-2
session.go
session.go
+107
-52
task_pool.go
task_pool.go
+131
-0
No files found.
change_log.md
View file @
7886bc27
...
@@ -14,6 +14,10 @@
...
@@ -14,6 +14,10 @@
## develop history ##
## develop history ##
---
---
-
2019/06/08
> Improvement
*
add task worker pool
-
2019/06/07
-
2019/06/07
> Improvement
> Improvement
*
use time.After instead of wheel.After
*
use time.After instead of wheel.After
...
...
client.go
View file @
7886bc27
...
@@ -350,7 +350,6 @@ func (c *client) connect() {
...
@@ -350,7 +350,6 @@ func (c *client) connect() {
}
}
err
=
c
.
newSession
(
ss
)
err
=
c
.
newSession
(
ss
)
if
err
==
nil
{
if
err
==
nil
{
// ss.RunEventLoop()
ss
.
(
*
session
)
.
run
()
ss
.
(
*
session
)
.
run
()
c
.
Lock
()
c
.
Lock
()
if
c
.
ssMap
==
nil
{
if
c
.
ssMap
==
nil
{
...
@@ -421,6 +420,7 @@ func (c *client) stop() {
...
@@ -421,6 +420,7 @@ func (c *client) stop() {
s
.
Close
()
s
.
Close
()
}
}
c
.
ssMap
=
nil
c
.
ssMap
=
nil
c
.
Unlock
()
c
.
Unlock
()
})
})
}
}
...
...
getty.go
View file @
7886bc27
...
@@ -13,7 +13,9 @@ import (
...
@@ -13,7 +13,9 @@ import (
"compress/flate"
"compress/flate"
"net"
"net"
"time"
"time"
)
import
(
perrors
"github.com/pkg/errors"
perrors
"github.com/pkg/errors"
)
)
...
@@ -152,6 +154,7 @@ type Session interface {
...
@@ -152,6 +154,7 @@ type Session interface {
SetRQLen
(
int
)
SetRQLen
(
int
)
SetWQLen
(
int
)
SetWQLen
(
int
)
SetWaitTime
(
time
.
Duration
)
SetWaitTime
(
time
.
Duration
)
SetTaskPool
(
*
TaskPool
)
GetAttribute
(
interface
{})
interface
{}
GetAttribute
(
interface
{})
interface
{}
SetAttribute
(
interface
{},
interface
{})
SetAttribute
(
interface
{},
interface
{})
...
...
options.go
View file @
7886bc27
...
@@ -10,6 +10,10 @@
...
@@ -10,6 +10,10 @@
package
getty
package
getty
import
(
"fmt"
)
/////////////////////////////////////////
/////////////////////////////////////////
// Server Options
// Server Options
/////////////////////////////////////////
/////////////////////////////////////////
...
@@ -105,3 +109,54 @@ func WithRootCertificateFile(cert string) ClientOption {
...
@@ -105,3 +109,54 @@ func WithRootCertificateFile(cert string) ClientOption {
o
.
cert
=
cert
o
.
cert
=
cert
}
}
}
}
/////////////////////////////////////////
// Task Pool Options
/////////////////////////////////////////
type
TaskPoolOptions
struct
{
tQLen
int
// task queue length
tQNumber
int
// task queue number
tQPoolSize
int
// task pool size
}
func
(
o
*
TaskPoolOptions
)
validate
()
{
if
o
.
tQPoolSize
<
1
{
panic
(
fmt
.
Sprintf
(
"[getty][task_pool] illegal pool size %d"
,
o
.
tQPoolSize
))
}
if
o
.
tQLen
<
1
{
o
.
tQLen
=
defaultTaskQLen
}
if
o
.
tQNumber
<
1
{
o
.
tQNumber
=
defaultTaskQNumber
}
if
o
.
tQNumber
>
o
.
tQPoolSize
{
o
.
tQNumber
=
o
.
tQPoolSize
}
}
type
TaskPoolOption
func
(
*
TaskPoolOptions
)
// @size is the task queue pool size
func
WithTaskPoolTaskPoolSize
(
size
int
)
TaskPoolOption
{
return
func
(
o
*
TaskPoolOptions
)
{
o
.
tQPoolSize
=
size
}
}
// @length is the task queue length
func
WithTaskPoolTaskQueueLength
(
length
int
)
TaskPoolOption
{
return
func
(
o
*
TaskPoolOptions
)
{
o
.
tQLen
=
length
}
}
// @number is the task queue number
func
WithTaskPoolTaskQueueNumber
(
number
int
)
TaskPoolOption
{
return
func
(
o
*
TaskPoolOptions
)
{
o
.
tQNumber
=
number
}
}
server.go
View file @
7886bc27
...
@@ -252,7 +252,6 @@ func (s *server) runTcpEventLoop(newSession NewSessionCallback) {
...
@@ -252,7 +252,6 @@ func (s *server) runTcpEventLoop(newSession NewSessionCallback) {
continue
continue
}
}
delay
=
0
delay
=
0
// client.RunEventLoop()
client
.
(
*
session
)
.
run
()
client
.
(
*
session
)
.
run
()
}
}
}()
}()
...
@@ -323,7 +322,6 @@ func (s *wsHandler) serveWSRequest(w http.ResponseWriter, r *http.Request) {
...
@@ -323,7 +322,6 @@ func (s *wsHandler) serveWSRequest(w http.ResponseWriter, r *http.Request) {
if
ss
.
(
*
session
)
.
maxMsgLen
>
0
{
if
ss
.
(
*
session
)
.
maxMsgLen
>
0
{
conn
.
SetReadLimit
(
int64
(
ss
.
(
*
session
)
.
maxMsgLen
))
conn
.
SetReadLimit
(
int64
(
ss
.
(
*
session
)
.
maxMsgLen
))
}
}
// ss.RunEventLoop()
ss
.
(
*
session
)
.
run
()
ss
.
(
*
session
)
.
run
()
}
}
...
...
session.go
View file @
7886bc27
...
@@ -29,6 +29,7 @@ const (
...
@@ -29,6 +29,7 @@ const (
netIOTimeout
=
1e9
// 1s
netIOTimeout
=
1e9
// 1s
period
=
60
*
1e9
// 1 minute
period
=
60
*
1e9
// 1 minute
pendingDuration
=
3e9
pendingDuration
=
3e9
defaultQLen
=
1024
defaultSessionName
=
"session"
defaultSessionName
=
"session"
defaultTCPSessionName
=
"tcp-session"
defaultTCPSessionName
=
"tcp-session"
defaultUDPSessionName
=
"udp-session"
defaultUDPSessionName
=
"udp-session"
...
@@ -45,24 +46,36 @@ const (
...
@@ -45,24 +46,36 @@ const (
type
session
struct
{
type
session
struct
{
name
string
name
string
endPoint
EndPoint
endPoint
EndPoint
maxMsgLen
int32
// net read Write
// net read Write
Connection
Connection
// pkgHandler ReadWriter
listener
EventListener
// codec
reader
Reader
// @reader should be nil when @conn is a gettyWSConn object.
reader
Reader
// @reader should be nil when @conn is a gettyWSConn object.
writer
Writer
writer
Writer
listener
EventListener
once
sync
.
Once
done
chan
struct
{}
// errFlag bool
period
time
.
Duration
// read & write
wait
time
.
Duration
rQ
chan
interface
{}
rQ
chan
interface
{}
wQ
chan
interface
{}
wQ
chan
interface
{}
// handle logic
maxMsgLen
int32
// task queue
tPool
*
TaskPool
// heartbeat
period
time
.
Duration
// done
wait
time
.
Duration
once
sync
.
Once
done
chan
struct
{}
// attribute
// attribute
attrs
*
ValuesContext
attrs
*
ValuesContext
// goroutines sync
// goroutines sync
grNum
int32
grNum
int32
lock
sync
.
RWMutex
lock
sync
.
RWMutex
...
@@ -72,10 +85,14 @@ func newSession(endPoint EndPoint, conn Connection) *session {
...
@@ -72,10 +85,14 @@ func newSession(endPoint EndPoint, conn Connection) *session {
ss
:=
&
session
{
ss
:=
&
session
{
name
:
defaultSessionName
,
name
:
defaultSessionName
,
endPoint
:
endPoint
,
endPoint
:
endPoint
,
maxMsgLen
:
maxReadBufLen
,
Connection
:
conn
,
Connection
:
conn
,
done
:
make
(
chan
struct
{}),
maxMsgLen
:
maxReadBufLen
,
period
:
period
,
period
:
period
,
done
:
make
(
chan
struct
{}),
wait
:
pendingDuration
,
wait
:
pendingDuration
,
attrs
:
NewValuesContext
(
nil
),
attrs
:
NewValuesContext
(
nil
),
}
}
...
@@ -115,7 +132,6 @@ func (s *session) Reset() {
...
@@ -115,7 +132,6 @@ func (s *session) Reset() {
s
.
name
=
defaultSessionName
s
.
name
=
defaultSessionName
s
.
once
=
sync
.
Once
{}
s
.
once
=
sync
.
Once
{}
s
.
done
=
make
(
chan
struct
{})
s
.
done
=
make
(
chan
struct
{})
// s.errFlag = false
s
.
period
=
period
s
.
period
=
period
s
.
wait
=
pendingDuration
s
.
wait
=
pendingDuration
s
.
attrs
=
NewValuesContext
(
nil
)
s
.
attrs
=
NewValuesContext
(
nil
)
...
@@ -190,30 +206,51 @@ func (s *session) IsClosed() bool {
...
@@ -190,30 +206,51 @@ func (s *session) IsClosed() bool {
}
}
// set maximum pacakge length of every pacakge in (EventListener)OnMessage(@pkgs)
// set maximum pacakge length of every pacakge in (EventListener)OnMessage(@pkgs)
func
(
s
*
session
)
SetMaxMsgLen
(
length
int
)
{
s
.
maxMsgLen
=
int32
(
length
)
}
func
(
s
*
session
)
SetMaxMsgLen
(
length
int
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
maxMsgLen
=
int32
(
length
)
}
// set session name
// set session name
func
(
s
*
session
)
SetName
(
name
string
)
{
s
.
name
=
name
}
func
(
s
*
session
)
SetName
(
name
string
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
name
=
name
}
// set EventListener
// set EventListener
func
(
s
*
session
)
SetEventListener
(
listener
EventListener
)
{
func
(
s
*
session
)
SetEventListener
(
listener
EventListener
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
listener
=
listener
s
.
listener
=
listener
}
}
// set package handler
// set package handler
func
(
s
*
session
)
SetPkgHandler
(
handler
ReadWriter
)
{
func
(
s
*
session
)
SetPkgHandler
(
handler
ReadWriter
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
reader
=
handler
s
.
reader
=
handler
s
.
writer
=
handler
s
.
writer
=
handler
// s.pkgHandler = handler
}
}
// set Reader
// set Reader
func
(
s
*
session
)
SetReader
(
reader
Reader
)
{
func
(
s
*
session
)
SetReader
(
reader
Reader
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
reader
=
reader
s
.
reader
=
reader
}
}
// set Writer
// set Writer
func
(
s
*
session
)
SetWriter
(
writer
Writer
)
{
func
(
s
*
session
)
SetWriter
(
writer
Writer
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
writer
=
writer
s
.
writer
=
writer
}
}
...
@@ -224,8 +261,8 @@ func (s *session) SetCronPeriod(period int) {
...
@@ -224,8 +261,8 @@ func (s *session) SetCronPeriod(period int) {
}
}
s
.
lock
.
Lock
()
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
period
=
time
.
Duration
(
period
)
*
time
.
Millisecond
s
.
period
=
time
.
Duration
(
period
)
*
time
.
Millisecond
s
.
lock
.
Unlock
()
}
}
// set @session's read queue size
// set @session's read queue size
...
@@ -235,9 +272,9 @@ func (s *session) SetRQLen(readQLen int) {
...
@@ -235,9 +272,9 @@ func (s *session) SetRQLen(readQLen int) {
}
}
s
.
lock
.
Lock
()
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
rQ
=
make
(
chan
interface
{},
readQLen
)
s
.
rQ
=
make
(
chan
interface
{},
readQLen
)
s
.
lock
.
Unlock
()
log
.
Debug
(
"%s, [session.SetRQLen] rQ{len:%d, cap:%d}"
,
s
.
Stat
(),
len
(
s
.
rQ
),
cap
(
s
.
rQ
))
log
.
Debugf
(
"%s, [session.SetRQLen] rQ{len:%d, cap:%d}"
,
s
.
Stat
(),
len
(
s
.
rQ
),
cap
(
s
.
rQ
))
}
}
// set @session's Write queue size
// set @session's Write queue size
...
@@ -247,9 +284,9 @@ func (s *session) SetWQLen(writeQLen int) {
...
@@ -247,9 +284,9 @@ func (s *session) SetWQLen(writeQLen int) {
}
}
s
.
lock
.
Lock
()
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
wQ
=
make
(
chan
interface
{},
writeQLen
)
s
.
wQ
=
make
(
chan
interface
{},
writeQLen
)
s
.
lock
.
Unlock
()
log
.
Debug
(
"%s, [session.SetWQLen] wQ{len:%d, cap:%d}"
,
s
.
Stat
(),
len
(
s
.
wQ
),
cap
(
s
.
wQ
))
log
.
Debugf
(
"%s, [session.SetWQLen] wQ{len:%d, cap:%d}"
,
s
.
Stat
(),
len
(
s
.
wQ
),
cap
(
s
.
wQ
))
}
}
// set maximum wait time when session got error or got exit signal
// set maximum wait time when session got error or got exit signal
...
@@ -259,8 +296,16 @@ func (s *session) SetWaitTime(waitTime time.Duration) {
...
@@ -259,8 +296,16 @@ func (s *session) SetWaitTime(waitTime time.Duration) {
}
}
s
.
lock
.
Lock
()
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
wait
=
waitTime
s
.
wait
=
waitTime
s
.
lock
.
Unlock
()
}
// set task pool
func
(
s
*
session
)
SetTaskPool
(
p
*
TaskPool
)
{
s
.
lock
.
Lock
()
defer
s
.
lock
.
Unlock
()
s
.
tPool
=
p
}
}
// set attribute of key @session:key
// set attribute of key @session:key
...
@@ -400,12 +445,6 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) error {
...
@@ -400,12 +445,6 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) error {
// func (s *session) RunEventLoop() {
// func (s *session) RunEventLoop() {
func
(
s
*
session
)
run
()
{
func
(
s
*
session
)
run
()
{
if
s
.
rQ
==
nil
||
s
.
wQ
==
nil
{
errStr
:=
fmt
.
Sprintf
(
"session{name:%s, rQ:%#v, wQ:%#v}"
,
s
.
name
,
s
.
rQ
,
s
.
wQ
)
log
.
Error
(
errStr
)
panic
(
errStr
)
}
if
s
.
Connection
==
nil
||
s
.
listener
==
nil
||
s
.
writer
==
nil
{
if
s
.
Connection
==
nil
||
s
.
listener
==
nil
||
s
.
writer
==
nil
{
errStr
:=
fmt
.
Sprintf
(
"session{name:%s, conn:%#v, listener:%#v, writer:%#v}"
,
errStr
:=
fmt
.
Sprintf
(
"session{name:%s, conn:%#v, listener:%#v, writer:%#v}"
,
s
.
name
,
s
.
Connection
,
s
.
listener
,
s
.
writer
)
s
.
name
,
s
.
Connection
,
s
.
listener
,
s
.
writer
)
...
@@ -413,6 +452,14 @@ func (s *session) run() {
...
@@ -413,6 +452,14 @@ func (s *session) run() {
panic
(
errStr
)
panic
(
errStr
)
}
}
if
s
.
wQ
==
nil
{
s
.
wQ
=
make
(
chan
interface
{},
defaultQLen
)
}
if
s
.
rQ
==
nil
&&
s
.
tPool
==
nil
{
s
.
rQ
=
make
(
chan
interface
{},
defaultQLen
)
}
// call session opened
// call session opened
s
.
UpdateActive
()
s
.
UpdateActive
()
if
err
:=
s
.
listener
.
OnOpen
(
s
);
err
!=
nil
{
if
err
:=
s
.
listener
.
OnOpen
(
s
);
err
!=
nil
{
...
@@ -420,6 +467,7 @@ func (s *session) run() {
...
@@ -420,6 +467,7 @@ func (s *session) run() {
return
return
}
}
// start read/write gr
atomic
.
AddInt32
(
&
(
s
.
grNum
),
2
)
atomic
.
AddInt32
(
&
(
s
.
grNum
),
2
)
go
s
.
handleLoop
()
go
s
.
handleLoop
()
go
s
.
handlePackage
()
go
s
.
handlePackage
()
...
@@ -448,10 +496,8 @@ func (s *session) handleLoop() {
...
@@ -448,10 +496,8 @@ func (s *session) handleLoop() {
}
}
grNum
=
atomic
.
AddInt32
(
&
(
s
.
grNum
),
-
1
)
grNum
=
atomic
.
AddInt32
(
&
(
s
.
grNum
),
-
1
)
// if !s.errFlag {
s
.
listener
.
OnClose
(
s
)
s
.
listener
.
OnClose
(
s
)
// }
log
.
Info
(
"%s, [session.handleLoop] goroutine exit now, left gr num %d"
,
s
.
Stat
(),
grNum
)
log
.
Infof
(
"%s, [session.handleLoop] goroutine exit now, left gr num %d"
,
s
.
Stat
(),
grNum
)
s
.
gc
()
s
.
gc
()
}()
}()
...
@@ -516,6 +562,14 @@ LOOP:
...
@@ -516,6 +562,14 @@ LOOP:
}
}
}
}
func
(
s
*
session
)
addTask
(
pkg
interface
{})
{
if
s
.
tPool
!=
nil
{
s
.
tPool
.
AddTask
(
task
{
session
:
s
,
pkg
:
pkg
})
}
else
{
s
.
rQ
<-
pkg
}
}
func
(
s
*
session
)
handlePackage
()
{
func
(
s
*
session
)
handlePackage
()
{
var
(
var
(
err
error
err
error
...
@@ -593,8 +647,6 @@ func (s *session) handleTCPPackage() error {
...
@@ -593,8 +647,6 @@ func (s *session) handleTCPPackage() error {
break
break
}
}
log
.
Errorf
(
"%s, [session.conn.read] = error:%+v"
,
s
.
sessionToken
(),
err
)
log
.
Errorf
(
"%s, [session.conn.read] = error:%+v"
,
s
.
sessionToken
(),
err
)
// for (Codec)OnErr
// s.errFlag = true
exit
=
true
exit
=
true
}
}
break
break
...
@@ -619,8 +671,6 @@ func (s *session) handleTCPPackage() error {
...
@@ -619,8 +671,6 @@ func (s *session) handleTCPPackage() error {
if
err
!=
nil
{
if
err
!=
nil
{
log
.
Warnf
(
"%s, [session.handleTCPPackage] = len{%d}, error:%+v"
,
log
.
Warnf
(
"%s, [session.handleTCPPackage] = len{%d}, error:%+v"
,
s
.
sessionToken
(),
pkgLen
,
err
)
s
.
sessionToken
(),
pkgLen
,
err
)
// for (Codec)OnErr
// s.errFlag = true
exit
=
true
exit
=
true
break
break
}
}
...
@@ -630,7 +680,7 @@ func (s *session) handleTCPPackage() error {
...
@@ -630,7 +680,7 @@ func (s *session) handleTCPPackage() error {
}
}
// handle case 3
// handle case 3
s
.
UpdateActive
()
s
.
UpdateActive
()
s
.
rQ
<-
pkg
s
.
addTask
(
pkg
)
pktBuf
.
Next
(
pkgLen
)
pktBuf
.
Next
(
pkgLen
)
// continue to handle case 4
// continue to handle case 4
}
}
...
@@ -705,7 +755,7 @@ func (s *session) handleUDPPackage() error {
...
@@ -705,7 +755,7 @@ func (s *session) handleUDPPackage() error {
}
}
s
.
UpdateActive
()
s
.
UpdateActive
()
s
.
rQ
<-
UDPContext
{
Pkg
:
pkg
,
PeerAddr
:
addr
}
s
.
addTask
(
UDPContext
{
Pkg
:
pkg
,
PeerAddr
:
addr
})
}
}
return
perrors
.
WithStack
(
err
)
return
perrors
.
WithStack
(
err
)
...
@@ -735,7 +785,6 @@ func (s *session) handleWSPackage() error {
...
@@ -735,7 +785,6 @@ func (s *session) handleWSPackage() error {
if
err
!=
nil
{
if
err
!=
nil
{
log
.
Warnf
(
"%s, [session.handleWSPackage] = error{%+s}"
,
log
.
Warnf
(
"%s, [session.handleWSPackage] = error{%+s}"
,
s
.
sessionToken
(),
err
)
s
.
sessionToken
(),
err
)
// s.errFlag = true
return
perrors
.
WithStack
(
err
)
return
perrors
.
WithStack
(
err
)
}
}
s
.
UpdateActive
()
s
.
UpdateActive
()
...
@@ -749,9 +798,10 @@ func (s *session) handleWSPackage() error {
...
@@ -749,9 +798,10 @@ func (s *session) handleWSPackage() error {
s
.
sessionToken
(),
length
,
err
)
s
.
sessionToken
(),
length
,
err
)
continue
continue
}
}
s
.
rQ
<-
unmarshalPkg
s
.
addTask
(
unmarshalPkg
)
}
else
{
}
else
{
s
.
rQ
<-
pkg
s
.
addTask
(
pkg
)
}
}
}
}
...
@@ -784,10 +834,14 @@ func (s *session) gc() {
...
@@ -784,10 +834,14 @@ func (s *session) gc() {
s
.
lock
.
Lock
()
s
.
lock
.
Lock
()
if
s
.
attrs
!=
nil
{
if
s
.
attrs
!=
nil
{
s
.
attrs
=
nil
s
.
attrs
=
nil
if
s
.
wQ
!=
nil
{
close
(
s
.
wQ
)
close
(
s
.
wQ
)
s
.
wQ
=
nil
s
.
wQ
=
nil
}
if
s
.
rQ
!=
nil
{
close
(
s
.
rQ
)
close
(
s
.
rQ
)
s
.
rQ
=
nil
s
.
rQ
=
nil
}
s
.
Connection
.
close
((
int
)((
int64
)(
s
.
wait
)))
s
.
Connection
.
close
((
int
)((
int64
)(
s
.
wait
)))
}
}
s
.
lock
.
Unlock
()
s
.
lock
.
Unlock
()
...
@@ -797,5 +851,6 @@ func (s *session) gc() {
...
@@ -797,5 +851,6 @@ func (s *session) gc() {
// or (session)handleLoop automatically. It's thread safe.
// or (session)handleLoop automatically. It's thread safe.
func
(
s
*
session
)
Close
()
{
func
(
s
*
session
)
Close
()
{
s
.
stop
()
s
.
stop
()
log
.
Infof
(
"%s closed now. its current gr num is %d"
,
s
.
sessionToken
(),
atomic
.
LoadInt32
(
&
(
s
.
grNum
)))
log
.
Info
(
"%s closed now. its current gr num is %d"
,
s
.
sessionToken
(),
atomic
.
LoadInt32
(
&
(
s
.
grNum
)))
}
}
task_pool.go
0 → 100644
View file @
7886bc27
package
getty
import
(
"sync"
"sync/atomic"
)
const
(
defaultTaskQNumber
=
10
defaultTaskQLen
=
128
)
// task t
type
task
struct
{
session
*
session
pkg
interface
{}
}
// task pool: manage task ts
type
TaskPool
struct
{
TaskPoolOptions
idx
uint32
// round robin index
qArray
[]
chan
task
wg
sync
.
WaitGroup
once
sync
.
Once
done
chan
struct
{}
}
// build a task pool
func
NewTaskPool
(
opts
...
TaskPoolOption
)
*
TaskPool
{
var
tOpts
TaskPoolOptions
for
_
,
opt
:=
range
opts
{
opt
(
&
tOpts
)
}
tOpts
.
validate
()
p
:=
&
TaskPool
{
TaskPoolOptions
:
tOpts
,
qArray
:
make
([]
chan
task
,
tOpts
.
tQNumber
),
done
:
make
(
chan
struct
{}),
}
for
i
:=
0
;
i
<
p
.
tQNumber
;
i
++
{
p
.
qArray
[
i
]
=
make
(
chan
task
,
p
.
tQLen
)
}
return
p
}
// start task pool
func
(
p
*
TaskPool
)
start
()
{
for
i
:=
0
;
i
<
p
.
tQPoolSize
;
i
++
{
p
.
wg
.
Add
(
1
)
workerID
:=
i
q
:=
p
.
qArray
[
workerID
%
p
.
tQNumber
]
go
p
.
run
(
int
(
workerID
),
q
)
}
}
// worker
func
(
p
*
TaskPool
)
run
(
id
int
,
q
chan
task
)
{
defer
p
.
wg
.
Done
()
var
(
ok
bool
t
task
)
for
{
select
{
case
<-
p
.
done
:
if
0
<
len
(
q
)
{
log
.
Warn
(
"[getty][task_pool] task worker %d exit now while its task buffer length %d is greater than 0"
,
id
,
len
(
q
))
}
else
{
log
.
Info
(
"[getty][task_pool] task worker %d exit now"
,
id
)
}
return
case
t
,
ok
=
<-
q
:
if
ok
{
t
.
session
.
listener
.
OnMessage
(
t
.
session
,
t
.
pkg
)
}
}
}
}
// add task
func
(
p
*
TaskPool
)
AddTask
(
t
task
)
{
id
:=
atomic
.
AddUint32
(
&
p
.
idx
,
1
)
%
uint32
(
p
.
tQNumber
)
select
{
case
<-
p
.
done
:
return
case
p
.
qArray
[
id
]
<-
t
:
}
}
// stop all tasks
func
(
p
*
TaskPool
)
stop
()
{
select
{
case
<-
p
.
done
:
return
default
:
p
.
once
.
Do
(
func
()
{
close
(
p
.
done
)
})
}
}
// check whether the session has been closed.
func
(
p
*
TaskPool
)
IsClosed
()
bool
{
select
{
case
<-
p
.
done
:
return
true
default
:
return
false
}
}
func
(
p
*
TaskPool
)
Close
()
{
p
.
stop
()
p
.
wg
.
Wait
()
for
i
:=
range
p
.
qArray
{
close
(
p
.
qArray
[
i
])
}
}
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment