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