cis-kuldeep/PyBitmessage-2021-04-27
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

use helper_sql in class_singleWorker

Browse Source
This commit is contained in:
Grant T. Olson 2013-08-29 07:27:30 -04:00
parent d879e35e26
commit 1fb11495a6
1 changed files with 51 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

186
src/class_singleWorker.py
View File

@ -10,6 +10,7 @@ import sys
from class_addressGenerator import pointMult from class_addressGenerator import pointMult
import tr import tr
from debug import logger from debug import logger
from helper_sql import *
# This thread, of which there is only one, does the heavy lifting: # This thread, of which there is only one, does the heavy lifting:
# calculating POWs. # calculating POWs.
@ -22,35 +23,23 @@ class singleWorker(threading.Thread):
threading.Thread.__init__(self) threading.Thread.__init__(self)
def run(self): def run(self):
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put(
'''SELECT toripe FROM sent WHERE ((status='awaitingpubkey' OR status='doingpubkeypow') AND folder='sent')''') '''SELECT toripe FROM sent WHERE ((status='awaitingpubkey' OR status='doingpubkeypow') AND folder='sent')''')
shared.sqlSubmitQueue.put('')
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: for row in queryreturn:
toripe, = row toripe, = row
shared.neededPubkeys[toripe] = 0 shared.neededPubkeys[toripe] = 0
# Initialize the shared.ackdataForWhichImWatching data structure using data # Initialize the shared.ackdataForWhichImWatching data structure using data
# from the sql database. # from the sql database.
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put(
'''SELECT ackdata FROM sent where (status='msgsent' OR status='doingmsgpow')''') '''SELECT ackdata FROM sent where (status='msgsent' OR status='doingmsgpow')''')
shared.sqlSubmitQueue.put('')
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: for row in queryreturn:
ackdata, = row ackdata, = row
print 'Watching for ackdata', ackdata.encode('hex') print 'Watching for ackdata', ackdata.encode('hex')
shared.ackdataForWhichImWatching[ackdata] = 0 shared.ackdataForWhichImWatching[ackdata] = 0
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put(
'''SELECT DISTINCT toaddress FROM sent WHERE (status='doingpubkeypow' AND folder='sent')''') '''SELECT DISTINCT toaddress FROM sent WHERE (status='doingpubkeypow' AND folder='sent')''')
shared.sqlSubmitQueue.put('')
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: for row in queryreturn:
toaddress, = row toaddress, = row
self.requestPubKey(toaddress) self.requestPubKey(toaddress)
@ -248,26 +237,19 @@ class singleWorker(threading.Thread):
if shared.safeConfigGetBoolean(myAddress, 'chan'): if shared.safeConfigGetBoolean(myAddress, 'chan'):
payload = '\x00' * 8 + payload # Attach a fake nonce on the front payload = '\x00' * 8 + payload # Attach a fake nonce on the front
# just so that it is in the correct format. # just so that it is in the correct format.
t = (hash,payload,embeddedTime,'yes') sqlExecute('''INSERT INTO pubkeys VALUES (?,?,?,?)''',
shared.sqlLock.acquire() hash,
shared.sqlSubmitQueue.put('''INSERT INTO pubkeys VALUES (?,?,?,?)''') payload,
shared.sqlSubmitQueue.put(t) embeddedTime,
shared.sqlReturnQueue.get() 'yes')
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.config.set( shared.config.set(
myAddress, 'lastpubkeysendtime', str(int(time.time()))) myAddress, 'lastpubkeysendtime', str(int(time.time())))
with open(shared.appdata + 'keys.dat', 'wb') as configfile: with open(shared.appdata + 'keys.dat', 'wb') as configfile:
shared.config.write(configfile) shared.config.write(configfile)
def sendBroadcast(self): def sendBroadcast(self):
shared.sqlLock.acquire() queryreturn = sqlQuery(
t = ('broadcastqueued',) '''SELECT fromaddress, subject, message, ackdata FROM sent WHERE status=? and folder='sent' ''', 'broadcastqueued')
shared.sqlSubmitQueue.put(
'''SELECT fromaddress, subject, message, ackdata FROM sent WHERE status=? and folder='sent' ''')
shared.sqlSubmitQueue.put(t)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: for row in queryreturn:
fromaddress, subject, body, ackdata = row fromaddress, subject, body, ackdata = row
status, addressVersionNumber, streamNumber, ripe = decodeAddress( status, addressVersionNumber, streamNumber, ripe = decodeAddress(
@ -355,79 +337,49 @@ class singleWorker(threading.Thread):
# Update the status of the message in the 'sent' table to have # Update the status of the message in the 'sent' table to have
# a 'broadcastsent' status # a 'broadcastsent' status
shared.sqlLock.acquire() sqlExecute(
t = (inventoryHash,'broadcastsent', int( 'UPDATE sent SET msgid=?, status=?, lastactiontime=? WHERE ackdata=?',
time.time()), ackdata) inventoryHash,
shared.sqlSubmitQueue.put( 'broadcastsent',
'UPDATE sent SET msgid=?, status=?, lastactiontime=? WHERE ackdata=?') int(time.time()),
shared.sqlSubmitQueue.put(t) ackdata)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
def sendMsg(self): def sendMsg(self):
# Check to see if there are any messages queued to be sent # Check to see if there are any messages queued to be sent
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put(
'''SELECT DISTINCT toaddress FROM sent WHERE (status='msgqueued' AND folder='sent')''') '''SELECT DISTINCT toaddress FROM sent WHERE (status='msgqueued' AND folder='sent')''')
shared.sqlSubmitQueue.put('')
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: # For each address to which we need to send a message, check to see if we have its pubkey already. for row in queryreturn: # For each address to which we need to send a message, check to see if we have its pubkey already.
toaddress, = row toaddress, = row
toripe = decodeAddress(toaddress)[3] toripe = decodeAddress(toaddress)[3]
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put( '''SELECT hash FROM pubkeys WHERE hash=? ''', toripe)
'''SELECT hash FROM pubkeys WHERE hash=? ''')
shared.sqlSubmitQueue.put((toripe,))
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
if queryreturn != []: # If we have the needed pubkey, set the status to doingmsgpow (we'll do it further down) if queryreturn != []: # If we have the needed pubkey, set the status to doingmsgpow (we'll do it further down)
t = (toaddress,) sqlExecute(
shared.sqlLock.acquire() '''UPDATE sent SET status='doingmsgpow' WHERE toaddress=? AND status='msgqueued' ''',
shared.sqlSubmitQueue.put( toaddress)
'''UPDATE sent SET status='doingmsgpow' WHERE toaddress=? AND status='msgqueued' ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
else: # We don't have the needed pubkey. Set the status to 'awaitingpubkey' and request it if we haven't already else: # We don't have the needed pubkey. Set the status to 'awaitingpubkey' and request it if we haven't already
if toripe in shared.neededPubkeys: if toripe in shared.neededPubkeys:
# We already sent a request for the pubkey # We already sent a request for the pubkey
t = (toaddress,) sqlExecute(
shared.sqlLock.acquire() '''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='msgqueued' ''', toaddress)
shared.sqlSubmitQueue.put(
'''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='msgqueued' ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(('updateSentItemStatusByHash', ( shared.UISignalQueue.put(('updateSentItemStatusByHash', (
toripe, tr.translateText("MainWindow",'Encryption key was requested earlier.')))) toripe, tr.translateText("MainWindow",'Encryption key was requested earlier.'))))
else: else:
# We have not yet sent a request for the pubkey # We have not yet sent a request for the pubkey
t = (toaddress,) sqlExecute(
shared.sqlLock.acquire() '''UPDATE sent SET status='doingpubkeypow' WHERE toaddress=? AND status='msgqueued' ''',
shared.sqlSubmitQueue.put( toaddress)
'''UPDATE sent SET status='doingpubkeypow' WHERE toaddress=? AND status='msgqueued' ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(('updateSentItemStatusByHash', ( shared.UISignalQueue.put(('updateSentItemStatusByHash', (
toripe, tr.translateText("MainWindow",'Sending a request for the recipient\'s encryption key.')))) toripe, tr.translateText("MainWindow",'Sending a request for the recipient\'s encryption key.'))))
self.requestPubKey(toaddress) self.requestPubKey(toaddress)
shared.sqlLock.acquire()
# Get all messages that are ready to be sent, and also all messages # Get all messages that are ready to be sent, and also all messages
# which we have sent in the last 28 days which were previously marked # which we have sent in the last 28 days which were previously marked
# as 'toodifficult'. If the user as raised the maximum acceptable # as 'toodifficult'. If the user as raised the maximum acceptable
# difficulty then those messages may now be sendable. # difficulty then those messages may now be sendable.
shared.sqlSubmitQueue.put( queryreturn = sqlQuery(
'''SELECT toaddress, toripe, fromaddress, subject, message, ackdata, status FROM sent WHERE (status='doingmsgpow' or status='forcepow' or (status='toodifficult' and lastactiontime>?)) and folder='sent' ''') '''SELECT toaddress, toripe, fromaddress, subject, message, ackdata, status FROM sent WHERE (status='doingmsgpow' or status='forcepow' or (status='toodifficult' and lastactiontime>?)) and folder='sent' ''',
shared.sqlSubmitQueue.put((int(time.time()) - 2419200,)) int(time.time()) - 2419200)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
for row in queryreturn: # For each message we need to send.. for row in queryreturn: # For each message we need to send..
toaddress, toripe, fromaddress, subject, message, ackdata, status = row toaddress, toripe, fromaddress, subject, message, ackdata, status = row
# There is a remote possibility that we may no longer have the # There is a remote possibility that we may no longer have the
@ -436,12 +388,9 @@ class singleWorker(threading.Thread):
# user sends a message but doesn't let the POW function finish, # user sends a message but doesn't let the POW function finish,
# then leaves their client off for a long time which could cause # then leaves their client off for a long time which could cause
# the needed pubkey to expire and be deleted. # the needed pubkey to expire and be deleted.
shared.sqlLock.acquire() queryreturn = sqlQuery(
shared.sqlSubmitQueue.put( '''SELECT hash FROM pubkeys WHERE hash=? ''',
'''SELECT hash FROM pubkeys WHERE hash=? ''') toripe)
shared.sqlSubmitQueue.put((toripe,))
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
if queryreturn == [] and toripe not in shared.neededPubkeys: if queryreturn == [] and toripe not in shared.neededPubkeys:
# We no longer have the needed pubkey and we haven't requested # We no longer have the needed pubkey and we haven't requested
# it. # it.
@ -449,14 +398,8 @@ class singleWorker(threading.Thread):
sys.stderr.write( sys.stderr.write(
'For some reason, the status of a message in our outbox is \'doingmsgpow\' even though we lack the pubkey. Here is the RIPE hash of the needed pubkey: %s\n' % toripe.encode('hex')) 'For some reason, the status of a message in our outbox is \'doingmsgpow\' even though we lack the pubkey. Here is the RIPE hash of the needed pubkey: %s\n' % toripe.encode('hex'))
t = (toaddress,) sqlExecute(
shared.sqlLock.acquire() '''UPDATE sent SET status='msgqueued' WHERE toaddress=? AND status='doingmsgpow' ''', toaddress)
shared.sqlSubmitQueue.put(
'''UPDATE sent SET status='msgqueued' WHERE toaddress=? AND status='doingmsgpow' ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(('updateSentItemStatusByHash', ( shared.UISignalQueue.put(('updateSentItemStatusByHash', (
toripe, tr.translateText("MainWindow",'Sending a request for the recipient\'s encryption key.')))) toripe, tr.translateText("MainWindow",'Sending a request for the recipient\'s encryption key.'))))
self.requestPubKey(toaddress) self.requestPubKey(toaddress)
@ -475,21 +418,15 @@ class singleWorker(threading.Thread):
# mark the pubkey as 'usedpersonally' so that we don't ever delete # mark the pubkey as 'usedpersonally' so that we don't ever delete
# it. # it.
shared.sqlLock.acquire() sqlExecute(
t = (toripe,) '''UPDATE pubkeys SET usedpersonally='yes' WHERE hash=?''',
shared.sqlSubmitQueue.put( toripe)
'''UPDATE pubkeys SET usedpersonally='yes' WHERE hash=?''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
# Let us fetch the recipient's public key out of our database. If # Let us fetch the recipient's public key out of our database. If
# the required proof of work difficulty is too hard then we'll # the required proof of work difficulty is too hard then we'll
# abort. # abort.
shared.sqlSubmitQueue.put( queryreturn = sqlQuery(
'SELECT transmitdata FROM pubkeys WHERE hash=?') 'SELECT transmitdata FROM pubkeys WHERE hash=?',
shared.sqlSubmitQueue.put((toripe,)) toripe)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlLock.release()
if queryreturn == []: if queryreturn == []:
with shared.printLock: with shared.printLock:
sys.stderr.write( sys.stderr.write(
@ -559,14 +496,9 @@ class singleWorker(threading.Thread):
if (requiredAverageProofOfWorkNonceTrialsPerByte > shared.config.getint('bitmessagesettings', 'maxacceptablenoncetrialsperbyte') and shared.config.getint('bitmessagesettings', 'maxacceptablenoncetrialsperbyte') != 0) or (requiredPayloadLengthExtraBytes > shared.config.getint('bitmessagesettings', 'maxacceptablepayloadlengthextrabytes') and shared.config.getint('bitmessagesettings', 'maxacceptablepayloadlengthextrabytes') != 0): if (requiredAverageProofOfWorkNonceTrialsPerByte > shared.config.getint('bitmessagesettings', 'maxacceptablenoncetrialsperbyte') and shared.config.getint('bitmessagesettings', 'maxacceptablenoncetrialsperbyte') != 0) or (requiredPayloadLengthExtraBytes > shared.config.getint('bitmessagesettings', 'maxacceptablepayloadlengthextrabytes') and shared.config.getint('bitmessagesettings', 'maxacceptablepayloadlengthextrabytes') != 0):
# The demanded difficulty is more than we are willing # The demanded difficulty is more than we are willing
# to do. # to do.
shared.sqlLock.acquire() sqlExecute(
t = (ackdata,) '''UPDATE sent SET status='toodifficult' WHERE ackdata=? ''',
shared.sqlSubmitQueue.put( ackdata)
'''UPDATE sent SET status='toodifficult' WHERE ackdata=? ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(('updateSentItemStatusByAckdata', (ackdata, tr.translateText("MainWindow", "Problem: The work demanded by the recipient (%1 and %2) is more difficult than you are willing to do.").arg(str(float(requiredAverageProofOfWorkNonceTrialsPerByte) / shared.networkDefaultProofOfWorkNonceTrialsPerByte)).arg(str(float( shared.UISignalQueue.put(('updateSentItemStatusByAckdata', (ackdata, tr.translateText("MainWindow", "Problem: The work demanded by the recipient (%1 and %2) is more difficult than you are willing to do.").arg(str(float(requiredAverageProofOfWorkNonceTrialsPerByte) / shared.networkDefaultProofOfWorkNonceTrialsPerByte)).arg(str(float(
requiredPayloadLengthExtraBytes) / shared.networkDefaultPayloadLengthExtraBytes)).arg(unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'), localtime(int(time.time()))), 'utf-8'))))) requiredPayloadLengthExtraBytes) / shared.networkDefaultPayloadLengthExtraBytes)).arg(unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'), localtime(int(time.time()))), 'utf-8')))))
continue continue
@ -694,13 +626,7 @@ class singleWorker(threading.Thread):
try: try:
encrypted = highlevelcrypto.encrypt(payload,"04"+pubEncryptionKeyBase256.encode('hex')) encrypted = highlevelcrypto.encrypt(payload,"04"+pubEncryptionKeyBase256.encode('hex'))
except: except:
shared.sqlLock.acquire() sqlExecute('''UPDATE sent SET status='badkey' WHERE ackdata=?''', ackdata)
t = (ackdata,)
shared.sqlSubmitQueue.put('''UPDATE sent SET status='badkey' WHERE ackdata=?''')
shared.sqlSubmitQueue.put(t)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(ackdata,tr.translateText("MainWindow",'Problem: The recipient\'s encryption key is no good. Could not encrypt message. %1').arg(unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8'))))) shared.UISignalQueue.put(('updateSentItemStatusByAckdata',(ackdata,tr.translateText("MainWindow",'Problem: The recipient\'s encryption key is no good. Could not encrypt message. %1').arg(unicode(strftime(shared.config.get('bitmessagesettings', 'timeformat'),localtime(int(time.time()))),'utf-8')))))
continue continue
encryptedPayload = embeddedTime + encodeVarint(toStreamNumber) + encrypted encryptedPayload = embeddedTime + encodeVarint(toStreamNumber) + encrypted
@ -741,13 +667,8 @@ class singleWorker(threading.Thread):
newStatus = 'msgsentnoackexpected' newStatus = 'msgsentnoackexpected'
else: else:
newStatus = 'msgsent' newStatus = 'msgsent'
shared.sqlLock.acquire() sqlExecute('''UPDATE sent SET msgid=?, status=? WHERE ackdata=?''',
t = (inventoryHash,newStatus,ackdata,) inventoryHash,newStatus,ackdata)
shared.sqlSubmitQueue.put('''UPDATE sent SET msgid=?, status=? WHERE ackdata=?''')
shared.sqlSubmitQueue.put(t)
queryreturn = shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
def requestPubKey(self, toAddress): def requestPubKey(self, toAddress):
toStatus, addressVersionNumber, streamNumber, ripe = decodeAddress( toStatus, addressVersionNumber, streamNumber, ripe = decodeAddress(
@ -789,14 +710,9 @@ class singleWorker(threading.Thread):
shared.broadcastToSendDataQueues(( shared.broadcastToSendDataQueues((
streamNumber, 'sendinv', inventoryHash)) streamNumber, 'sendinv', inventoryHash))
t = (toAddress,) sqlExecute(
shared.sqlLock.acquire() '''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='doingpubkeypow' ''',
shared.sqlSubmitQueue.put( toAddress)
'''UPDATE sent SET status='awaitingpubkey' WHERE toaddress=? AND status='doingpubkeypow' ''')
shared.sqlSubmitQueue.put(t)
shared.sqlReturnQueue.get()
shared.sqlSubmitQueue.put('commit')
shared.sqlLock.release()
shared.UISignalQueue.put(( shared.UISignalQueue.put((
'updateStatusBar', tr.translateText("MainWindow",'Broacasting the public key request. This program will auto-retry if they are offline.'))) 'updateStatusBar', tr.translateText("MainWindow",'Broacasting the public key request. This program will auto-retry if they are offline.')))