-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
urso
committed
Jul 13, 2017
1 parent
8b994f2
commit c407f56
Showing
3 changed files
with
217 additions
and
146 deletions.
There are no files selected for viewing
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 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,183 @@ | ||
| package membroker | ||
|
|
||
| import ( | ||
| "time" | ||
| ) | ||
|
|
||
| type eventLoop struct { | ||
| broker *Broker | ||
|
|
||
| // active broker API channels | ||
| events chan pushRequest | ||
| get chan getRequest | ||
| pubCancel chan producerCancelRequest | ||
|
|
||
| // ack handling | ||
| acks chan int // ackloop -> eventloop : total number of events ACKed by outputs | ||
| schedACKS chan chanList // eventloop -> ackloop : active list of batches to be acked | ||
| pendingACKs chanList // ordered list of active batches to be send to the ackloop | ||
| ackSeq uint // ack batch sequence number to validate ordering | ||
|
|
||
| // buffer flush timer state | ||
| timer *time.Timer | ||
| idleC <-chan time.Time | ||
| forceFlush bool | ||
| } | ||
|
|
||
| func newEventLoop(b *Broker) *eventLoop { | ||
| l := &eventLoop{ | ||
| broker: b, | ||
| events: b.events, | ||
| pubCancel: b.pubCancel, | ||
| acks: b.acks, | ||
| } | ||
|
|
||
| if to := b.idleTimeout; to > 0 { | ||
| // create initialy 'stopped' timer -> reset will be used | ||
| // on timer object, if flush timer becomes active. | ||
| l.timer = time.NewTimer(to) | ||
| if !l.timer.Stop() { | ||
| <-l.timer.C | ||
| } | ||
| } | ||
|
|
||
| return l | ||
| } | ||
|
|
||
| func (l *eventLoop) run() { | ||
| broker := l.broker | ||
| for { | ||
| select { | ||
| case <-broker.done: | ||
| return | ||
|
|
||
| case req := <-l.events: // producer pushing new event | ||
| l.handleInsert(&req) | ||
|
|
||
| case req := <-l.pubCancel: // producer cancellig active events | ||
| l.handleCancel(&req) | ||
|
|
||
| case req := <-l.get: // consumer asking for next batch | ||
| l.handleConsumer(&req) | ||
|
|
||
| case <-l.idleC: | ||
| // handle flush timer being triggered -> pending events can be forwarded via 'get' | ||
| l.enableFlushEvents() | ||
|
|
||
| case l.schedACKS <- l.pendingACKs: | ||
| // on send complete list of pending batches has been forwarded -> clear list and queue | ||
| l.schedACKS = nil | ||
| l.pendingACKs = chanList{} | ||
|
|
||
| case count := <-l.acks: | ||
| l.handleACK(count) | ||
|
|
||
| } | ||
|
|
||
| // update get and idle timer after state machine | ||
| l.get = broker.requests | ||
| if !l.forceFlush { | ||
| avail := broker.avail() | ||
| if avail == 0 || broker.totalAvail() < broker.minEvents { | ||
| l.get = nil | ||
|
|
||
| if avail > 0 { | ||
| l.startFlushTimer() | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (l *eventLoop) handleInsert(req *pushRequest) { | ||
| // log := l.broker.logger | ||
| // log.Debugf("push event: %v\t%v\t%p\n", req.event, req.seq, req.state) | ||
|
|
||
| if avail, ok := l.broker.insert(req); ok && avail == 0 { | ||
| // log.Debugf("buffer: all regions full") | ||
|
|
||
| // no more space to accept new events -> unset events queue for time being | ||
| l.events = nil | ||
| } | ||
| } | ||
|
|
||
| func (l *eventLoop) handleCancel(req *producerCancelRequest) { | ||
| // log := l.broker.logger | ||
| // log.Debug("handle cancel request") | ||
|
|
||
| var ( | ||
| removed int | ||
| broker = l.broker | ||
| ) | ||
|
|
||
| if st := req.state; st != nil { | ||
| st.cancelled = true | ||
| removed = broker.cancel(st) | ||
| } | ||
|
|
||
| // signal cancel request being finished | ||
| if req.resp != nil { | ||
| req.resp <- producerCancelResponse{ | ||
| removed: removed, | ||
| } | ||
| } | ||
|
|
||
| // re-enable pushRequest if buffer can take new events | ||
| if !broker.full() { | ||
| l.events = broker.events | ||
| } | ||
| } | ||
|
|
||
| func (l *eventLoop) handleConsumer(req *getRequest) { | ||
| // log := l.broker.logger | ||
|
|
||
| start, buf := l.broker.get(req.sz) | ||
| count := len(buf) | ||
| if count == 0 { | ||
| panic("empty batch returned") | ||
| } | ||
|
|
||
| // log.Debug("newACKChan: ", b.ackSeq, count) | ||
| ackCH := newACKChan(l.ackSeq, start, count) | ||
| l.ackSeq++ | ||
|
|
||
| req.resp <- getResponse{buf, ackCH} | ||
| l.pendingACKs.append(ackCH) | ||
| l.schedACKS = l.broker.scheduledACKs | ||
|
|
||
| l.stopFlushTimer() | ||
| } | ||
|
|
||
| func (l *eventLoop) handleACK(count int) { | ||
| // log := l.broker.logger | ||
| // log.Debug("receive buffer ack:", count) | ||
|
|
||
| // Give broker/buffer a chance to clean up most recent ACKs | ||
| // After handling ACKs some buffer has been freed up | ||
| // -> always reenable producers | ||
| broker := l.broker | ||
| broker.cleanACKs(count) | ||
| l.events = l.broker.events | ||
| } | ||
|
|
||
| func (l *eventLoop) enableFlushEvents() { | ||
| l.forceFlush = true | ||
| l.idleC = nil | ||
| } | ||
|
|
||
| func (l *eventLoop) stopFlushTimer() { | ||
| l.forceFlush = false | ||
| if l.idleC != nil { | ||
| l.idleC = nil | ||
| if !l.timer.Stop() { | ||
| <-l.timer.C | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (l *eventLoop) startFlushTimer() { | ||
| if l.idleC == nil && l.timer != nil { | ||
| l.timer.Reset(l.broker.idleTimeout) | ||
| l.idleC = l.timer.C | ||
| } | ||
| } |
Oops, something went wrong.