lorenz.stechauner/BSc-Thesis
1
0
Fork 0
Code Activity

proj: Implement semaphore functions

Browse Source
This commit is contained in:
Lorenz Stechauner
2025-02-17 15:00:30 +01:00
parent 9104c555eb
commit cc2b270d8f
4 changed files with 653 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
proj/server/src/intercept.py
View File

@@ -1,12 +1,18 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Optional, TypedDict
from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn
import argparse
import os
import re
type Pointer[T] = tuple[int, T]
type Flags = tuple[int, list[str]]
StructTimeSpec = TypedDict('StructTimeSpec', {'tv_sec': int, 'tv_nsec': int})
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
pass
@@ -100,12 +106,35 @@ class Handler(StreamRequestHandler):
if idx < len(argument) and argument[idx] == ',':
idx += 1
return (val, list(l)), idx
elif argument[idx] == '|':
m = re.match(r'^[| A-Za-z0-9_]*', argument[idx:])
flags = m.group(0)
if not flags.startswith('|') or not flags.endswith('|'):
raise ValueError()
idx += len(flags)
if idx < len(argument) and argument[idx] == ',':
idx += 1
flags = [f.strip() for f in flags[1:-1].split('|') if len(f.strip()) > 0]
return (val, flags), idx
elif argument[idx] == '"':
s, i = Handler.parse_str(argument[idx:])
idx += i
if idx < len(argument) and argument[idx] == ',':
idx += 1
return (val, s), idx
elif argument[idx] == '{':
m = re.match(r'^[^}]*', argument[idx:])
value = m.group(0)
if not value.startswith('{') or not value.endswith('}'):
raise ValueError()
idx += len(value)
if idx < len(argument) and argument[idx] == ',':
idx += 1
entries = {}
for e in [v.strip() for v in value[1:-1].split(',') if len(e.strip()) > 0]:
k, v = e.split(':', 1)
entries[k.strip()] = int(v.strip(), 0)
return (val, entries), idx
else:
raise ValueError()
@@ -133,7 +162,7 @@ class Handler(StreamRequestHandler):
try:
func = getattr(self, f'before_{func_name}')
if callable(func):
command = func(*args)
command = func(*args) or 'ok'
else:
command = 'ok'
except AttributeError:
@@ -155,6 +184,42 @@ class Handler(StreamRequestHandler):
pass
print(f'[{self.pid}] -> {ret}')
def before_malloc(self, size: int) -> str: pass
def after_malloc(self, size: int,
ret_value: int, errno: str = None) -> None: pass
def before_calloc(self, nmemb: int, size: int) -> str: pass
def after_calloc(self, nmemb: int, size: int,
ret_value: int, errno: str = None) -> None: pass
def before_realloc(self, ptr: int, size: int) -> str: pass
def after_realloc(self, ptr: int, size: int,
ret_value: int, errno: str = None) -> None: pass
def before_reallocarray(self, ptr: int, nmemb: int, size: int) -> str: pass
def after_reallocarray(self, ptr: int, nmemb: int, size: int,
ret_value: int, errno: str = None) -> None: pass
def before_free(self, ptr: int) -> str: pass
def after_free(self, ptr: int) -> None: pass
def before_getopt(self, argc: int, argv: Pointer[list[Pointer[bytes]]], optstring: Pointer[bytes]) -> str: pass
def after_getopt(self, argc: int, argv: Pointer[list[Pointer[bytes]]], optstring: Pointer[bytes],
ret_value: int) -> None: pass
def before_close(self, fildes: int) -> str: pass
def after_close(self, fildes: int,
ret_value: int, errno: str = None) -> None: pass
def before_sem_init(self, sem: int, pshared: int, value: int) -> str: pass
def after_sem_init(self, sem: int, pshared: int, value: int,
ret_value: int, errno: str = None) -> None: pass
def before_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int]) -> str: pass
def after_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int],
ret_value: int, errno: str = None) -> None: pass
def before_sem_post(self, sem: int) -> str: pass
def after_sem_post(self, sem: int,
ret_value: int, errno: str = None) -> None: pass
def before_sem_wait(self, sem: int) -> str: pass
def after_sem_wait(self, sem: int,
ret_value: int, errno: str = None) -> None: pass
def before_sem_timedwait(self, sem: int, abs_timeout: Pointer[StructTimeSpec]): pass
def after_sem_timedwait(self, sem: int, abs_timeout: Pointer[StructTimeSpec],
ret_value: int, errno: str = None): pass
class MemoryAllocationTester(Handler):
allocated: dict[int, int]
@@ -184,20 +249,20 @@ class MemoryAllocationTester(Handler):
if total > self.max_allocated:
self.max_allocated = total
def after_malloc(self, size: int, ret_value: int) -> None:
def after_malloc(self, size, ret_value, errno=None) -> None:
self.num_malloc += 1
if ret_value != 0:
print(ret_value)
self.allocated[ret_value] = size
self.update_max_allocated()
def after_calloc(self, nmemb: int, size: int, ret_value: int) -> None:
def after_calloc(self, nmemb, size, ret_value, errno=None) -> None:
self.num_malloc += 1
if ret_value != 0:
self.allocated[ret_value] = nmemb * size
self.update_max_allocated()
def after_realloc(self, ptr: int, size: int, ret_value: int) -> None:
def after_realloc(self, ptr, size, ret_value, errno=None) -> None:
self.num_realloc += 1
if ptr != 0:
if ret_value != 0:
@@ -205,7 +270,15 @@ class MemoryAllocationTester(Handler):
self.allocated[ret_value] = size
self.update_max_allocated()
def after_free(self, ptr: int) -> None:
def after_reallocarray(self, ptr, nmemb, size, ret_value, errno=None) -> None:
self.num_realloc += 1
if ptr != 0:
if ret_value != 0:
del self.allocated[ptr]
self.allocated[ret_value] = nmemb * size
self.update_max_allocated()
def after_free(self, ptr) -> None:
self.num_free += 1
if ptr != 0:
del self.allocated[ptr]

6
proj/server/src/server.py
View File

@@ -20,18 +20,18 @@ class TestHandler(intercept.Handler):
else:
print("All blocks free'd!")
def after_malloc(self, size: int, ret_value: int) -> None:
def after_malloc(self, size, ret_value, errno=None) -> None:
if ret_value != 0:
self.allocated[ret_value] = size
def before_free(self, ptr: int) -> str:
def before_free(self, ptr) -> str:
if ptr != 0:
del self.allocated[ptr]
return 'ok'
class GetoptHandler(intercept.Handler):
def after_getopt(self, argc: int, argv: tuple[int, list[tuple[int, bytes]]], optstring: tuple[int, bytes], ret_value: int) -> None:
def after_getopt(self, argc, argv, optstring, ret_value) -> None:
print(argc, [v[1] for v in argv[1]], optstring[1], ret_value)