p2pool/worker_interface.py

   1 from __future__ import division
   2
   3 import base64
   4 import json
   5 import random
   6
   7 from twisted.internet import defer, reactor
   8 from twisted.python import log
   9
  10 import p2pool
  11 from p2pool import data as p2pool_data
  12 from p2pool.util import jsonrpc, deferred_resource, variable
  13 from p2pool.bitcoin import data as bitcoin_data
  14
  15 def get_username(request):
  16     try:
  17         return base64.b64decode(request.getHeader('Authorization').split(' ', 1)[1]).split(':')[0]
  18     except: # XXX
  19         return None
  20
  21 def get_payout_script(request, net):
  22     user = get_username(request)
  23     if user is None:
  24         return None
  25     try:
  26         return bitcoin_data.pubkey_hash_to_script2(bitcoin_data.address_to_pubkey_hash(user, net))
  27     except: # XXX blah
  28         return None
  29
  30 def get_memory(request):
  31     if request.getHeader('X-Miner-Extensions') is not None and 'workidentifier' in request.getHeader('X-Miner-Extensions').split(' '):
  32         return 0
  33     if request.getHeader('X-Work-Identifier') is not None:
  34         return 0
  35     user_agent = request.getHeader('User-Agent')
  36     user_agent2 = '' if user_agent is None else user_agent.lower()
  37     if 'java' in user_agent2 or 'diablominer' in user_agent2: return 0 # hopefully diablominer...
  38     if 'cpuminer' in user_agent2: return 0
  39     if 'ufasoft' in user_agent2: return 0 # not confirmed
  40     if 'cgminer' in user_agent2: return 1
  41     if 'poclbm' in user_agent2: return 1
  42     if 'phoenix' in user_agent2: return 2
  43     print 'Unknown miner User-Agent:', repr(user_agent)
  44     return 0
  45
  46 def get_id(request):
  47     return request.getClientIP(), request.getHeader('Authorization')
  48
  49 class Holds(object):
  50     def __init__(self):
  51         self.holds = {}
  52
  53     @defer.inlineCallbacks
  54     def wait_hold(self, request_id):
  55         while request_id in self.holds:
  56             yield self.holds[request_id].get_deferred()
  57
  58     def set_hold(self, request_id, dt):
  59         if request_id in self.holds:
  60             raise ValueError('hold already present!')
  61         self.holds[request_id] = variable.Event()
  62         self.holds[request_id].status = 0
  63         def cb():
  64             if self.holds[request_id].status != 0:
  65                 raise AssertionError()
  66             self.holds[request_id].status = 1
  67             self.holds.pop(request_id).happened()
  68         reactor.callLater(dt, cb)
  69
  70 class LongPollingWorkerInterface(deferred_resource.DeferredResource):
  71     def __init__(self, parent):
  72         self.parent = parent
  73
  74     @defer.inlineCallbacks
  75     def render_GET(self, request):
  76         request.setHeader('Content-Type', 'application/json')
  77         request.setHeader('X-Long-Polling', '/long-polling')
  78         request.setHeader('X-Roll-NTime', 'expire=60')
  79         try:
  80             try:
  81                 request.write(json.dumps({
  82                     'jsonrpc': '2.0',
  83                     'id': 0,
  84                     'result': (yield self.parent.getwork(request, long_poll=True)),
  85                     'error': None,
  86                 }))
  87             except jsonrpc.Error:
  88                 raise
  89             except Exception:
  90                 log.err(None, 'Squelched long polling error:')
  91                 raise jsonrpc.Error(-32099, u'Unknown error')
  92         except jsonrpc.Error, e:
  93             request.write(json.dumps({
  94                 'jsonrpc': '2.0',
  95                 'id': 0,
  96                 'result': None,
  97                 'error': e._to_obj(),
  98             }))
  99     render_POST = render_GET
 100
 101 class WorkerInterface(jsonrpc.Server):
 102     def __init__(self, work, compute, response_callback, net):
 103         jsonrpc.Server.__init__(self)
 104
 105         self.work = work
 106         self.compute = compute
 107         self.response_callback = response_callback
 108         self.net = net
 109         self.holds = Holds()
 110         self.last_cache_invalidation = {}
 111
 112         self.putChild('long-polling', LongPollingWorkerInterface(self))
 113         self.putChild('', self)
 114
 115     @defer.inlineCallbacks
 116     def rpc_getwork(self, request, data=None):
 117         request.setHeader('X-Long-Polling', '/long-polling')
 118         request.setHeader('X-Roll-NTime', 'expire=60')
 119
 120         if data is not None:
 121             defer.returnValue(self.response_callback(data, get_username(request), self.net))
 122
 123         defer.returnValue((yield self.getwork(request)))
 124     rpc_getwork.takes_request = True
 125
 126     @defer.inlineCallbacks
 127     def getwork(self, request, long_poll=False):
 128         request_id = get_id(request)
 129         memory = get_memory(request)
 130
 131         id = random.randrange(10000)
 132         if p2pool.DEBUG:
 133             print 'POLL %i START long_poll=%r user_agent=%r x-work-identifier=%r user=%r' % (id, long_poll, request.getHeader('User-Agent'), request.getHeader('X-Work-Identifier'), get_username(request))
 134
 135         if request_id not in self.last_cache_invalidation:
 136             self.last_cache_invalidation[request_id] = variable.Variable((None, None))
 137
 138         yield self.holds.wait_hold(request_id)
 139         work = self.work.value
 140         thought_work = self.last_cache_invalidation[request_id].value
 141
 142         if long_poll and work == thought_work[-1]:
 143             if p2pool.DEBUG:
 144                 print 'POLL %i WAITING user=%r' % (id, get_username(request))
 145             yield defer.DeferredList([self.work.changed.get_deferred(), self.last_cache_invalidation[request_id].changed.get_deferred()], fireOnOneCallback=True)
 146         work = self.work.value
 147
 148         if thought_work[-1] is not None and work != thought_work[-1] and any(x is None or work['previous_block'] == x['previous_block'] for x in thought_work[-memory or len(thought_work):]):
 149             # clients won't believe the update
 150             work = work.copy()
 151             work['previous_block'] = random.randrange(2**256)
 152             if p2pool.DEBUG:
 153                 print 'POLL %i FAKED user=%r' % (id, get_username(request))
 154             self.holds.set_hold(request_id, .01)
 155         res = self.compute(work, get_payout_script(request, self.net))
 156
 157         self.last_cache_invalidation[request_id].set((thought_work[-1], work))
 158         if p2pool.DEBUG:
 159             print 'POLL %i END %s user=%r' % (id, p2pool_data.format_hash(work['best_share_hash']), get_username(request))
 160
 161         if request.getHeader('X-All-Targets') is None and res.share_target > self.net.MAX_TARGET:
 162             res = res.update(share_target=self.net.MAX_TARGET)
 163
 164         defer.returnValue(res.getwork(identifier=str(work['best_share_hash'])))