-
Notifications
You must be signed in to change notification settings - Fork 0
/
valueaccessor.go
106 lines (98 loc) · 2.52 KB
/
valueaccessor.go
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
package gostore
import (
"time"
)
// valueAccessor provides an alternate interface for reading (and optionally,
// modifying) the Value of a key-mapped StoreMapValue. Transactions can request a
// valueAccessor through the logManager for either reading, or for reading/writing.
// This interface exists to regulate access to the Value, specifically disallow
// readers when there is a writer transaction, and vice-versa.
type valueAccessor struct {
c chan Value
done chan bool
isWAccessor bool
}
func getRAccessor(smv storeMapValue) *valueAccessor {
select {
case smv.ping <- struct{}{}:
default:
initializeAccessorChan(smv)
}
return <-smv.rAccessorChan
}
func getWAccessor(smv storeMapValue) *valueAccessor {
select {
case smv.ping <- struct{}{}:
default:
initializeAccessorChan(smv)
}
return <-smv.wAccessorChan
}
func (va *valueAccessor) promoteToWAccessor(smv storeMapValue) *valueAccessor {
if va.isWAccessor {
return va
}
va.done <- true
return getWAccessor(smv)
}
func connectAccessor(va *valueAccessor, smv storeMapValue, connDone chan bool) {
v := smv.value
for {
select {
case va.c <- v:
case v = <-func() chan Value {
if va.isWAccessor {
return va.c
}
return nil
}():
case d := <-va.done:
close(va.c)
close(va.done)
if va.isWAccessor && d {
smv.value = v
}
connDone <- true
}
}
}
var accessorTimeout = time.Minute * 10
// initializeAccessorChan for a StoreMapValue starts up a goroutine that regulates
// the provision of read and write accessors to the StoreMapValue. Essentially, it
// behaves like a RWMutex.
func initializeAccessorChan(smv storeMapValue) {
go func() {
numReaders := 0
connDone := make(chan bool)
for {
va := valueAccessor{make(chan Value), make(chan bool), false}
select {
case smv.wAccessorChan <- func() *valueAccessor {
if numReaders == 0 { // only allowed if there are no read accessors
return &va
}
select {}
}():
va.isWAccessor = true
go connectAccessor(&va, smv, connDone)
<-connDone // block since no one else can access while there is a write accessor
case smv.rAccessorChan <- &va: // Read Access
numReaders++
va.isWAccessor = false
go connectAccessor(&va, smv, connDone)
case <-connDone: // Commit/Abort
numReaders--
case <-smv.ping:
case <-func() <-chan time.Time {
if numReaders == 0 { // only allowed if there are no accessors
return time.After(accessorTimeout)
}
return nil
}(): // Timeout
break
}
}
close(connDone)
}()
return
}