proj: Implement shm_open, shm_unlink
This commit is contained in:
@@ -1,16 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from typing import Optional, TypedDict
|
||||
from typing import Optional, TypedDict, NotRequired
|
||||
from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
|
||||
|
||||
type Pointer[T] = tuple[int, T]
|
||||
type Flags = tuple[int, list[str]]
|
||||
type Pointer = int
|
||||
type PointerTo[T] = tuple[Pointer, T]
|
||||
type Flags = tuple[Pointer, list[str]]
|
||||
StructTimeSpec = TypedDict('StructTimeSpec', {'tv_sec': int, 'tv_nsec': int})
|
||||
StructSigAction = TypedDict('StructSigAction', {'sa_flags': Flags, 'sa_handler': NotRequired[Pointer], 'sa_sigaction': NotRequired[Pointer], 'sa_mask': list[str]})
|
||||
|
||||
|
||||
class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
|
||||
@@ -201,30 +203,30 @@ class Handler(StreamRequestHandler):
|
||||
def before_malloc(self, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_malloc(self, size: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_calloc(self, nmemb: int, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_calloc(self, nmemb: int, size: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_realloc(self, ptr: int, size: int) -> str:
|
||||
def before_realloc(self, ptr: Pointer, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_realloc(self, ptr: int, size: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
def after_realloc(self, ptr: Pointer, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_reallocarray(self, ptr: int, nmemb: int, size: int) -> str:
|
||||
def before_reallocarray(self, ptr: Pointer, nmemb: int, size: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_reallocarray(self, ptr: int, nmemb: int, size: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
def after_reallocarray(self, ptr: Pointer, nmemb: int, size: int,
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_free(self, ptr: int) -> str:
|
||||
def before_free(self, ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_free(self, ptr: int) -> None:
|
||||
def after_free(self, ptr: Pointer) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_getopt(self, argc: int, argv: Pointer[list[Pointer[bytes]]], optstring: Pointer[bytes]) -> str:
|
||||
def before_getopt(self, argc: int, argv: PointerTo[list[PointerTo[bytes]]], optstring: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_getopt(self, argc: int, argv: Pointer[list[Pointer[bytes]]], optstring: Pointer[bytes],
|
||||
def after_getopt(self, argc: int, argv: PointerTo[list[PointerTo[bytes]]], optstring: PointerTo[bytes],
|
||||
ret_value: int) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_close(self, fildes: int) -> str:
|
||||
@@ -232,6 +234,11 @@ class Handler(StreamRequestHandler):
|
||||
def after_close(self, fildes: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sigaction(self, sig: int, act: PointerTo[StructSigAction], oact_ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sigaction(self, sig: int, act: PointerTo[StructSigAction], oact_ptr: Pointer,
|
||||
ret_value: int, errno: str = None, oact: StructSigAction = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_init(self, sem: int, pshared: int, value: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_init(self, sem: int, pshared: int, value: int,
|
||||
@@ -240,48 +247,58 @@ class Handler(StreamRequestHandler):
|
||||
def before_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_open(self, name: str, oflag: Flags, mode: Optional[int], value: Optional[int],
|
||||
ret_value: Pointer, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_post(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_post(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_post(self, sem: int) -> str:
|
||||
def before_sem_wait(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_post(self, sem: int,
|
||||
def after_sem_wait(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_wait(self, sem: int) -> str:
|
||||
def before_sem_trywait(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_wait(self, sem: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_trywait(self, sem: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_trywait(self, sem: int,
|
||||
def after_sem_trywait(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_timedwait(self, sem: int, abs_timeout: Pointer[StructTimeSpec]) -> str:
|
||||
def before_sem_timedwait(self, sem: Pointer, abs_timeout: PointerTo[StructTimeSpec]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_timedwait(self, sem: int, abs_timeout: Pointer[StructTimeSpec],
|
||||
def after_sem_timedwait(self, sem: Pointer, abs_timeout: PointerTo[StructTimeSpec],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_getvalue(self, sem: int, value_ptr: int) -> str:
|
||||
def before_sem_getvalue(self, sem: Pointer, value_ptr: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_getvalue(self, sem:int, value_ptr: int,
|
||||
def after_sem_getvalue(self, sem: Pointer, value_ptr: Pointer,
|
||||
ret_value: int, errno: str = None, value: int = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_close(self, sem: int) -> str:
|
||||
def before_sem_close(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_close(self, sem: int,
|
||||
def after_sem_close(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_unlink(self, name: Pointer[bytes]) -> str:
|
||||
def before_sem_unlink(self, name: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_unlink(self, name: Pointer[bytes],
|
||||
def after_sem_unlink(self, name: PointerTo[bytes],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_sem_destroy(self, sem: int) -> str:
|
||||
def before_sem_destroy(self, sem: Pointer) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_sem_destroy(self, sem: int,
|
||||
def after_sem_destroy(self, sem: Pointer,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_shm_open(self, name: PointerTo[bytes], oflag: Flags, mode: int) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_shm_open(self, name: PointerTo[bytes], oflag: Flags, mode: int,
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
def before_shm_unlink(self, name: PointerTo[bytes]) -> str:
|
||||
raise NotImplementedError()
|
||||
def after_shm_unlink(self, name: PointerTo[bytes],
|
||||
ret_value: int, errno: str = None) -> None:
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class MemoryAllocationTester(Handler):
|
||||
|
||||
Reference in New Issue
Block a user