cdht_2.py

import threading, time, sys, socket
# import pickle


host = '127.0.0.1'
base_port = 50000
is_quit = False
# peer_id = int(sys.argv[1])
# peer_port = int(sys.argv[1]) + base_port_num
# successor_1 = int(sys.argv[2]) + base_port_num
# successor_2 = int(sys.argv[3]) + base_port_num

# peer_id = sys.argv[1]
# peer_port = base_port + int(peer_id)
#
# peer_successor_1 = int(sys.argv[2])
# peer_successor_1_port = peer_successor_1 + base_port
#
# peer_successor_2 = int(sys.argv[3])
# peer_successor_2_port = peer_successor_2 + base_port

peer_id = 3
peer_port = base_port + int(peer_id)
peer_successor_1 = 4
peer_successor_1_port = peer_successor_1 + base_port
peer_successor_2 = 1
peer_successor_2_port = peer_successor_2 + base_port
# # print (peer_id)

# UDP socket
# UDPsocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# UDPsocket.bind((host, peer_port))

print("Listening...")


class UDPClient(threading.Thread):  # threading.Thread
    def __init__(self,):
        threading.Thread.__init__(self)
        self._running = True
        # self.client = client
        # self.host = host
        # self.successor_1 = successor_1
        # self.successor_2 = successor_2

    def terminate(self):
        self._running = False

    def run(self):
        global peer_successor_1, peer_successor_2, peer_successor_1_port, peer_successor_2_port
        client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        client.settimeout(1)
        seq = 0
        resp_seq = {}
        while True:
            if not self._running:
                break
            data = [seq, peer_id]
            client.sendto(bytes(data), (host, peer_successor_1_port))
            client.sendto(bytes(data), (host, peer_successor_2_port))
            seq += 1
            time.sleep(5)
            for _ in range(2):
                try:
                    data, addr = client.recvfrom(1024)
                    data = list(data)
                    resp_seq[data[1]] = data[0]
                    print('A ping response message was received from Peer {}'.format(data[1]))
                except:
                    pass
            for e in resp_seq:
                if seq - resp_seq[e] >= 5:
                    print('Peer {} is no longer alive.'.format(e))
                    data = [e]
                    if e == peer_successor_1:
                        response = TCPClient_request(host, peer_successor_2_port, data)
                        response = list(response)
                        peer_successor_1 = peer_successor_2
                        peer_successor_2 = response[0]

                    elif e == peer_successor_2:
                        response = TCPClient_request(host, peer_successor_1_port, data)
                        response = list(response)
                        peer_successor_2 = response[1]
                        # TCPClient_response(host, peer_successor_1_port)
                    peer_successor_1_port = peer_successor_1 + base_port
                    peer_successor_2_port = peer_successor_2 + base_port
                    seq = 0  #initialize the seq
                    resp_seq = {}
                    print('My first successor is now peer {}.'.format(peer_successor_1))
                    print('My second successor is now peer {}.'.format(peer_successor_2))
                    break
                    # TCPClient_request(host, peer_successor_1_port, data)




class UDPServer(threading.Thread):  # threading.Thread
    def __init__(self, ):
        threading.Thread.__init__(self)
        self._running = True
        # self.server = server
        # self.host = host

    def terminate(self):
        self._running = False

    def run(self):
        predecessor = []
        global predecessor_1, predecessor_2

        server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        server.bind((host, peer_port))
        while True:
            if not self._running:
                break
            try:
                data, addr = server.recvfrom(1024)
                data = list(data)
                # if data[0] == 0:  #this is a request message
                print('A ping request message was received from Peer {}'.format(data[1]))
                # time.sleep(5)
                if data[1] not in predecessor:
                    predecessor.append(data[1])
                    if len(predecessor) >= 3:
                        predecessor = [data[1]]
                    elif len(predecessor) == 2:
                        predecessor_1 = predecessor[0]
                        predecessor_2 = predecessor[1]
                        # print(predecessor_1, predecessor_2)
                    else:
                        pass
                data = [data[0], peer_id]
                server.sendto(bytes(data), addr)
            except:
                continue


def TCPClient_request(host, port, data):
    tcpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    tcpsocket.connect((host, port))
    tcpsocket.send(bytes(data))
    response = tcpsocket.recv(1024)
    tcpsocket.close()
    if response:
        return response

# def TCPClient_response(host, port):
#     tcpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
#     tcpsocket.connect((host, port))
#
#     tcpsocket.close()
#     return data


class TCPClient(threading.Thread):
    def __init__(self, ):
        threading.Thread.__init__(self)
        self._running = True
        # self.peer_successor_1_port = peer_successor_1_port

    def terminate(self):
        self._running = False

    def run(self):
        global is_quit
        while True:
            if not self._running:
                break
            try:
                request = input('')
                request = request.split()
                # print('request', request)
                if len(request) == 0:
                    continue
                if request[0].lower() == 'request':
                    if len(request) > 2 or len(request) <= 1 or int(request[1]) < 0 or int(request[1]) > 9999:
                        raise ValueError
                    filename = int(request[1])
                    if filename == peer_id:
                        print('File {} is here.'.format(filename))
                    else:
                        data = [filename, peer_id]
                        print('File request message for {} has been sent to my successor.'.format(filename))
                        TCPClient_request(host, peer_successor_1_port, data)
                elif request[0].lower() == 'quit':
                    data = [0, peer_id, peer_successor_1, peer_successor_2]
                    is_quit = True
                    # print('global flag changed')
                    TCPClient_request(host, predecessor_1 + 50000, data)
                    TCPClient_request(host, predecessor_2 + 50000, data)

            except ValueError:
                print('Wrong input.')


class TCPServer(threading.Thread):
    def __init__(self,):
        threading.Thread.__init__(self)
        self._running = True
        # self.peer_successor_1_port = peer_successor_1_port

    def terminate(self):
        self._running = False

    def run(self):
        global peer_successor_1, peer_successor_2, peer_successor_1_port, peer_successor_2_port
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.bind((host, peer_port))
        server.listen(1)
        while True:
            if not self._running:
                break
            connectionSocket, addr = server.accept()
            data = connectionSocket.recv(1024)
            data = list(data)
            if data[0] == 0: # a peer quit gracefully
                print('Peer {} will depart from the network.'.format(data[1]))
                if peer_successor_1 == data[1]:
                    peer_successor_1 = data[2]
                    peer_successor_2 = data[3]
                elif peer_successor_2 == data[1]:
                    peer_successor_2 = data[2]

                peer_successor_1_port = peer_successor_1 + base_port
                peer_successor_2_port = peer_successor_2 + base_port
                print('My first successor is now peer {}.'.format(peer_successor_1))
                print('My second successor is now peer {}.'.format(peer_successor_2))

            elif len(data) == 1: # a peer died without notifying others
                data = [peer_successor_1, peer_successor_2]
                connectionSocket.send(bytes(data))

            else:  # request file, data[0]=filename, data[1]=predecessor
                if data[0] % 256 <= peer_id and data[0] % 256 > data[1] \
                        or data[0] % 256 > data[1] and peer_id < data[1] \
                        or data[0] % 256 <= peer_id and data[1] > peer_id:
                    print('File {} is here.'.format(data[0]))
                else:
                    print('File {} is not stored here.'.format(data[0]))
                    TCPClient_request(host, peer_successor_1_port, [data[0], peer_id])
                    # connectionSocket.send(bytes([data[0], peer_id]))
                    print('File request message has been forwarded to my successor.')
            connectionSocket.close()


try:
    udpclient_threading = UDPClient()
    udpserver_threading = UDPServer()

    tcpclient_threading = TCPClient()
    tcpserver_threading = TCPServer()

    udpclient_threading.start()
    udpserver_threading.start()
    tcpclient_threading.start()
    tcpserver_threading.start()
    # print('all threadings start')
    while True:
        if is_quit:
            # print('is_quit')
            udpclient_threading.terminate()
            udpserver_threading.terminate()
            tcpclient_threading.terminate()
            tcpserver_threading.terminate()
            break


except:
    pass
    # continue