This repository has been archived by the owner on Dec 22, 2023. It is now read-only.
Use channel in messaging between goroutine in stead of zmq.REQ #168
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This will properly handle worker crash on init by timeout. And in case of all workers crash, the channeler will not crash. Just all RPC will timeout. Also add mutex lock to make the workerQueue goroutine safe.
Since it will throw to the end-user. Providing meaningful exception with user term is better.
|
timeout when running the test with the i hope these are helpful: |
cheungpat
changed the title
cheungpat
reviewed
Oct 23, 2016
| } | ||
| heartbeatAt = time.Now().Add(HeartbeatInterval) | ||
| } | ||
|
|
||
| lb.workers.Purge() | ||
| lb.workers.Unlock() |
There was a problem hiding this comment.
should defer lb.workers.Unlock()
There was a problem hiding this comment.
This is within the for control. And the defer is bind to func closure scope. If we want to use defer here. We need to create a large function scope. Which I don't prefer.
| delete(lb.addressChan, address) | ||
| respChan <- []byte{0} | ||
| case <-lb.stop: | ||
| break ChannelerLoop |
There was a problem hiding this comment.
I think simply returns here
| break ChannelerLoop | ||
| } | ||
| } | ||
| lb.logger.Infof("zmq channler stopped %p!\n", lb) |
There was a problem hiding this comment.
this should be defer lb.logger.Infof(...)
| frames = append([][]byte{lb.workers.Next()}, frames...) | ||
| lb.backend.SendMessage(frames) | ||
| lb.logger.Debugf("zmq/broker: server => plugin: %#x, %s\n", frames[0], frames) | ||
| backend.SendMessage(frames) | ||
| case nil: |
There was a problem hiding this comment.
you can also quit the loop by checking the stop channel here
There was a problem hiding this comment.
No. This switch case is only zmq socket. Not go chan. So we cannot consume the chan here.
| timeout chan string | ||
| workers workerQueue | ||
| logger *logrus.Entry | ||
| stop chan int |
There was a problem hiding this comment.
while you still have fresh memory what this struct does, could you add some comments to the meaning of the struct members?
There was a problem hiding this comment.
if you have still got time: there are two components to this broker: one is a broker and the other is the channeler. There are some struct members that are only used in one component but not the other, perhaps we could make the distinction more obvious by creating separate structs for these two components.
|
if you enable edits on the branch for the pull request, I could make assorted changes such as typos on the pull request on your behalf |
cheungpat
reviewed
Oct 24, 2016
| // workerQueue is not goroutine safe. To use it safely across goroutine. | ||
| // Please use the Lock/Unlock interace before manupliate the queue item via | ||
| // methods like Add/Tick/Purge. | ||
| // Comsuming method Next is the only method will acquire the mutex lock |
|
The timeout is because Travis running the test with Setting |
|
Making another PR. |
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This will resolve socket exhaust problem on high concurrency.
connects #160