From 2a1b6dd86d92e1c7272e4b5713996d4daac13ca4 Mon Sep 17 00:00:00 2001 From: Jonathan Warren Date: Wed, 13 Nov 2013 22:44:42 -0500 Subject: [PATCH] some initial objectProcessorThread work --- src/class_receiveDataThread.py | 42 +++++++++++++--------------------- src/shared.py | 17 ++++++++++++++ 2 files changed, 33 insertions(+), 26 deletions(-) diff --git a/src/class_receiveDataThread.py b/src/class_receiveDataThread.py index b1a62261..01d6428a 100644 --- a/src/class_receiveDataThread.py +++ b/src/class_receiveDataThread.py @@ -158,7 +158,7 @@ class receiveDataThread(threading.Thread): elif remoteCommand == 'getdata\x00\x00\x00\x00\x00' and self.connectionIsOrWasFullyEstablished: self.recgetdata(self.data[24:self.payloadLength + 24]) elif remoteCommand == 'msg\x00\x00\x00\x00\x00\x00\x00\x00\x00' and self.connectionIsOrWasFullyEstablished: - self.recmsg(self.data[24:self.payloadLength + 24]) + self.recmsg(self.data[:self.payloadLength + 24]) elif remoteCommand == 'broadcast\x00\x00\x00' and self.connectionIsOrWasFullyEstablished: self.recbroadcast(self.data[24:self.payloadLength + 24]) elif remoteCommand == 'ping\x00\x00\x00\x00\x00\x00\x00\x00' and self.connectionIsOrWasFullyEstablished: @@ -221,19 +221,6 @@ class receiveDataThread(threading.Thread): self.data = self.ackDataThatWeHaveYetToSend.pop() self.processData() - def isProofOfWorkSufficient( - self, - data, - nonceTrialsPerByte=0, - payloadLengthExtraBytes=0): - if nonceTrialsPerByte < shared.networkDefaultProofOfWorkNonceTrialsPerByte: - nonceTrialsPerByte = shared.networkDefaultProofOfWorkNonceTrialsPerByte - if payloadLengthExtraBytes < shared.networkDefaultPayloadLengthExtraBytes: - payloadLengthExtraBytes = shared.networkDefaultPayloadLengthExtraBytes - POW, = unpack('>Q', hashlib.sha512(hashlib.sha512(data[ - :8] + hashlib.sha512(data[8:]).digest()).digest()).digest()[0:8]) - # print 'POW:', POW - return POW <= 2 ** 64 / ((len(data) + payloadLengthExtraBytes) * (nonceTrialsPerByte)) def sendpong(self): print 'Sending pong' @@ -342,7 +329,7 @@ class receiveDataThread(threading.Thread): def recbroadcast(self, data): self.messageProcessingStartTime = time.time() # First we must check to make sure the proof of work is sufficient. - if not self.isProofOfWorkSufficient(data): + if not shared.isProofOfWorkSufficient(data): print 'Proof of work in broadcast message insufficient.' return readPosition = 8 # bypass the nonce @@ -858,13 +845,14 @@ class receiveDataThread(threading.Thread): # We have received a msg message. def recmsg(self, data): + readPosition = 24 # bypass the network header self.messageProcessingStartTime = time.time() # First we must check to make sure the proof of work is sufficient. - if not self.isProofOfWorkSufficient(data): + if not shared.isProofOfWorkSufficient(data[readPosition:readPosition+10]): print 'Proof of work in msg message insufficient.' return - readPosition = 8 + readPosition += 8 # bypass the POW nonce embeddedTime, = unpack('>I', data[readPosition:readPosition + 4]) # This section is used for the transition from 32 bit time to 64 bit @@ -905,12 +893,15 @@ class receiveDataThread(threading.Thread): shared.inventorySets[self.streamNumber].add(self.inventoryHash) shared.inventoryLock.release() self.broadcastinv(self.inventoryHash) - shared.numberOfMessagesProcessed += 1 - shared.UISignalQueue.put(( - 'updateNumberOfMessagesProcessed', 'no data')) + #shared.numberOfMessagesProcessed += 1 + #shared.UISignalQueue.put(( + # 'updateNumberOfMessagesProcessed', 'no data')) - self.processmsg( - readPosition, data) # When this function returns, we will have either successfully processed the message bound for us, ignored it because it isn't bound for us, or found problem with the message that warranted ignoring it. + #self.processmsg( + # readPosition, data) # When this function returns, we will have either successfully processed the message bound for us, ignored it because it isn't bound for us, or found problem with the message that warranted ignoring it. + shared.objectProcessorQueue.put(data) + with shard.printLock: + print 'Size of objectProcessorQueue is:', sys.getsizeof(shared.objectProcessorQueue) # Let us now set lengthOfTimeWeShouldUseToProcessThisMessage. If we # haven't used the specified amount of time, we shall sleep. These @@ -1104,7 +1095,7 @@ class receiveDataThread(threading.Thread): toAddress, 'noncetrialsperbyte') requiredPayloadLengthExtraBytes = shared.config.getint( toAddress, 'payloadlengthextrabytes') - if not self.isProofOfWorkSufficient(encryptedData, requiredNonceTrialsPerByte, requiredPayloadLengthExtraBytes): + if not shared.isProofOfWorkSufficient(encryptedData, requiredNonceTrialsPerByte, requiredPayloadLengthExtraBytes): print 'Proof of work in msg message insufficient only because it does not meet our higher requirement.' return blockMessage = False # Gets set to True if the user shouldn't see the message according to black or white lists. @@ -1283,7 +1274,7 @@ class receiveDataThread(threading.Thread): if len(data) < 146 or len(data) > 420: # sanity check return # We must check to make sure the proof of work is sufficient. - if not self.isProofOfWorkSufficient(data): + if not shared.isProofOfWorkSufficient(data): print 'Proof of work in pubkey message insufficient.' return @@ -1567,10 +1558,9 @@ class receiveDataThread(threading.Thread): # the messages that require it. self.possibleNewPubkey(address = fromAddress) - # We have received a getpubkey message def recgetpubkey(self, data): - if not self.isProofOfWorkSufficient(data): + if not shared.isProofOfWorkSufficient(data): print 'Proof of work in getpubkey message insufficient.' return if len(data) < 34: diff --git a/src/shared.py b/src/shared.py index 2404ff15..21d1d3ac 100644 --- a/src/shared.py +++ b/src/shared.py @@ -72,6 +72,9 @@ daemon = False inventorySets = {} # key = streamNumer, value = a set which holds the inventory object hashes that we are aware of. This is used whenever we receive an inv message from a peer to check to see what items are new to us. We don't delete things out of it; instead, the singleCleaner thread clears and refills it every couple hours. needToWriteKnownNodesToDisk = False # If True, the singleCleaner will write it to disk eventually. maximumLengthOfTimeToBotherResendingMessages = 0 +objectProcessorQueue = Queue.Queue( + ) # receiveDataThreads dump objects they hear on the network into this queue to be processed. +streamsInWhichIAmParticipating = {} #If changed, these values will cause particularly unexpected behavior: You won't be able to either send or receive messages because the proof of work you do (or demand) won't match that done or demanded by others. Don't change them! networkDefaultProofOfWorkNonceTrialsPerByte = 320 #The amount of work that should be performed (and demanded) per byte of the payload. Double this number to double the work. @@ -277,6 +280,20 @@ def reloadBroadcastSendersForWhichImWatching(): privEncryptionKey = doubleHashOfAddressData[:32] MyECSubscriptionCryptorObjects[tag] = highlevelcrypto.makeCryptor(privEncryptionKey.encode('hex')) +def isProofOfWorkSufficient( + self, + data, + nonceTrialsPerByte=0, + payloadLengthExtraBytes=0): + if nonceTrialsPerByte < shared.networkDefaultProofOfWorkNonceTrialsPerByte: + nonceTrialsPerByte = shared.networkDefaultProofOfWorkNonceTrialsPerByte + if payloadLengthExtraBytes < shared.networkDefaultPayloadLengthExtraBytes: + payloadLengthExtraBytes = shared.networkDefaultPayloadLengthExtraBytes + POW, = unpack('>Q', hashlib.sha512(hashlib.sha512(data[ + :8] + hashlib.sha512(data[8:]).digest()).digest()).digest()[0:8]) + # print 'POW:', POW + return POW <= 2 ** 64 / ((len(data) + payloadLengthExtraBytes) * (nonceTrialsPerByte)) + def doCleanShutdown(): global shutdown shutdown = 1 #Used to tell proof of work worker threads to exit.