separated out Tracker's delta handling into TrackerView class
[p2pool.git] / p2pool / util / forest.py
1 '''
2 forest data structure
3 '''
4
5 import itertools
6
7 from p2pool.util import skiplist, variable
8
9
10 class TrackerSkipList(skiplist.SkipList):
11     def __init__(self, tracker):
12         skiplist.SkipList.__init__(self)
13         self.tracker = tracker
14         
15         self.tracker.removed.watch_weakref(self, lambda self, item: self.forget_item(item.hash))
16     
17     def previous(self, element):
18         return self.tracker._delta_type.from_element(self.tracker.items[element]).tail
19
20
21 class DistanceSkipList(TrackerSkipList):
22     def get_delta(self, element):
23         return element, 1, self.previous(element)
24     
25     def combine_deltas(self, (from_hash1, dist1, to_hash1), (from_hash2, dist2, to_hash2)):
26         if to_hash1 != from_hash2:
27             raise AssertionError()
28         return from_hash1, dist1 + dist2, to_hash2
29     
30     def initial_solution(self, start, (n,)):
31         return 0, start
32     
33     def apply_delta(self, (dist1, to_hash1), (from_hash2, dist2, to_hash2), (n,)):
34         if to_hash1 != from_hash2:
35             raise AssertionError()
36         return dist1 + dist2, to_hash2
37     
38     def judge(self, (dist, hash), (n,)):
39         if dist > n:
40             return 1
41         elif dist == n:
42             return 0
43         else:
44             return -1
45     
46     def finalize(self, (dist, hash), (n,)):
47         assert dist == n
48         return hash
49
50 def get_attributedelta_type(attrs): # attrs: {name: func}
51     class ProtoAttributeDelta(object):
52         __slots__ = ['head', 'tail'] + attrs.keys()
53         
54         @classmethod
55         def get_none(cls, element_id):
56             return cls(element_id, element_id, **dict((k, 0) for k in attrs))
57         
58         @classmethod
59         def from_element(cls, item):
60             return cls(item.hash, item.previous_hash, **dict((k, v(item)) for k, v in attrs.iteritems()))
61         
62         def __init__(self, head, tail, **kwargs):
63             self.head, self.tail = head, tail
64             for k, v in kwargs.iteritems():
65                 setattr(self, k, v)
66         
67         def __add__(self, other):
68             assert self.tail == other.head
69             return self.__class__(self.head, other.tail, **dict((k, getattr(self, k) + getattr(other, k)) for k in attrs))
70         
71         def __sub__(self, other):
72             if self.head == other.head:
73                 return self.__class__(other.tail, self.tail, **dict((k, getattr(self, k) - getattr(other, k)) for k in attrs))
74             elif self.tail == other.tail:
75                 return self.__class__(self.head, other.head, **dict((k, getattr(self, k) - getattr(other, k)) for k in attrs))
76             else:
77                 raise AssertionError()
78         
79         def __repr__(self):
80             return '%s(%r, %r%s)' % (self.__class__, self.head, self.tail, ''.join(', %s=%r' % (k, getattr(self, k)) for k in attrs))
81     ProtoAttributeDelta.attrs = attrs
82     return ProtoAttributeDelta
83
84 AttributeDelta = get_attributedelta_type(dict(
85     height=lambda item: 1,
86 ))
87
88 class TrackerView(object):
89     def __init__(self, tracker, delta_type):
90         self._tracker = tracker
91         self._delta_type = delta_type
92         
93         self._deltas = {} # item_hash -> delta, ref
94         self._reverse_deltas = {} # ref -> set of item_hashes
95         
96         self._ref_generator = itertools.count()
97         self._delta_refs = {} # ref -> delta
98         self._reverse_delta_refs = {} # delta.tail -> ref
99         
100         self._tracker.preremove_special.watch_weakref(self, lambda self, item: self._handle_preremove_special(item))
101         self._tracker.postremove_special.watch_weakref(self, lambda self, item: self._handle_postremove_special(item))
102         self._tracker.removed.watch_weakref(self, lambda self, item: self._handle_removed(item))
103     
104     def _handle_preremove_special(self, item):
105         delta = self._delta_type.from_element(item)
106         
107         # move delta refs referencing children down to this, so they can be moved up in one step
108         if delta.tail in self._reverse_delta_refs:
109             for x in list(self._reverse_deltas.get(self._reverse_delta_refs.get(delta.head, object()), set())):
110                 self.get_last(x)
111             assert delta.head not in self._reverse_delta_refs, list(self._reverse_deltas.get(self._reverse_delta_refs.get(delta.head, None), set()))
112     
113     def _handle_postremove_special(self, item):
114         delta = self._delta_type.from_element(item)
115         
116         # move ref pointing to this up
117         if delta.tail in self._reverse_delta_refs:
118             assert delta.head not in self._reverse_delta_refs, list(self._reverse_deltas.get(self._reverse_delta_refs.get(delta.head, object()), set()))
119             
120             ref = self._reverse_delta_refs[delta.tail]
121             cur_delta = self._delta_refs[ref]
122             assert cur_delta.tail == delta.tail
123             self._delta_refs[ref] = cur_delta - self._delta_type.from_element(item)
124             assert self._delta_refs[ref].tail == delta.head
125             del self._reverse_delta_refs[delta.tail]
126             self._reverse_delta_refs[delta.head] = ref
127     
128     def _handle_removed(self, item):
129         delta = self._delta_type.from_element(item)
130         
131         # delete delta entry and ref if it is empty
132         if delta.head in self._deltas:
133             delta1, ref = self._deltas.pop(delta.head)
134             self._reverse_deltas[ref].remove(delta.head)
135             if not self._reverse_deltas[ref]:
136                 del self._reverse_deltas[ref]
137                 delta2 = self._delta_refs.pop(ref)
138                 del self._reverse_delta_refs[delta2.tail]
139     
140     
141     def get_height(self, item_hash):
142         return self.get_delta_to_last(item_hash).height
143     
144     def get_work(self, item_hash):
145         return self.get_delta_to_last(item_hash).work
146     
147     def get_last(self, item_hash):
148         return self.get_delta_to_last(item_hash).tail
149     
150     def get_height_and_last(self, item_hash):
151         delta = self.get_delta_to_last(item_hash)
152         return delta.height, delta.tail
153     
154     def _get_delta(self, item_hash):
155         if item_hash in self._deltas:
156             delta1, ref = self._deltas[item_hash]
157             delta2 = self._delta_refs[ref]
158             res = delta1 + delta2
159         else:
160             res = self._delta_type.from_element(self._tracker.items[item_hash])
161         assert res.head == item_hash
162         return res
163     
164     def _set_delta(self, item_hash, delta):
165         other_item_hash = delta.tail
166         if other_item_hash not in self._reverse_delta_refs:
167             ref = self._ref_generator.next()
168             assert ref not in self._delta_refs
169             self._delta_refs[ref] = self._delta_type.get_none(other_item_hash)
170             self._reverse_delta_refs[other_item_hash] = ref
171             del ref
172         
173         ref = self._reverse_delta_refs[other_item_hash]
174         ref_delta = self._delta_refs[ref]
175         assert ref_delta.tail == other_item_hash
176         
177         if item_hash in self._deltas:
178             prev_ref = self._deltas[item_hash][1]
179             self._reverse_deltas[prev_ref].remove(item_hash)
180             if not self._reverse_deltas[prev_ref] and prev_ref != ref:
181                 self._reverse_deltas.pop(prev_ref)
182                 x = self._delta_refs.pop(prev_ref)
183                 self._reverse_delta_refs.pop(x.tail)
184         self._deltas[item_hash] = delta - ref_delta, ref
185         self._reverse_deltas.setdefault(ref, set()).add(item_hash)
186     
187     def get_delta_to_last(self, item_hash):
188         assert isinstance(item_hash, (int, long, type(None)))
189         delta = self._delta_type.get_none(item_hash)
190         updates = []
191         while delta.tail in self._tracker.items:
192             updates.append((delta.tail, delta))
193             this_delta = self._get_delta(delta.tail)
194             delta += this_delta
195         for update_hash, delta_then in updates:
196             self._set_delta(update_hash, delta - delta_then)
197         return delta
198     
199     def get_delta(self, item, ancestor):
200         assert self._tracker.is_child_of(ancestor, item)
201         return self.get_delta_to_last(item) - self.get_delta_to_last(ancestor)
202
203 class Tracker(object):
204     def __init__(self, items=[], delta_type=AttributeDelta):
205         self.items = {} # hash -> item
206         self.reverse = {} # delta.tail -> set of item_hashes
207         
208         self.heads = {} # head hash -> tail_hash
209         self.tails = {} # tail hash -> set of head hashes
210         
211         self.added = variable.Event()
212         self.preremove_special = variable.Event()
213         self.postremove_special = variable.Event()
214         self.removed = variable.Event()
215         
216         self.get_nth_parent_hash = DistanceSkipList(self)
217         
218         self._delta_type = delta_type
219         self._default_view = TrackerView(self, delta_type)
220         
221         for item in items:
222             self.add(item)
223     
224     def __getattr__(self, name):
225         attr = getattr(self._default_view, name)
226         setattr(self, name, attr)
227         return attr
228     
229     def add(self, item):
230         assert not isinstance(item, (int, long, type(None)))
231         delta = self._delta_type.from_element(item)
232         
233         if delta.head in self.items:
234             raise ValueError('item already present')
235         
236         if delta.head in self.tails:
237             heads = self.tails.pop(delta.head)
238         else:
239             heads = set([delta.head])
240         
241         if delta.tail in self.heads:
242             tail = self.heads.pop(delta.tail)
243         else:
244             tail = self.get_last(delta.tail)
245         
246         self.items[delta.head] = item
247         self.reverse.setdefault(delta.tail, set()).add(delta.head)
248         
249         self.tails.setdefault(tail, set()).update(heads)
250         if delta.tail in self.tails[tail]:
251             self.tails[tail].remove(delta.tail)
252         
253         for head in heads:
254             self.heads[head] = tail
255         
256         self.added.happened(item)
257     
258     def remove(self, item_hash):
259         assert isinstance(item_hash, (int, long, type(None)))
260         if item_hash not in self.items:
261             raise KeyError()
262         
263         item = self.items[item_hash]
264         del item_hash
265         
266         delta = self._delta_type.from_element(item)
267         
268         children = self.reverse.get(delta.head, set())
269         
270         if delta.head in self.heads and delta.tail in self.tails:
271             tail = self.heads.pop(delta.head)
272             self.tails[tail].remove(delta.head)
273             if not self.tails[delta.tail]:
274                 self.tails.pop(delta.tail)
275         elif delta.head in self.heads:
276             tail = self.heads.pop(delta.head)
277             self.tails[tail].remove(delta.head)
278             if self.reverse[delta.tail] != set([delta.head]):
279                 pass # has sibling
280             else:
281                 self.tails[tail].add(delta.tail)
282                 self.heads[delta.tail] = tail
283         elif delta.tail in self.tails and len(self.reverse[delta.tail]) <= 1:
284             self.preremove_special.happened(item)
285             
286             heads = self.tails.pop(delta.tail)
287             for head in heads:
288                 self.heads[head] = delta.head
289             self.tails[delta.head] = set(heads)
290             
291             self.postremove_special.happened(item)
292         else:
293             raise NotImplementedError()
294         
295         self.items.pop(delta.head)
296         self.reverse[delta.tail].remove(delta.head)
297         if not self.reverse[delta.tail]:
298             self.reverse.pop(delta.tail)
299         
300         self.removed.happened(item)
301     
302     def get_chain(self, start_hash, length):
303         assert length <= self.get_height(start_hash)
304         for i in xrange(length):
305             yield self.items[start_hash]
306             start_hash = self._delta_type.from_element(self.items[start_hash]).tail
307     
308     def is_child_of(self, item_hash, possible_child_hash):
309         height, last = self.get_height_and_last(item_hash)
310         child_height, child_last = self.get_height_and_last(possible_child_hash)
311         if child_last != last:
312             return None # not connected, so can't be determined
313         height_up = child_height - height
314         return height_up >= 0 and self.get_nth_parent_hash(possible_child_hash, height_up) == item_hash
315
316 class SubsetTracker(Tracker):
317     def __init__(self, subset_of, **kwargs):
318         Tracker.__init__(self, **kwargs)
319         self.get_nth_parent_hash = subset_of.get_nth_parent_hash # overwrites Tracker.__init__'s
320         self._subset_of = subset_of
321     
322     def add(self, item):
323         delta = self._delta_type.from_element(item)
324         if self._subset_of is not None:
325             assert delta.head in self._subset_of.items
326         Tracker.add(self, item)
327     
328     def remove(self, item_hash):
329         if self._subset_of is not None:
330             assert item_hash in self._subset_of.items
331         Tracker.remove(self, item_hash)