Skip to content

Commit af94d63

Browse files
letterbeezpsletterbeezps
authored
Merge pull request #12 from letterbeezps/dev
Dev
2 parents 0e4918c + bcf83d6 commit af94d63

File tree

10 files changed

+87
-163
lines changed

10 files changed

+87
-163
lines changed

‎engine/engine.go

+4-1
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@ type Engine interface {
1818
// requested
1919
Delete(key string)
2020

21+
// requested
22+
DeleteReal(key string)
23+
2124
// option
2225
// Scan(start, end internal.Bound, iter func(key string, value []byte) bool)
2326

@@ -58,7 +61,7 @@ func EngineTestGetSet(engine Engine, t *testing.T) {
5861
assert.False(t, ok)
5962
assert.Equal(t, []byte{}, ret)
6063

61-
engine.Delete("a")
64+
engine.DeleteReal("a")
6265
ret, ok = engine.Get("a")
6366
assert.False(t, ok)
6467
assert.Equal(t, []byte{}, ret)

‎engine/memory/memory.go

+4
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,10 @@ func (m *Memory) Set(key string, value []byte) {
3333
}
3434

3535
func (m *Memory) Delete(key string) {
36+
m.Data.Set(key, []byte{})
37+
}
38+
39+
func (m *Memory) DeleteReal(key string) {
3640
m.Data.Delete(key)
3741
}
3842

‎internal/merge_iterator.go

+3-7
Original file line numberDiff line numberDiff line change
@@ -9,16 +9,12 @@ type TwoMergeIterator struct {
99
First iface.Iterator
1010
Second iface.Iterator
1111
ChooseFirst bool
12-
Less func(f, s string) bool
13-
Equal func(f, s string) bool
1412
}
1513

16-
func NewTwoMergeIterstor(f, s iface.Iterator, less, equal func(f, s string) bool) (*TwoMergeIterator, error) {
14+
func NewTwoMergeIterstor(f, s iface.Iterator) (*TwoMergeIterator, error) {
1715
ret := &TwoMergeIterator{
1816
First: f,
1917
Second: s,
20-
Less: less,
21-
Equal: equal,
2218
}
2319
if err := ret.skipSecond(); err != nil {
2420
return nil, errors.Wrap(err, "skipSecond")
@@ -34,11 +30,11 @@ func (iter *TwoMergeIterator) chooseFirst() bool {
3430
if !iter.Second.IsValid() {
3531
return true
3632
}
37-
return iter.Less(iter.First.Key(), iter.Second.Key())
33+
return iter.First.Key() < iter.Second.Key()
3834
}
3935

4036
func (iter *TwoMergeIterator) skipSecond() error {
41-
if iter.First.IsValid() && iter.Second.IsValid() && iter.Equal(iter.First.Key(), iter.Second.Key()) {
37+
if iter.First.IsValid() && iter.Second.IsValid() && iter.First.Key() == iter.Second.Key() {
4238
return iter.Second.Next()
4339
}
4440
return nil

‎mvcc/error.go

+4-3
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,8 @@ package mvcc
33
import "errors"
44

55
var (
6-
ErrorSerialization = errors.New("serialization")
7-
ErrorReadOnly = errors.New("can't update with read only tx")
8-
ErrorNotSupportNoBound = errors.New("tx iterator not support NoBound")
6+
ErrorSerialization = errors.New("serialization")
7+
ErrorReadOnly = errors.New("can't update with read only tx")
8+
ErrorNotSupportNoBound = errors.New("tx iterator not support NoBound")
9+
ErrorTxWriteKeyNotAtCache = errors.New("tx write key not at cache")
910
)

‎mvcc/key.go

-27
Original file line numberDiff line numberDiff line change
@@ -50,33 +50,6 @@ func decodeTxActiveKey(key string) (TXID, error) {
5050
return id, nil
5151
}
5252

53-
//////// txWrite ////////////
54-
55-
func encodeTxWriteKey(txId TXID, key string) (string, error) {
56-
k, err := orderedcode.Append(nil, orderedcode.Decr(uint64(txId)))
57-
if err != nil {
58-
return "", errors.Wrap(err, "getTxWriteKey")
59-
}
60-
ret := fmt.Sprintf("%s_%s", TxWritePrefix, string(k))
61-
if key != "" {
62-
ret = fmt.Sprintf("%s_%s", ret, key)
63-
}
64-
return ret, nil
65-
}
66-
67-
func decodeTxWriteKey(key string) (TXID, string, error) {
68-
keys := strings.Split(key, "_")
69-
if len(keys) < 3 {
70-
return 0, "", errors.New(fmt.Sprintf("bad format of TxWriteKey: %s", key))
71-
}
72-
var id uint64
73-
_, err := orderedcode.Parse(keys[1], orderedcode.Decr(&id))
74-
if err != nil {
75-
return 0, "", errors.Wrap(err, fmt.Sprintf("parse failed: %s", key))
76-
}
77-
return id, strings.Join(keys[2:], "_"), nil
78-
}
79-
8053
//////// txKey ////////////
8154

8255
func encodeTxKey(txId TXID, key string) (string, error) {

‎mvcc/key_test.go

+10-10
Original file line numberDiff line numberDiff line change
@@ -71,16 +71,16 @@ func Test_activeKey(t *testing.T) {
7171
assert.Equal(t, uint64(12), id)
7272
}
7373

74-
func Test_txWritKey(t *testing.T) {
75-
k, err := encodeTxWriteKey(12, "aa_bb")
76-
assert.Nil(t, err)
77-
t.Log(k)
78-
79-
id, origin_key, err := decodeTxWriteKey(k)
80-
assert.Nil(t, err)
81-
assert.Equal(t, uint64(12), id)
82-
assert.Equal(t, "aa_bb", origin_key)
83-
}
74+
// func Test_txWritKey(t *testing.T) {
75+
// k, err := encodeTxWriteKey(12, "aa_bb")
76+
// assert.Nil(t, err)
77+
// t.Log(k)
78+
79+
// id, origin_key, err := decodeTxWriteKey(k)
80+
// assert.Nil(t, err)
81+
// assert.Equal(t, uint64(12), id)
82+
// assert.Equal(t, "aa_bb", origin_key)
83+
// }
8484

8585
func Test_txKey(t *testing.T) {
8686
k, err := encodeTxKey(12, "aa_bb_cc")

‎mvcc/mvcc.go

+3
Original file line numberDiff line numberDiff line change
@@ -11,19 +11,22 @@ import (
1111
type MVCC struct {
1212
Lock *sync.Mutex
1313
Engine engine.Engine
14+
Cache engine.Engine
1415
}
1516

1617
func NewMVCC() *MVCC {
1718
return &MVCC{
1819
Lock: &sync.Mutex{},
1920
Engine: memory.NewMemory(),
21+
Cache: memory.NewMemory(),
2022
}
2123
}
2224

2325
func (m *MVCC) Begin(readOnly bool) (*TX, error) {
2426
tx := &TX{
2527
Lock: m.Lock,
2628
Engine: m.Engine,
29+
Cache: m.Cache,
2730
}
2831
err := tx.Begin(readOnly)
2932
if err != nil {

‎mvcc/tx.go

+42-58
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@ import (
1414

1515
type TX struct {
1616
Lock *sync.Mutex
17-
Engine engine.Engine
17+
Cache engine.Engine // store active data, which will be written to engine after commit or deleted with rollback
18+
Engine engine.Engine // store stable data.
1819
State *TxState
1920
}
2021

@@ -35,24 +36,25 @@ func (tx *TX) Begin(readOnly bool) error {
3536
return errors.Wrap(err, "scanActive")
3637
}
3738
tx.State = &TxState{
38-
TxID: txId,
39-
ReadOnly: readOnly,
40-
ActiveTx: active,
39+
TxID: txId,
40+
ReadOnly: readOnly,
41+
ActiveTx: active,
42+
ActiveKeys: map[string]struct{}{},
4143
}
4244
if !readOnly {
4345
activeKey, err := encodeTxActiveKey(txId)
4446
if err != nil {
4547
return errors.Wrap(err, fmt.Sprintf("getTxActiveKey with %d", txId))
4648
}
47-
tx.Engine.Set(activeKey, []byte{})
49+
tx.Cache.Set(activeKey, []byte{})
4850
}
4951
return nil
5052
}
5153

5254
func (tx *TX) scanActive() (map[TXID]struct{}, error) {
5355
start := internal.NewBound(TxActivePrefix, internal.Include)
5456
end := internal.NewBound(TxActivePrefixEnd, internal.Exclude)
55-
iter := tx.Engine.Iter(start, end)
57+
iter := tx.Cache.Iter(start, end)
5658
ret := map[TXID]struct{}{}
5759
for iter.IsValid() {
5860
key := iter.Key()
@@ -86,7 +88,12 @@ func (tx *TX) Write(key string, value []byte) error {
8688
}
8789
start := internal.NewBound(startTxKey, internal.Include)
8890
end := internal.NewBound(endTxKey, internal.Include)
89-
iter := tx.Engine.Iter(start, end)
91+
cacheIter := tx.Cache.Iter(start, end)
92+
engineIter := tx.Engine.Iter(start, end)
93+
iter, err := internal.NewTwoMergeIterstor(cacheIter, engineIter)
94+
if err != nil {
95+
return errors.Wrap(err, "NewTwoMergeIterstor")
96+
}
9097
for iter.IsValid() {
9198
check_id, _, err := decodeTxKey(iter.Key())
9299
if err != nil {
@@ -98,16 +105,12 @@ func (tx *TX) Write(key string, value []byte) error {
98105
iter.Next()
99106
}
100107

101-
tnWriteKey, err := encodeTxWriteKey(tx.State.TxID, key)
102-
if err != nil {
103-
return errors.Wrap(err, fmt.Sprintf("encodeTxWriteKey with key: %s", key))
104-
}
105-
tx.Engine.Set(tnWriteKey, []byte{})
106108
txKey, err := encodeTxKey(tx.State.TxID, key)
107109
if err != nil {
108110
return errors.Wrap(err, fmt.Sprintf("encodeTxKey with key: %s", key))
109111
}
110-
tx.Engine.Set(txKey, value)
112+
tx.Cache.Set(txKey, value)
113+
tx.State.ActiveKeys[txKey] = struct{}{}
111114
return nil
112115
}
113116

@@ -123,67 +126,38 @@ func (tx *TX) Commit() error {
123126
if tx.State.ReadOnly {
124127
return errors.Wrap(ErrorReadOnly, fmt.Sprintf("tx with id %d is read only, not need commit", tx.State.TxID))
125128
}
126-
startKey, err := encodeTxWriteKey(tx.State.TxID, "")
127-
if err != nil {
128-
return errors.Wrap(err, fmt.Sprintf("encodeTxWriteKey with id: %d", tx.State.TxID))
129-
}
130-
endKey := getPrefixEnd(startKey)
131-
start := internal.NewBound(startKey, internal.Include)
132-
end := internal.NewBound(endKey, internal.Exclude)
133-
iter := tx.Engine.Iter(start, end)
134-
removeKeys := []string{}
135-
for iter.IsValid() {
136-
removeKeys = append(removeKeys, iter.Key())
137-
iter.Next()
138-
}
139-
for _, key := range removeKeys {
140-
tx.Engine.Delete(key)
129+
for k := range tx.State.ActiveKeys {
130+
if v, ok := tx.Cache.Get(k); ok {
131+
tx.Engine.Set(k, v)
132+
tx.Cache.DeleteReal(k)
133+
} else {
134+
return ErrorTxWriteKeyNotAtCache
135+
}
141136
}
142-
143137
activeKey, err := encodeTxActiveKey(tx.State.TxID)
144138
if err != nil {
145139
return errors.Wrap(err, fmt.Sprintf("getTxActiveKey with %d", tx.State.TxID))
146140
}
147-
tx.Engine.Delete(activeKey)
141+
tx.Cache.DeleteReal(activeKey)
148142
return nil
149143
}
150144

151145
func (tx *TX) RollBack() error {
152146
if tx.State.ReadOnly {
153147
return errors.Wrap(ErrorReadOnly, fmt.Sprintf("tx with id %d is read only, not need rollback", tx.State.TxID))
154148
}
155-
startKey, err := encodeTxWriteKey(tx.State.TxID, "")
156-
if err != nil {
157-
return errors.Wrap(err, fmt.Sprintf("encodeTxWriteKey with id: %d", tx.State.TxID))
158-
}
159-
endKey := getPrefixEnd(startKey)
160-
start := internal.NewBound(startKey, internal.Include)
161-
end := internal.NewBound(endKey, internal.Exclude)
162-
iter := tx.Engine.Iter(start, end)
163-
removeKeys := []string{}
164-
for iter.IsValid() {
165-
removeKeys = append(removeKeys, iter.Key())
166-
_, origin_key, err := decodeTxWriteKey(iter.Key())
167-
if err != nil {
168-
return errors.Wrap(err, fmt.Sprintf("decodeTxWriteKey with key: %s", iter.Key()))
169-
}
170-
txKey, err := encodeTxKey(tx.State.TxID, origin_key)
171-
if err != nil {
172-
return errors.Wrap(err, fmt.Sprintf("encodeTxKey with key: %s", txKey))
149+
for k := range tx.State.ActiveKeys {
150+
if _, ok := tx.Cache.Get(k); ok {
151+
tx.Cache.DeleteReal(k)
152+
} else {
153+
return ErrorTxWriteKeyNotAtCache
173154
}
174-
removeKeys = append(removeKeys, txKey)
175-
iter.Next()
176-
}
177-
178-
for _, key := range removeKeys {
179-
tx.Engine.Delete(key)
180155
}
181-
182156
activeKey, err := encodeTxActiveKey(tx.State.TxID)
183157
if err != nil {
184158
return errors.Wrap(err, fmt.Sprintf("getTxActiveKey with %d", tx.State.TxID))
185159
}
186-
tx.Engine.Delete(activeKey)
160+
tx.Cache.DeleteReal(activeKey)
187161
return nil
188162
}
189163

@@ -198,7 +172,12 @@ func (tx *TX) Get(key string) ([]byte, error) {
198172
}
199173
start := internal.NewBound(startTxKey, internal.Include)
200174
end := internal.NewBound(endTxKey, internal.Include)
201-
iter := tx.Engine.Iter(start, end)
175+
cacheIter := tx.Cache.Iter(start, end)
176+
engineIter := tx.Engine.Iter(start, end)
177+
iter, err := internal.NewTwoMergeIterstor(cacheIter, engineIter)
178+
if err != nil {
179+
return nil, errors.Wrap(err, "NewTwoMergeIterstor")
180+
}
202181
for iter.IsValid() {
203182
check_id, _, err := decodeTxKey(iter.Key())
204183
if err != nil {
@@ -228,13 +207,18 @@ func (tx *TX) Iter(start, end string) (iface.Iterator, error) {
228207

229208
endEngineKey := internal.NewBound(endTxKey, internal.Include)
230209

210+
cacheIter := tx.Cache.Iter(startEngineKey, endEngineKey)
231211
engineIter := tx.Engine.Iter(startEngineKey, endEngineKey)
212+
iter, err := internal.NewTwoMergeIterstor(cacheIter, engineIter)
213+
if err != nil {
214+
return nil, errors.Wrap(err, "NewTwoMergeIterstor")
215+
}
232216

233217
ret := &TXIterator{
234218
State: tx.State,
235219
Start: start,
236220
End: end,
237-
EngineIterator: engineIter,
221+
EngineIterator: iter,
238222
}
239223
err = ret.Next()
240224
if err != nil {

‎mvcc/tx_state.go

+4-3
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,10 @@ package mvcc
33
type TXID = uint64
44

55
type TxState struct {
6-
TxID TXID
7-
ReadOnly bool
8-
ActiveTx map[TXID]struct{}
6+
TxID TXID
7+
ReadOnly bool
8+
ActiveTx map[TXID]struct{}
9+
ActiveKeys map[string]struct{} // keys writen by currency tx
910
}
1011

1112
// check whether the given txId is visible to currenct tx

0 commit comments

Comments
 (0)