efficiency improvement - don't pass transactions to old peers
[p2pool.git] / p2pool / p2p.py
1 from __future__ import division
2
3 import math
4 import random
5 import time
6
7 from twisted.internet import defer, protocol, reactor
8 from twisted.python import failure, log
9
10 import p2pool
11 from p2pool import data as p2pool_data
12 from p2pool.bitcoin import data as bitcoin_data
13 from p2pool.util import deferral, p2protocol, pack, variable
14
15 class PeerMisbehavingError(Exception):
16     pass
17
18
19 def fragment(f, **kwargs):
20     try:
21         return f(**kwargs)
22     except p2protocol.TooLong:
23         att(f, **dict((k, v[:len(v)//2]) for k, v in kwargs.iteritems()))
24         return att(f, **dict((k, v[len(v)//2:]) for k, v in kwargs.iteritems()))
25
26 class Protocol(p2protocol.Protocol):
27     max_remembered_txs_size = 2500000
28     
29     def __init__(self, node, incoming):
30         p2protocol.Protocol.__init__(self, node.net.PREFIX, 1000000, node.traffic_happened)
31         self.node = node
32         self.incoming = incoming
33         
34         self.other_version = None
35         self.connected2 = False
36     
37     def connectionMade(self):
38         p2protocol.Protocol.connectionMade(self)
39         
40         self.factory.proto_made_connection(self)
41         
42         self.connection_lost_event = variable.Event()
43         
44         self.addr = self.transport.getPeer().host, self.transport.getPeer().port
45         
46         self.send_version(
47             version=8,
48             services=0,
49             addr_to=dict(
50                 services=0,
51                 address=self.transport.getPeer().host,
52                 port=self.transport.getPeer().port,
53             ),
54             addr_from=dict(
55                 services=0,
56                 address=self.transport.getHost().host,
57                 port=self.transport.getHost().port,
58             ),
59             nonce=self.node.nonce,
60             sub_version=p2pool.__version__,
61             mode=1,
62             best_share_hash=self.node.best_share_hash_func(),
63         )
64         
65         self.timeout_delayed = reactor.callLater(10, self._connect_timeout)
66         
67         self.get_shares = deferral.GenericDeferrer(
68             max_id=2**256,
69             func=lambda id, hashes, parents, stops: self.send_sharereq(id=id, hashes=hashes, parents=parents, stops=stops),
70             timeout=15,
71             on_timeout=self.transport.loseConnection,
72         )
73         
74         self.remote_tx_hashes = set() # view of peer's known_txs # not actually initially empty, but sending txs instead of tx hashes won't hurt
75         self.remote_remembered_txs_size = 0
76         
77         self.remembered_txs = {} # view of peer's mining_txs
78         self.remembered_txs_size = 0
79         self.known_txs_cache = {}
80     
81     def _connect_timeout(self):
82         self.timeout_delayed = None
83         print 'Handshake timed out, disconnecting from %s:%i' % self.addr
84         self.transport.loseConnection()
85     
86     def packetReceived(self, command, payload2):
87         try:
88             if command != 'version' and not self.connected2:
89                 raise PeerMisbehavingError('first message was not version message')
90             p2protocol.Protocol.packetReceived(self, command, payload2)
91         except PeerMisbehavingError, e:
92             print 'Peer %s:%i misbehaving, will drop and ban. Reason:' % self.addr, e.message
93             self.badPeerHappened()
94     
95     def badPeerHappened(self):
96         if p2pool.DEBUG:
97             print "Bad peer banned:", self.addr
98         self.transport.loseConnection()
99         self.node.bans[self.transport.getPeer().host] = time.time() + 60*60
100     
101     def _timeout(self):
102         self.timeout_delayed = None
103         print 'Connection timed out, disconnecting from %s:%i' % self.addr
104         self.transport.loseConnection()
105     
106     message_version = pack.ComposedType([
107         ('version', pack.IntType(32)),
108         ('services', pack.IntType(64)),
109         ('addr_to', bitcoin_data.address_type),
110         ('addr_from', bitcoin_data.address_type),
111         ('nonce', pack.IntType(64)),
112         ('sub_version', pack.VarStrType()),
113         ('mode', pack.IntType(32)), # always 1 for legacy compatibility
114         ('best_share_hash', pack.PossiblyNoneType(0, pack.IntType(256))),
115     ])
116     def handle_version(self, version, services, addr_to, addr_from, nonce, sub_version, mode, best_share_hash):
117         if self.other_version is not None:
118             raise PeerMisbehavingError('more than one version message')
119         if version < 4:
120             raise PeerMisbehavingError('peer too old')
121         
122         self.other_version = version
123         self.other_sub_version = sub_version[:512]
124         self.other_services = services
125         
126         if nonce == self.node.nonce:
127             raise PeerMisbehavingError('was connected to self')
128         if nonce in self.node.peers:
129             if p2pool.DEBUG:
130                 print 'Detected duplicate connection, disconnecting from %s:%i' % self.addr
131             self.transport.loseConnection()
132             return
133         
134         self.nonce = nonce
135         self.connected2 = True
136         
137         self.timeout_delayed.cancel()
138         self.timeout_delayed = reactor.callLater(100, self._timeout)
139         
140         old_dataReceived = self.dataReceived
141         def new_dataReceived(data):
142             if self.timeout_delayed is not None:
143                 self.timeout_delayed.reset(100)
144             old_dataReceived(data)
145         self.dataReceived = new_dataReceived
146         
147         self.factory.proto_connected(self)
148         
149         self._stop_thread = deferral.run_repeatedly(lambda: [
150             self.send_ping(),
151         random.expovariate(1/100)][-1])
152         
153         self._stop_thread2 = deferral.run_repeatedly(lambda: [
154             self.send_addrme(port=self.node.port),
155         random.expovariate(1/(100*len(self.node.peers) + 1))][-1])
156         
157         if best_share_hash is not None:
158             self.node.handle_share_hashes([best_share_hash], self)
159         
160         if self.other_version < 8:
161             return
162         
163         def update_remote_view_of_my_known_txs(before, after):
164             added = set(after) - set(before)
165             removed = set(before) - set(after)
166             if added:
167                 self.send_have_tx(tx_hashes=list(added))
168             if removed:
169                 self.send_losing_tx(tx_hashes=list(removed))
170                 
171                 # cache forgotten txs here for a little while so latency of "losing_tx" packets doesn't cause problems
172                 key = max(self.known_txs_cache) + 1 if self.known_txs_cache else 0
173                 self.known_txs_cache[key] = dict((h, before[h]) for h in removed)
174                 reactor.callLater(20, self.known_txs_cache.pop, key)
175         watch_id = self.node.known_txs_var.transitioned.watch(update_remote_view_of_my_known_txs)
176         self.connection_lost_event.watch(lambda: self.node.known_txs_var.transitioned.unwatch(watch_id))
177         
178         self.send_have_tx(tx_hashes=self.node.known_txs_var.value.keys())
179         
180         def update_remote_view_of_my_mining_txs(before, after):
181             added = set(after) - set(before)
182             removed = set(before) - set(after)
183             if added:
184                 self.remote_remembered_txs_size += sum(len(bitcoin_data.tx_type.pack(after[x])) for x in added)
185                 assert self.remote_remembered_txs_size <= self.max_remembered_txs_size
186                 fragment(self.send_remember_tx, tx_hashes=[x for x in added if x in self.remote_tx_hashes], txs=[after[x] for x in added if x not in self.remote_tx_hashes])
187             if removed:
188                 self.send_forget_tx(tx_hashes=removed)
189                 self.remote_remembered_txs_size -= sum(len(bitcoin_data.tx_type.pack(before[x])) for x in removed)
190         watch_id2 = self.node.mining_txs_var.transitioned.watch(update_remote_view_of_my_mining_txs)
191         self.connection_lost_event.watch(lambda: self.node.mining_txs_var.transitioned.unwatch(watch_id2))
192         
193         self.remote_remembered_txs_size += sum(len(bitcoin_data.tx_type.pack(x)) for x in self.node.mining_txs_var.value.values())
194         assert self.remote_remembered_txs_size <= self.max_remembered_txs_size
195         fragment(self.send_remember_tx, tx_hashes=[], txs=self.node.mining_txs_var.value.values())
196     
197     message_ping = pack.ComposedType([])
198     def handle_ping(self):
199         pass
200     
201     message_addrme = pack.ComposedType([
202         ('port', pack.IntType(16)),
203     ])
204     def handle_addrme(self, port):
205         host = self.transport.getPeer().host
206         #print 'addrme from', host, port
207         if host == '127.0.0.1':
208             if random.random() < .8 and self.node.peers:
209                 random.choice(self.node.peers.values()).send_addrme(port=port) # services...
210         else:
211             self.node.got_addr((self.transport.getPeer().host, port), self.other_services, int(time.time()))
212             if random.random() < .8 and self.node.peers:
213                 random.choice(self.node.peers.values()).send_addrs(addrs=[
214                     dict(
215                         address=dict(
216                             services=self.other_services,
217                             address=host,
218                             port=port,
219                         ),
220                         timestamp=int(time.time()),
221                     ),
222                 ])
223     
224     message_addrs = pack.ComposedType([
225         ('addrs', pack.ListType(pack.ComposedType([
226             ('timestamp', pack.IntType(64)),
227             ('address', bitcoin_data.address_type),
228         ]))),
229     ])
230     def handle_addrs(self, addrs):
231         for addr_record in addrs:
232             self.node.got_addr((addr_record['address']['address'], addr_record['address']['port']), addr_record['address']['services'], min(int(time.time()), addr_record['timestamp']))
233             if random.random() < .8 and self.node.peers:
234                 random.choice(self.node.peers.values()).send_addrs(addrs=[addr_record])
235     
236     message_getaddrs = pack.ComposedType([
237         ('count', pack.IntType(32)),
238     ])
239     def handle_getaddrs(self, count):
240         if count > 100:
241             count = 100
242         self.send_addrs(addrs=[
243             dict(
244                 timestamp=int(self.node.addr_store[host, port][2]),
245                 address=dict(
246                     services=self.node.addr_store[host, port][0],
247                     address=host,
248                     port=port,
249                 ),
250             ) for host, port in
251             self.node.get_good_peers(count)
252         ])
253     
254     message_shares = pack.ComposedType([
255         ('shares', pack.ListType(p2pool_data.share_type)),
256     ])
257     def handle_shares(self, shares):
258         self.node.handle_shares([p2pool_data.load_share(share, self.node.net, self) for share in shares if share['type'] not in [6, 7]], self)
259     
260     def sendShares(self, shares, tracker, known_txs, include_txs_with=[]):
261         if not shares:
262             return defer.succeed(None)
263         
264         if self.other_version >= 8:
265             tx_hashes = set()
266             for share in shares:
267                 if share.hash in include_txs_with:
268                     tx_hashes.update(share.get_other_tx_hashes(tracker))
269             
270             hashes_to_send = [x for x in tx_hashes if x not in self.node.mining_txs_var.value and x in known_txs]
271             
272             new_remote_remembered_txs_size = self.remote_remembered_txs_size + sum(len(bitcoin_data.tx_type.pack(known_txs[x])) for x in hashes_to_send)
273             if new_remote_remembered_txs_size > self.max_remembered_txs_size:
274                 raise ValueError('shares have too many txs')
275             self.remote_remembered_txs_size = new_remote_remembered_txs_size
276             
277             fragment(self.send_remember_tx, tx_hashes=[x for x in hashes_to_send if x in self.remote_tx_hashes], txs=[known_txs[x] for x in hashes_to_send if x not in self.remote_tx_hashes])
278         
279         res = fragment(self.send_shares, shares=[share.as_share() for share in shares])
280         
281         if self.other_version >= 8:
282             res = self.send_forget_tx(tx_hashes=hashes_to_send)
283             
284             self.remote_remembered_txs_size -= sum(len(bitcoin_data.tx_type.pack(known_txs[x])) for x in hashes_to_send)
285         
286         return res
287     
288     
289     message_sharereq = pack.ComposedType([
290         ('id', pack.IntType(256)),
291         ('hashes', pack.ListType(pack.IntType(256))),
292         ('parents', pack.VarIntType()),
293         ('stops', pack.ListType(pack.IntType(256))),
294     ])
295     def handle_sharereq(self, id, hashes, parents, stops):
296         shares = self.node.handle_get_shares(hashes, parents, stops, self)
297         try:
298             self.send_sharereply(id=id, result='good', shares=[share.as_share() for share in shares])
299         except p2protocol.TooLong:
300             self.send_sharereply(id=id, result='too long', shares=[])
301     
302     message_sharereply = pack.ComposedType([
303         ('id', pack.IntType(256)),
304         ('result', pack.EnumType(pack.VarIntType(), {0: 'good', 1: 'too long', 2: 'unk2', 3: 'unk3', 4: 'unk4', 5: 'unk5', 6: 'unk6'})),
305         ('shares', pack.ListType(p2pool_data.share_type)),
306     ])
307     def handle_sharereply(self, id, result, shares):
308         if result == 'good':
309             res = [p2pool_data.load_share(share, self.node.net, self) for share in shares if share['type'] not in [6, 7]]
310         else:
311             res = failure.Failure("sharereply result: " + result)
312         self.get_shares.got_response(id, res)
313     
314     
315     message_bestblock = pack.ComposedType([
316         ('header', bitcoin_data.block_header_type),
317     ])
318     def handle_bestblock(self, header):
319         self.node.handle_bestblock(header, self)
320     
321     
322     message_have_tx = pack.ComposedType([
323         ('tx_hashes', pack.ListType(pack.IntType(256))),
324     ])
325     def handle_have_tx(self, tx_hashes):
326         self.remote_tx_hashes.update(tx_hashes)
327     message_losing_tx = pack.ComposedType([
328         ('tx_hashes', pack.ListType(pack.IntType(256))),
329     ])
330     def handle_losing_tx(self, tx_hashes):
331         self.remote_tx_hashes.difference_update(tx_hashes)
332     
333     
334     message_remember_tx = pack.ComposedType([
335         ('tx_hashes', pack.ListType(pack.IntType(256))),
336         ('txs', pack.ListType(bitcoin_data.tx_type)),
337     ])
338     def handle_remember_tx(self, tx_hashes, txs):
339         for tx_hash in tx_hashes:
340             if tx_hash not in self.remembered_txs:
341                 if tx_hash in self.node.known_txs_var.value:
342                     tx = self.node.known_txs_var.value[tx_hash]
343                 else:
344                     for cache in self.known_txs_cache.itervalues():
345                         if tx_hash in cache:
346                             tx = cache[tx_hash]
347                             print 'Transaction rescued from peer latency cache!'
348                             break
349                     else:
350                         print >>sys.stderr, 'Peer referenced unknown transaction, disconnecting'
351                         self.transport.loseConnection()
352                         return
353                 
354                 self.remembered_txs[tx_hash] = tx
355                 self.remembered_txs_size += len(bitcoin_data.tx_type.pack(tx))
356         new_known_txs = dict(self.node.known_txs_var.value)
357         for tx in txs:
358             tx_hash = bitcoin_data.hash256(bitcoin_data.tx_type.pack(tx))
359             if tx_hash not in self.remembered_txs:
360                 self.remembered_txs[tx_hash] = tx
361                 self.remembered_txs_size += len(bitcoin_data.tx_type.pack(tx))
362             new_known_txs[tx_hash] = tx
363         self.node.known_txs_var.set(new_known_txs)
364         if self.remembered_txs_size >= self.max_remembered_txs_size:
365             raise PeerMisbehavingError('too much transaction data stored')
366     message_forget_tx = pack.ComposedType([
367         ('tx_hashes', pack.ListType(pack.IntType(256))),
368     ])
369     def handle_forget_tx(self, tx_hashes):
370         for tx_hash in tx_hashes:
371             self.remembered_txs_size -= len(bitcoin_data.tx_type.pack(self.remembered_txs[tx_hash]))
372             assert self.remembered_txs_size >= 0
373             del self.remembered_txs[tx_hash]
374     
375     
376     def connectionLost(self, reason):
377         self.connection_lost_event.happened()
378         if self.timeout_delayed is not None:
379             self.timeout_delayed.cancel()
380         if self.connected2:
381             self.factory.proto_disconnected(self, reason)
382             self._stop_thread()
383             self._stop_thread2()
384             self.connected2 = False
385         self.factory.proto_lost_connection(self, reason)
386         if p2pool.DEBUG:
387             print "Peer connection lost:", self.addr, reason
388         self.get_shares.respond_all(reason)
389     
390     @defer.inlineCallbacks
391     def do_ping(self):
392         start = reactor.seconds()
393         yield self.get_shares(hashes=[0], parents=0, stops=[])
394         end = reactor.seconds()
395         defer.returnValue(end - start)
396
397 class ServerFactory(protocol.ServerFactory):
398     def __init__(self, node, max_conns):
399         self.node = node
400         self.max_conns = max_conns
401         
402         self.conns = {}
403         self.running = False
404     
405     def buildProtocol(self, addr):
406         if sum(self.conns.itervalues()) >= self.max_conns or self.conns.get(self._host_to_ident(addr.host), 0) >= 3:
407             return None
408         if addr.host in self.node.bans and self.node.bans[addr.host] > time.time():
409             return None
410         p = Protocol(self.node, True)
411         p.factory = self
412         if p2pool.DEBUG:
413             print "Got peer connection from:", addr
414         return p
415     
416     def _host_to_ident(self, host):
417         a, b, c, d = host.split('.')
418         return a, b
419     
420     def proto_made_connection(self, proto):
421         ident = self._host_to_ident(proto.transport.getPeer().host)
422         self.conns[ident] = self.conns.get(ident, 0) + 1
423     def proto_lost_connection(self, proto, reason):
424         ident = self._host_to_ident(proto.transport.getPeer().host)
425         self.conns[ident] -= 1
426         if not self.conns[ident]:
427             del self.conns[ident]
428     
429     def proto_connected(self, proto):
430         self.node.got_conn(proto)
431     def proto_disconnected(self, proto, reason):
432         self.node.lost_conn(proto, reason)
433     
434     def start(self):
435         assert not self.running
436         self.running = True
437         
438         def attempt_listen():
439             if self.running:
440                 self.listen_port = reactor.listenTCP(self.node.port, self)
441         deferral.retry('Error binding to P2P port:', traceback=False)(attempt_listen)()
442     
443     def stop(self):
444         assert self.running
445         self.running = False
446         
447         return self.listen_port.stopListening()
448
449 class ClientFactory(protocol.ClientFactory):
450     def __init__(self, node, desired_conns, max_attempts):
451         self.node = node
452         self.desired_conns = desired_conns
453         self.max_attempts = max_attempts
454         
455         self.attempts = set()
456         self.conns = set()
457         self.running = False
458     
459     def _host_to_ident(self, host):
460         a, b, c, d = host.split('.')
461         return a, b
462     
463     def buildProtocol(self, addr):
464         p = Protocol(self.node, False)
465         p.factory = self
466         return p
467     
468     def startedConnecting(self, connector):
469         ident = self._host_to_ident(connector.getDestination().host)
470         if ident in self.attempts:
471             raise AssertionError('already have attempt')
472         self.attempts.add(ident)
473     
474     def clientConnectionFailed(self, connector, reason):
475         self.attempts.remove(self._host_to_ident(connector.getDestination().host))
476     
477     def clientConnectionLost(self, connector, reason):
478         self.attempts.remove(self._host_to_ident(connector.getDestination().host))
479     
480     def proto_made_connection(self, proto):
481         pass
482     def proto_lost_connection(self, proto, reason):
483         pass
484     
485     def proto_connected(self, proto):
486         self.conns.add(proto)
487         self.node.got_conn(proto)
488     def proto_disconnected(self, proto, reason):
489         self.conns.remove(proto)
490         self.node.lost_conn(proto, reason)
491     
492     def start(self):
493         assert not self.running
494         self.running = True
495         self._stop_thinking = deferral.run_repeatedly(self._think)
496     def stop(self):
497         assert self.running
498         self.running = False
499         self._stop_thinking()
500     
501     def _think(self):
502         try:
503             if len(self.conns) < self.desired_conns and len(self.attempts) < self.max_attempts and self.node.addr_store:
504                 (host, port), = self.node.get_good_peers(1)
505                 
506                 if self._host_to_ident(host) in self.attempts:
507                     pass
508                 elif host in self.node.bans and self.node.bans[host] > time.time():
509                     pass
510                 else:
511                     #print 'Trying to connect to', host, port
512                     reactor.connectTCP(host, port, self, timeout=5)
513         except:
514             log.err()
515         
516         return random.expovariate(1/1)
517
518 class SingleClientFactory(protocol.ReconnectingClientFactory):
519     def __init__(self, node):
520         self.node = node
521     
522     def buildProtocol(self, addr):
523         p = Protocol(self.node, incoming=False)
524         p.factory = self
525         return p
526     
527     def proto_made_connection(self, proto):
528         pass
529     def proto_lost_connection(self, proto, reason):
530         pass
531     
532     def proto_connected(self, proto):
533         self.resetDelay()
534         self.node.got_conn(proto)
535     def proto_disconnected(self, proto, reason):
536         self.node.lost_conn(proto, reason)
537
538 class Node(object):
539     def __init__(self, best_share_hash_func, port, net, addr_store={}, connect_addrs=set(), desired_outgoing_conns=10, max_outgoing_attempts=30, max_incoming_conns=50, preferred_storage=1000, traffic_happened=variable.Event(), known_txs_var=variable.Variable({}), mining_txs_var=variable.Variable({})):
540         self.best_share_hash_func = best_share_hash_func
541         self.port = port
542         self.net = net
543         self.addr_store = dict(addr_store)
544         self.connect_addrs = connect_addrs
545         self.preferred_storage = preferred_storage
546         self.traffic_happened = traffic_happened
547         self.known_txs_var = known_txs_var
548         self.mining_txs_var = mining_txs_var
549         
550         self.nonce = random.randrange(2**64)
551         self.peers = {}
552         self.bans = {} # address -> end_time
553         self.clientfactory = ClientFactory(self, desired_outgoing_conns, max_outgoing_attempts)
554         self.serverfactory = ServerFactory(self, max_incoming_conns)
555         self.running = False
556     
557     def start(self):
558         if self.running:
559             raise ValueError('already running')
560         
561         self.clientfactory.start()
562         self.serverfactory.start()
563         self.singleclientconnectors = [reactor.connectTCP(addr, port, SingleClientFactory(self)) for addr, port in self.connect_addrs]
564         
565         self.running = True
566         
567         self._stop_thinking = deferral.run_repeatedly(self._think)
568     
569     def _think(self):
570         try:
571             if len(self.addr_store) < self.preferred_storage and self.peers:
572                 random.choice(self.peers.values()).send_getaddrs(count=8)
573         except:
574             log.err()
575         
576         return random.expovariate(1/20)
577     
578     @defer.inlineCallbacks
579     def stop(self):
580         if not self.running:
581             raise ValueError('already stopped')
582         
583         self.running = False
584         
585         self._stop_thinking()
586         yield self.clientfactory.stop()
587         yield self.serverfactory.stop()
588         for singleclientconnector in self.singleclientconnectors:
589             yield singleclientconnector.factory.stopTrying()
590             yield singleclientconnector.disconnect()
591         del self.singleclientconnectors
592     
593     def got_conn(self, conn):
594         if conn.nonce in self.peers:
595             raise ValueError('already have peer')
596         self.peers[conn.nonce] = conn
597         
598         print '%s connection to peer %s:%i established. p2pool version: %i %r' % ('Incoming' if conn.incoming else 'Outgoing', conn.addr[0], conn.addr[1], conn.other_version, conn.other_sub_version)
599     
600     def lost_conn(self, conn, reason):
601         if conn.nonce not in self.peers:
602             raise ValueError('''don't have peer''')
603         if conn is not self.peers[conn.nonce]:
604             raise ValueError('wrong conn')
605         del self.peers[conn.nonce]
606         
607         print 'Lost peer %s:%i - %s' % (conn.addr[0], conn.addr[1], reason.getErrorMessage())
608     
609     
610     def got_addr(self, (host, port), services, timestamp):
611         if (host, port) in self.addr_store:
612             old_services, old_first_seen, old_last_seen = self.addr_store[host, port]
613             self.addr_store[host, port] = services, old_first_seen, max(old_last_seen, timestamp)
614         else:
615             self.addr_store[host, port] = services, timestamp, timestamp
616     
617     def handle_shares(self, shares, peer):
618         print 'handle_shares', (shares, peer)
619     
620     def handle_share_hashes(self, hashes, peer):
621         print 'handle_share_hashes', (hashes, peer)
622     
623     def handle_get_shares(self, hashes, parents, stops, peer):
624         print 'handle_get_shares', (hashes, parents, stops, peer)
625     
626     def handle_bestblock(self, header, peer):
627         print 'handle_bestblock', header
628     
629     def get_good_peers(self, max_count):
630         t = time.time()
631         return [x[0] for x in sorted(self.addr_store.iteritems(), key=lambda (k, (services, first_seen, last_seen)):
632             -math.log(max(3600, last_seen - first_seen))/math.log(max(3600, t - last_seen))*random.expovariate(1)
633         )][:max_count]