mining/DB_Mysql.py

   1 import time
   2 from stratum import settings
   3 import stratum.logger
   4 log = stratum.logger.get_logger('DB_Mysql')
   5
   6 import MySQLdb
   7
   8 class DB_Mysql():
   9     def __init__(self):
  10         log.debug("Connecting to DB")
  11         self.dbh = MySQLdb.connect(settings.DB_MYSQL_HOST,settings.DB_MYSQL_USER,settings.DB_MYSQL_PASS,settings.DB_MYSQL_DBNAME)
  12         self.dbc = self.dbh.cursor()
  13
  14     def updateStats(self,averageOverTime):
  15         log.debug("Updating Stats")
  16         # Note: we are using transactions... so we can set the speed = 0 and it doesn't take affect until we are commited.
  17         #self.dbc.execute("update pool_worker set speed = 0, alive = 0");
  18         #stime = '%.0f' % ( time.time() - averageOverTime );
  19         #self.dbc.execute("select username,SUM(difficulty) from shares where time > FROM_UNIXTIME(%s) group by username", (stime,))
  20         #total_speed = 0
  21         #for name,shares in self.dbc.fetchall():
  22         #    speed = int(int(shares) * pow(2,32)) / ( int(averageOverTime) * 1000 * 1000)
  23         #    total_speed += speed
  24         #    self.dbc.execute("update pool_worker set speed = %s, alive = 1 where username = %s", (speed,name))
  25         #self.dbc.execute("update pool set value = %s where parameter = 'pool_speed'",[total_speed])
  26         self.dbh.commit()
  27
  28     def archive_check(self):
  29         # Check for found shares to archive
  30         #self.dbc.execute("select time from shares where upstream_result = 1 order by time limit 1")
  31         #data = self.dbc.fetchone()
  32         #if data is None or (data[0] + settings.ARCHIVE_DELAY) > time.time() :
  33         #    return False
  34         return data[0]
  35
  36     def archive_found(self,found_time):
  37         self.dbc.execute("insert into shares_archive_found select * from shares where upstream_result = 'Y' and time <= FROM_UNIXTIME(%s)", (found_time,))
  38         self.dbh.commit()
  39
  40     def archive_to_db(self,found_time):
  41         self.dbc.execute("insert into shares_archive select * from shares where time <= FROM_UNIXTIME(%s)",(found_time,))
  42         self.dbh.commit()
  43
  44     def archive_cleanup(self,found_time):
  45         self.dbc.execute("delete from shares where time <= FROM_UNIXTIME(%s)",(found_time,))
  46         self.dbh.commit()
  47
  48     def archive_get_shares(self,found_time):
  49         #self.dbc.execute("select * from shares where time <= FROM_UNIXTIME(%s)",(found_time,))
  50         return self.dbc
  51
  52     def import_shares(self,data):
  53         log.debug("Importing Shares")
  54 #              0           1            2          3          4         5        6  7            8         9              10
  55 #       data: [worker_name,block_header,block_hash,difficulty,timestamp,is_valid,ip,block_height,prev_hash,invalid_reason,best_diff]
  56         checkin_times = {}
  57         total_shares = 0
  58         best_diff = 0
  59         for k,v in enumerate(data):
  60                 self.dbc.execute("insert into shares (rem_host,username,our_result,upstream_result,reason,share_diff) VALUES " +\
  61                         "(%s,%s,%s,%s,%s,%s)",
  62                         (v[6],v[0],v[5],'N',v[9],v[3]) )
  63
  64         self.dbh.commit()
  65
  66
  67     def found_block(self,data):
  68         # Note: difficulty = -1 here
  69         self.dbc.execute("update shares set upstream_result = %s, solution = %s where time = FROM_UNIXTIME(%s) and username = %s limit 1",
  70                 (data[5],data[2],data[4],data[0]))
  71         self.dbh.commit()
  72
  73     def delete_user(self,username):
  74         log.debug("Deleting Username")
  75         self.dbc.execute("delete from pool_worker where username = %s",
  76                 (username ))
  77         self.dbh.commit()
  78
  79     def insert_user(self,username,password):
  80         log.debug("Adding Username/Password")
  81         self.dbc.execute("insert into pool_worker (username,password) VALUES (%s,%s)",
  82                 (username, password ))
  83         self.dbh.commit()
  84
  85     def update_user(self,username,password):
  86         log.debug("Updating Username/Password")
  87         self.dbc.execute("update pool_worker set password = %(pass)s where username = %(uname)s",
  88                 (username, password ))
  89         self.dbh.commit()
  90
  91     def update_worker_diff(self,username,diff):
  92         if settings.DATABASE_EXTEND == True :
  93             self.dbc.execute("update pool_worker set difficulty = %s where username = %s",(diff,username))
  94             self.dbh.commit()
  95
  96     def clear_worker_diff(self):
  97         if settings.DATABASE_EXTEND == True :
  98             self.dbc.execute("update pool_worker set difficulty = 0")
  99             self.dbh.commit()
 100
 101     def check_password(self,username,password):
 102         log.debug("Checking Username/Password")
 103         self.dbc.execute("select COUNT(*) from pool_worker where username = %s and password = %s",
 104                 (username, password ))
 105         data = self.dbc.fetchone()
 106         if data[0] > 0 :
 107             return True
 108         return False
 109
 110     def update_pool_info(self,pi):
 111         self.dbc.executemany("update pool set value = %s where parameter = %s",[(pi['blocks'],"bitcoin_blocks"),
 112                 (pi['balance'],"bitcoin_balance"),
 113                 (pi['connections'],"bitcoin_connections"),
 114                 (pi['difficulty'],"bitcoin_difficulty"),
 115                 (time.time(),"bitcoin_infotime")
 116                 ])
 117         self.dbh.commit()
 118
 119     def get_pool_stats(self):
 120         self.dbc.execute("select * from pool")
 121         ret = {}
 122         for data in self.dbc.fetchall():
 123             ret[data[0]] = data[1]
 124         return ret
 125
 126     def get_workers_stats(self):
 127         self.dbc.execute("select username,speed,last_checkin,total_shares,total_rejects,total_found,alive,difficulty from pool_worker")
 128         ret = {}
 129         for data in self.dbc.fetchall():
 130             ret[data[0]] = { "username" : data[0],
 131                 "speed" : data[1],
 132                 "last_checkin" : time.mktime(data[2].timetuple()),
 133                 "total_shares" : data[3],
 134                 "total_rejects" : data[4],
 135                 "total_found" : data[5],
 136                 "alive" : data[6],
 137                 "difficulty" : data[7] }
 138         return ret
 139
 140     def close(self):
 141         self.dbh.close()
 142
 143     def check_tables(self):
 144         log.debug("Checking Tables")
 145
 146         # Do we have our tables?
 147         shares_exist = False
 148         self.dbc.execute("select COUNT(*) from INFORMATION_SCHEMA.STATISTICS " +\
 149                 "where table_schema = %(schema)s and table_name = 'shares' and index_name = 'shares_username'",
 150                 {"schema": settings.DB_MYSQL_DBNAME })
 151         data = self.dbc.fetchone()
 152         if data[0] <= 0 :
 153             self.update_version_1()     # no, we don't, so create them
 154
 155         if settings.DATABASE_EXTEND == True :
 156             self.update_tables()
 157
 158     def update_tables(self):
 159         version = 0
 160         current_version = 6
 161         while version < current_version :
 162             self.dbc.execute("select value from pool where parameter = 'DB Version'")
 163             data = self.dbc.fetchone()
 164             version = int(data[0])
 165             if version < current_version :
 166                 log.info("Updating Database from %i to %i" % (version, version +1))
 167                 getattr(self, 'update_version_' + str(version) )()
 168
 169     def update_version_1(self):
 170         if settings.DATABASE_EXTEND == True :
 171             self.dbc.execute("create table if not exists shares " +\
 172                 "(id serial primary key,time timestamp,rem_host TEXT, username TEXT, our_result BOOLEAN, upstream_result BOOLEAN, reason TEXT, solution TEXT, " +\
 173                 "block_num INTEGER, prev_block_hash TEXT, useragent TEXT, difficulty INTEGER) ENGINE = MYISAM;")
 174             self.dbc.execute("create index shares_username ON shares(username(10))")
 175
 176             self.dbc.execute("create table if not exists pool_worker" +\
 177                 "(id serial primary key,username TEXT, password TEXT, speed INTEGER, last_checkin timestamp" +\
 178                 ") ENGINE = MYISAM")
 179             self.dbc.execute("create index pool_worker_username ON pool_worker(username(10))")
 180
 181             self.dbc.execute("create table if not exists pool(parameter TEXT, value TEXT)")
 182             self.dbc.execute("alter table pool_worker add total_shares INTEGER default 0")
 183             self.dbc.execute("alter table pool_worker add total_rejects INTEGER default 0")
 184             self.dbc.execute("alter table pool_worker add total_found INTEGER default 0")
 185             self.dbc.execute("insert into pool (parameter,value) VALUES ('DB Version',2)")
 186
 187         else :
 188             self.dbc.execute("create table if not exists shares" + \
 189                 "(id serial,time timestamp,rem_host TEXT, username TEXT, our_result INTEGER, upstream_result INTEGER, reason TEXT, solution TEXT) ENGINE = MYISAM")
 190             self.dbc.execute("create index shares_username ON shares(username(10))")
 191             self.dbc.execute("create table if not exists pool_worker(id serial,username TEXT, password TEXT) ENGINE = MYISAM")
 192             self.dbc.execute("create index pool_worker_username ON pool_worker(username(10))")
 193         self.dbh.commit()
 194
 195
 196     def update_version_2(self):
 197         log.info("running update 2")
 198         self.dbc.executemany("insert into pool (parameter,value) VALUES (%s,%s)",[('bitcoin_blocks',0),
 199                 ('bitcoin_balance',0),
 200                 ('bitcoin_connections',0),
 201                 ('bitcoin_difficulty',0),
 202                 ('pool_speed',0),
 203                 ('pool_total_found',0),
 204                 ('round_shares',0),
 205                 ('round_progress',0),
 206                 ('round_start',time.time())
 207                 ])
 208         self.dbc.execute("update pool set value = 3 where parameter = 'DB Version'")
 209         self.dbh.commit()
 210
 211     def update_version_3(self):
 212         log.info("running update 3")
 213         self.dbc.executemany("insert into pool (parameter,value) VALUES (%s,%s)",[
 214                 ('round_best_share',0),
 215                 ('bitcoin_infotime',0)
 216                 ])
 217         self.dbc.execute("alter table pool_worker add alive BOOLEAN")
 218         self.dbc.execute("update pool set value = 4 where parameter = 'DB Version'")
 219         self.dbh.commit()
 220
 221     def update_version_4(self):
 222         log.info("running update 4")
 223         self.dbc.execute("alter table pool_worker add difficulty INTEGER default 0")
 224         self.dbc.execute("create table if not exists shares_archive " +\
 225                 "(id serial primary key,time timestamp,rem_host TEXT, username TEXT, our_result BOOLEAN, upstream_result BOOLEAN, reason TEXT, solution TEXT, " +\
 226                 "block_num INTEGER, prev_block_hash TEXT, useragent TEXT, difficulty INTEGER) ENGINE = MYISAM;")
 227         self.dbc.execute("create table if not exists shares_archive_found " +\
 228                 "(id serial primary key,time timestamp,rem_host TEXT, username TEXT, our_result BOOLEAN, upstream_result BOOLEAN, reason TEXT, solution TEXT, " +\
 229                 "block_num INTEGER, prev_block_hash TEXT, useragent TEXT, difficulty INTEGER) ENGINE = MYISAM;")
 230         self.dbc.execute("update pool set value = 5 where parameter = 'DB Version'")
 231         self.dbh.commit()
 232
 233     def update_version_5(self):
 234         log.info("running update 5")
 235         # Adding Primary key to table: pool
 236         self.dbc.execute("alter table pool add primary key (parameter(100))")
 237         self.dbh.commit()
 238         # Adjusting indicies on table: shares
 239         self.dbc.execute("DROP INDEX shares_username ON shares")
 240         self.dbc.execute("CREATE INDEX shares_time_username ON shares(time,username(10))")
 241         self.dbc.execute("CREATE INDEX shares_upstreamresult ON shares(upstream_result)")
 242         self.dbh.commit()
 243
 244         self.dbc.execute("update pool set value = 6 where parameter = 'DB Version'")
 245         self.dbh.commit()
 246