6766cd9c6169dd8301cd6ff6c052d663175b0541
[quix0rs-apt-p2p.git] / apt_dht_Khashmir / actions.py
1 ## Copyright 2002-2004 Andrew Loewenstern, All Rights Reserved
2 # see LICENSE.txt for license information
3
4 from twisted.internet import reactor
5 from twisted.python import log
6
7 from khash import intify
8 from util import uncompact
9
10 class ActionBase:
11     """ base class for some long running asynchronous proccesses like finding nodes or values """
12     def __init__(self, caller, target, callback, config):
13         self.caller = caller
14         self.target = target
15         self.config = config
16         self.num = intify(target)
17         self.found = {}
18         self.queried = {}
19         self.answered = {}
20         self.callback = callback
21         self.outstanding = 0
22         self.finished = 0
23     
24         def sort(a, b, num=self.num):
25             """ this function is for sorting nodes relative to the ID we are looking for """
26             x, y = num ^ a.num, num ^ b.num
27             if x > y:
28                 return 1
29             elif x < y:
30                 return -1
31             return 0
32         self.sort = sort
33         
34     def goWithNodes(self, t):
35         pass
36     
37     
38
39 FIND_NODE_TIMEOUT = 15
40
41 class FindNode(ActionBase):
42     """ find node action merits it's own class as it is a long running stateful process """
43     def handleGotNodes(self, dict):
44         _krpc_sender = dict['_krpc_sender']
45         dict = dict['rsp']
46         n = self.caller.Node(dict["id"], _krpc_sender[0], _krpc_sender[1])
47         self.caller.insertNode(n)
48         if dict["id"] in self.found:
49             self.found[dict["id"]].updateToken(dict.get('token', ''))
50         l = dict["nodes"]
51         if self.finished or self.answered.has_key(dict["id"]):
52             # a day late and a dollar short
53             return
54         self.outstanding = self.outstanding - 1
55         self.answered[dict["id"]] = 1
56         for compact_node in l:
57             node = uncompact(compact_node)
58             n = self.caller.Node(node)
59             if not self.found.has_key(n.id):
60                 self.found[n.id] = n
61         self.schedule()
62         
63     def schedule(self):
64         """
65             send messages to new peers, if necessary
66         """
67         if self.finished:
68             return
69         l = self.found.values()
70         l.sort(self.sort)
71         for node in l[:self.config['K']]:
72             if node.id == self.target:
73                 self.finished=1
74                 return self.callback([node])
75             if (not self.queried.has_key(node.id)) and node.id != self.caller.node.id:
76                 #xxxx t.timeout = time.time() + FIND_NODE_TIMEOUT
77                 df = node.findNode(self.target, self.caller.node.id)
78                 df.addCallbacks(self.handleGotNodes, self.makeMsgFailed(node))
79                 self.outstanding = self.outstanding + 1
80                 self.queried[node.id] = 1
81             if self.outstanding >= self.config['CONCURRENT_REQS']:
82                 break
83         assert self.outstanding >=0
84         if self.outstanding == 0:
85             ## all done!!
86             self.finished=1
87             reactor.callLater(0, self.callback, l[:self.config['K']])
88     
89     def makeMsgFailed(self, node):
90         def defaultGotNodes(err, self=self, node=node):
91             log.msg("find failed (%s) %s/%s" % (self.config['PORT'], node.host, node.port))
92             log.err(err)
93             self.caller.table.nodeFailed(node)
94             self.outstanding = self.outstanding - 1
95             self.schedule()
96         return defaultGotNodes
97     
98     def goWithNodes(self, nodes):
99         """
100             this starts the process, our argument is a transaction with t.extras being our list of nodes
101             it's a transaction since we got called from the dispatcher
102         """
103         for node in nodes:
104             if node.id == self.caller.node.id:
105                 continue
106             else:
107                 self.found[node.id] = node
108         
109         self.schedule()
110     
111
112 get_value_timeout = 15
113 class GetValue(FindNode):
114     def __init__(self, caller, target, callback, config, find="findValue"):
115         FindNode.__init__(self, caller, target, callback, config)
116         self.findValue = find
117             
118     """ get value task """
119     def handleGotNodes(self, dict):
120         _krpc_sender = dict['_krpc_sender']
121         dict = dict['rsp']
122         n = self.caller.Node(dict["id"], _krpc_sender[0], _krpc_sender[1])
123         self.caller.insertNode(n)
124         if self.finished or self.answered.has_key(dict["id"]):
125             # a day late and a dollar short
126             return
127         self.outstanding = self.outstanding - 1
128         self.answered[dict["id"]] = 1
129         # go through nodes
130         # if we have any closer than what we already got, query them
131         if dict.has_key('nodes'):
132             for compact_node in dict['nodes']:
133                 node = uncompact(compact_node)
134                 n = self.caller.Node(node)
135                 if not self.found.has_key(n.id):
136                     self.found[n.id] = n
137         elif dict.has_key('values'):
138             def x(y, z=self.results):
139                 if not z.has_key(y):
140                     z[y] = 1
141                     return y
142                 else:
143                     return None
144             z = len(dict['values'])
145             v = filter(None, map(x, dict['values']))
146             if(len(v)):
147                 reactor.callLater(0, self.callback, self.target, v)
148         self.schedule()
149         
150     ## get value
151     def schedule(self):
152         if self.finished:
153             return
154         l = self.found.values()
155         l.sort(self.sort)
156         
157         for node in l[:self.config['K']]:
158             if (not self.queried.has_key(node.id)) and node.id != self.caller.node.id:
159                 #xxx t.timeout = time.time() + GET_VALUE_TIMEOUT
160                 try:
161                     f = getattr(node, self.findValue)
162                 except AttributeError:
163                     log.msg("findValue %s doesn't have a %s method!" % (node, self.findValue))
164                 else:
165                     df = f(self.target, self.caller.node.id)
166                     df.addCallback(self.handleGotNodes)
167                     df.addErrback(self.makeMsgFailed(node))
168                     self.outstanding = self.outstanding + 1
169                     self.queried[node.id] = 1
170             if self.outstanding >= self.config['CONCURRENT_REQS']:
171                 break
172         assert self.outstanding >=0
173         if self.outstanding == 0:
174             ## all done, didn't find it!!
175             self.finished=1
176             reactor.callLater(0, self.callback, self.target, [])
177
178     ## get value
179     def goWithNodes(self, nodes, found=None):
180         self.results = {}
181         if found:
182             for n in found:
183                 self.results[n] = 1
184         for node in nodes:
185             if node.id == self.caller.node.id:
186                 continue
187             else:
188                 self.found[node.id] = node
189             
190         self.schedule()
191
192
193 class StoreValue(ActionBase):
194     def __init__(self, caller, target, value, callback, config, store="storeValue"):
195         ActionBase.__init__(self, caller, target, callback, config)
196         self.value = value
197         self.stored = []
198         self.store = store
199         
200     def storedValue(self, t, node):
201         self.outstanding -= 1
202         self.caller.insertNode(node)
203         if self.finished:
204             return
205         self.stored.append(t)
206         if len(self.stored) >= self.config['STORE_REDUNDANCY']:
207             self.finished=1
208             self.callback(self.target, self.value, self.stored)
209         else:
210             if not len(self.stored) + self.outstanding >= self.config['STORE_REDUNDANCY']:
211                 self.schedule()
212         return t
213     
214     def storeFailed(self, t, node):
215         log.msg("store failed %s/%s" % (node.host, node.port))
216         self.caller.nodeFailed(node)
217         self.outstanding -= 1
218         if self.finished:
219             return t
220         self.schedule()
221         return t
222     
223     def schedule(self):
224         if self.finished:
225             return
226         num = self.config['CONCURRENT_REQS'] - self.outstanding
227         if num > self.config['STORE_REDUNDANCY']:
228             num = self.config['STORE_REDUNDANCY']
229         for i in range(num):
230             try:
231                 node = self.nodes.pop()
232             except IndexError:
233                 if self.outstanding == 0:
234                     self.finished = 1
235                     self.callback(self.target, self.value, self.stored)
236             else:
237                 if not node.id == self.caller.node.id:
238                     self.outstanding += 1
239                     try:
240                         f = getattr(node, self.store)
241                     except AttributeError:
242                         log.msg("%s doesn't have a %s method!" % (node, self.store))
243                     else:
244                         df = f(self.target, self.value, node.token, self.caller.node.id)
245                         df.addCallback(self.storedValue, node=node)
246                         df.addErrback(self.storeFailed, node=node)
247                     
248     def goWithNodes(self, nodes):
249         self.nodes = nodes
250         self.nodes.sort(self.sort)
251         self.schedule()
252
253
254 class KeyExpirer:
255     def __init__(self, store, config):
256         self.store = store
257         self.config = config
258         self.next_expire = reactor.callLater(self.config['KEINITIAL_DELAY'], self.doExpire)
259     
260     def doExpire(self):
261         self.store.expireValues(self.config['KE_AGE'])
262         self.next_expire = reactor.callLater(self.config['KE_DELAY'], self.doExpire)
263         
264     def shutdown(self):
265         try:
266             self.next_expire.cancel()
267         except:
268             pass