Skip to content

readloop speedup #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
toli-belo wants to merge 2 commits into
base: master
Choose a base branch
from
Open

readloop speedup #10

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a465fcc
Remove extra rw checking and test it out
Apr 23, 2019
dea540e
Add pooling of read buffer in worker
toli-belo Apr 24, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 18 additions & 13 deletions worker/agent.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,16 +24,18 @@ type agent struct {
rw *bufio.ReadWriter
worker *Worker
in chan []byte
readBuffPool *sync.Pool
net, addr string
}

// Create the agent of job server.
func newAgent(net, addr string, worker *Worker) (a *agent, err error) {
a = &agent{
net: net,
addr: addr,
worker: worker,
in: make(chan []byte, rt.QueueSize),
net: net,
addr: addr,
worker: worker,
in: make(chan []byte, rt.QueueSize),
readBuffPool: &sync.Pool{New: func() interface{} { return rt.NewBuffer(rt.BufferSize) }},
}
return
}
Expand All @@ -55,11 +57,14 @@ func (a *agent) work() {
var err error
var data, leftdata []byte
startRw := a.loadRw()
if startRw == nil {
return // we are reconnecting at the moment, a new go routine will run on work() when reconnected...
}

// exit the loop if connection has been replaced because reconnect will launch a new work() thread
for startRw == a.loadRw() && !a.worker.isShuttingDown() {
for !a.worker.isShuttingDown() {

if data, err = a.read(); err != nil {
if data, err = a.read(startRw); err != nil {
if opErr, ok := err.(*net.OpError); ok {
if opErr.Temporary() {
a.worker.Log(Info, "opErr.Temporary():", a.addr)
Expand All @@ -68,12 +73,12 @@ func (a *agent) work() {
a.worker.Log(Info, "got permanent network error with server:", a.addr, "comm thread exiting.")
a.reconnectError(err)
// else - we're probably dc'ing due to a Close()
break
return
}
} else {
a.worker.Log(Info, "got error", err.Error(), "with server:", a.addr, "comm thread exiting...")
a.reconnectError(err)
break
return
}
}
if len(leftdata) > 0 { // some data left for processing
Expand All @@ -86,7 +91,7 @@ func (a *agent) work() {
for {
if inpack, l, err = decodeInPack(data); err != nil {
a.reconnectError(err)
break
return
} else {
leftdata = nil
inpack.a = a
Expand All @@ -99,7 +104,6 @@ func (a *agent) work() {
}
}
}

}
}

Expand Down Expand Up @@ -247,13 +251,14 @@ func (a *agent) Connect() {
}

// read length bytes from the socket
func (a *agent) read() (data []byte, err error) {
func (a *agent) read(myRw *bufio.ReadWriter) (data []byte, err error) {
n := 0

tmp := rt.NewBuffer(rt.BufferSize)
tmp := a.readBuffPool.Get().([]byte)
defer a.readBuffPool.Put(tmp)

var buf bytes.Buffer

myRw := a.loadRw()
// read the header so we can get the length of the data
if n, err = myRw.Read(tmp); err != nil {
return
Expand Down