DISTRIBUTED

SYSTEM
LAB FILE
BCE- C 661

Submitted By: Submitted To:

MAHBOOB RAZA Mr.Abhishant sharma
B.Tech, CSE, VI Sem Assistant Professor
Roll no: 226301124 CSE Department, FET, GKV
DEPARTMENT OF COMPUTER SCIENCE &
ENGINEERING
FACULTY OF ENGINEERING AND TECHNOLOGY
GURUKUL KANGRI UNIVERSITY
 Certificate by Student

This is to certify that I, MAHBOOB RAZA student of CSE Third Year (Session
2022-2026) has completed the enclosed program all by myself and have not
submitted these programs elsewhere for any other purpose.

Name : MAHBOOB RAZA

Branch : CSE, VI th Sem
Roll No : 226301124

Certificate by Subject Teacher

This is to certify that MAHBOOB RAZA who is a student of CSE Third year
(Session 2022- 2026) has completed all the programs enclosed in this file under
my supervision and has not submitted these programs elsewhere for award of any
degree as per best of my knowledge.

Mr. Abhishant sharma

Assistant Professor
Department of Computer Science & Engineering
Faculty of Engineering and Technology
Gurukula Kangri Deemed to be University, Haridwar
 INDEX
S.no Program Title Page
number
1. Non-token based Mutual Exclusion (Ricart-Agrawala Algorithm)

2. Lamport’s Logical Clock

3. Edge Chasing Deadlock Detection

4. Locking Algorithm

5. Remote Method Invocation ( RMI )

6. Remote Procedure Calls ( RPC )

7. Chat Server

8. Termination Detection (Diffusion)

1. Non-token based Mutual Exclusion (Ricart-Agrawala Algorithm)

Steps

1. send_request()

2. wait for replies from all

3. enter_critical_section()

4. send_reply_to_others()

Program :---
import threading

import time

import random

class Process(threading.Thread):

def __init__(self, pid, processes):

super().__init__()

self.pid = pid

self.clock = 0

self.queue = []

self.processes = processes

def request_cs(self):

self.clock += 1

print(f"Process {self.pid} requesting CS at time {self.clock}")

for p in self.processes:

if p.pid != self.pid:

p.receive_request(self.pid, self.clock)

def receive_request(self, pid, timestamp):

self.clock = max(self.clock, timestamp) + 1

print(f"Process {self.pid} received request from {pid} at time {timestamp}")

self.queue.append((timestamp, pid))

def run(self):
 time.sleep(random.uniform(1, 2)) # Simulate delay before requesting CS

self.request_cs()

time.sleep(1) # Simulate being in the critical section

print(f"Process {self.pid} exiting CS")

# Create processes

processes = []

for i in range(3):

processes.append(Process(i, [])) # Initialize with empty process list

# Set the full list of processes now that it's created

for p in processes:

p.processes = processes

# Start all threads

for p in processes:

p.start()

# Wait for all threads to finish

for p in processes:

p.join()

Output :-

2. Lamport’s Logical Clock

Steps
clock += 1

clock = max(clock, received_clock) + 1

Program :-

class LamportClock:

def __init__(self):

self.time = 0

def tick(self):

self.time += 1

def send_event(self):

self.tick()

return self.time

def receive_event(self, sender_time):

self.time = max(self.time, sender_time) + 1

# Example usage

p1 = LamportClock()

p2 = LamportClock()

print("P1 sends message")

t1 = p1.send_event()

print("P2 receives message")

p2.receive_event(t1)

print(f"Final P1: {p1.time}, P2: {p2.time}")

output :-
3. Edge Chasing Deadlock Detection
Steps

if receives_probe and it is blocked:

forward probe to next waiting process

if receives_own_probe:

deadlock detected

Program :-
wait_for = {

0: 1,

1: 2,

2: 0 # forms a cycle (deadlock)

def detect_deadlock(initiator):

visited = set()

def forward(pid):

if pid in visited:

return

visited.add(pid)

if wait_for.get(pid) is None:

return

next_pid = wait_for[pid]

if next_pid == initiator:
 print("Deadlock detected!")

else:

forward(next_pid)

forward(initiator)

# Start deadlock detection from process 0

detect_deadlock(0)

Output :-

4. Locking Algorithm
Steps

# Simple Lock

lock = False

if not lock:

lock = True

# critical section

lock = False

Program :-
import threading

import time # Import time module for sleep

lock = threading.Lock()
def critical_section(pid):

with lock:

print(f"Process {pid} entering critical section")

time.sleep(1) # Simulate work in the critical section

print(f"Process {pid} exiting critical section")

# Create and start threads

threads = [threading.Thread(target=critical_section, args=(i,)) for i in range(3)]

for t in threads:

t.start()

# Wait for all threads to finish

for t in threads:

t.join()

Output :-

5. RMI (Remote Method Invocation)

Steps

Install pyro5

Setup RMI server

Setup RMI client

Program :-

RMI SERVER

# rmi_server.py
import Pyro5.api

@Pyro5.api.expose

class Calculator:

def add(self, a, b):

return a + b

daemon = Pyro5.api.Daemon() # Start the Pyro daemon

uri = daemon.register(Calculator) # Register the class as a Pyro object

print("RMI Server is ready. Object URI =", uri)

daemon.requestLoop()

RMI CLIENT

# rmi_client.py

import Pyro5.api

uri = input("Enter the URI of the remote object: ") # Paste the printed URI from the server

calculator = Pyro5.api.Proxy(uri) # Create a proxy

print("5 + 3 =", calculator.add(5, 3))

OUTPUT:

6.RPC (Remote Procedure Call)

Steps

First Setup RPC server

After that RPC client

Program :-

RPC SERVER

# rpc_server.py

from xmlrpc.server import SimpleXMLRPCServer

def add(x, y):

return x + y

server = SimpleXMLRPCServer(("localhost", 8000))

print("RPC Server listening on port 8000...")

server.register_function(add, "add")

server.serve_forever()

RPC CLIENT

# rpc_client.py

import xmlrpc.client

proxy = xmlrpc.client.ServerProxy("http://localhost:8000/")

result = proxy.add(10, 20)

print("10 + 20 =", result)

OUTPUT:
7. Chat Server
Steps

# Server

socket.bind()

socket.listen()

conn, addr = socket.accept()

conn.send("Hello client".encode())

# Client

socket.connect()

msg = socket.recv(1024).decode()

Program :-
# server.py

import socket

import threading

clients = []

def handle_client(client):

while True:

msg = client.recv(1024).decode()

if not msg: break

for c in clients:

if c != client:

c.send(msg.encode())

server = socket.socket()
server.bind(('localhost', 5555))

server.listen()

print("Chat server started...")

while True:

client, addr = server.accept()

clients.append(client)

threading.Thread(target=handle_client, args=(client,)).start()

# client.py

import socket

import threading

client = socket.socket()

client.connect(('localhost', 5555))

def listen():

while True:

msg = client.recv(1024).decode()

print("Received:", msg)

threading.Thread(target=listen).start()

while True:

msg = input()

client.send(msg.encode())

Ouput :-
8. Termination Detection (Diffusion)

Program :-
class Node:

def __init__(self, pid):

self.pid = pid

self.children = []

self.ack_received = 0

def send_work(self):

print(f"Node {self.pid} sending work to children")

for child in self.children:

child.send_work()

self.receive_ack()

def receive_ack(self):

self.ack_received += 1

print(f"Node {self.pid} received ACK")

 if self.pid == 0 and self.ack_received == len(self.children):

print("Termination Detected at root")

# Create nodes

root = Node(0)

child1 = Node(1)

child2 = Node(2)

# Assign tree structure

root.children = [child1, child2]

# Start

root.send_work()

Output :-