Skip to content
GitLab
Explore
Sign in
Primary navigation
Search or go to…
Project
D
DBE-DS-Group1-WS24
Manage
Activity
Members
Labels
Plan
Issues
Issue boards
Milestones
Wiki
Requirements
Code
Merge requests
Repository
Branches
Commits
Tags
Repository graph
Compare revisions
Snippets
Locked files
Build
Pipelines
Jobs
Pipeline schedules
Test cases
Artifacts
Deploy
Releases
Package registry
Container Registry
Model registry
Operate
Environments
Terraform modules
Monitor
Incidents
Analyze
Value stream analytics
Contributor analytics
CI/CD analytics
Repository analytics
Code review analytics
Issue analytics
Insights
Model experiments
Help
Help
Support
GitLab documentation
Compare GitLab plans
Community forum
Contribute to GitLab
Provide feedback
Keyboard shortcuts
?
Snippets
Groups
Projects
Show more breadcrumbs
Robin Leber
DBE-DS-Group1-WS24
Commits
c5f75fbd
Commit
c5f75fbd
authored
1 year ago
by
Quoc Nguyen Dao
Browse files
Options
Downloads
Patches
Plain Diff
Server Server discovery fix
parent
a61a8d32
No related branches found
Branches containing commit
No related tags found
1 merge request
!2
leader election und heartbeat
Changes
2
Hide whitespace changes
Inline
Side-by-side
Showing
2 changed files
client.py
+33
-44
33 additions, 44 deletions
client.py
server.py
+120
-70
120 additions, 70 deletions
server.py
with
153 additions
and
114 deletions
client.py
+
33
−
44
View file @
c5f75fbd
...
...
@@ -2,11 +2,9 @@ import socket
import
threading
from
datetime
import
datetime
# Broadcast address and port for client broadcasts
BROADCAST_IP_CLIENT
=
"
192.168.0.255
"
BROADCAST_PORT_CLIENT
=
5973
# Broadcast address and port
BROADCAST_IP
=
"
192.168.0.255
"
BROADCAST_PORT
=
5973
# Local host information
MY_HOST
=
socket
.
gethostname
()
...
...
@@ -19,62 +17,53 @@ class Client():
self
.
currentLeader
=
''
self
.
server_socket
=
None
# print the current date and time
def
printwt
(
self
,
msg
):
current_date_time
=
datetime
.
now
().
strftime
(
'
%Y-%m-%d %H:%M:%S
'
)
print
(
f
'
[
{
current_date_time
}
]
{
msg
}
'
)
# dynamic discoverey: client sends request to server group and gets the IP of server as reply
def
BroadcastSendAndReceive
(
self
):
message
=
'
New client wants to connect:
'
+
MY_IP
# Create a UDP socket
client_
broadcast_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
broadcast_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
# Send message on broadcast address
client_
broadcast_socket
.
sendto
(
str
.
encode
(
message
),
(
BROADCAST_IP
_CLIENT
,
BROADCAST_PORT
_CLIENT
))
broadcast_socket
.
sendto
(
str
.
encode
(
message
),
(
BROADCAST_IP
,
BROADCAST_PORT
))
# Send broadcast message
self
.
printwt
(
"
Sent my IP to server group
"
)
# while True:
# Receive server leader's IP
try
:
data
,
addr
=
client_broadcast_socket
.
recvfrom
(
1024
)
if
data
:
self
.
current_leader
=
data
.
decode
()
self
.
printwt
(
f
'
Got Server Leader address:
{
self
.
current_leader
}
'
)
# receive reply data (server IP) from the other participants
# reply, addr = client_broadcast_socket.recvfrom(1024)
# if reply:
# # decode received data
# reply_addr = reply.decode()
# self.currentLeader = reply_addr
# self.printwt(f'Got Leader address: {self.currentLeader}')
# Connect to the server using TCP
self
.
server_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_STREAM
)
self
.
server_socket
.
connect
((
self
.
currentLeader
,
5555
))
print
(
"
You have entered the chat room
"
)
# Starte einen Thread, um Nachrichten zu empfangen
receive_thread
=
threading
.
Thread
(
target
=
self
.
receive_messages
)
receive_thread
.
start
()
# Main thread for sending messages
while
True
:
message
=
input
()
timestamped_message
=
f
'
[
{
datetime
.
now
().
strftime
(
"
%Y-%m-%d %H
:
%
M
:
%
S
"
)
}
]
{
message
}
'
self
.
server_socket
.
send
(
timestamped_message
.
encode
(
'
utf-8
'
))
while
True
:
try
:
# receive reply data (server IP) from the other participants
reply
,
addr
=
broadcast_socket
.
recvfrom
(
1024
)
if
reply
:
# decode received data
reply_addr
=
reply
.
decode
()
self
.
currentLeader
=
reply_addr
self
.
printwt
(
f
'
Got Leader address:
{
self
.
currentLeader
}
'
)
# Connect to the server using TCP
self
.
server_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_STREAM
)
self
.
server_socket
.
connect
((
self
.
currentLeader
,
5555
))
print
(
"
You have entered the chat room
"
)
# Starte einen Thread, um Nachrichten zu empfangen
receive_thread
=
threading
.
Thread
(
target
=
self
.
receive_messages
)
receive_thread
.
start
()
# Haupt-Thread zum Senden von Nachrichten
while
True
:
message
=
input
()
timestamped_message
=
f
'
[
{
datetime
.
now
().
strftime
(
"
%Y-%m-%d %H
:
%
M
:
%
S
"
)
}
]
{
message
}
'
self
.
server_socket
.
send
(
timestamped_message
.
encode
(
'
utf-8
'
))
except
socket
.
timeout
:
pass
except
socket
.
timeout
:
pass
def
receive_messages
(
self
):
while
True
:
...
...
This diff is collapsed.
Click to expand it.
server.py
+
120
−
70
View file @
c5f75fbd
...
...
@@ -3,10 +3,13 @@ import threading
from
datetime
import
datetime
import
time
# Listening port for client broadcasts
BROADCAST_PORT_CLIENT
=
5973
# Listening port for server broadcasts
BROADCAST_PORT_SERVER
=
5974
BROADCAST_IP
=
"
192.168.0.255
"
# Listening port Server Discovery
SERVER_BROADCAST_PORT
=
5000
# Listening port Client Discovery
CLIENT_BROADCAST_PORT
=
5973
# Local host information
MY_HOST
=
socket
.
gethostname
()
...
...
@@ -14,83 +17,142 @@ MY_IP = socket.gethostbyname(MY_HOST)
class
Server
():
def
__init__
(
self
):
self
.
group_members
=
{}
# Dictionary to store group members with their IPs as keys
self
.
leader_IP
=
''
# fix the leader IP
self
.
serverList
=
[]
# list if servers and their addresses
self
.
clients
=
[]
self
.
is_leader
=
False
# Flag to indicate if this server is the leader
self
.
informServer
=
False
self
.
isLeader
=
False
# New variable to track if the server is the leader
def
printwt
(
self
,
msg
):
current_date_time
=
datetime
.
now
().
strftime
(
'
%Y-%m-%d %H:%M:%S
'
)
print
(
f
'
[
{
current_date_time
}
]
{
msg
}
'
)
def
print_group_view
(
self
):
print
(
"
Group view is:
"
,
self
.
serverList
)
def
announce_leader
(
self
):
# Announce leader to clients
while
True
:
if
self
.
is_leader
:
self
.
broadcast_leader
()
time
.
sleep
(
5
)
#This function enables the server to listen to the server broadcast port and reply the ip address
def
BroadcastListenAndReply
(
self
):
def
broadcast_leader
(
self
):
# Broadcast the leader's IP to clients
for
client_socket
in
self
.
clients
:
try
:
client_socket
.
send
(
self
.
leader_IP
.
encode
())
except
:
self
.
clients
.
remove
(
client_socket
)
# if my IP is not in the server list add it
if
MY_IP
not
in
self
.
serverList
:
self
.
serverList
.
append
(
MY_IP
)
def
ListenForServerBroadcastAndReply
(
self
):
# Listen to server broadcasts and reply with own IP
server_broadcast_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
server_broadcast_socket
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_REUSEADDR
,
1
)
server_broadcast_socket
.
bind
((
MY_IP
,
BROADCAST_PORT_SERVER
))
# create socket bind to server address
broadcast_listen_sock
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
broadcast_listen_sock
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_REUSEADDR
,
1
)
broadcast_listen_sock
.
bind
((
''
,
SERVER_BROADCAST_PORT
))
while
True
:
data
,
addr
=
server_broadcast_socket
.
recvfrom
(
1024
)
if
data
:
# Decode received data (new server's IP)
new_server_ip
=
data
.
decode
()
self
.
group_members
[
new_server_ip
]
=
datetime
.
now
().
strftime
(
'
%Y-%m-%d %H:%M:%S
'
)
try
:
data
,
address
=
broadcast_listen_sock
.
recvfrom
(
1024
)
# Respond to the new member with own IP
server_broadcast_socket
.
sendto
(
MY_IP
.
encode
(),
addr
)
self
.
printwt
(
f
'
New
group member found
:
{
new
_s
erver_
ip
}
'
)
self
.
print_group_members
()
if
data
and
not
self
.
informServer
:
newServer_address
=
data
.
decode
(
)
self
.
printwt
(
f
'
New
participant wants to connect
:
{
new
S
erver_
address
}
'
)
self
.
isLeader
=
False
# if the decoded address is not in the server list add it and print the list
def
print_group_members
(
self
):
self
.
printwt
(
"
Current group members:
"
)
for
member
,
timestamp
in
self
.
group_members
.
items
():
self
.
printwt
(
f
'
{
member
}
(joined at
{
timestamp
}
)
'
)
if
newServer_address
not
in
self
.
serverList
:
self
.
serverList
.
append
(
newServer_address
)
reply_message
=
MY_IP
broadcast_listen_sock
.
sendto
(
str
.
encode
(
reply_message
),
address
)
# self.SendClientListUpdate(newServer_address)
self
.
informServer
=
True
self
.
printwt
(
'
Replied my IP to new participant
'
)
except
socket
.
timeout
:
break
def
become_leader
(
self
):
# Announce to the group that this server is the leader
self
.
is_leader
=
True
self
.
leader_IP
=
MY_IP
self
.
broadcast_leader
()
time
.
sleep
(
1
)
#this function enables the server to send a broadcast to the server group and receive the answers of existing members
def
BroadcastSendAndReceive
(
self
):
# create socket
broadcast_group
=
(
BROADCAST_IP
,
SERVER_BROADCAST_PORT
)
broadcast_send_sock
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
broadcast_send_sock
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_BROADCAST
,
1
)
broadcast_send_sock
.
settimeout
(
2
)
# ...
message
=
MY_IP
broadcast_send_sock
.
sendto
(
message
.
encode
(),
broadcast_group
)
self
.
printwt
(
"
Sent my IP to server group
"
)
# if my IP is not in the server list add it
if
MY_IP
not
in
self
.
serverList
:
self
.
serverList
.
append
(
MY_IP
)
# listen for IPs from existing servers
maxLoop
=
5
currentLoop
=
0
# Anzahl der eingehenden Antworten initialisieren
num_responses
=
0
while
currentLoop
<
maxLoop
:
while
True
:
currentLoop
+=
1
try
:
# receive reply data from the other participants
reply
,
address
=
broadcast_send_sock
.
recvfrom
(
1024
)
if
reply
:
reply_address
=
reply
.
decode
()
# if reply address is not in the server list, add it
if
reply_address
not
in
self
.
serverList
:
self
.
serverList
.
append
(
reply_address
)
# Erhöhe die Anzahl der eingehenden Antworten
num_responses
+=
1
except
socket
.
timeout
:
break
if
num_responses
==
1
:
broadcast_send_sock
.
close
()
self
.
isLeader
=
True
self
.
leader_IP
=
MY_IP
# Hier wird die IP-Adresse des Leaders zugewiesen
self
.
printwt
(
f
'
I am the only server in the system, so the leader IP is:
{
self
.
leader_IP
}
'
)
time
.
sleep
(
1
)
self
.
print_group_view
()
# Listen to client broadcast (request) and reply with Server IP
def
l
isten
_for_client_broadcasts
(
self
):
def
L
isten
ForClientAndReply
(
self
):
# Create a UDP socket
c
li
ent_broadca
st_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
list
en
_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_DGRAM
)
# Set the socket to broadcast and enable reusing addresses
client_broadcast_socket
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_REUSEADDR
,
1
)
listen_socket
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_BROADCAST
,
1
)
listen_socket
.
setsockopt
(
socket
.
SOL_SOCKET
,
socket
.
SO_REUSEADDR
,
1
)
# Bind socket to address and port
c
li
ent_broadca
st_socket
.
bind
((
MY_IP
,
BROADCAST_PORT
_CLIENT
))
list
en
_socket
.
bind
((
MY_IP
,
CLIENT_
BROADCAST_PORT
))
print
(
"
Listening to broadcast messages
"
)
#
print("Listening to broadcast messages")
# Receiving broadcast massage
while
True
:
data
,
addr
=
c
li
ent_broadca
st_socket
.
recvfrom
(
1024
)
data
,
addr
=
list
en
_socket
.
recvfrom
(
1024
)
if
data
and
self
.
is_leader
:
client_broadcast_socket
.
sendto
(
self
.
leader_IP
.
encode
(),
addr
)
if
data
:
self
.
printwt
(
data
.
decode
())
# if Iam the leader, answer the client including my IP
if
MY_IP
==
self
.
leader_IP
:
reply_message
=
MY_IP
listen_socket
.
sendto
(
str
.
encode
(
reply_message
),
addr
)
self
.
printwt
(
'
Replied my IP to new client
'
)
def
handle_client
(
self
,
client_socket
,
client_address
):
...
...
@@ -120,32 +182,20 @@ class Server():
if
__name__
==
'
__main__
'
:
server
=
Server
()
thread2
=
threading
.
Thread
(
target
=
server
.
BroadcastListenAndReply
)
thread2
.
start
()
# Start a thread to announce the leader to clients
leader_thread
=
threading
.
Thread
(
target
=
server
.
announce_leader
)
leader_thread
.
start
()
# Start a thread to listen for server broadcasts
server_broadcast_thread
=
threading
.
Thread
(
target
=
server
.
ListenForServerBroadcastAndReply
)
server_broadcast_thread
.
start
()
client_broadcast_thread
=
threading
.
Thread
(
target
=
server
.
listen_for_client_broadcasts
)
client_broadcast_thread
.
start
()
thread3
=
threading
.
Thread
(
target
=
server
.
BroadcastSendAndReceive
)
thread3
.
start
()
thread1
=
threading
.
Thread
(
target
=
server
.
ListenForClientAndReply
)
thread1
.
start
()
# Socket erstellen und binden
server_socket
=
socket
.
socket
(
socket
.
AF_INET
,
socket
.
SOCK_STREAM
)
server_socket
.
bind
((
MY_IP
,
5555
))
server_socket
.
listen
(
5
)
# Check if this server is the first in the group and become the leader
time
.
sleep
(
3
)
# Wait for other servers to potentially announce themselves
if
not
server
.
group_members
:
server
.
become_leader
()
while
True
:
client_socket
,
client_address
=
server_socket
.
accept
()
...
...
This diff is collapsed.
Click to expand it.
Preview
0%
Loading
Try again
or
attach a new file
.
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Save comment
Cancel
Please
register
or
sign in
to comment