Worker Design

[lkevinzc]

Base structures:

• the user code should implement worker by inheriting from WorkerBase
• if the user wants to do fancier stuff like replacing default json into something like msgpack or bjson, they just write their own pack and unpack methods

class ServiceBase:
    """Defines the default py-ingress/egress methods"""

    def unpack(self):  # ingress
        # by default using pydantic class
        json.loads()

    def pack(self):  # egress
        # by default using pydantic class
        json.dumps()


class WorkerBase(ServiceBase):
    """Defines the default py-ipc de/serialize methods"""

    example = ...

    def __init__(self):
        self.stage = "x"  # updated when doing the server.append_worker, used to determine pack/unpack; {"ingress", "x", "egress"}
        self.max_batch_size = 1  # updated when doing the server.append_worker, used before __call__ to index element

    def serialize(self, data):
        return pickle.dumps(data, protocol=pickle.HIGHEST_PROTOCOL)

    def deserialize(self, data):
        return pickle.loads(data)


# users' implementation:
# 1) no customization
class NormalStage(WorkerBase):
    def __init__():
        pass

    def __call__():
        pass


# 2) custimized ingress/egress methods for potential speedup
class CustomizedService(WorkerBase):
    def unpack():  # ingress
        msgpack.packb()

    def pack():  # egress
        msgpack.unpackb()


class CustomizedStage(CumtomizedService):
    def __init__():
        pass

    def __call__():
        pass