Merge branch 'main' of https://github.com/harshit-soora/Distributed-Storage-System into main

Author: ksaini-personal

monitor/monitor_gossip.py

@@ -1,40 +1,42 @@
 import socket
 import pickle
 import os
-from transfer import _send_msg, _recv_msg, _wait_recv_msg
-from info import HEARTBEAT_PORT, num_objects_per_file, max_num_objects_per_pg, MSG_SIZE, HEADERSIZE
-
-def __init__():
+import sys
 
-	#This will help us to know if the storage is alive or not
-	for i in range(4):
-		self.storage_node[i] = True
+from monitor_replicate import recovery
 
+sys.path.insert(1, '../utils/')
+from transfer import _send_msg, _recv_msg, _wait_recv_msg
+from info import mds_ip, monitor_ip, storage_ip, num_objects_per_file, max_num_objects_per_pg, MSG_SIZE, HEADERSIZE
 
-def recovery():
-	#Will start the recovery protocol
-	pass
 
-def gossip(c, msg):
+def gossip(c, msg, live_osd):
 	node_ip = msg["ip"]
+	crash_osd_id = msg["osd_id"]
+
+	if live_osd[crash_osd_id] == False:
+		return 
+	
+	print(f"Need to start the recovery protocol for {node_ip} {crash_osd_id}")
+
+	hashtable = {"pg_id1":[("osd_id1", True), ("osd_id2", True), ("osd_id4", True)]}
+	# Use this in maincode
+	# hashtable_file = open('hashtable', 'rb')
+	# hashtable_dump = hashtable_file.read()
+	# hashtable = pickle.load(hashtable_dump)
+	# hashtable_file.close()
 
-	i = 0
-	for node in storage_ip:
-		if storage_ip[node] == node_ip:
-			# self.storage_node[i] = False
 
-			print(f"Need to start the recovery protocol for {node_ip} storage node number {i+1}")
-			recovery()
+	live_osd[crash_osd_id] = False
+	rf = 3
 
-			break
+	hastable = recovery(crash_osd_id, hashtable, live_osd, rf)
+	print(hashtable)
 
-		i += 1
 
-	res = {"type": "ACK"}
-	_send_msg(c, res)
 
 
-def heartbeat_protocol(soc):
+def heartbeat_protocol(soc, live_osd):
 	# Establish connection with client. 
 	c, addr = soc.accept()  
 	print(f"\nGot connection from {addr}")
@@ -56,7 +58,7 @@ def heartbeat_protocol(soc):
 		elif msg["type"] == "FAIL":
 			res = {"type": "ACK"}
 			_send_msg(c, res)
-			# gossip(c, msg)
+			gossip(c, msg, live_osd)
 
 		c.close()
 
@@ -66,13 +68,15 @@ def heartbeat_protocol(soc):
 	return
 
 
+def run_main_monitor_gossip_cum_recovery():
+	# Make sure to maintain this 
+	live_osd = {"osd_id1":True, "osd_id2":True, "osd_id3":True, "osd_id4":True}
 
-if __name__ == "__main__":
 	s = socket.socket()         
 	print ("Socket successfully created")
 
 	# reserve a port on your computer in our 
-	port = HEARTBEAT_PORT
+	port =  monitor_ip["primary"]["port"]       
 
 	# Next bind to the port 
 	# we have not typed any ip in the ip field 
@@ -83,14 +87,13 @@ def heartbeat_protocol(soc):
 	print ("Monitor socket binded to %s" %(port)) 
 
 	# put the socket into listening mode 
-	s.listen(5)     
-	print ("socket is listening")            
+	s.listen(5)              
 
 	# a forever loop until we interrupt it or 
 	# an error occurs 
 	while True: 
 
-		heartbeat_protocol(s)
+		heartbeat_protocol(s, live_osd)
 
 	s.close()
 	# Close the connection with the client 

monitor/monitor_replicate.py

@@ -1,38 +1,42 @@
 import socket
 import pickle
-from transfer import _send_msg, _recv_msg
+import time
+from transfer import _send_msg, _recv_msg, _wait_recv_msg
 from info import mds_ip, monitor_ip, storage_ip, num_objects_per_file, max_num_objects_per_pg, MSG_SIZE, HEADERSIZE
 
 
 def send_replicate_request(pg_id, master_osd, clone_osd):
 	soc = socket.socket() 
 	soc.settimeout(5)        
-	print(f"Socket successfully created for replicating pg_id {pg_id} from master osd {master_osd} to clone osd {clone_osd}")
+	print(f"Socket successfully created for replicating {pg_id} from master {master_osd} to clone {clone_osd}")
 
 	# reserve a port on your computer in our 
 	ip_add = storage_ip[master_osd]["ip"] 
-	port =  storage_ip[master_osd]["port"]       
-	  
+	port =  storage_ip[master_osd]["port"] + 10    
+	
 	flag = -1
 	try :
 		soc.connect((ip_add, port))	
-		soc.timeout(None)
+		soc.settimeout(None)
 
-		print(f"Connecting with osd {osd_id}...")
+		print(f"Connecting with {master_osd} on port {port}...")
 
-		res = {"type": "REPLICATE", "pg_id" : pg_id, "osd" : osd_id}
+		res = {"type": "REPLICATE", "pg_id" : pg_id, "osd_id" : clone_osd}
 		_send_msg(soc, res)
 		print("Message send for replication ")
-		time.sleep(3)
+
+		# time.sleep(5)
 
-		msg = _recv_msg(c, MSG_SIZE)
+		msg = _wait_recv_msg(soc, MSG_SIZE)
+		print(msg)
 		if msg == None:
 			flag = 0
 		elif msg["type"] == "ACK": 
 			flag = 1
 
-	except : 
-		print("Didn't Connect! [Timeout] Primary Monitor is Down")
+	except Exception as e: 
+		print(e)
+		print(f"Didn't Connect! [Timeout] master {master_osd} is down,  port {port}")
 
 	soc.close()
 	return flag
@@ -56,10 +60,12 @@ def replicate(pg_id, osd_id_master_list, osd_id_clone_dict):
 					add_entry.append([clone, True])
 					break
 				elif ret == -1:
+					print("may try send_replicate_request again or continue to next master")
 					# may try send_replicate_request again
 					# or continue to next master
 					continue
 				elif ret == 0:
+					print("eans master osd server break before sending the ack")
 					# means master osd server break before sending the ack
 					continue
 				else:
@@ -82,6 +88,7 @@ def recovery(crash_osd_id, hashtable, live_osd, replication_factor):
 	rf = replication_factor
 
 	# Check the hashtable for pg_id dependency 
+	new_hashtable = hashtable
 	for pg_id in hashtable.keys():
 		list_of_replicated_osds = hashtable[pg_id]
 		list_of_master = []
@@ -105,6 +112,8 @@ def recovery(crash_osd_id, hashtable, live_osd, replication_factor):
 			# Thus clone_osd dict() will only have the osd that are live and 
 			# don't have that pg_id which was there is crashed node "crash_osd_id"
 
-			for entry in new_entries
-				hashtable[pg_id].append(entry)
+			for entry in new_entries:
+				new_hashtable[pg_id].append(entry)
+
 
+	return new_hashtable

osd/storage_gossip.py

@@ -3,28 +3,34 @@
 import threading
 import time
 import os
+
+from storage_replication import replicate_pg
+
+sys.path.insert(1, '../utils/')
 from transfer import _send_msg, _recv_msg, _wait_recv_msg
-from info import MONITOR_IPs, OSD_IPs, HEARTBEAT_PORT, num_objects_per_file, max_num_objects_per_pg, MSG_SIZE, HEADERSIZE
+from info import mds_ip, monitor_ip, storage_ip, num_objects_per_file, max_num_objects_per_pg, MSG_SIZE, HEADERSIZE
+
+
 
-STORAGE_ID = 1
+STORAGE_ID = "osd_id1"
 
 
-def report_monitor(node_ip, node_id):
+def report_monitor(node_ip, osd_id):
 	flag = 0
 	#Will call monitor to state about the down node
 	soc = socket.socket()
 	soc.settimeout(5)
 	print ("Socket successfully created for Recovery: Primary")
 
-	monitor_1 = MONITOR_IPs["primary"]
+	monitor_1 = monitor_ip["primary"]
 
 	try :
 		soc.connect((monitor_1["ip"], monitor_1["port"]))	
 		# soc.timeout(None)
 
 		print(f"Connecting Primary monitor...")
 
-		res = {"type": "FAIL", "ip" : node_ip, "id" : node_id}
+		res = {"type": "FAIL", "ip" : node_ip, "osd_id" : osd_id}
 		_send_msg(soc, res)
 
 		msg = _wait_recv_msg(soc, MSG_SIZE)
@@ -47,14 +53,14 @@ def report_monitor(node_ip, node_id):
 	print ("Socket successfully created for Recovery: Backup")
 
 
-	monitor_2 = MONITOR_IPs["backup"]
+	monitor_2 = monitor_ip["backup"]
 
 	try : 
 		soc.connect((monitor_2["ip"], monitor_2["port"]))	
 		soc.settimeout(None)			
 		print(f"Connecting Backup monitor...")
 
-		res = {"type": "FAIL", "ip" : node_ip, "id" : node_id}
+		res = {"type": "FAIL", "ip" : node_ip, "osd_id" : osd_id}
 		_send_msg(soc, res)
 
 		msg = _wait_recv_msg(soc, MSG_SIZE)
@@ -79,35 +85,36 @@ def recv_gossip():
 
 		i=0
 		for i in range(4):
-			if i+1 != STORAGE_ID:
-				node_ip =  OSD_IPs[i+1]["ip"]
-				port = OSD_IPs[i+1]["port"]
+			curr_osd = "osd_id" + str(i+1)
+			if curr_osd != STORAGE_ID:
+				node_ip =  storage_ip[curr_osd]["ip"]
+				port = storage_ip[curr_osd]["port"]
 
 				soc = socket.socket()
 				soc.settimeout(5)
-				print(f"\n\nSocket successfully created for Gossip with osd {i+1}")
+				print(f"\n\nSocket successfully created for Gossip with osd {curr_osd}")
 
 				try :
 					soc.connect((node_ip, port))
 					soc.settimeout(None)
 
-					print(f"Connecting {node_ip} storage node number {i+1} port {port}")
+					print(f"Connecting {node_ip} storage node number {curr_osd} port {port}")
 
 					msg = {"type": "ALIVE"}
 					_send_msg(soc, msg)
 
 					rec = _wait_recv_msg(soc, MSG_SIZE)
 					print(msg)
 					if rec == None: 
-						print(f"Didn't receive data to Storage {i+1} ip {node_ip}! [Timeout] ")
-						report_monitor(node_ip, i+1)
+						print(f"Didn't receive data to Storage {curr_osd} ip {node_ip}! [Timeout] ")
+						report_monitor(node_ip, curr_osd)
 
 					elif rec["type"] != "ACK": 
-						report_monitor(node_ip, i+1)
+						report_monitor(node_ip, curr_osd)
 
 				except :	
-					print(f"Didn't Connect to Storage {i+1} ip {node_ip}! [Timeout]")
-					report_monitor(node_ip, i+1)
+					print(f"Didn't Connect to Storage {curr_osd} ip {node_ip}! [Timeout]")
+					report_monitor(node_ip, curr_osd)
 
 				soc.close()	
 
@@ -121,13 +128,12 @@ def send_heartbeat():
 
 	s = socket.socket()         
 	print ("Socket successfully created for Heartbeat")
-	port =  HEARTBEAT_PORT
+	port =  storage_ip[STORAGE_ID]["port"] 
 	s.bind(('', port))         
 	print ("Socket binded to %s" %(port)) 
 
 	# put the socket into listening mode 
 	s.listen(5)     
-	print ("Socket is listening") 
 
 	while True:
 		# Establish connection with client. 
@@ -154,9 +160,29 @@ def send_heartbeat():
 			os._exit(1)
 
 
-if __name__ == "__main__":
+def read_write_pg_replication():
+	#This will check for incoming messages 
+	#from other nodes to replicate certain data 
+
+	s = socket.socket()         
+	print ("Socket successfully created for Read-Write only for replication")
+	# As mentioned in the info file 
+	# Read/write request will be send on "port + 10"
+	port =  storage_ip[STORAGE_ID]["port"] + 10
+	s.bind(('', port))         
+	print ("Socket binded to %s" %(port)) 
+	  
+	# put the socket into listening mode 
+	s.listen(5)     
+	
+	while True:
+		replicate_pg(s)
+
 
+def _run_main_osd_gossip_cum_recovery():
 	p1 = threading.Thread(target=send_heartbeat)
 	p2 = threading.Thread(target=recv_gossip)
+	p3 = threading.Thread(target=read_write_pg_replication)
 	p1.start()
-	p2.start()
+	p2.start()
+	p3.start()