Commit 4399335d authored by muXxer's avatar muXxer

Fixed some race conditions

parent 906e4f71
......@@ -36,13 +36,15 @@ type PowClient struct {
Verbose bool // Print info messages
connection *grpc.ClientConn // Connection to the powsrv.io
connectionError error
connectionErrorOcurred chan bool
connectionErrorOcurred chan struct{}
powStream api.Pow_DoPowClient // Stream for DoPow
requestID uint64
requestIDLock *sync.Mutex
outgoingReq chan PendingPowRequest // Outgoing Requests to the powsrv.io
pendingRequests map[uint64]PendingPowRequest
pendingRequestsLock *sync.Mutex
signalClosed bool
signalChanQuit chan struct{}
}
func (client *PowClient) Init() error {
......@@ -58,7 +60,10 @@ func (client *PowClient) Init() error {
return err
}
client.connectionErrorOcurred = make(chan bool, 1)
client.connectionErrorOcurred = make(chan struct{})
client.signalClosed = false
client.signalChanQuit = make(chan struct{})
client.requestID = 0
client.requestIDLock = &sync.Mutex{}
......@@ -83,14 +88,20 @@ func (client *PowClient) Init() error {
}
func (client *PowClient) Close() {
client.pendingRequestsLock.Lock()
for _, pendingRequest := range client.pendingRequests {
close(pendingRequest.powResponseChan)
}
client.pendingRequestsLock.Unlock()
if !client.signalClosed {
client.signalClosed = true
close(client.signalChanQuit)
client.pendingRequestsLock.Lock()
for _, pendingRequest := range client.pendingRequests {
close(pendingRequest.powResponseChan)
}
client.pendingRequestsLock.Unlock()
client.powStream.CloseSend()
client.connection.Close()
client.powStream.CloseSend()
client.connection.Close()
}
}
func (client *PowClient) printMsg(msg string) {
......@@ -119,7 +130,6 @@ func (client *PowClient) handleConnection() {
// read done.
client.printMsg("Connection to server closed")
client.connectionError = err
client.connectionErrorOcurred <- true
close(client.connectionErrorOcurred)
return
}
......@@ -128,7 +138,6 @@ func (client *PowClient) handleConnection() {
close(client.outgoingReq)
client.printMsg(fmt.Sprintf("Failed to receive: %v", err))
client.connectionError = err
client.connectionErrorOcurred <- true
close(client.connectionErrorOcurred)
return
}
......@@ -138,7 +147,9 @@ func (client *PowClient) handleConnection() {
client.pendingRequestsLock.Lock()
pendingPowRequest, ok := client.pendingRequests[incomingRes.Identifier]
if ok {
pendingPowRequest.powResponseChan <- incomingRes
if !client.signalClosed {
pendingPowRequest.powResponseChan <- incomingRes
}
delete(client.pendingRequests, incomingRes.Identifier)
}
client.pendingRequestsLock.Unlock()
......@@ -186,6 +197,10 @@ func (client *PowClient) PowFunc(trytes trinary.Trytes, minWeightMagnitude int,
client.outgoingReq <- pendingPowRequest
select {
case <-client.signalChanQuit:
return "", errors.New("PowClient was closed")
case powResponse, hasMore := <-pendingPowRequest.powResponseChan:
if hasMore {
// Got response
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment