actions.py

   1 from time import time
   2
   3 from const import reactor
   4 import const
   5
   6 from hash import intify
   7 from knode import KNode as Node
   8 from ktable import KTable, K
   9
  10 class ActionBase:
  11         """ base class for some long running asynchronous proccesses like finding nodes or values """
  12         def __init__(self, table, target, callback):
  13                 self.table = table
  14                 self.target = target
  15                 self.num = intify(target)
  16                 self.found = {}
  17                 self.queried = {}
  18                 self.answered = {}
  19                 self.callback = callback
  20                 self.outstanding = 0
  21                 self.finished = 0
  22
  23                 def sort(a, b, num=self.num):
  24                         """ this function is for sorting nodes relative to the ID we are looking for """
  25                         x, y = num ^ a.num, num ^ b.num
  26                         if x > y:
  27                                 return 1
  28                         elif x < y:
  29                                 return -1
  30                         return 0
  31                 self.sort = sort
  32
  33         def goWithNodes(self, t):
  34                 pass
  35
  36
  37
  38 FIND_NODE_TIMEOUT = 15
  39
  40 class FindNode(ActionBase):
  41         """ find node action merits it's own class as it is a long running stateful process """
  42         def handleGotNodes(self, args):
  43                 l, sender = args
  44                 sender = Node().initWithDict(sender)
  45                 self.table.table.insertNode(sender)
  46                 if self.finished or self.answered.has_key(sender.id):
  47                         # a day late and a dollar short
  48                         return
  49                 self.outstanding = self.outstanding - 1
  50                 self.answered[sender.id] = 1
  51                 for node in l:
  52                         n = Node().initWithDict(node)
  53                         if not self.found.has_key(n.id):
  54                                 self.found[n.id] = n
  55                 self.schedule()
  56
  57         def schedule(self):
  58                 """
  59                         send messages to new peers, if necessary
  60                 """
  61                 if self.finished:
  62                         return
  63                 l = self.found.values()
  64                 l.sort(self.sort)
  65
  66                 for node in l[:K]:
  67                         if node.id == self.target:
  68                                 self.finished=1
  69                                 return self.callback([node])
  70                         if (not self.queried.has_key(node.id)) and node.id != self.table.node.id:
  71                                 #xxxx t.timeout = time.time() + FIND_NODE_TIMEOUT
  72                                 df = node.findNode(self.target, self.table.node.senderDict())
  73                                 df.addCallbacks(self.handleGotNodes, self.makeMsgFailed(node))
  74                                 self.outstanding = self.outstanding + 1
  75                                 self.queried[node.id] = 1
  76                         if self.outstanding >= const.CONCURRENT_REQS:
  77                                 break
  78                 assert(self.outstanding) >=0
  79                 if self.outstanding == 0:
  80                         ## all done!!
  81                         self.finished=1
  82                         reactor.callFromThread(self.callback, l[:K])
  83
  84         def makeMsgFailed(self, node):
  85                 def defaultGotNodes(err, self=self, node=node):
  86                         self.table.table.nodeFailed(node)
  87                         self.outstanding = self.outstanding - 1
  88                         self.schedule()
  89                 return defaultGotNodes
  90
  91         def goWithNodes(self, nodes):
  92                 """
  93                         this starts the process, our argument is a transaction with t.extras being our list of nodes
  94                         it's a transaction since we got called from the dispatcher
  95                 """
  96                 for node in nodes:
  97                         if node.id == self.table.node.id:
  98                                 continue
  99                         else:
 100                                 self.found[node.id] = node
 101
 102                 self.schedule()
 103
 104
 105 GET_VALUE_TIMEOUT = 15
 106 class GetValue(FindNode):
 107     """ get value task """
 108     def handleGotNodes(self, args):
 109                 l, sender = args
 110                 sender = Node().initWithDict(sender)
 111                 self.table.table.insertNode(sender)
 112                 if self.finished or self.answered.has_key(sender.id):
 113                         # a day late and a dollar short
 114                         return
 115                 self.outstanding = self.outstanding - 1
 116                 self.answered[sender.id] = 1
 117                 # go through nodes
 118                 # if we have any closer than what we already got, query them
 119                 if l.has_key('nodes'):
 120                         for node in l['nodes']:
 121                                 n = Node().initWithDict(node)
 122                                 if not self.found.has_key(n.id):
 123                                         self.found[n.id] = n
 124                 elif l.has_key('values'):
 125                         def x(y, z=self.results):
 126                                 y = y.decode('base64')
 127                                 if not z.has_key(y):
 128                                         z[y] = 1
 129                                         return y
 130                                 else:
 131                                         return None
 132                         v = filter(None, map(x, l['values']))
 133                         if(len(v)):
 134                                 reactor.callFromThread(self.callback, v)
 135                 self.schedule()
 136
 137     ## get value
 138     def schedule(self):
 139                 if self.finished:
 140                         return
 141                 l = self.found.values()
 142                 l.sort(self.sort)
 143
 144                 for node in l[:K]:
 145                         if (not self.queried.has_key(node.id)) and node.id != self.table.node.id:
 146                                 #xxx t.timeout = time.time() + GET_VALUE_TIMEOUT
 147                                 df = node.findValue(self.target, self.table.node.senderDict())
 148                                 df.addCallback(self.handleGotNodes)
 149                                 df.addErrback(self.makeMsgFailed(node))
 150                                 self.outstanding = self.outstanding + 1
 151                                 self.queried[node.id] = 1
 152                         if self.outstanding >= const.CONCURRENT_REQS:
 153                                 break
 154                 assert(self.outstanding) >=0
 155                 if self.outstanding == 0:
 156                         ## all done, didn't find it!!
 157                         self.finished=1
 158                         reactor.callFromThread(self.callback,[])
 159
 160     ## get value
 161     def goWithNodes(self, nodes, found=None):
 162                 self.results = {}
 163                 if found:
 164                         for n in found:
 165                                 self.results[n] = 1
 166                 for node in nodes:
 167                         if node.id == self.table.node.id:
 168                                 continue
 169                         else:
 170                                 self.found[node.id] = node
 171
 172                 self.schedule()
 173
 174
 175
 176 class KeyExpirer:
 177         def __init__(self, store):
 178                 self.store = store
 179                 reactor.callLater(const.KEINITIAL_DELAY, self.doExpire)
 180
 181         def doExpire(self):
 182                 self.cut = "%0.6f" % (time() - const.KE_AGE)
 183                 self._expire()
 184
 185         def _expire(self):
 186                 c = self.store.cursor()
 187                 s = "delete from kv where time < '%s';" % self.cut
 188                 c.execute(s)
 189                 reactor.callLater(const.KE_DELAY, self.doExpire)
 190