Running HeartBeat & Replication...Yayyyyygit status

Author: harshit-soora

Test Code/Monitor/__pycache__/info.cpython-38.pyc

@@ -1,7 +1,7 @@
 import socket
 import pickle
 import os
-from transfer import _send_msg, _recv_msg
+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
 
 
@@ -39,7 +39,8 @@ def replicate_pg(soc):
 			print(f"Socket created to send request to SAVE data at OSD {osd_id}")
 
 			ip_add = storage_ip[osd_id]["ip"]
-			port = storage_ip[osd_id]["port"]
+			port = storage_ip[osd_id]["port"] + 10
+			# Remember replication is on normal port + 10
 
 			try :
 				new_soc.connect((ip_add, port))
@@ -72,13 +73,18 @@ def replicate_pg(soc):
 			pg_id = msg["pg_id"]
 			pg = msg["pg"]
 
-			file = open("./data/"+ pg_id, 'wb')
+			try :
+				file = open("./data/"+ pg_id, 'wb')
+				print("Replicated data is saved")
 
-			pg_dump = pickle.dump(pg)
-			file.write(pg_dump)
+				pg_dump = pickle.dump(pg)
+				file.write(pg_dump)
+
+			except Exception as e:
+				print(e)
 
 			msg = {"type":"ACK"}
-			_send_msg(msg)
+			_send_msg(c, msg)
 			print("Write successful..send back Ack to the master osd")
 
 
@@ -87,5 +93,5 @@ def replicate_pg(soc):
 
 		c.close()
 
-		print(f"Exiting from pid os.getpid() ..")
+		print(f"Exiting from pid {os.getpid()} ..")
 		os._exit(1)

Test Code/Monitor/__pycache__/monitor_replicate.cpython-38.pyc

@@ -0,0 +1 @@
+�}�.

Test Code/Monitor/__pycache__/transfer.cpython-38.pyc

@@ -0,0 +1,52 @@
+"""
+@author: Kartik Saini
+
+"""
+
+mds_ip = {"primary":{"ip":"", "port":0}, "backup":{"ip":"", "port":0}}
+
+num_objects_per_file = 1
+
+max_num_objects_per_pg = 1
+
+_port = 9999
+
+MSG_SIZE = 1024
+HEADERSIZE = 10
+
+"""
+Message Name/type Harshit is using :
+ALIVE = will be send to check the heartbeat
+ACK = will be send in return of heartbeat 
+FAIL = will be send by storage node to monitor along with the IP or storage number
+"""
+
+dir_tree = {
+            "dir1":{},
+            "dir2":{}
+            }
+
+"""
+PG -> osds hash table on monitor
+
+OSDs -> ip, port .. for live osds on monitor
+
+OSD -> replicas ... for monitoring on OSD
+
+"""
+
+# Update all the stroage node IP everytime
+storage_1 = {"ip":"127.0.0.1", "port":10001}
+storage_2 = {"ip":"127.0.0.1", "port":10002}
+storage_3 = {"ip":"127.0.0.1", "port":10003}
+storage_4 = {"ip":"4.0.0.0", "port":10004}
+
+storage_ip = {"osd_id1":storage_1,"osd_id2":storage_2,"osd_id3":storage_3,"osd_id4":storage_4}
+
+# add 10 to port to get read/write IP
+
+
+# Update all the monitor node IP
+monitor_1 = {"ip":"127.0.0.1", "port":4002}
+monitor_2 = {"ip":"127.0.0.1", "port":4002}
+monitor_ip = {"primary": monitor_1, "backup": monitor_2}

Test Code/Storage-1/__pycache__/info.cpython-38.pyc

@@ -0,0 +1,185 @@
+import socket
+import pickle
+import threading
+import time
+import os
+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
+
+from storage_replication import replicate_pg
+
+STORAGE_ID = "osd_id2"
+
+
+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_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, "osd_id" : osd_id}
+		_send_msg(soc, res)
+		
+		msg = _wait_recv_msg(soc, MSG_SIZE)
+		print(msg)
+		if msg == None:
+			pass
+		elif msg["type"] == "ACK": 
+
+			soc.close()
+			return
+
+	except Exception as e: 
+		print(e)
+		print("Didn't Connect! [Timeout] Primary Monitor is Down")
+
+	soc.close()
+
+	soc = socket.socket()
+	soc.settimeout(5)
+	print ("Socket successfully created for Recovery: 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, "osd_id" : osd_id}
+		_send_msg(soc, res)
+
+		msg = _wait_recv_msg(soc, MSG_SIZE)
+		print(msg)
+		if msg == None:
+			pass
+		elif msg["type"] == "ACK": 
+			return
+
+	except:
+		print("MAY GOD HELP US!! WE ARE DOOMED\n\n")
+
+	soc.close()
+
+
+
+def recv_gossip():
+
+	while True:
+		time.sleep(10)
+		# Wait for 10 sec to  run this protocol
+
+		i=0
+		for i in range(4):
+			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 {curr_osd}")
+			
+				try :
+					soc.connect((node_ip, port))
+					soc.settimeout(None)
+
+					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 {curr_osd} ip {node_ip}! [Timeout] ")
+						report_monitor(node_ip, curr_osd)
+
+					elif rec["type"] != "ACK": 
+						report_monitor(node_ip, curr_osd)
+					
+				except :	
+					print(f"Didn't Connect to Storage {curr_osd} ip {node_ip}! [Timeout]")
+					report_monitor(node_ip, curr_osd)
+			
+				soc.close()	
+
+			time.sleep(3)
+
+
+
+def send_heartbeat():
+	#This will check for incoming messages 
+	#from other nodes and reply 
+
+	s = socket.socket()         
+	print ("Socket successfully created for Heartbeat")
+	port =  storage_ip[STORAGE_ID]["port"] 
+	s.bind(('', port))         
+	print ("Socket binded to %s" %(port)) 
+	  
+	# put the socket into listening mode 
+	s.listen(5)     
+
+	while True:
+		# Establish connection with client. 
+		c, addr = s.accept()     
+		print(f"\nGot connection from {addr}")
+
+		n = os.fork()
+		if n == 0:
+			print(F"Inside child process {os.getpid()}")
+			msg = _recv_msg(c, MSG_SIZE)	
+			print(msg)
+
+			if msg == None:
+				print(f"Didn't receive data from ip {addr}! [Timeout] ")
+				report_monitor(addr, None)
+
+			elif msg["type"] == "ALIVE": 
+				res = {"type": "ACK"}
+				_send_msg(c, res)
+
+			c.close()
+
+			print(f"Exiting from pid {os.getpid()} ..\n\n")
+			os._exit(1)
+
+
+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)
+
+
+if __name__ == "__main__":
+
+	p1 = threading.Thread(target=send_heartbeat)
+	p2 = threading.Thread(target=recv_gossip)
+	p3 = threading.Thread(target=read_write_pg_replication)
+	p1.start()
+	p2.start()
+	p3.start()

Test Code/Storage-1/__pycache__/storage_replication.cpython-38.pyc

@@ -0,0 +1,97 @@
+import socket
+import pickle
+import os
+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 replicate_pg(soc):
+	# This will be always true loop that will run in parallel
+
+	c, addr = soc.accept() 
+	print (f"Got connection from {addr} to do the replication")
+
+	n = os.fork()
+
+	if n==0:
+		msg = _recv_msg(c, MSG_SIZE)
+		print(msg)
+
+		if msg == None:
+			print(f"Didn't receive data! [Timeout] {addr}")
+			msg = {"type":"SEND AGAIN"}
+			_send_msg(c, msg)
+
+		elif msg["type"] == "REPLICATE":		
+			# Can do one thing here
+			# call the func that act as access this pg_id for the client
+			# then call other function that act as send this pg to osd to do a write
+			pg_id = msg["pg_id"]
+			osd_id = msg["osd_id"]
+
+			file = open("./data/"+pg_id, 'rb')
+
+			pg_b = file.read()
+			pg = pickle.loads(pg_b)
+
+			## Connect to this osd_id with new socket
+			new_soc = socket.socket()
+			print(f"Socket created to send request to SAVE data at OSD {osd_id}")
+
+			ip_add = storage_ip[osd_id]["ip"]
+			port = storage_ip[osd_id]["port"] + 10
+			# Remember replication is on normal port + 10
+
+			try :
+				new_soc.connect((ip_add, port))
+				print(f"Connection made with {ip_add} on {port}")
+
+				msg = {"type":"SAVE", "pg_id":pg_id, "pg": pg}
+				_send_msg(new_soc, msg)
+				print("Msg send to SAVE the data")
+
+				msg = _wait_recv_msg(new_soc, MSG_SIZE)
+				print("Msg received !")
+
+			except Exception as e:
+				print(e)
+
+			if msg == None:
+				res = {"type": "REPLICATION FAIL"}
+				_send_msg(c, res)
+				print("Fail Msg send to the monitor")
+
+			elif msg["type"] == "ACK":
+				res = {"type": "ACK"}
+				_send_msg(c, res)
+				print("Success Msg send to the monitor")
+
+			
+		elif msg["type"] == "SAVE":
+			# This will take request from other osd to save some data
+			# This is basic replication strategy 
+			pg_id = msg["pg_id"]
+			pg = msg["pg"]
+
+			try :
+				file = open("./data/"+ pg_id, 'wb')
+				print("Replicated data is saved")
+
+				pg_dump = pickle.dump(pg)
+				file.write(pg_dump)
+
+			except Exception as e:
+				print(e)
+
+			msg = {"type":"ACK"}
+			_send_msg(c, msg)
+			print("Write successful..send back Ack to the master osd")
+
+
+		else:
+			print("[ERROR] Check the code in replication")
+
+		c.close()
+
+		print(f"Exiting from pid {os.getpid()} ..")
+		os._exit(1)