1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
package ethdb
import (
"sync"
"time"
"github.com/ethereum/go-ethereum/compression/rle"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/iterator"
)
type LDBDatabase struct {
fn string
mu sync.Mutex
db *leveldb.DB
queue map[string][]byte
quit chan struct{}
}
func NewLDBDatabase(file string) (*LDBDatabase, error) {
// Open the db
db, err := leveldb.OpenFile(file, nil)
if err != nil {
return nil, err
}
database := &LDBDatabase{
fn: file,
db: db,
quit: make(chan struct{}),
}
database.makeQueue()
go database.update()
return database, nil
}
func (self *LDBDatabase) makeQueue() {
self.queue = make(map[string][]byte)
}
func (self *LDBDatabase) Put(key []byte, value []byte) {
self.mu.Lock()
defer self.mu.Unlock()
self.queue[string(key)] = value
/*
value = rle.Compress(value)
err := self.db.Put(key, value, nil)
if err != nil {
fmt.Println("Error put", err)
}
*/
}
func (self *LDBDatabase) Get(key []byte) ([]byte, error) {
self.mu.Lock()
defer self.mu.Unlock()
// Check queue first
if dat, ok := self.queue[string(key)]; ok {
return dat, nil
}
dat, err := self.db.Get(key, nil)
if err != nil {
return nil, err
}
return rle.Decompress(dat)
}
func (self *LDBDatabase) Delete(key []byte) error {
self.mu.Lock()
defer self.mu.Unlock()
// make sure it's not in the queue
delete(self.queue, string(key))
return self.db.Delete(key, nil)
}
func (self *LDBDatabase) LastKnownTD() []byte {
data, _ := self.Get([]byte("LTD"))
if len(data) == 0 {
data = []byte{0x0}
}
return data
}
func (self *LDBDatabase) NewIterator() iterator.Iterator {
return self.db.NewIterator(nil, nil)
}
func (self *LDBDatabase) Flush() error {
self.mu.Lock()
defer self.mu.Unlock()
batch := new(leveldb.Batch)
for key, value := range self.queue {
batch.Put([]byte(key), rle.Compress(value))
}
self.makeQueue() // reset the queue
return self.db.Write(batch, nil)
}
func (self *LDBDatabase) Close() {
self.quit <- struct{}{}
<-self.quit
glog.V(logger.Info).Infoln("flushed and closed db:", self.fn)
}
func (self *LDBDatabase) update() {
ticker := time.NewTicker(1 * time.Minute)
done:
for {
select {
case <-ticker.C:
if err := self.Flush(); err != nil {
glog.V(logger.Error).Infof("error: flush '%s': %v\n", self.fn, err)
}
case <-self.quit:
break done
}
}
if err := self.Flush(); err != nil {
glog.V(logger.Error).Infof("error: flush '%s': %v\n", self.fn, err)
}
// Close the leveldb database
self.db.Close()
self.quit <- struct{}{}
}
|
