Files
@ 13603ac261f1
Branch filter:
Location: Morevna/src/client.py - annotation
13603ac261f1
5.9 KiB
text/x-python
minor enhancements
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | ee936b917440 34f4027c1bd6 cd2ba192bf12 34f4027c1bd6 b052f27e1cbc 34f4027c1bd6 34f4027c1bd6 ccbe369ce439 41ea9614ce8c 4b88aca70fbc 8bb6a904d50b ccbe369ce439 bb3b53ee15d6 34f4027c1bd6 34f4027c1bd6 362cff560740 362cff560740 362cff560740 75e070b6b447 9f2b0a4f3538 75e070b6b447 cd2ba192bf12 cd2ba192bf12 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 cd2ba192bf12 02ea4fed2520 7e101f53704e 9f2b0a4f3538 68becf4f98c2 02ea4fed2520 02ea4fed2520 02ea4fed2520 02ea4fed2520 02ea4fed2520 02ea4fed2520 02ea4fed2520 34f4027c1bd6 75e070b6b447 02ea4fed2520 34f4027c1bd6 34f4027c1bd6 75e070b6b447 8bb6a904d50b 6c8e994fd906 75e070b6b447 cd2ba192bf12 362cff560740 362cff560740 362cff560740 362cff560740 362cff560740 ccbe369ce439 ccbe369ce439 362cff560740 362cff560740 ccbe369ce439 ccbe369ce439 362cff560740 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 b73a5d69a11b 8bb6a904d50b b73a5d69a11b b73a5d69a11b 34f4027c1bd6 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 5813971dbecc 0b67ddd4455a 5813971dbecc 5813971dbecc 5813971dbecc 34f4027c1bd6 6c8e994fd906 5813971dbecc 6c8e994fd906 5813971dbecc 34f4027c1bd6 2e5828ec7d49 5813971dbecc 5813971dbecc 5813971dbecc ee936b917440 ee936b917440 2e5828ec7d49 2e5828ec7d49 5813971dbecc 5813971dbecc 5813971dbecc 2e5828ec7d49 4b88aca70fbc 34f4027c1bd6 095908159393 095908159393 095908159393 2e5828ec7d49 34f4027c1bd6 b73a5d69a11b b73a5d69a11b a4bfaabaeabb 34f4027c1bd6 b052f27e1cbc 6c8e994fd906 34f4027c1bd6 0b67ddd4455a 87a9ced6e7b5 87a9ced6e7b5 0b67ddd4455a 87a9ced6e7b5 a4bfaabaeabb a4bfaabaeabb 34f4027c1bd6 a4bfaabaeabb 87a9ced6e7b5 a4bfaabaeabb 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 4b88aca70fbc 34f4027c1bd6 362cff560740 34f4027c1bd6 dad65188b1a0 b73a5d69a11b 3d0876534e40 0ad71a952f92 0ad71a952f92 0ad71a952f92 0ad71a952f92 0ad71a952f92 0ad71a952f92 0ad71a952f92 3d0876534e40 bb3b53ee15d6 3d0876534e40 3d0876534e40 6c8e994fd906 3d0876534e40 0b67ddd4455a 0b67ddd4455a bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 8bb6a904d50b bb3b53ee15d6 bb3b53ee15d6 3d0876534e40 bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 bb3b53ee15d6 3d0876534e40 3d0876534e40 6c8e994fd906 3d0876534e40 3d0876534e40 3d0876534e40 362cff560740 6c8e994fd906 8bb6a904d50b 75e070b6b447 8bb6a904d50b 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 87a9ced6e7b5 6c8e994fd906 6c8e994fd906 | import collections
import socket
import ssl
import logging as log
from datetime import datetime
import config as conf
import status
import stats
from util import Progress
from hashtree import HashTree,hashBlock
from netnode import BaseConnection,NetNode,FailedConnection,LockedException,IncompatibleException
from datafile import DataFile
class DeniedConnection(Exception): pass
class Connection(BaseConnection):
def __init__(self,host,port):
super().__init__()
sock=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sslContext=ssl.create_default_context(cafile=conf.peers)
sslContext.check_hostname=False
sslContext.load_cert_chain(conf.certfile,conf.keyfile)
self._socket=sslContext.wrap_socket(sock)
try:
self._socket.connect((host,port))
except (ConnectionRefusedError,OSError) as e:
log.exception(e)
print("Couldn't connect to {0}:{1}".format(host,port))
raise FailedConnection()
except ssl.SSLError as e:
log.exception(e)
print("Error creating SSL connection to {0}:{1}".format(host,port))
raise FailedConnection()
self.createNetworkers()
print("Connected to {0}".format(host))
class Client(NetNode):
def __init__(self,filename,treeFile=""):
print(datetime.now(), "initializing...")
super().__init__(filename,treeFile)
def init(self,action):
jsonData={"command":"init", "blockSize":self._tree.BLOCK_SIZE, "blockCount":self._tree.leafCount, "version":conf.version, "action":action}
self._outcoming.writeMsg(jsonData)
jsonData,binData=self._incoming.readMsg()
if jsonData["command"]=="deny":
if jsonData["status"]==status.incompatible.version:
raise DeniedConnection("Incompatible client version. Consider upgrading it.")
raise DeniedConnection()
assert jsonData["command"]=="init"
if jsonData["version"]<conf.lowestCompatible:
raise IncompatibleException("Incompatible server version. Consider upgrading it.")
## Asks server for node hashes to determine which are to be transferred.
#
# Uses a binary HashTree, where item at k is hash of items at 2k+1, 2k+2.
#
# Requests nodes in order of a batch DFS. Needs stack of size O(treeDepth*batchSize). Nodes in each tree level are accessed in order.
def negotiate(self):
localTree=self._tree
blocksToTransfer=[]
nodeStack=collections.deque([0]) # root
# determine which blocks to send
print(datetime.now(), "negotiating:")
progress=Progress(localTree.leafCount)
while len(nodeStack)>0:
indices=[]
for i in range(conf.batchSize.hash):
indices.append(nodeStack.pop())
if len(nodeStack)==0: break
self._outcoming.writeMsg({"command":"req", "index":indices, "dataType":"hash"})
jsonData,binData=self._incoming.readMsg()
assert jsonData["index"]==indices
assert jsonData["dataType"]=="hash"
stats.logExchangedNode(len(indices))
frontier=[]
for (j,i) in enumerate(indices):
(j1,j2)=[HashTree.HASH_LEN*ji for ji in (j,j+1)]
if localTree.store[i]!=binData[j1:j2]:
# ie. 0-6 nodes, 7-14 leaves. 2*6+2<15
if 2*i+2<len(localTree.store): # inner node
frontier.append(2*i+1)
frontier.append(2*i+2)
else:
blocksToTransfer.append(i-localTree.leafStart) # leaf
progress.p(i-localTree.leafStart)
nodeStack.extend(reversed(frontier))
progress.done()
size=stats.formatBytes(len(blocksToTransfer)*self._tree.BLOCK_SIZE)
print(datetime.now(), "{0} to transfer".format(size))
return blocksToTransfer
def sendData(self,blocksToTransfer):
log.info(blocksToTransfer)
dataFile=DataFile.open(self._filename)
print(datetime.now(), "sending data:")
progress=Progress(len(blocksToTransfer))
for k in range(0,len(blocksToTransfer),conf.batchSize.data):
indices=[]
blocks=[]
for j in range(conf.batchSize.data):
if k+j>=len(blocksToTransfer): break
i=blocksToTransfer[k+j]
block=dataFile.readFrom(i)
indices.append(i)
blocks.append(block)
log.info("block #{0}: {1}...{2}".format(i,block[:5],block[-5:]))
progress.p(k+j)
if indices: self._sendData(indices,blocks)
progress.done()
self._outcoming.writeMsg({"command":"end","action":"push"})
log.info("closing session...")
dataFile.close()
def pullData(self,blocksToTransfer,ignoreLock=False):
if not ignoreLock:
try:
self._lock()
except LockedException:
print("The file is locked. Either (a) there's another pull going on (then wait or kill it), or (b) a previous pull ended prematurely and the file is probably corrupt (then repeat pull with -f for force).")
return
log.info(blocksToTransfer)
dataFile=DataFile.open(self._filename, mode="rb+")
print(datetime.now(), "receiving data:")
progress=Progress(len(blocksToTransfer))
for k in range(0,len(blocksToTransfer),conf.batchSize.data):
indices=blocksToTransfer[k:k+conf.batchSize.data]
self._outcoming.writeMsg({"command":"req", "index":indices, "dataType":"data"})
jsonData,binData=self._incoming.readMsg()
assert jsonData["command"]=="send" and jsonData["index"]==indices and jsonData["dataType"]=="data", jsonData
for (j,i) in enumerate(indices):
block=binData[j*HashTree.BLOCK_SIZE:(j+1)*HashTree.BLOCK_SIZE]
dataFile.writeAt(i,block)
if self._treeFile:
self._newLeaves[i+self._tree.leafStart]=hashBlock(block)
log.info("block #{0}: {1}...{2}".format(i,block[:5],block[-5:]))
stats.logTransferredBlock()
progress.p(k+j)
progress.done()
self._outcoming.writeMsg({"command":"end"})
log.info("closing session...")
dataFile.close()
self._unlock()
if self._treeFile:
self._updateTree()
def _sendData(self,indices,blocks):
jsonData={"command":"send", "index":indices, "dataType":"data"}
binData=b"".join(blocks)
self._outcoming.writeMsg(jsonData,binData)
stats.logTransferredBlock(len(indices))
jsonData,binData=self._incoming.readMsg()
assert jsonData["command"]=="ack" and jsonData["index"]==indices, jsonData
def setConnection(self,connection):
(self._incoming,self._outcoming)=connection
|