Remote image
+
+This domain is for use in illustrative examples in documents. You may use this + domain in literature without prior coordination or asking for permission.
+
+
+
+
+
diff --git a/public/ulyssis.png b/public/ulyssis.png
new file mode 100644
index 0000000..14819f5
Binary files /dev/null and b/public/ulyssis.png differ
diff --git a/server.py b/server.py
index 0d5c7ce..fea876d 100644
--- a/server.py
+++ b/server.py
@@ -1,39 +1,90 @@
#!/usr/bin/env python3
-
+import argparse
+import logging
+import multiprocessing
import socket
+import sys
-# socket heeft een listening and accept method
-import time
+from server.httpserver import HTTPServer
-SERVER = "127.0.0.1" #dynamisch fixen in project
-PORT = 5055
-server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-ADDR = (SERVER, PORT) # hier wordt de socket gebonden aan mijn IP adres, dit moet wel anders
-server.bind(ADDR) # in het project gebeuren
+def main():
+ parser = argparse.ArgumentParser(description='HTTP Server')
+ parser.add_argument("--verbose", "-v", action='count', default=0, help="Increase verbosity level of logging")
+ parser.add_argument("--workers", "-w",
+ help="The amount of worker processes. This is by default based on the number of cpu threads.",
+ type=int)
+ parser.add_argument("--port", "-p", help="The port to listen on", default=8000)
+ arguments = parser.parse_args()
-HEADER = 64 # maximum size messages
-FORMAT = 'utf-8' # recieving images through this format does not work
-DISCONNECT_MESSAGE = "DISCONNECT!" # special message for disconnecting client and server
+ logging_level = logging.ERROR - (10 * arguments.verbose)
+ logging.basicConfig(level=logging_level)
+ logging.debug("Arguments: %s", arguments)
-# function for starting server
-def start():
- pass
- server.listen()
- while True: # infinite loop in which server accept incoming connections, we want to run it forever
- conn, addr = server.accept() # Server blocks untill a client connects
- print("new connection: ", addr[0], " connected.")
- connected = True
- while connected: # while client is connected, we want to recieve messages
- msg = conn.recv(HEADER).decode(FORMAT).rstrip() # Argument is maximum size of msg (in project look into details of accp), decode is for converting bytes to strings, rstrip is for stripping messages for special hidden characters
- print("message: ", msg)
- for i in range(0,10):
- conn.send(b"test")
- time.sleep(1)
+ # Set workers
+ if arguments.workers:
+ workers = int(arguments.workers)
+ else:
+ workers = multiprocessing.cpu_count()
- break
- print("close connection ", addr[0], " disconnected.")
- conn.close()
+ # Set port
+ if arguments.port:
+ port = int(arguments.port)
+ else:
+ port = 8000
-print("server is starting ... ")
-start()
\ No newline at end of file
+ # Get hostname and address
+ hostname = socket.gethostname()
+ address = socket.gethostbyname(hostname)
+ server = HTTPServer(address, port, workers, logging_level)
+ server.start()
+
+
+try:
+ if __name__ == '__main__':
+ main()
+except Exception as e:
+ print("[ABRT] Internal error: " + str(e), file=sys.stderr)
+ logging.debug("Internal error", exc_info=e)
+ sys.exit(70)
+
+# import socket
+#
+# # Get hostname and address
+# hostname = socket.gethostname()
+# address = socket.gethostbyname(hostname)
+#
+# # socket heeft een listening and accept method
+#
+# SERVER = "127.0.0.1" # dynamisch fixen in project
+# PORT = 5055
+# server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+#
+# ADDR = (SERVER, PORT) # hier wordt de socket gebonden aan mijn IP adres, dit moet wel anders
+# server.bind(ADDR) # in het project gebeuren
+#
+# HEADER = 64 # maximum size messages
+# FORMAT = 'utf-8'
+# DISCONNECT_MESSAGE = "DISCONNECT!" # special message for disconnecting client and server
+#
+#
+# # function for starting server
+# def start():
+# pass
+# server.listen()
+# while True: # infinite loop in which server accept incoming connections, we want to run it forever
+# conn, addr = server.accept() # Server blocks untill a client connects
+# print("new connection: ", addr[0], " connected.")
+# connected = True
+# while connected: # while client is connected, we want to recieve messages
+# msg = conn.recv(HEADER).decode(
+# FORMAT).rstrip() # Argument is maximum size of msg (in project look into details of accp), decode is for converting bytes to strings, rstrip is for stripping messages for special hidden characters
+# print("message: ", msg)
+# if msg == DISCONNECT_MESSAGE:
+# connected = False
+# print("close connection ", addr[0], " disconnected.")
+# conn.close()
+#
+#
+# print("server is starting ... ")
+# start()
diff --git a/server/RequestHandler.py b/server/RequestHandler.py
new file mode 100644
index 0000000..d3ce433
--- /dev/null
+++ b/server/RequestHandler.py
@@ -0,0 +1,57 @@
+import logging
+from logging import Logger
+from socket import socket
+from typing import Union
+from urllib.parse import ParseResultBytes, ParseResult
+
+from httplib import parser
+from httplib.exceptions import MethodNotAllowed, BadRequest
+from httplib.httpsocket import HTTPSocket
+from httplib.retriever import Retriever
+
+METHODS = ("GET", "HEAD", "PUT", "POST")
+
+
+class RequestHandler:
+ conn: HTTPSocket
+ logger: Logger
+
+ def __init__(self, conn: socket, logger, host):
+ self.conn = HTTPSocket(conn, host)
+ self.logger = logger
+
+ def listen(self):
+ self.logger.debug("Parsing request line")
+ logging.debug("test logger")
+ (method, target, version) = parser.parse_request_line(self.conn)
+ headers = parser.parse_request_headers(self.conn)
+
+ self._validate_request(method, target, version, headers)
+
+ self.logger.debug("Parsed request-line: version: %s, target: %r", method, target)
+ headers = parser.get_headers(self.conn)
+ self.logger.debug("Parsed headers: %r", headers)
+ retriever = Retriever.create(self.conn, headers)
+ body = retriever.retrieve()
+
+ self.logger.debug("body: %r", body)
+
+ def _check_request_line(self, method: str, target: Union[ParseResultBytes, ParseResult], version):
+
+ if method not in METHODS:
+ raise MethodNotAllowed(METHODS)
+
+ # only origin-form and absolute-form are allowed
+ if len(target.path) < 1 or target.path[0] != "/" or \
+ target.netloc not in ("http", "https") and target.hostname == "":
+ raise BadRequest()
+
+ if version not in ("1.0", "1.1"):
+ raise BadRequest()
+
+ def _validate_request(self, method, target, version, headers):
+
+ self._check_request_line(method, target, version)
+
+ if version == "1.1" and "host" not in headers:
+ raise BadRequest()
diff --git a/server/httpserver.py b/server/httpserver.py
new file mode 100644
index 0000000..f4b3ebd
--- /dev/null
+++ b/server/httpserver.py
@@ -0,0 +1,94 @@
+import logging
+import multiprocessing as mp
+import socket
+import time
+from multiprocessing.context import Process
+from multiprocessing.queues import Queue
+from multiprocessing.synchronize import Event
+
+from server import worker
+
+
+class HTTPServer:
+ address: str
+ port: int
+ workers = []
+ worker_count: int
+ server: socket
+
+ _dispatch_queue: Queue
+ _stop_event: Event
+
+ def __init__(self, address: str, port: int, worker_count, logging_level):
+ self.address = address
+ self.port = port
+ self.worker_count = worker_count
+ self.logging_level = logging_level
+
+ mp.set_start_method("spawn")
+ self._dispatch_queue = mp.Queue()
+ self._stop_event = mp.Event()
+
+ def start(self):
+ try:
+ self.__do_start()
+ except KeyboardInterrupt:
+ self.__shutdown()
+
+ def __do_start(self):
+ # Create socket
+
+ self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.server.bind((self.address, self.port))
+
+ self.__create_workers()
+
+ self.__listen()
+
+ def __listen(self):
+
+ self.server.listen()
+ logging.debug("Listening for connections")
+
+ while True:
+ if self._dispatch_queue.qsize() > self.worker_count:
+ time.sleep(0.01)
+ continue
+
+ conn, addr = self.server.accept()
+ logging.info("New connection: %s", addr[0])
+ self._dispatch_queue.put((conn, addr))
+ logging.debug("Dispatched connection %s", addr)
+
+ def __shutdown(self):
+
+ # Set stop event
+ self._stop_event.set()
+
+ # Wake up workers
+ logging.debug("Waking up workers")
+ for p in self.workers:
+ self._dispatch_queue.put((None, None))
+
+ logging.debug("Closing dispatch queue")
+ self._dispatch_queue.close()
+
+ logging.debug("Waiting for workers to shutdown")
+ p: Process
+ for p in self.workers:
+ p.join()
+ p.terminate()
+
+ logging.debug("Shutting down socket")
+ self.server.shutdown(socket.SHUT_RDWR)
+ self.server.close()
+
+ def __create_workers(self):
+ for i in range(self.worker_count):
+ logging.debug("Creating worker: %d", i + 1)
+ p = mp.Process(target=worker.worker,
+ args=(self.address, i + 1, self.logging_level, self._dispatch_queue, self._stop_event))
+ p.start()
+ self.workers.append(p)
+
+ time.sleep(0.1)
diff --git a/server/worker.py b/server/worker.py
new file mode 100644
index 0000000..c63b1d1
--- /dev/null
+++ b/server/worker.py
@@ -0,0 +1,83 @@
+import logging
+import multiprocessing
+import multiprocessing as mp
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from logging import Logger
+from socket import socket
+
+from server.RequestHandler import RequestHandler
+
+THREAD_LIMIT = 20
+
+
+def worker(address, name, log_level, queue: mp.Queue, stop_event: mp.Event):
+ logging.basicConfig(level=log_level)
+ logger = multiprocessing.log_to_stderr(level=log_level)
+ runner = Worker(address, name, logger, queue, stop_event)
+ runner.logger.debug("Worker %s started", name)
+
+ try:
+ runner.run()
+ except KeyboardInterrupt:
+ logger.debug("Ctrl+C pressed, terminating")
+ runner.shutdown()
+
+
+class Worker:
+
+ host: str
+ name: str
+ logger: Logger
+ queue: mp.Queue
+ executor: ThreadPoolExecutor
+ stop_event: mp.Event
+
+ finished_queue: mp.Queue
+
+ def __init__(self, host, name, logger, queue: mp.Queue, stop_event: mp.Event):
+ self.host = host
+ self.name = name
+ self.logger = logger
+ self.queue = queue
+ self.executor = ThreadPoolExecutor(THREAD_LIMIT)
+ self.stop_event = stop_event
+ self.finished_queue = mp.Queue()
+
+ for i in range(THREAD_LIMIT):
+ self.finished_queue.put(i)
+
+ def run(self):
+ while not self.stop_event.is_set():
+
+ # Blocks until thread is free
+ self.finished_queue.get()
+ # Blocks until new client connects
+ conn, addr = self.queue.get()
+
+ if conn is None or addr is None:
+ break
+
+ self.logger.debug("Received new client: %s", addr)
+
+ # submit client to thread
+ print(threading.get_ident())
+ self.executor.submit(self._handle_client, conn, addr)
+
+ self.shutdown()
+
+ def _handle_client(self, conn: socket, addr):
+ try:
+ self.logger.debug("Handling client: %s", addr)
+
+ handler = RequestHandler(conn, self.logger, self.host)
+ handler.listen()
+ except Exception as e:
+ self.logger.debug("Internal error", exc_info=e)
+
+ # Finished, put back into queue
+ self.finished_queue.put(threading.get_ident())
+
+ def shutdown(self):
+ self.logger.info("shutting down")
+ self.executor.shutdown()
diff --git a/server_flow.md b/server_flow.md
new file mode 100644
index 0000000..f486721
--- /dev/null
+++ b/server_flow.md
@@ -0,0 +1,4 @@
+# Flow
+- listen
+- dispatch asap
+- throw error if too full
\ No newline at end of file