Morevna Changeset - 83dc5e1e183e

Changeset - 83dc5e1e183e

Parent rev.

Child rev.

[Not reviewed]

default

0 3 0

Laman - 5 years ago 2020-05-24 21:23:39

regularly flushing the hash tree to the disk during a data transfer

3 files changed with 16 insertions and 2 deletions:

src/client.py

src/netnode.py

src/server.py

0 comments (0 inline, 0 general)

src/client.py

➞

Show inline comments

@@ @@ -148,25 +148,30 @@ class Client(NetNode): @@
 		log.info(blocks_to_transfer)
 		data_file = DataFile.open(self._filename, mode="rb+")
 		print(datetime.now(), "receiving data:")
 		progress = Progress(len(blocks_to_transfer))
 		last_flushed = datetime.now().timestamp()
 		for k in range(0, len(blocks_to_transfer), conf.batch_size.data):
 			indices = blocks_to_transfer[k:k+conf.batch_size.data]
 			self._outcoming.write_msg({"command": "req", "index": indices, "dataType": "data"})
 			json_data, bin_data = self._incoming.read_msg()
 			assert json_data["command"]=="send" and json_data["index"]==indices and json_data["dataType"]=="data", json_data
 			for (j, i) in enumerate(indices):
 				block = bin_data[j*HashTree.BLOCK_SIZE:(j+1)*HashTree.BLOCK_SIZE]
 				data_file.write_at(i, block)
 				log.info("block #{0}: {1}...{2}".format(i, block[:5], block[-5:]))
 				if self._tree_file:
 					self._new_leaves[i+self._tree.leaf_start] = hash_block(block)
 				log.info("block #{0}: {1}...{2}".format(i, block[:5], block[-5:]))
 					t = datetime.now().timestamp()
 					if t-last_flushed >= 60 and self._tree_file:
 						self._update_tree()
 						last_flushed = t
 				stats.log_transferred_block()
 				progress.p(k+j)
 		progress.done()
 		self._outcoming.write_msg({"command": "end"})

src/netnode.py

➞

Show inline comments

@@ @@ -68,7 +68,8 @@ class NetNode: @@
 		os.remove(lock_file)
 	def _update_tree(self):
 		log.info("updating hash tree...")
 		self._tree.batch_update(self._new_leaves.items())
 		self._tree.save(self._tree_file)
 		self._new_leaves = dict()
 		log.info("tree updated")

src/server.py

➞

Show inline comments

 import socket
 from datetime import datetime
 import socket
 import ssl
 import multiprocessing
 import logging as log
 from hashtree import hash_block
 from netnode import BaseConnection,NetNode
@@ @@ -59,12 +60,13 @@ class Server(NetNode): @@
 		super().__init__(filename, tree_file)
 		(self._incoming, self._outcoming) = connection
 		self.BLOCK_SIZE = self._tree.BLOCK_SIZE
 		self._data_file_obj = None
 		self._last_flushed = 0
 	@staticmethod
 	def run(connection, *args):
 		with connection as c:
 			s = Server(c, *args)
 			s.serve()
@@ @@ -91,12 +93,13 @@ class Server(NetNode): @@
 				self._outcoming.write_msg({"command": "deny", "status": status.incompatible.version})
 			if json_data["action"]=="pull" and self.is_locked():
 				self._outcoming.write_msg({"command": "deny", "status": status.locked})
 			if json_data["action"]=="push" and not self.is_locked():
 				self._lock()
 			self._last_flushed = datetime.now().timestamp()
 			self._outcoming.write_msg({"command": "init", "version": conf.version})
 		elif json_data["command"]=="req":
 			if json_data["dataType"]=="data":
 				self._outcoming.write_msg(*self._request_data(json_data["index"]))
 			else:
@@ @@ -143,12 +146,17 @@ class Server(NetNode): @@
 		for (i, k) in enumerate(indices):
 			block = bin_data[i*self.BLOCK_SIZE:(i+1)*self.BLOCK_SIZE]
 			self._data_file.write_at(k, block)
 			if self._tree_file:
 				self._new_leaves[k+self._tree.leaf_start] = hash_block(block)
 		t = datetime.now().timestamp()
 		if t-self._last_flushed>=60 and self._tree_file:
 			self._update_tree()
 			self._last_flushed = t
 		return ({"command": "ack", "index": indices},)
 	def _finalize(self):
 		log.info("closing session...")
 		self._data_file.close()
 		self._data_file_obj = None

0 comments (0 inline, 0 general)