-
Notifications
You must be signed in to change notification settings - Fork 1
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
Showing
18 changed files
with
825 additions
and
5 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,8 @@ | ||
| # pkg | ||
| go common package | ||
|
|
||
| - Feature | ||
| - dispatcher | ||
| - log | ||
| - pool | ||
| - rate |
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,2 @@ | ||
| # Dispatch Goroutine | ||
| ref: github.com/istio/istio/mixer/pkg/runtime/dispatcher |
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,17 @@ | ||
| package dispatcher | ||
|
|
||
| type dispatchState struct { | ||
| session *session | ||
| handler DispatchHandler | ||
| err error | ||
| } | ||
|
|
||
| func (ds *dispatchState) clear() { | ||
| ds.session = nil | ||
| ds.err = nil | ||
| } | ||
|
|
||
| func (ds *dispatchState) invokeHandler(p interface{}) { | ||
| ds.err = ds.handler(p) | ||
| ds.session.completed <- ds | ||
| } |
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,62 @@ | ||
| package dispatcher | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/hb-go/pkg/pool" | ||
| ) | ||
|
|
||
| type DispatchHandler func(interface{}) error | ||
|
|
||
| type Dispatcher struct { | ||
| // pool of sessions | ||
| sessionPool sync.Pool | ||
|
|
||
| // pool of dispatch states | ||
| statePool sync.Pool | ||
|
|
||
| // pool of goroutines | ||
| gp *pool.GoroutinePool | ||
| } | ||
|
|
||
| func NewDispatcher(handlerGP *pool.GoroutinePool) *Dispatcher { | ||
| d := &Dispatcher{ | ||
| gp: handlerGP, | ||
| } | ||
|
|
||
| d.sessionPool.New = func() interface{} { return &session{} } | ||
| d.statePool.New = func() interface{} { return &dispatchState{} } | ||
| return d | ||
| } | ||
|
|
||
| func (d *Dispatcher) Dispatch(handlers ...DispatchHandler) error { | ||
| s := d.getSession() | ||
|
|
||
| s.handlers = handlers | ||
| err := s.dispatch() | ||
|
|
||
| d.putSession(s) | ||
| return err | ||
| } | ||
|
|
||
| func (d *Dispatcher) getSession() *session { | ||
| s := d.sessionPool.Get().(*session) | ||
| s.dispatcher = d | ||
| return s | ||
| } | ||
|
|
||
| func (d *Dispatcher) putSession(s *session) { | ||
| s.clear() | ||
| d.sessionPool.Put(s) | ||
| } | ||
|
|
||
| func (d *Dispatcher) getDispatchState() *dispatchState { | ||
| ds := d.statePool.Get().(*dispatchState) | ||
|
|
||
| return ds | ||
| } | ||
|
|
||
| func (d *Dispatcher) putDispatchState(ds *dispatchState) { | ||
| ds.clear() | ||
| d.statePool.Put(ds) | ||
| } |
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,78 @@ | ||
| package dispatcher | ||
|
|
||
| import ( | ||
| "log" | ||
|
|
||
| "github.com/hashicorp/go-multierror" | ||
| ) | ||
|
|
||
| const queueAllocSize = 64 | ||
|
|
||
| type session struct { | ||
| dispatcher *Dispatcher | ||
| handlers []DispatchHandler | ||
|
|
||
| activeDispatches int | ||
| completed chan *dispatchState | ||
|
|
||
| err error | ||
| } | ||
|
|
||
| func (s *session) clear() { | ||
| s.dispatcher = nil | ||
| s.activeDispatches = 0 | ||
| s.handlers = nil | ||
| s.err = nil | ||
|
|
||
| // Drain the channel | ||
| exit := false | ||
| for !exit { | ||
| select { | ||
| case <-s.completed: | ||
| log.Printf("Leaked dispatch state discovered!") | ||
| continue | ||
| default: | ||
| exit = true | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (s *session) ensureParallelism(minParallelism int) { | ||
| // Resize the channel to accommodate the parallelism, if necessary. | ||
| if cap(s.completed) < minParallelism { | ||
| allocSize := ((minParallelism / queueAllocSize) + 1) * queueAllocSize | ||
| s.completed = make(chan *dispatchState, allocSize) | ||
| } | ||
| } | ||
|
|
||
| func (s *session) dispatch() error { | ||
| s.ensureParallelism(len(s.handlers)) | ||
|
|
||
| for _, h := range s.handlers { | ||
| ds := s.dispatcher.getDispatchState() | ||
| ds.handler = h | ||
| s.dispatchToHandler(ds) | ||
| } | ||
|
|
||
| s.waitForDispatched() | ||
| return s.err | ||
| } | ||
|
|
||
| func (s *session) dispatchToHandler(ds *dispatchState) { | ||
| s.activeDispatches++ | ||
| ds.session = s | ||
| s.dispatcher.gp.ScheduleWork(ds.invokeHandler, nil) | ||
| } | ||
|
|
||
| func (s *session) waitForDispatched() { | ||
| for s.activeDispatches > 0 { | ||
| state := <-s.completed | ||
| s.activeDispatches-- | ||
|
|
||
| if state.err != nil { | ||
| s.err = multierror.Append(s.err, state.err) | ||
| } | ||
|
|
||
| s.dispatcher.putDispatchState(state) | ||
| } | ||
| } |
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,16 @@ | ||
| package dispatcher | ||
|
|
||
| import ( | ||
| "context" | ||
| "google.golang.org/grpc" | ||
| ) | ||
|
|
||
| type ( | ||
| NewClientFunc func(*grpc.ClientConn) interface{} | ||
| WorkerFunc func(interface{}) error | ||
| ) | ||
|
|
||
| type Service struct { | ||
| Ctx context.Context | ||
| Call func(*grpc.ClientConn) interface{} | ||
| } |
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
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,95 @@ | ||
| package log | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "log" | ||
| "os" | ||
| ) | ||
|
|
||
| type defaultLogger struct { | ||
| *log.Logger | ||
| calldepth int | ||
| } | ||
|
|
||
| func NewLogger() *defaultLogger { | ||
| return &defaultLogger{ | ||
| Logger: log.New(os.Stderr, "", log.Ldate|log.Ltime|log.Lmicroseconds|log.Llongfile), | ||
| calldepth: 3, | ||
| } | ||
| } | ||
|
|
||
| func (l *defaultLogger) SetCalldepth(calldepth int) { | ||
| l.calldepth = calldepth | ||
| } | ||
|
|
||
| func (l *defaultLogger) Debug(v ...interface{}) { | ||
| l.output(DEBUG, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Debugf(format string, v ...interface{}) { | ||
| l.outputf(DEBUG, format, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Info(v ...interface{}) { | ||
| l.output(INFO, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Infof(format string, v ...interface{}) { | ||
| l.outputf(INFO, format, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Warn(v ...interface{}) { | ||
| l.output(WARN, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Warnf(format string, v ...interface{}) { | ||
| l.outputf(WARN, format, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Error(v ...interface{}) { | ||
| l.output(ERROR, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Errorf(format string, v ...interface{}) { | ||
| l.outputf(ERROR, format, v...) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Fatal(v ...interface{}) { | ||
| l.output(fatalLvl, v...) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Fatalf(format string, v ...interface{}) { | ||
| l.outputf(fatalLvl, format, v...) | ||
| os.Exit(1) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Panic(v ...interface{}) { | ||
| s := fmt.Sprint(v...) | ||
| l.output(panicLvl, s) | ||
| panic(s) | ||
| } | ||
|
|
||
| func (l *defaultLogger) Panicf(format string, v ...interface{}) { | ||
| s := fmt.Sprintf(format, v...) | ||
| l.output(panicLvl, s) | ||
| panic(s) | ||
| } | ||
|
|
||
| func (l *defaultLogger) output(lvl Lvl, v ...interface{}) { | ||
| if lvl < level { | ||
| return | ||
| } | ||
| l.Output(l.calldepth, header(lvl, fmt.Sprint(v...))) | ||
| } | ||
|
|
||
| func (l *defaultLogger) outputf(lvl Lvl, format string, v ...interface{}) { | ||
| if lvl < level { | ||
| return | ||
| } | ||
| l.Output(l.calldepth, header(lvl, fmt.Sprintf(format, v...))) | ||
| } | ||
|
|
||
| func header(lvl Lvl, msg string) string { | ||
| return fmt.Sprintf("[%s] %s", lvl.String(), msg) | ||
| } |
Oops, something went wrong.