mirror of
https://github.com/cwinfo/yggdrasil-go.git
synced 2024-11-25 23:01:38 +00:00
Make use of metadata cache
This commit is contained in:
parent
d07e0ddfa0
commit
d9884a5cac
@ -322,8 +322,12 @@ func (a *admin) init(c *Core, listenaddr string) {
|
||||
return admin_info{}, err
|
||||
}
|
||||
})
|
||||
a.addHandler("getMeta", []string{"box_pub_key", "coords"}, func(in admin_info) (admin_info, error) {
|
||||
result, err := a.admin_getMeta(in["box_pub_key"].(string), in["coords"].(string))
|
||||
a.addHandler("getMeta", []string{"box_pub_key", "coords", "[nocache]"}, func(in admin_info) (admin_info, error) {
|
||||
var nocache bool
|
||||
if in["nocache"] != nil {
|
||||
nocache = in["nocache"].(string) == "true"
|
||||
}
|
||||
result, err := a.admin_getMeta(in["box_pub_key"].(string), in["coords"].(string), nocache)
|
||||
if err == nil {
|
||||
var m map[string]interface{}
|
||||
if err = json.Unmarshal(result, &m); err == nil {
|
||||
@ -813,13 +817,18 @@ func (a *admin) admin_dhtPing(keyString, coordString, targetString string) (dhtR
|
||||
return dhtRes{}, errors.New(fmt.Sprintf("DHT ping timeout: %s", keyString))
|
||||
}
|
||||
|
||||
func (a *admin) admin_getMeta(keyString, coordString string) (metadataPayload, error) {
|
||||
func (a *admin) admin_getMeta(keyString, coordString string, nocache bool) (metadataPayload, error) {
|
||||
var key boxPubKey
|
||||
if keyBytes, err := hex.DecodeString(keyString); err != nil {
|
||||
return metadataPayload{}, err
|
||||
} else {
|
||||
copy(key[:], keyBytes)
|
||||
}
|
||||
if !nocache {
|
||||
if response, err := a.core.metadata.getCachedMetadata(key); err == nil {
|
||||
return response, nil
|
||||
}
|
||||
}
|
||||
var coords []byte
|
||||
for _, cstr := range strings.Split(strings.Trim(coordString, "[]"), " ") {
|
||||
if cstr == "" {
|
||||
|
@ -1,6 +1,7 @@
|
||||
package yggdrasil
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
@ -11,36 +12,51 @@ type metadata struct {
|
||||
myMetadataMutex sync.RWMutex
|
||||
callbacks map[boxPubKey]metadataCallback
|
||||
callbacksMutex sync.Mutex
|
||||
cache map[boxPubKey]metadataPayload
|
||||
cache map[boxPubKey]metadataCached
|
||||
cacheMutex sync.RWMutex
|
||||
}
|
||||
|
||||
type metadataPayload []byte
|
||||
|
||||
type metadataCached struct {
|
||||
payload metadataPayload
|
||||
created time.Time
|
||||
}
|
||||
|
||||
type metadataCallback struct {
|
||||
call func(meta *metadataPayload)
|
||||
created time.Time
|
||||
}
|
||||
|
||||
// Initialises the metadata cache/callback stuff
|
||||
// Initialises the metadata cache/callback maps, and starts a goroutine to keep
|
||||
// the cache/callback maps clean of stale entries
|
||||
func (m *metadata) init(core *Core) {
|
||||
m.core = core
|
||||
m.callbacks = make(map[boxPubKey]metadataCallback)
|
||||
m.cache = make(map[boxPubKey]metadataPayload)
|
||||
m.cache = make(map[boxPubKey]metadataCached)
|
||||
|
||||
go func() {
|
||||
for {
|
||||
m.callbacksMutex.Lock()
|
||||
for boxPubKey, callback := range m.callbacks {
|
||||
if time.Since(callback.created) > time.Minute {
|
||||
delete(m.callbacks, boxPubKey)
|
||||
}
|
||||
}
|
||||
time.Sleep(time.Second * 5)
|
||||
m.callbacksMutex.Unlock()
|
||||
m.cacheMutex.Lock()
|
||||
for boxPubKey, cache := range m.cache {
|
||||
if time.Since(cache.created) > time.Hour {
|
||||
delete(m.cache, boxPubKey)
|
||||
}
|
||||
}
|
||||
m.cacheMutex.Unlock()
|
||||
time.Sleep(time.Second * 30)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Add a callback
|
||||
// Add a callback for a metadata lookup
|
||||
func (m *metadata) addCallback(sender boxPubKey, call func(meta *metadataPayload)) {
|
||||
m.callbacksMutex.Lock()
|
||||
defer m.callbacksMutex.Unlock()
|
||||
@ -60,14 +76,14 @@ func (m *metadata) callback(sender boxPubKey, meta metadataPayload) {
|
||||
}
|
||||
}
|
||||
|
||||
// Get the metadata
|
||||
// Get the current node's metadata
|
||||
func (m *metadata) getMetadata() metadataPayload {
|
||||
m.myMetadataMutex.RLock()
|
||||
defer m.myMetadataMutex.RUnlock()
|
||||
return m.myMetadata
|
||||
}
|
||||
|
||||
// Set the metadata
|
||||
// Set the current node's metadata
|
||||
func (m *metadata) setMetadata(meta metadataPayload) {
|
||||
m.myMetadataMutex.Lock()
|
||||
defer m.myMetadataMutex.Unlock()
|
||||
@ -78,20 +94,23 @@ func (m *metadata) setMetadata(meta metadataPayload) {
|
||||
func (m *metadata) addCachedMetadata(key boxPubKey, payload metadataPayload) {
|
||||
m.cacheMutex.Lock()
|
||||
defer m.cacheMutex.Unlock()
|
||||
m.cache[key] = payload
|
||||
m.cache[key] = metadataCached{
|
||||
created: time.Now(),
|
||||
payload: payload,
|
||||
}
|
||||
}
|
||||
|
||||
// Get a metadata entry from the cache
|
||||
func (m *metadata) getCachedMetadata(key boxPubKey) metadataPayload {
|
||||
func (m *metadata) getCachedMetadata(key boxPubKey) (metadataPayload, error) {
|
||||
m.cacheMutex.RLock()
|
||||
defer m.cacheMutex.RUnlock()
|
||||
if meta, ok := m.cache[key]; ok {
|
||||
return meta
|
||||
return meta.payload, nil
|
||||
}
|
||||
return metadataPayload{}
|
||||
return metadataPayload{}, errors.New("No cache entry found")
|
||||
}
|
||||
|
||||
// Handles a meta request/response.
|
||||
// Handles a meta request/response - called from the router
|
||||
func (m *metadata) handleMetadata(meta *sessionMeta) {
|
||||
if meta.IsResponse {
|
||||
m.callback(meta.SendPermPub, meta.Metadata)
|
||||
@ -101,7 +120,7 @@ func (m *metadata) handleMetadata(meta *sessionMeta) {
|
||||
}
|
||||
}
|
||||
|
||||
// Send metadata request or response
|
||||
// Send metadata request or response - called from the router
|
||||
func (m *metadata) sendMetadata(key boxPubKey, coords []byte, isResponse bool) {
|
||||
table := m.core.switchTable.table.Load().(lookupTable)
|
||||
meta := sessionMeta{
|
||||
|
Loading…
Reference in New Issue
Block a user