aboutsummaryrefslogtreecommitdiff
path: root/rwtransaction.go
blob: 145e0cf6621c7c69459e13e4e48a5dc3303595b1 (plain) (blame)
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
package bolt

import (
	"sort"
	"unsafe"
)

// RWTransaction represents a transaction that can read and write data.
// Only one read/write transaction can be active for a DB at a time.
type RWTransaction struct {
	Transaction
	branches map[pgid]*branch
	leafs    map[pgid]*leaf
}

// init initializes the transaction.
func (t *RWTransaction) init(db *DB) {
	t.Transaction.init(db)
	t.pages = make(map[pgid]*page)

	// Copy the meta and increase the transaction id. 
	t.meta = &meta{}
	db.meta().copy(t.meta)
	t.meta.txnid += txnid(2)
}

// CreateBucket creates a new bucket.
func (t *RWTransaction) CreateBucket(name string) error {
	// Check if bucket already exists.
	if b := t.Bucket(name); b != nil {
		return &Error{"bucket already exists", nil}
	} else if len(name) == 0 {
		return &Error{"bucket name cannot be blank", nil}
	} else if len(name) > MaxBucketNameSize {
		return &Error{"bucket name too large", nil}
	}

	// Create a blank root leaf page.
	p := t.allocate(1)
	p.flags = p_leaf

	// Add bucket to system page.
	t.sys.put(name, &bucket{root: p.id})

	return nil
}

// DropBucket deletes a bucket.
func (t *RWTransaction) DeleteBucket(name string) error {
	// Remove from system page.
	t.sys.del(name)

	// TODO: Delete entry from system bucket.
	// TODO: Free all pages.
	// TODO: Remove cursor.
	return nil
}

func (t *RWTransaction) Put(name string, key []byte, value []byte) error {
	b := t.Bucket(name)
	if b == nil {
		return &Error{"bucket not found", nil}
	}

	// Validate the key and data size.
	if len(key) == 0 {
		return &Error{"key required", nil}
	} else if len(key) > MaxKeySize {
		return &Error{"key too large", nil}
	} else if len(value) > MaxDataSize {
		return &Error{"data too large", nil}
	}

	// Insert a new node.
	c := b.Cursor()
	c.Goto(key)
	t.leaf(c).put(key, value)

	return nil
}

func (t *RWTransaction) Delete(name string, key []byte) error {
	// TODO: Traverse to the correct node.
	// TODO: If missing, exit.
	// TODO: Remove node from page.
	// TODO: If page is empty then add it to the freelist.
	return nil
}

// Commit writes all changes to disk.
func (t *RWTransaction) Commit() error {
	// TODO(benbjohnson): Use vectorized I/O to write out dirty pages.

	// TODO: Rebalance.

	// Spill data onto dirty pages.
	t.spill()

	// Spill system page.
	p := t.allocate((t.sys.size() / t.db.pageSize) + 1)
	t.sys.write(p)

	// Write dirty pages to disk.
	if err := t.write(); err != nil {
		return err
	}

	// Update the meta.
	t.meta.sys = p.id

	// Write meta to disk.
	if err := t.writeMeta(); err != nil {
		return err
	}

	return nil
}

func (t *RWTransaction) Rollback() {
	t.close()
}

func (t *RWTransaction) close() {
	// Clear temporary pages.
	t.leafs = nil

	// TODO: Release writer lock.
}

// allocate returns a contiguous block of memory starting at a given page.
func (t *RWTransaction) allocate(count int) *page {
	// TODO(benbjohnson): Use pages from the freelist.

	// Allocate a set of contiguous pages from the end of the file.
	buf := make([]byte, count*t.db.pageSize)
	p := (*page)(unsafe.Pointer(&buf[0]))
	p.id = t.meta.pgid
	p.overflow = uint32(count - 1)

	// Increment the last page id.
	t.meta.pgid += pgid(count)

	// Save it in our page cache.
	t.pages[p.id] = p

	return p
}

// spill writes all the leafs and branches to dirty pages.
func (t *RWTransaction) spill() {
	// Spill leafs first.
	for _, l := range t.leafs {
		t.spillLeaf(l)
	}

	// Sort branches by highest depth first.
	branches := make(branches, 0, len(t.branches))
	for _, b := range t.branches {
		branches = append(branches, b)
	}
	sort.Sort(branches)

	// Spill branches by deepest first.
	for _, b := range branches {
		t.spillBranch(b)
	}
}

// spillLeaf writes a leaf to one or more dirty pages.
func (t *RWTransaction) spillLeaf(l *leaf) {
	parent := l.parent

	// Split leaf, if necessary.
	leafs := l.split(t.db.pageSize)

	// TODO: If this is a root leaf and we split then add a parent branch.

	// Process each resulting leaf.
	previd := leafs[0].pgid
	for index, l := range leafs {
		// Allocate contiguous space for the leaf.
		p := t.allocate((l.size() / t.db.pageSize) + 1)

		// Write the leaf to the page.
		l.write(p)

		// Insert or replace the node in the parent branch with the new pgid.
		if parent != nil {
			parent.put(previd, p.id, l.items[0].key, (index == 0))
			previd = l.pgid
		}
	}
}

// spillBranch writes a branch to one or more dirty pages.
func (t *RWTransaction) spillBranch(l *branch) {
	warn("[pending] RWTransaction.spillBranch()") // TODO
}

// write writes any dirty pages to disk.
func (t *RWTransaction) write() error {
	// TODO(benbjohnson): If our last page id is greater than the mmap size then lock the DB and resize.

	// Sort pages by id.
	pages := make(pages, 0, len(t.pages))
	for _, p := range t.pages {
		pages = append(pages, p)
	}
	sort.Sort(pages)

	// Write pages to disk in order.
	for _, p := range pages {
		size := (int(p.overflow) + 1) * t.db.pageSize
		buf := (*[maxAllocSize]byte)(unsafe.Pointer(p))[:size]
		t.db.file.WriteAt(buf, int64(p.id)*int64(t.db.pageSize))
	}

	return nil
}

// writeMeta writes the meta to the disk.
func (t *RWTransaction) writeMeta() error {
	// Create a temporary buffer for the meta page.
	buf := make([]byte, t.db.pageSize)
	p := t.db.pageInBuffer(buf, 0)
	t.meta.write(p)

	// Write the meta page to file.
	t.db.metafile.WriteAt(buf, int64(p.id)*int64(t.db.pageSize))

	return nil
}

// leaf retrieves a leaf object based on the current position of a cursor.
func (t *RWTransaction) leaf(c *Cursor) *leaf {
	e := c.stack[len(c.stack)-1]
	id := e.page.id

	// Retrieve leaf if it has already been fetched.
	if l := t.leafs[id]; l != nil {
		return l
	}

	// Otherwise create a leaf and cache it.
	l := &leaf{}
	l.read(t.page(id))
	l.parent = t.branch(c.stack[:len(c.stack)-1])
	t.leafs[id] = l

	return l
}

// branch retrieves a branch object based a cursor stack.
// This should only be called from leaf().
func (t *RWTransaction) branch(stack []elem) *branch {
	if len(stack) == 0 {
		return nil
	}

	// Retrieve branch if it has already been fetched.
	e := &stack[len(stack)-1]
	id := e.page.id
	if b := t.branches[id]; b != nil {
		return b
	}

	// Otherwise create a branch and cache it.
	b := &branch{}
	b.read(t.page(id))
	b.depth = len(stack) - 1
	b.parent = t.branch(stack[:len(stack)-1])
	t.branches[id] = b

	return b
}