Morevna Changeset - 9f2b0a4f3538

Changeset - 9f2b0a4f3538

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Laman - 7 years ago 2017-10-31 19:45:36

push to multiple servers

6 files changed with 18 insertions and 11 deletions:

src/client.py

src/morevna.py

src/netnode.py

src/server.py

src/tests/test_overall.py

src/util.py

0 comments (0 inline, 0 general)

src/client.py

➞

Show inline comments

@@ @@ -4,35 +4,35 @@ 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
 class Connection(BaseConnection):
 	def __init__(self):
+	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((conf.hosts[0], conf.port))
+			self._socket.connect((host,port))
 		except ConnectionRefusedError:
 			print("Couldn't connect to {0}:{1}".format(conf.hosts[0],conf.port))
 			sys.exit(1)
 		self.createNetworkers()
 class Client(NetNode):
 	def __init__(self,filename,treeFile=""):
 		print(datetime.now(), "initializing...")
 		super().__init__(filename,treeFile)

src/morevna.py

➞

Show inline comments

 import sys
 import os.path
 import logging as log
 from argparse import ArgumentParser
 from util import spawnDaemon
+from util import spawnDaemon, splitHost
 import config as conf
 import stats
 from hashtree import HashTree
 from client import Client, Connection as ClientConnection
 from server import Miniserver
 def _checkFile(f):
 	if not os.path.isfile(f):
 		print("invalid file specified:",f,file=sys.stderr)
 		sys.exit(1)
@@ @@ -24,44 +24,45 @@ def buildTree(args): @@
 		dataMod=os.stat(args.datafile).st_mtime
 		if dataMod<treeMod and not args.force:
 			print("tree file is up to date")
 			return
 	tree=HashTree.fromFile(args.datafile)
 	tree.save(args.treefile)
 def push(args):
 	_checkFile(args.datafile)
 	if args.tree:
 		_checkFile(args.tree)
-	if args.host: conf.hosts.insert(0,args.host)
+	if args.host: conf.hosts=[args.host]
 	if args.port: conf.port=args.port
 	c=Client(args.datafile,args.tree)
 	with ClientConnection() as con:
 	for host in conf.hosts:
 		with ClientConnection(*splitHost(host,conf.port)) as con:
 		c.setConnection(con)
 		blocksToTransfer=c.negotiate()
 		c.sendData(blocksToTransfer)
 	print()
 	print(stats.report())
 def pull(args):
 	_checkFile(args.datafile)
 	if args.tree:
 		_checkFile(args.tree)
-	if args.host: conf.hosts.insert(0,args.host)
+	if args.host: conf.hosts=[args.host]
 	if args.port: conf.port=args.port
 	c=Client(args.datafile,args.tree)
 	with ClientConnection() as con:
+	with ClientConnection(*splitHost(conf.hosts[0],conf.port)) as con:
 		c.setConnection(con)
 		blocksToTransfer=c.negotiate()
 		c.pullData(blocksToTransfer)
 	print()
 	print(stats.report())
 def serve(args):
 	_checkFile(args.datafile)
 	if args.tree:
 		_checkFile(args.tree)
 	if args.host: conf.hosts.insert(0,args.host)
 	if args.port: conf.port=args.port

src/netnode.py

➞

Show inline comments

@@ @@ -17,25 +17,25 @@ class BaseConnection: # abstract @@
 		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):
 		try:
 			self._socket.shutdown(socket.SHUT_RDWR)
 			self._socket.close()
 		except OSError:
-			log.warning("encountered an error when shutting down the connection")
+			log.warning("broken connection")
 class NetNode:
 	def __init__(self,filename,treeFile=""):
 		self._incoming=None
 		self._outcoming=None
 		self._filename=filename
 		self._treeFile=treeFile
 		if treeFile:
 			self._tree=HashTree.load(treeFile)

src/server.py

➞

Show inline comments

@@ @@ -55,25 +55,25 @@ class Server(NetNode): @@
 		s=Server(*args)
 		s.serve()
 	@property
 	def _dataFile(self):
 		if not self._dataFileHandle:
 			self._dataFileHandle=open(self._filename, mode="rb+")
 		return self._dataFileHandle
 	def serve(self):
 		try:
 			while self._serveOne(): pass
 		except AssertionError as e:
+		except (AssertionError,ConnectionResetError) as e:
 			log.warning(e)
 	def _serveOne(self):
 		jsonData,binData=self._incoming.readMsg()
 		if jsonData["command"]=="init":
 			assert jsonData["blockSize"]==self.BLOCK_SIZE
 			assert jsonData["blockCount"]==self._tree.leafCount
 			self._outcoming.writeMsg({"command": "ack"})
 		elif jsonData["command"]=="req":
 			if jsonData["dataType"]=="data":

src/tests/test_overall.py

➞

Show inline comments

@@ @@ -59,38 +59,38 @@ class TestMorevna(TestCase): @@
 		self.assertEqual(*compareFiles(refFile,treeFile))
 		os.remove(treeFile)
 	def test_push(self):
 		ms=Miniserver(filename)
 		p=multiprocessing.Process(target=ms.serve)
 		p.start()
 		for clientFile in ("test2.img","test3.img","test4.img"):
 			clientFile=os.path.join(dataDir,clientFile)
 			c=Client(clientFile)
 			with ClientConnection() as con:
+			with ClientConnection("127.0.0.1",config.port) as con:
 				c.setConnection(con)
 				blocksToTransfer=c.negotiate()
 				c.sendData(blocksToTransfer)
 			self.assertEqual(*compareFiles(clientFile,filename))
 		p.terminate()
 		p.join()
 	def test_pull(self):
 		serverFile=os.path.join(dataDir,"test3.img")
 		ms=Miniserver(serverFile)
 		p=multiprocessing.Process(target=ms.serve)
 		p.start()
 		c=Client(filename)
 		with ClientConnection() as con:
+		with ClientConnection("127.0.0.1",config.port) as con:
 			c.setConnection(con)
 			blocksToTransfer=c.negotiate()
 			c.pullData(blocksToTransfer)
 		self.assertEqual(*compareFiles(serverFile,filename))
 		p.terminate()
 		p.join()

src/util.py

➞

Show inline comments

@@ @@ -24,24 +24,30 @@ def spawnDaemon(fun): @@
 			print("[{0}] server running".format(pid))
 			sys.exit(0)
 	except OSError as e:
 		print("fork #2 failed: {0} ({1})".format(e.errno,e.strerror),file=sys.stderr)
 		sys.exit(1)
 	fun()
 	# all done
 	os._exit(os.EX_OK)
 def splitHost(host,defaultPort=0):
 	address,_,port=host.partition(":")
 	if not port: port=defaultPort
 	return (address,port)
 class Progress:
 	def __init__(self,n,i0=0):
 		self._n=n
 		self._i0=i0
 		self._i=i0
 		self._last=""
 	def p(self,i):
 		i0=self._i0
 		n=self._n
 		assert i0<=i<n or n<i<=i0, (i0,i,n)

0 comments (0 inline, 0 general)