4
\$\begingroup\$

What am I doing?

UDP/TCP server in Python (2.7 / 3.7)

How am I doing?

Using threads

Where am I going to execute it?

In a virtual machine running linux in AWS or Google Cloud or Azure ()

What is going to received?

Streaming data simultaneously from many devices (we could say a few thousand devices)

What contains the incoming data?

A JSON or CSV line of text (no more than 50 fields / 3000 characters)

What this server need?

Low latency to process all the incoming data in the minimum possible time

What am I going to do with the data?

Send one by one (just receive it) to a streaming module in cloud to analyze it

Will I need to communicate to the device?

Yes. Is not implemented now, but for the future, each time a device send a message to the server, the server will check some parameters of the device


I have read many articles and question about threads vs multiprocessing in python. It looks like multiprocessing is "better" in performance than threads, but I'm not sure if for this use of case with high IO, threads would be more appropriate


The code (actually for Azure)

import json
import threading
import SocketServer
from azure.eventhub import EventHubClient, EventData

ADDRESS = ""
USER = ""
KEY = ""

class ThreadedUDPRequestHandler(SocketServer.BaseRequestHandler):

    def handle(self):
        data = self.request[0].strip()
        current_thread = threading.current_thread()
        print("Thread: {} client: {}, wrote: {}".format(current_thread.name, self.client_address, data))
        Split = threading.Thread(target=ParseIncomingData,args=(data,))
        Split.start()



class ThreadedUDPServer(SocketServer.ThreadingMixIn, SocketServer.UDPServer):
    pass


def SendToEventHub(data):
    if not ADDRESS:
        raise ValueError("No EventHubs URL supplied.")
    client = EventHubClient(ADDRESS, debug=False, username=USER, password=KEY)
    sender = client.add_sender(partition="0")
    client.run()
    try:
        print("Thread: {}, Sending message: {}".format(threading.current_thread().name,data))
        sender.send(EventData(data))
    except:
        raise
    finally:
        client.stop()


def ParseIncomingData(message):
    sender = threading.Thread(target=SendToEventHub, args=(json.dumps(message),))
    sender.start()


if __name__ == "__main__":
    HOST, PORT = "0.0.0.0", 6071
    try:
        serverUDP = ThreadedUDPServer((HOST, PORT), ThreadedUDPRequestHandler)
        server_thread_UDP = threading.Thread(target=serverUDP.serve_forever)
        server_thread_UDP.daemon = True
        server_thread_UDP.start()
        serverUDP.serve_forever()
    except (KeyboardInterrupt, SystemExit):
        serverUDP.shutdown()
        serverUDP.server_close()
        exit()

With high input messages, this is an example of the output

enter image description here


Testing with asyncio

Test 1

I have never worked with asyncio so I am sure the code is not good, but I would like to know which would be the way to use it

def MDIParseIncomingDataAzure(message):
    print("Thread: {}, MDIParseIncomingDataAzure".format(current_thread().name))
    SendToEventHub(payload)


def SendToEventHub(data):
        print("Thread: {}, SendToEventHub".format(current_thread().name))


class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        print('Thread: {}. datagram_received '.format(current_thread().name))
        MDIParseIncomingDataAzure(data)



async def main():
    print("Starting UDP server")
    loop = asyncio.get_running_loop()

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(),
        local_addr=('0.0.0.0', 6071))

    try:
        await asyncio.sleep(3600)  # Serve for 1 hour.
    finally:
        transport.close()


asyncio.run(main())

The output of this code (which run slow step by step)

test1


Test 2

def MDIParseIncomingDataAzure(message,loop):
    asyncio.set_event_loop(loop)
    print("Thread: {}, MDIParseIncomingDataAzure".format(current_thread().name))
    new_loop = asyncio.new_event_loop()
    t = Thread(target=SendToEventHub, args=(payload, new_loop,))
    t.start()
    loop.run_forever()


def SendToEventHub(data,loop):
    asyncio.set_event_loop(loop)
    print("Thread: {}, SendToEventHub".format(current_thread().name))


class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        print('Thread: {}. datagram_received '.format(current_thread().name))
        new_loop = asyncio.new_event_loop()
        t = Thread(target=MDIParseIncomingDataAzure, args=(data,new_loop,))
        t.start()



async def main():
    print("Starting UDP server")
    loop = asyncio.get_running_loop()

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(),
        local_addr=('0.0.0.0', 6071))

    try:
        await asyncio.sleep(3600)  # Serve for 1 hour.
    finally:
        transport.close()


asyncio.run(main())

The output:

test2

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.