Final Running Replication Code

Author: harshit-soora

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

@@ -36,17 +36,17 @@
 """
 
 # Update all the stroage node IP everytime
-storage_1 = {"ip":"127.0.0.1", "port":7001}
-storage_2 = {"ip":"127.0.0.1", "port":7002}
-storage_3 = {"ip":"127.0.0.1", "port":7003}
-storage_4 = {"ip":"4.0.0.0", "port":7004}
+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":5002}
-monitor_2 = {"ip":"127.0.0.1", "port":5002}
+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/Monitor/__pycache__/monitor_replicate.cpython-38.pyc

@@ -1,5 +1,6 @@
 import socket
 import pickle
+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
 
@@ -12,26 +13,30 @@ def send_replicate_request(pg_id, master_osd, clone_osd):
 	# reserve a port on your computer in our 
 	ip_add = storage_ip[master_osd]["ip"] 
 	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(5)
 
-		msg = _wait_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] Master OSD 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
@@ -55,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:

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

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

Test Code/Monitor/info.py

@@ -36,17 +36,17 @@
 """
 
 # Update all the stroage node IP everytime
-storage_1 = {"ip":"127.0.0.1", "port":7001}
-storage_2 = {"ip":"127.0.0.1", "port":7002}
-storage_3 = {"ip":"127.0.0.1", "port":7003}
-storage_4 = {"ip":"4.0.0.0", "port":7004}
+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":5002}
-monitor_2 = {"ip":"127.0.0.1", "port":5002}
+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/Monitor/monitor_replicate.py

@@ -9,7 +9,7 @@ def replicate_pg(soc):
 	# This will be always true loop that will run in parallel
 
 	c, addr = soc.accept() 
-	print ('Got connection from', addr )
+	print (f"Got connection from {addr} to do the replication")
 
 	n = os.fork()
 
@@ -19,6 +19,8 @@ def replicate_pg(soc):
 
 		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
@@ -39,17 +41,19 @@ def replicate_pg(soc):
 			ip_add = storage_ip[osd_id]["ip"]
 			port = storage_ip[osd_id]["port"]
 
-			new_soc.connect((ip_add, port))
-			print(f"Connection made with {ip_add} on {port}")
+			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 = {"type":"SAVE", "pg_id":pg_id, "pg": pg}
+				_send_msg(new_soc, msg)
+				print("Msg send to SAVE the data")
 
-			time.sleep(3)
+				msg = _wait_recv_msg(new_soc, MSG_SIZE)
+				print("Msg received !")
 
-			msg = _recv_msg(new_soc, MSG_SIZE)
-			print("Msg received !")
+			except Exception as e:
+				print(e)
 
 			if msg == None:
 				res = {"type": "REPLICATION FAIL"}