Changeset - bb3b53ee15d6
[Not reviewed]
default
0 3 1
Laman - 7 years ago 2018-01-21 00:42:05

batch pull
4 files changed with 50 insertions and 18 deletions:
0 comments (0 inline, 0 general)
src/client.py
Show inline comments
 
@@ -7,12 +7,13 @@ 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):
 
@@ -139,34 +140,33 @@ class Client(NetNode):
 
			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
 
		dataFile=DataFile.open(self._filename, mode="rb+")
 

	
 
		print(datetime.now(), "receiving data:")
 
		progress=Progress(len(blocksToTransfer))
 
		for (k,i2) in enumerate(blocksToTransfer):
 
			self._outcoming.writeMsg({"command":"req", "index":i2, "dataType":"data"})
 

	
 
		for k in range(0,len(blocksToTransfer),conf.batchSize):
 
			indices=blocksToTransfer[k:k+conf.batchSize]
 
			self._outcoming.writeMsg({"command":"req", "index":indices, "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)
 
			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[i2+self._tree.leafStart]=hashBlock(binData)
 
					self._newLeaves[i+self._tree.leafStart]=hashBlock(block)
 

	
 
			log.info("block #{0}: {1}...{2}".format(i2,binData[:5],binData[-5:]))
 
				log.info("block #{0}: {1}...{2}".format(i,block[:5],block[-5:]))
 

	
 
			stats.logTransferredBlock()
 
			i1=i2
 
			progress.p(k)
 
				progress.p(k+j)
 
		progress.done()
 

	
 
		self._outcoming.writeMsg({"command":"end"})
 

	
 
		log.info("closing session...")
 
		dataFile.close()
src/datafile.py
Show inline comments
 
new file 100644
 
from hashtree import HashTree
 

	
 

	
 
BLOCK_SIZE=HashTree.BLOCK_SIZE
 

	
 

	
 
class DataFile:
 
	def __init__(self,fileHandle):
 
		self._lastIndex=0
 
		self._f=fileHandle
 

	
 
	@staticmethod
 
	def open(filename,mode="rb"):
 
		return DataFile(open(filename,mode=mode))
 

	
 
	def writeAt(self,i,blockData):
 
		if i!=self._lastIndex+1:
 
			self._f.seek(i*BLOCK_SIZE)
 
		self._f.write(blockData)
 
		self._lastIndex=i
 

	
 
	def readFrom(self,i,byteCount=BLOCK_SIZE):
 
		if i!=self._lastIndex+1:
 
			self._f.seek(i*BLOCK_SIZE)
 
		self._lastIndex=i
 
		return self._f.read(byteCount)
 

	
 
	def close(self):
 
		self._f.close()
src/morevna.py
Show inline comments
 
@@ -4,13 +4,13 @@ import logging as log
 
from argparse import ArgumentParser
 

	
 
from util import spawnDaemon, splitHost
 
import config as conf
 
import stats
 
from hashtree import HashTree
 
from client import Client, Connection as ClientConnection, FailedConnection, DeniedConnection
 
from client import Client, Connection as ClientConnection, FailedConnection, DeniedConnection, IncompatibleException
 
from server import Miniserver
 

	
 

	
 
def _checkFile(f):
 
	if not os.path.isfile(f):
 
		print("invalid file specified:",f,file=sys.stderr)
 
@@ -50,12 +50,13 @@ def push(args):
 
			print(stats.report())
 
			print()
 
		except FailedConnection: pass
 
		except DeniedConnection as e:
 
			print("Server {0}:{1} denied connection.".format(*host))
 
			print(e)
 
		except IncompatibleException as e: print(e)
 

	
 
def pull(args):
 
	_checkFile(args.datafile)
 
	if args.tree:
 
		_checkFile(args.tree)
 
	if args.host: conf.hosts=[args.host]
src/server.py
Show inline comments
 
@@ -120,17 +120,19 @@ class Server(NetNode):
 
		return (jsonResponse,binResponse)
 

	
 
	def _requestData(self,index):
 
		log.info("received request for data block #{0}".format(index))
 

	
 
		jsonResponse={"command":"send", "index":index, "dataType":"data"}
 
		if self._lastIndex+1!=index:
 
			self._dataFile.seek(index*self.BLOCK_SIZE)
 
		binResponse=self._dataFile.read(self.BLOCK_SIZE)
 
		blocks=[]
 
		for i in index:
 
			if self._lastIndex+1!=i:
 
				self._dataFile.seek(i*self.BLOCK_SIZE)
 
			blocks.append(self._dataFile.read(self.BLOCK_SIZE))
 

	
 
		return (jsonResponse,binResponse)
 
		return (jsonResponse,b"".join(blocks))
 

	
 
	def _receiveData(self,jsonData,binData):
 
		if not self.isLocked(): self._lock()
 
		log.info("received data block #{0}: {1}...{2}".format(jsonData["index"],binData[:5],binData[-5:]))
 

	
 
		indices=jsonData["index"]
0 comments (0 inline, 0 general)