From 6e669d4611be43b12f5e7469ac7c2e72fceb0bbd Mon Sep 17 00:00:00 2001 From: Milos Gajdos Date: Thu, 13 Jun 2019 12:09:49 +0100 Subject: [PATCH] Reorganised source. Renamed files. No Code change. --- router/{default.go => default_router.go} | 0 router/default_table.go | 251 +++++++++++++++++++++++ router/options.go | 4 +- router/query.go | 3 + router/rib.go | 3 + router/{entry.go => route.go} | 3 + router/router.go | 27 +-- router/table.go | 244 ++-------------------- router/table_watcher.go | 6 +- 9 files changed, 295 insertions(+), 246 deletions(-) rename router/{default.go => default_router.go} (100%) create mode 100644 router/default_table.go rename router/{entry.go => route.go} (94%) diff --git a/router/default.go b/router/default_router.go similarity index 100% rename from router/default.go rename to router/default_router.go diff --git a/router/default_table.go b/router/default_table.go new file mode 100644 index 00000000..b5d6dd8f --- /dev/null +++ b/router/default_table.go @@ -0,0 +1,251 @@ +package router + +import ( + "fmt" + "hash" + "hash/fnv" + "strings" + "sync" + + "github.com/google/uuid" + "github.com/olekukonko/tablewriter" +) + +// TableOptions are routing table options +type TableOptions struct{} + +// table is in memory routing table +type table struct { + // opts are table options + opts TableOptions + // m stores routing table map + m map[string]map[uint64]Route + // h hashes route entries + h hash.Hash64 + // w is a list of table watchers + w map[string]*tableWatcher + sync.RWMutex +} + +// newTable creates default routing table and returns it +func newTable(opts ...TableOption) Table { + // default options + var options TableOptions + + // apply requested options + for _, o := range opts { + o(&options) + } + + h := fnv.New64() + h.Reset() + + return &table{ + opts: options, + m: make(map[string]map[uint64]Route), + w: make(map[string]*tableWatcher), + h: h, + } +} + +// Init initializes routing table with options +func (t *table) Init(opts ...TableOption) error { + for _, o := range opts { + o(&t.opts) + } + return nil +} + +// Options returns routing table options +func (t *table) Options() TableOptions { + return t.opts +} + +// Add adds a route to the routing table +func (t *table) Add(r Route) error { + t.Lock() + defer t.Unlock() + + destAddr := r.Options().DestAddr + sum := t.hash(r) + + if _, ok := t.m[destAddr]; !ok { + t.m[destAddr] = make(map[uint64]Route) + t.m[destAddr][sum] = r + go t.sendResult(&Result{Action: "add", Route: r}) + return nil + } + + if _, ok := t.m[destAddr][sum]; ok && r.Options().Policy == OverrideIfExists { + t.m[destAddr][sum] = r + go t.sendResult(&Result{Action: "update", Route: r}) + return nil + } + + if r.Options().Policy == IgnoreIfExists { + return nil + } + + return ErrDuplicateRoute +} + +// Remove removes the route from the routing table +func (t *table) Remove(r Route) error { + t.Lock() + defer t.Unlock() + + destAddr := r.Options().DestAddr + sum := t.hash(r) + + if _, ok := t.m[destAddr]; !ok { + return ErrRouteNotFound + } + + delete(t.m[destAddr], sum) + go t.sendResult(&Result{Action: "remove", Route: r}) + + return nil +} + +// Update updates routing table with new route +func (t *table) Update(r Route) error { + t.Lock() + defer t.Unlock() + + destAddr := r.Options().DestAddr + sum := t.hash(r) + + if _, ok := t.m[destAddr]; !ok { + return ErrRouteNotFound + } + + if _, ok := t.m[destAddr][sum]; ok { + t.m[destAddr][sum] = r + go t.sendResult(&Result{Action: "update", Route: r}) + return nil + } + + return ErrRouteNotFound +} + +// Lookup queries routing table and returns all routes that match it +func (t *table) Lookup(q Query) ([]Route, error) { + t.RLock() + defer t.RUnlock() + + var results []Route + + for destAddr, routes := range t.m { + if q.Options().DestAddr != "*" { + if q.Options().DestAddr != destAddr { + continue + } + for _, route := range routes { + if q.Options().Network == "*" || q.Options().Network == route.Options().Network { + results = append(results, route) + } + } + } + + if q.Options().DestAddr == "*" { + for _, route := range routes { + if q.Options().Network == "*" || q.Options().Network == route.Options().Network { + results = append(results, route) + } + } + } + } + + if len(results) == 0 && q.Options().Policy != DiscardNoRoute { + return nil, ErrRouteNotFound + } + + return results, nil +} + +// Watch returns routing table entry watcher +func (t *table) Watch(opts ...WatchOption) (Watcher, error) { + // by default watch everything + wopts := WatchOptions{ + DestAddr: "*", + Network: "*", + } + + for _, o := range opts { + o(&wopts) + } + + watcher := &tableWatcher{ + opts: wopts, + resChan: make(chan *Result, 10), + done: make(chan struct{}), + } + + t.Lock() + t.w[uuid.New().String()] = watcher + t.Unlock() + + return watcher, nil +} + +// sendResult sends rules to all subscribe watchers +func (t *table) sendResult(r *Result) { + t.RLock() + defer t.RUnlock() + + for _, w := range t.w { + select { + case w.resChan <- r: + case <-w.done: + } + } +} + +// Size returns the size of the routing table +func (t *table) Size() int { + t.RLock() + defer t.RUnlock() + + return len(t.m) +} + +// String returns debug information +func (t *table) String() string { + t.RLock() + defer t.RUnlock() + + // this will help us build routing table string + sb := &strings.Builder{} + + // create nice table printing structure + table := tablewriter.NewWriter(sb) + table.SetHeader([]string{"Destination", "Gateway", "Network", "Metric"}) + + for _, destRoute := range t.m { + for _, route := range destRoute { + strRoute := []string{ + route.Options().DestAddr, + route.Options().Gateway.Address(), + route.Options().Gateway.Network(), + fmt.Sprintf("%d", route.Options().Metric), + } + table.Append(strRoute) + } + } + + // render table into sb + table.Render() + + return sb.String() +} + +// hash hashes the route using router gateway and network address +func (t *table) hash(r Route) uint64 { + gwAddr := r.Options().Gateway.Address() + netAddr := r.Options().Network + + t.h.Reset() + t.h.Write([]byte(gwAddr + netAddr)) + + return t.h.Sum64() +} diff --git a/router/options.go b/router/options.go index 9774f237..c3b858a0 100644 --- a/router/options.go +++ b/router/options.go @@ -12,7 +12,7 @@ var ( DefaultNetworkAddress = ":9094" ) -// Options allows to set router options +// Options are router options type Options struct { // ID is router ID ID string @@ -82,7 +82,7 @@ func NetworkRegistry(r registry.Registry) Option { } // RouterIB allows to configure RIB -func RouterIB(r RIB) Option { +func RouterRIB(r RIB) Option { return func(o *Options) { o.RIB = r } diff --git a/router/query.go b/router/query.go index 2cc0bf24..313278d0 100644 --- a/router/query.go +++ b/router/query.go @@ -10,6 +10,9 @@ const ( ClosestMatch ) +// QueryOption is used to define query options +type QueryOption func(*QueryOptions) + // QueryOptions allow to define routing table query options type QueryOptions struct { // DestAddr defines destination address diff --git a/router/rib.go b/router/rib.go index 1c6f538e..2e1f7cfe 100644 --- a/router/rib.go +++ b/router/rib.go @@ -12,6 +12,9 @@ type RIB interface { String() string } +// RIBOptopn is used to configure RIB +type RIBOption func(*RIBOptions) + // RIBOptions allow to set RIB sources. type RIBOptions struct { // Source defines RIB source URL diff --git a/router/entry.go b/router/route.go similarity index 94% rename from router/entry.go rename to router/route.go index 3b94a562..9ef5b19d 100644 --- a/router/entry.go +++ b/router/route.go @@ -14,6 +14,9 @@ const ( ErrIfExists ) +// RouteOption is used to define routing table entry options +type RouteOption func(*RouteOptions) + // RouteOptions defines micro network routing table route options type RouteOptions struct { // DestAddr is destination address diff --git a/router/router.go b/router/router.go index 419b62d3..5e100c6b 100644 --- a/router/router.go +++ b/router/router.go @@ -1,19 +1,26 @@ // Package router provides an interface for micro network router package router +import "errors" + +var ( + // ErrNotImplemented is returned when some functionality has not been implemented + ErrNotImplemented = errors.New("not implemented") +) + // Router is micro network router type Router interface { // Init initializes the router with options Init(...Option) error // Options returns the router options Options() Options - // Table returns routing table + // Table returns the router routing table Table() Table - // Address returns router adddress + // Address returns the router adddress Address() string - // Gossip returns router gossip address + // Gossip returns the router gossip address Gossip() string - // Network returns router network address + // Network returns the router network address Network() string // Start starts the router Start() error @@ -26,18 +33,6 @@ type Router interface { // Option used by the router type Option func(*Options) -// RIBOptopn is used to configure RIB -type RIBOption func(*RIBOptions) - -// RouteOption is used to define routing table entry options -type RouteOption func(*RouteOptions) - -// QueryOption is used to define query options -type QueryOption func(*QueryOptions) - -// WatchOption is used to define what routes to watch in the table -type WatchOption func(*WatchOptions) - // NewRouter creates new Router and returns it func NewRouter(opts ...Option) Router { return newRouter(opts...) diff --git a/router/table.go b/router/table.go index d246e177..e1361380 100644 --- a/router/table.go +++ b/router/table.go @@ -2,251 +2,41 @@ package router import ( "errors" - "fmt" - "hash" - "hash/fnv" - "strings" - "sync" - - "github.com/google/uuid" - "github.com/olekukonko/tablewriter" ) var ( - // ErrRouteNotFound is returned when no route was found + // ErrRouteNotFound is returned when no route was found in the routing table ErrRouteNotFound = errors.New("route not found") - // ErrDuplicateRoute is return when route already exists + // ErrDuplicateRoute is returned when the route already exists ErrDuplicateRoute = errors.New("duplicate route") - // ErrNotImplemented is returned when some functionality has not been implemented - ErrNotImplemented = errors.New("not implemented") ) -// Table is routing table +// Table defines routing table interface type Table interface { - // Add adds new route to the table + // Init initializes the router with options + Init(...TableOption) error + // Options returns the router options + Options() TableOptions + // Add adds new route to the routing table Add(Route) error - // Remove removes existing route from the table + // Remove removes existing route from the routing table Remove(Route) error - // Update updates route in the table + // Update updates route in the routing table Update(Route) error - // Lookup looks up routes in the table + // Lookup looks up routes in the routing table and returns them Lookup(Query) ([]Route, error) - // Watch returns a watcher which allows you to track updates to the table + // Watch returns a watcher which allows to track updates to the routing table Watch(opts ...WatchOption) (Watcher, error) - // Size returns the size of the table + // Size returns the size of the routing table Size() int // String prints the routing table String() string } -// table is routing table -type table struct { - // m stores routing table map - m map[string]map[uint64]Route - // h hashes route entries - h hash.Hash64 - // w is a list of table watchers - w map[string]*tableWatcher - sync.RWMutex -} +// TableOption used by the routing table +type TableOption func(*TableOptions) // NewTable creates new routing table and returns it -func NewTable() Table { - h := fnv.New64() - h.Reset() - - return &table{ - m: make(map[string]map[uint64]Route), - w: make(map[string]*tableWatcher), - h: h, - } -} - -// Add adds a route to the routing table -func (t *table) Add(r Route) error { - t.Lock() - defer t.Unlock() - - destAddr := r.Options().DestAddr - sum := t.hash(r) - - if _, ok := t.m[destAddr]; !ok { - t.m[destAddr] = make(map[uint64]Route) - t.m[destAddr][sum] = r - go t.sendResult(&Result{Action: "add", Route: r}) - return nil - } - - if _, ok := t.m[destAddr][sum]; ok && r.Options().Policy == OverrideIfExists { - t.m[destAddr][sum] = r - go t.sendResult(&Result{Action: "update", Route: r}) - return nil - } - - if r.Options().Policy == IgnoreIfExists { - return nil - } - - return ErrDuplicateRoute -} - -// Remove removes the route from the routing table -func (t *table) Remove(r Route) error { - t.Lock() - defer t.Unlock() - - destAddr := r.Options().DestAddr - sum := t.hash(r) - - if _, ok := t.m[destAddr]; !ok { - return ErrRouteNotFound - } - - delete(t.m[destAddr], sum) - go t.sendResult(&Result{Action: "remove", Route: r}) - - return nil -} - -// Update updates routing table with new route -func (t *table) Update(r Route) error { - t.Lock() - defer t.Unlock() - - destAddr := r.Options().DestAddr - sum := t.hash(r) - - if _, ok := t.m[destAddr]; !ok { - return ErrRouteNotFound - } - - if _, ok := t.m[destAddr][sum]; ok { - t.m[destAddr][sum] = r - go t.sendResult(&Result{Action: "update", Route: r}) - return nil - } - - return ErrRouteNotFound -} - -// Lookup queries routing table and returns all routes that match it -func (t *table) Lookup(q Query) ([]Route, error) { - t.RLock() - defer t.RUnlock() - - var results []Route - - for destAddr, routes := range t.m { - if q.Options().DestAddr != "*" { - if q.Options().DestAddr != destAddr { - continue - } - for _, route := range routes { - if q.Options().Network == "*" || q.Options().Network == route.Options().Network { - results = append(results, route) - } - } - } - - if q.Options().DestAddr == "*" { - for _, route := range routes { - if q.Options().Network == "*" || q.Options().Network == route.Options().Network { - results = append(results, route) - } - } - } - } - - if len(results) == 0 && q.Options().Policy != DiscardNoRoute { - return nil, ErrRouteNotFound - } - - return results, nil -} - -// Watch returns routing table entry watcher -func (t *table) Watch(opts ...WatchOption) (Watcher, error) { - // by default watch everything - wopts := WatchOptions{ - DestAddr: "*", - Network: "*", - } - - for _, o := range opts { - o(&wopts) - } - - watcher := &tableWatcher{ - opts: wopts, - resChan: make(chan *Result, 10), - done: make(chan struct{}), - } - - t.Lock() - t.w[uuid.New().String()] = watcher - t.Unlock() - - return watcher, nil -} - -// sendResult sends rules to all subscribe watchers -func (t *table) sendResult(r *Result) { - t.RLock() - defer t.RUnlock() - - for _, w := range t.w { - select { - case w.resChan <- r: - case <-w.done: - } - } -} - -// Size returns the size of the routing table -func (t *table) Size() int { - t.RLock() - defer t.RUnlock() - - return len(t.m) -} - -// String returns debug information -func (t *table) String() string { - t.RLock() - defer t.RUnlock() - - // this will help us build routing table string - sb := &strings.Builder{} - - // create nice table printing structure - table := tablewriter.NewWriter(sb) - table.SetHeader([]string{"Destination", "Gateway", "Network", "Metric"}) - - for _, destRoute := range t.m { - for _, route := range destRoute { - strRoute := []string{ - route.Options().DestAddr, - route.Options().Gateway.Address(), - route.Options().Gateway.Network(), - fmt.Sprintf("%d", route.Options().Metric), - } - table.Append(strRoute) - } - } - - // render table into sb - table.Render() - - return sb.String() -} - -// hash hashes the route using router gateway and network address -func (t *table) hash(r Route) uint64 { - gwAddr := r.Options().Gateway.Address() - netAddr := r.Options().Network - - t.h.Reset() - t.h.Write([]byte(gwAddr + netAddr)) - - return t.h.Sum64() +func NewTable(opts ...TableOption) Table { + return newTable(opts...) } diff --git a/router/table_watcher.go b/router/table_watcher.go index 79aa7c46..78a87c96 100644 --- a/router/table_watcher.go +++ b/router/table_watcher.go @@ -9,7 +9,11 @@ var ( ErrWatcherStopped = errors.New("routing table watcher stopped") ) -// Watcher is an interface that returns updates to the routing table +// WatchOption is used to define what routes to watch in the table +type WatchOption func(*WatchOptions) + +// Watcher defines routing table watcher interface +// Watcher returns updates to the routing table type Watcher interface { // Next is a blocking call that returns watch result Next() (*Result, error)