Files
@ 1828ea5794c5
Branch filter:
Location: Morevna/src/client.py - annotation
1828ea5794c5
4.4 KiB
text/x-python
tests: redirected output and logging
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 | ee936b917440 34f4027c1bd6 cd2ba192bf12 34f4027c1bd6 b052f27e1cbc 34f4027c1bd6 34f4027c1bd6 41ea9614ce8c 4b88aca70fbc b052f27e1cbc 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 6c8e994fd906 cd2ba192bf12 cd2ba192bf12 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 cd2ba192bf12 026618d6681b 026618d6681b 026618d6681b 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 34f4027c1bd6 cd2ba192bf12 026618d6681b 34f4027c1bd6 34f4027c1bd6 b73a5d69a11b b73a5d69a11b 6c8e994fd906 6c8e994fd906 026618d6681b 34f4027c1bd6 6c8e994fd906 6c8e994fd906 cd2ba192bf12 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 2e5828ec7d49 b73a5d69a11b 6c8e994fd906 b73a5d69a11b b73a5d69a11b 34f4027c1bd6 b73a5d69a11b 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 34f4027c1bd6 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 5813971dbecc 7d21dd70864a 5813971dbecc 5813971dbecc 5813971dbecc 34f4027c1bd6 6c8e994fd906 5813971dbecc 6c8e994fd906 5813971dbecc 34f4027c1bd6 2e5828ec7d49 5813971dbecc 5813971dbecc 5813971dbecc ee936b917440 ee936b917440 2e5828ec7d49 2e5828ec7d49 5813971dbecc 5813971dbecc 5813971dbecc 2e5828ec7d49 4b88aca70fbc 34f4027c1bd6 2e5828ec7d49 34f4027c1bd6 b73a5d69a11b b73a5d69a11b 026618d6681b b73a5d69a11b 34f4027c1bd6 b052f27e1cbc 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 34f4027c1bd6 6c8e994fd906 34f4027c1bd6 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 4b88aca70fbc 34f4027c1bd6 6c8e994fd906 34f4027c1bd6 dad65188b1a0 b73a5d69a11b 3d0876534e40 3d0876534e40 3d0876534e40 3d0876534e40 3d0876534e40 3d0876534e40 3d0876534e40 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 6c8e994fd906 3d0876534e40 6c8e994fd906 6c8e994fd906 6c8e994fd906 3d0876534e40 6c8e994fd906 3d0876534e40 6c8e994fd906 6c8e994fd906 6c8e994fd906 3d0876534e40 3d0876534e40 6c8e994fd906 3d0876534e40 3d0876534e40 3d0876534e40 6c8e994fd906 6c8e994fd906 6c8e994fd906 | import collections
import socket
import ssl
import logging as log
from datetime import datetime
import config as conf
import stats
from util import Progress
from hashtree import HashTree
from networkers import NetworkReader,NetworkWriter
class Connection:
def __init__(self):
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)
self._socket.connect((conf.hosts[0], conf.port))
fr=self._socket.makefile(mode="rb")
fw=self._socket.makefile(mode="wb")
self.incoming=NetworkReader(fr)
self.outcoming=NetworkWriter(fw)
def __enter__(self):
return self.incoming,self.outcoming
def __exit__(self, exc_type, exc_val, exc_tb):
self._socket.shutdown(socket.SHUT_RDWR)
self._socket.close()
class Client:
def __init__(self,filename):
self._incoming=None
self._outcoming=None
self._filename=filename
print(datetime.now(), "initializing...")
self._localTree=HashTree.fromFile(self._filename)
## 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._localTree
blocksToTransfer=[]
nodeStack=collections.deque([0]) # root
# initialize session
jsonData={"command":"init", "blockSize":localTree.BLOCK_SIZE, "blockCount":localTree.leafCount, "version":conf.version}
self._outcoming.writeMsg(jsonData)
jsonData,binData=self._incoming.readMsg()
assert jsonData["command"]=="ack"
# determine which blocks to send
print(datetime.now(), "negotiating:")
progress=Progress(localTree.leafCount)
while len(nodeStack)>0:
indices=[]
for i in range(conf.batchSize):
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()
return blocksToTransfer
def sendData(self,blocksToTransfer):
log.info(blocksToTransfer)
dataFile=open(self._filename, mode="rb")
i1=-1
print(datetime.now(), "sending data:")
progress=Progress(len(blocksToTransfer))
for (k,i2) in enumerate(blocksToTransfer):
jsonData={"command":"send", "index":i2, "dataType":"data"}
if i1+1!=i2:
dataFile.seek(i2*HashTree.BLOCK_SIZE)
binData=dataFile.read(HashTree.BLOCK_SIZE)
log.info("block #{0}: {1}...{2}".format(i2,binData[:5],binData[-5:]))
self._outcoming.writeMsg(jsonData,binData)
stats.logTransferredBlock()
jsonData,binData=self._incoming.readMsg()
assert jsonData["command"]=="ack" and jsonData["index"]==i2, jsonData
i1=i2
progress.p(k)
progress.done()
self._outcoming.writeMsg({"command":"end"})
log.info("closing session...")
dataFile.close()
def pullData(self,blocksToTransfer):
log.info(blocksToTransfer)
dataFile=open(self._filename, mode="rb+")
i1=-1
print(datetime.now(), "receiving data:")
progress=Progress(len(blocksToTransfer))
for (k,i2) in enumerate(blocksToTransfer):
self._outcoming.writeMsg({"command":"req", "index":i2, "dataType":"data"})
jsonData,binData=self._incoming.readMsg()
assert jsonData["command"]=="send" and jsonData["index"]==i2 and jsonData["dataType"]=="data", jsonData
if i1+1!=i2:
dataFile.seek(i2*HashTree.BLOCK_SIZE)
dataFile.write(binData)
log.info("block #{0}: {1}...{2}".format(i2,binData[:5],binData[-5:]))
stats.logTransferredBlock()
i1=i2
progress.p(k)
progress.done()
self._outcoming.writeMsg({"command":"end"})
log.info("closing session...")
dataFile.close()
def setConnection(self,connection):
(self._incoming,self._outcoming)=connection
|