add files i forgot to last night
Jeromy committed
Aug 16, 2014 at 08:20 UTC
2f3121024a496cca0fa2cee0510de4870407fac1
2 files changed
+140
-20
routing/dht/mes_listener.go
new
+116
@@ -0,0 +1,116 @@
1
+package dht
2
+
3
+import (
4
+ "sync"
5
+ "time"
6
+
7
+ swarm "github.com/jbenet/go-ipfs/swarm"
8
+ u "github.com/jbenet/go-ipfs/util"
9
+)
10
+
11
+type MesListener struct {
12
+ listeners map[uint64]*listenInfo
13
+ haltchan chan struct{}
14
+ unlist chan uint64
15
+ nlist chan *listenInfo
16
+ send chan *respMes
17
+}
18
+
19
+// The listen info struct holds information about a message that is being waited for
20
+type listenInfo struct {
21
+ // Responses matching the listen ID will be sent through resp
22
+ resp chan *swarm.Message
23
+
24
+ // count is the number of responses to listen for
25
+ count int
26
+
27
+ // eol is the time at which this listener will expire
28
+ eol time.Time
29
+
30
+ // sendlock is used to prevent conditions where we try to send on the resp
31
+ // channel as its being closed by a timeout in another thread
32
+ sendLock sync.Mutex
33
+
34
+ closed bool
35
+
36
+ id uint64
37
+}
38
+
39
+func NewMesListener() *MesListener {
40
+ ml := new(MesListener)
41
+ ml.haltchan = make(chan struct{})
42
+ ml.listeners = make(map[uint64]*listenInfo)
43
+ ml.nlist = make(chan *listenInfo, 16)
44
+ ml.send = make(chan *respMes, 16)
45
+ ml.unlist = make(chan uint64, 16)
46
+ go ml.run()
47
+ return ml
48
+}
49
+
50
+func (ml *MesListener) Listen(id uint64, count int, timeout time.Duration) <-chan *swarm.Message {
51
+ li := new(listenInfo)
52
+ li.count = count
53
+ li.eol = time.Now().Add(timeout)
54
+ li.resp = make(chan *swarm.Message, count)
55
+ li.id = id
56
+ ml.nlist <- li
57
+ return li.resp
58
+}
59
+
60
+func (ml *MesListener) Unlisten(id uint64) {
61
+ ml.unlist <- id
62
+}
63
+
64
+type respMes struct {
65
+ id uint64
66
+ mes *swarm.Message
67
+}
68
+
69
+func (ml *MesListener) Respond(id uint64, mes *swarm.Message) {
70
+ ml.send <- &respMes{
71
+ id: id,
72
+ mes: mes,
73
+ }
74
+}
75
+
76
+func (ml *MesListener) Halt() {
77
+ ml.haltchan <- struct{}{}
78
+}
79
+
80
+func (ml *MesListener) run() {
81
+ for {
82
+ select {
83
+ case <-ml.haltchan:
84
+ return
85
+ case id := <-ml.unlist:
86
+ trg, ok := ml.listeners[id]
87
+ if !ok {
88
+ continue
89
+ }
90
+ close(trg.resp)
91
+ delete(ml.listeners, id)
92
+ case li := <-ml.nlist:
93
+ ml.listeners[li.id] = li
94
+ case s := <-ml.send:
95
+ trg, ok := ml.listeners[s.id]
96
+ if !ok {
97
+ u.DOut("Send with no listener.")
98
+ continue
99
+ }
100
+
101
+ if time.Now().After(trg.eol) {
102
+ close(trg.resp)
103
+ delete(ml.listeners, s.id)
104
+ continue
105
+ }
106
+
107
+ trg.resp <- s.mes
108
+ trg.count--
109
+
110
+ if trg.count == 0 {
111
+ close(trg.resp)
112
+ delete(ml.listeners, s.id)
113
+ }
114
+ }
115
+ }
116
+}
routing/dht/routing.go
+24
-20
@@ -21,6 +21,12 @@ import (
21
// Pool size is the number of nodes used for group find/set RPC calls
22
var PoolSize = 6
23
24
+// We put the 'K' in kademlia!
25
+var KValue = 10
26
+
27
+// Its in the paper, i swear
28
+var AlphaValue = 3
29
+
30
// TODO: determine a way of creating and managing message IDs
31
func GenerateMessageID() uint64 {
32
//return (uint64(rand.Uint32()) << 32) & uint64(rand.Uint32())
@@ -35,24 +41,25 @@ func GenerateMessageID() uint64 {
41
// This is the top level "Store" operation of the DHT
42
func (s *IpfsDHT) PutValue(key u.Key, value []byte) {
43
complete := make(chan struct{})
44
+ count := 0
45
for _, route := range s.routes {
39
- p := route.NearestPeer(kb.ConvertKey(key))
40
- if p == nil {
41
- s.network.Error(kb.ErrLookupFailure)
42
- go func() {
46
+ peers := route.NearestPeers(kb.ConvertKey(key), KValue)
47
+ for _, p := range peers {
48
+ if p == nil {
49
+ s.network.Error(kb.ErrLookupFailure)
50
+ continue
51
+ }
52
+ count++
53
+ go func(sp *peer.Peer) {
54
+ err := s.putValueToNetwork(sp, string(key), value)
55
+ if err != nil {
56
+ s.network.Error(err)
57
+ }
58
complete <- struct{}{}
44
- }()
45
- continue
59
+ }(p)
60
}
47
- go func() {
48
- err := s.putValueToNetwork(p, string(key), value)
49
- if err != nil {
50
- s.network.Error(err)
51
- }
52
- complete <- struct{}{}
53
- }()
61
}
55
- for _, _ = range s.routes {
62
+ for i := 0; i < count; i++ {
63
<-complete
64
}
65
}
@@ -150,15 +157,13 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
157
158
c := counter{}
159
153
- // This limit value is referred to as k in the kademlia paper
154
- limit := 20
160
count := 0
161
go func() {
162
for {
163
select {
164
case p := <-npeer_chan:
165
count++
161
- if count >= limit {
166
+ if count >= KValue {
167
break
168
}
169
c.Increment()
@@ -194,7 +199,7 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
199
200
for _, np := range peers {
201
// TODO: filter out peers that arent closer
197
- if !pset.Contains(np) && pset.Size() < limit {
202
+ if !pset.Contains(np) && pset.Size() < KValue {
203
pset.Add(np) //This is racey... make a single function to do operation
204
npeer_chan <- np
205
}
@@ -204,8 +209,7 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
209
}
210
}
211
207
- concurFactor := 3
208
- for i := 0; i < concurFactor; i++ {
212
+ for i := 0; i < AlphaValue; i++ {
213
go process()
214
}
215