Skip to content

Commit 7d8bee0

Browse files
bhmjbhmj
committed
add cache and www packages
1 parent 5502719 commit 7d8bee0

4 files changed

Lines changed: 576 additions & 0 deletions

File tree

‎cache/dbcache/dbcache.go‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
package dbcache
2+
3+
import (
4+
"path/filepath"
5+
"time"
6+
7+
"github.com/bhmj/goblocks/dbase/abstract"
8+
"github.com/bhmj/goblocks/file"
9+
"github.com/bhmj/goblocks/log"
10+
"github.com/bhmj/goblocks/www"
11+
)
12+
13+
type cache struct {
14+
db abstract.DB
15+
logger log.MetaLogger
16+
cacheDir string
17+
}
18+
19+
type Cache interface {
20+
GetURL(url string) (string, error)
21+
GetContent(url string) ([]byte, string, error)
22+
Cleanup()
23+
}
24+
25+
func New(db abstract.DB, logger log.MetaLogger, cacheDir string) Cache {
26+
return &cache{
27+
db: db,
28+
logger: logger,
29+
cacheDir: cacheDir,
30+
}
31+
}
32+
33+
func (c *cache) GetURL(url string) (extPath string, err error) {
34+
extPath, _ = c.getCacheRecord(url)
35+
fullPath := filepath.Join(c.cacheDir, extPath)
36+
if extPath != "" && file.Exists(fullPath) {
37+
return
38+
}
39+
extPath, contentType, fileSize, err := c.requestURL(url)
40+
if err != nil {
41+
return
42+
}
43+
44+
err = c.setCacheRecord(url, extPath, contentType, fileSize)
45+
return
46+
}
47+
48+
func (c *cache) GetContent(url string) (body []byte, contentType string, err error) {
49+
extPath, contentType := c.getCacheRecord(url)
50+
fullPath := filepath.Join(c.cacheDir, extPath)
51+
if extPath != "" && file.Exists(fullPath) {
52+
body, err = file.Read(fullPath)
53+
return
54+
}
55+
extPath, body, contentType, fileSize, err := c.fetchURL(url)
56+
if err != nil {
57+
return
58+
}
59+
err = c.setCacheRecord(url, extPath, contentType, fileSize)
60+
return
61+
}
62+
63+
type cacheRec struct {
64+
FilePath string `db:"file_path"`
65+
ContentType string `db:"content_type"`
66+
AddedAt time.Time `db:"added_at"`
67+
}
68+
69+
func (c *cache) getCacheRecord(url string) (string, string) {
70+
// TODO: add memory cache
71+
var entry cacheRec
72+
sql := `
73+
with upd as (
74+
update file_cache set
75+
last_read_at = now()
76+
where source_url = $1
77+
returning id
78+
)
79+
select file_path, content_type, added_at
80+
from file_cache
81+
where id = (select id from upd limit 1)`
82+
found, err := c.db.QueryRow(&entry, sql, url)
83+
if err != nil {
84+
c.logger.Error("getting cache record", log.Error(err))
85+
return "", ""
86+
}
87+
if !found {
88+
c.logger.Info("getting cache record: not found", log.String("url", url))
89+
}
90+
return entry.FilePath, entry.ContentType
91+
}
92+
93+
func (c *cache) setCacheRecord(url, extPath, contentType string, fileSize int64) error {
94+
// TODO: update memory cache
95+
sql := `
96+
insert into file_cache (
97+
source_url, file_path, content_type, file_size
98+
) values (
99+
$1, $2, $3, $4
100+
) on conflict (source_url) do update set
101+
file_path = excluded.file_path,
102+
content_type = excluded.content_type
103+
;`
104+
return c.db.Exec(sql, url, extPath, contentType, fileSize)
105+
}
106+
107+
func (c *cache) contentTypeUpdate(url, contentType string, fileSize int64) {
108+
sql := `update file_cache set content_type = $1, file_size = $2 where source_url = $3`
109+
err := c.db.Exec(sql, contentType, fileSize, url)
110+
if err != nil {
111+
c.logger.Error("updating content_type", log.Error(err))
112+
}
113+
}
114+
115+
func (c *cache) requestURL(url string) (string, string, int64, error) {
116+
path := time.Now().Format("2006-01-02")
117+
return www.EnqueueDownload(url, c.cacheDir, path, c.contentTypeUpdate)
118+
}
119+
120+
func (c *cache) fetchURL(url string) (string, []byte, string, int64, error) {
121+
path := time.Now().Format("2006-01-02")
122+
return www.DownloadContent(url, c.cacheDir, path)
123+
}
124+
125+
func (c *cache) Cleanup() {}

‎cache/memcache/memcache.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
package memcache
2+
3+
import (
4+
"reflect"
5+
"sync"
6+
"time"
7+
)
8+
9+
/*
10+
Memory cache for arbirtary data.
11+
Size-based eviction (no TTL).
12+
On eviction deletes LRU items amongst those older than `keepFirst` minutes.
13+
14+
TODO: add TTL and batch eviction.
15+
*/
16+
17+
const keepFirst = time.Minute * 5 // no-eviction time
18+
19+
type cacheRec struct {
20+
Value interface{} // 16
21+
ContentType string // 16
22+
AddedAt int64 // 8
23+
UseCount int // 8
24+
}
25+
26+
type cache struct {
27+
maxSize int
28+
size int
29+
storage map[string]*cacheRec
30+
sync.RWMutex
31+
}
32+
33+
type Cache interface {
34+
Get(key string) (interface{}, string, bool) // value, content-type, found
35+
Set(key string, value interface{}, contentType ...string)
36+
Del(key string)
37+
Cleanup()
38+
Size() int
39+
}
40+
41+
func RecordSize() int {
42+
return valueSize(cacheRec{})
43+
}
44+
45+
// New creates new memory cache with given maximum size in bytes
46+
func New(size int) Cache {
47+
return &cache{
48+
maxSize: size,
49+
size: 0,
50+
storage: make(map[string]*cacheRec),
51+
}
52+
}
53+
54+
// Get retrieves value from cache, nil if not found.
55+
func (c *cache) Get(key string) (interface{}, string, bool) {
56+
c.Lock()
57+
defer c.Unlock()
58+
if v, found := c.storage[key]; found {
59+
c.storage[key].UseCount++
60+
return v.Value, v.ContentType, true
61+
}
62+
return nil, "", false
63+
}
64+
65+
// Set stores a value in cache, evicting stale elements if needed. Can skip storing if the cache is full and no evictable entries found.
66+
func (c *cache) Set(key string, value interface{}, contentType ...string) {
67+
c.Lock()
68+
defer c.Unlock()
69+
70+
ct := ""
71+
for _, v := range contentType {
72+
ct = v
73+
break
74+
}
75+
prec, found := c.storage[key]
76+
useCount := 0
77+
addedAt := time.Now().Unix()
78+
if found {
79+
useCount = prec.UseCount
80+
addedAt = prec.AddedAt
81+
c.size -= valueSize(*prec)
82+
}
83+
84+
rec := cacheRec{Value: value, ContentType: ct, UseCount: useCount + 1, AddedAt: addedAt}
85+
sz := valueSize(rec)
86+
87+
for c.size+sz > c.maxSize {
88+
if !c.evict() {
89+
return
90+
}
91+
}
92+
c.storage[key] = &rec
93+
c.size += sz
94+
}
95+
96+
// Del deletes cache entry if exists.
97+
func (c *cache) Del(key string) {
98+
c.Lock()
99+
defer c.Unlock()
100+
if v, found := c.storage[key]; found {
101+
c.size -= valueSize(v)
102+
delete(c.storage, key)
103+
}
104+
}
105+
106+
// Cleanup empties cache.
107+
func (c *cache) Cleanup() {
108+
c.Lock()
109+
defer c.Unlock()
110+
for k := range c.storage {
111+
delete(c.storage, k)
112+
}
113+
}
114+
115+
// Size returns current cache size.
116+
func (c *cache) Size() int {
117+
c.RLock()
118+
defer c.RUnlock()
119+
return c.size
120+
}
121+
122+
// evict removes one less used stale entry.
123+
// It has O(n) complexity, but we don't care for it now.
124+
func (c *cache) evict() bool {
125+
var prey *cacheRec
126+
var key string
127+
from := time.Now().Add(-keepFirst).Unix()
128+
for k, v := range c.storage {
129+
if v.AddedAt < from && (prey == nil || v.UseCount < prey.UseCount) {
130+
prey = v
131+
key = k
132+
}
133+
}
134+
if prey != nil {
135+
sz := valueSize(*prey)
136+
c.size -= sz
137+
delete(c.storage, key)
138+
return true
139+
}
140+
return false
141+
}
142+
143+
// stale moves cache entry back in time so it can be evicted (test only function)
144+
func (c *cache) stale(key string) {
145+
c.Lock()
146+
defer c.Unlock()
147+
if v, found := c.storage[key]; found {
148+
c.storage[key] = &cacheRec{Value: v.Value, UseCount: v.UseCount, AddedAt: time.Now().Add(-keepFirst * 2).Unix()}
149+
}
150+
}
151+
152+
func valueSize(value interface{}) int {
153+
return int(reflect.TypeOf(value).Size())
154+
}

‎cache/memcache/memcache_test.go‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
package memcache
2+
3+
import (
4+
"testing"
5+
6+
"github.com/stretchr/testify/assert"
7+
)
8+
9+
func TestCache(t *testing.T) {
10+
a := assert.New(t)
11+
12+
recSize := RecordSize()
13+
mc := New(3 * recSize) // bytes
14+
15+
// fill up cache
16+
mc.Set("a", 123)
17+
mc.Set("b", 456)
18+
mc.Set("c", 789)
19+
20+
// check size
21+
sz := mc.Size()
22+
a.Equal(3*recSize, sz)
23+
24+
// value in cache
25+
v, _, _ := mc.Get("a")
26+
a.Equal(123, v.(int))
27+
28+
// stale "a" value
29+
raw := mc.(*cache)
30+
raw.stale("a")
31+
32+
// add element, evicting "a"
33+
mc.Set("d", 987)
34+
35+
// "d" is in cache
36+
_, _, found := mc.Get("d")
37+
a.True(found)
38+
39+
// value is evicted
40+
v, _, found = mc.Get("a")
41+
a.False(found)
42+
a.Nil(v)
43+
44+
// make "b" frequently used
45+
for i := 0; i < 100; i++ {
46+
mc.Get("b")
47+
}
48+
49+
// allow all cache entries to be evicted
50+
raw.stale("b")
51+
raw.stale("c")
52+
raw.stale("d")
53+
54+
// make new entries
55+
mc.Set("x", 654)
56+
mc.Set("y", 321)
57+
58+
// Two entries were evicted to free up space for the two new entries;
59+
// However, b should not be evicted since is was used much recently.
60+
61+
// "b" should stay
62+
v, _, _ = mc.Get("b")
63+
a.Equal(456, v.(int))
64+
}

0 commit comments

Comments
 (0)