Files @ 0ad71a952f92
Branch filter:

Location: Morevna/src/client.py - annotation

Laman
handled locked exception on pull
ee936b917440
34f4027c1bd6
cd2ba192bf12
34f4027c1bd6
b052f27e1cbc
34f4027c1bd6
34f4027c1bd6
41ea9614ce8c
4b88aca70fbc
8bb6a904d50b
0ad71a952f92
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
362cff560740
362cff560740
362cff560740
2e5828ec7d49
2e5828ec7d49
2e5828ec7d49
2e5828ec7d49
2e5828ec7d49
b73a5d69a11b
8bb6a904d50b
b73a5d69a11b
b73a5d69a11b
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
095908159393
095908159393
095908159393
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
362cff560740
34f4027c1bd6
dad65188b1a0
b73a5d69a11b
3d0876534e40
0ad71a952f92
0ad71a952f92
0ad71a952f92
0ad71a952f92
0ad71a952f92
0ad71a952f92
0ad71a952f92
3d0876534e40
3d0876534e40
3d0876534e40
3d0876534e40
3d0876534e40
6c8e994fd906
6c8e994fd906
6c8e994fd906
6c8e994fd906
6c8e994fd906
3d0876534e40
6c8e994fd906
6c8e994fd906
6c8e994fd906
3d0876534e40
8bb6a904d50b
8bb6a904d50b
8bb6a904d50b
6c8e994fd906
3d0876534e40
6c8e994fd906
6c8e994fd906
6c8e994fd906
3d0876534e40
3d0876534e40
6c8e994fd906
3d0876534e40
3d0876534e40
3d0876534e40
362cff560740
6c8e994fd906
8bb6a904d50b
75e070b6b447
8bb6a904d50b
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,hashBlock
from netnode import BaseConnection,NetNode,FailedConnection,LockedException


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":
			raise DeniedConnection()
		assert jsonData["command"]=="init"

	## 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):
				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=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","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=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)

			if self._treeFile:
				self._newLeaves[i2+self._tree.leafStart]=hashBlock(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()
		self._unlock()

		if self._treeFile:
			self._updateTree()

	def setConnection(self,connection):
		(self._incoming,self._outcoming)=connection