Repository: ihnokim/codepack (language: python)

Issue:
Feature docker runner



Discussion / hints:

Fix patch:
diff --git a/.gitignore b/.gitignore
index 0089d9a..bf820e9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,4 @@
 *.ipynb_checkpoints*
 *.idea*
 *.ipynb
+htmlcov/*
diff --git a/app/apiserver/__init__.py b/app/apiserver/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/apiserver/dependencies/__init__.py b/app/apiserver/dependencies/__init__.py
new file mode 100644
index 0000000..04acb61
--- /dev/null
+++ b/app/apiserver/dependencies/__init__.py
@@ -0,0 +1,4 @@
+from .common import Common
+
+
+common = Common()
diff --git a/app/apiserver/dependencies/common.py b/app/apiserver/dependencies/common.py
new file mode 100644
index 0000000..9f7536c
--- /dev/null
+++ b/app/apiserver/dependencies/common.py
@@ -0,0 +1,21 @@
+class Common:
+    def __init__(self):
+        self.variables = dict()
+
+    def __getitem__(self, item):
+        return self.variables[item]
+
+    def __setitem__(self, key, value):
+        self.variables[key] = value
+
+    def __getattr__(self, item):
+        return self.__getitem__(item)
+
+    def keys(self):
+        return self.variables.keys()
+
+    def values(self):
+        return self.variables.values()
+
+    def items(self):
+        return self.variables.items()
diff --git a/app/apiserver/main.py b/app/apiserver/main.py
new file mode 100644
index 0000000..d59d14f
--- /dev/null
+++ b/app/apiserver/main.py
@@ -0,0 +1,71 @@
+from codepack.scheduler import Scheduler
+from codepack.config import Default
+from fastapi import FastAPI, Request
+from codepack.employee import Supervisor
+from .routers import code, codepack, argpack, scheduler
+from .dependencies import common
+
+
+app = FastAPI()
+app.include_router(code.router)
+app.include_router(codepack.router)
+app.include_router(argpack.router)
+app.include_router(scheduler.router)
+
+
+@app.middleware('http')
+async def add_process_time_header(request: Request, call_next):
+    response = await call_next(request)
+    response.headers['Access-Control-Allow_Origin'] = '*'
+    response.headers['Access-Control-Allow_Headers'] = 'Content-Type,Authorization'
+    response.headers['Access-Control-Allow-Methods'] = 'GET,PUT,POST,DELETE'
+    return response
+
+
+@app.on_event('startup')
+async def startup():
+    Default.init()
+    config = Default.config.get_config(section='apiserver')
+    supervisor = config.get('supervisor', 'self')
+    _scheduler = config.get('scheduler', 'self')
+    if isinstance(supervisor, Supervisor):
+        common['supervisor'] = supervisor
+    elif isinstance(supervisor, str):
+        if supervisor == 'self':
+            common['supervisor'] = Default.get_employee('supervisor')
+        else:
+            raise NotImplementedError("supervisor should be 'self', not '%s'")
+    if isinstance(_scheduler, Scheduler):
+        common['scheduler'] = _scheduler
+        common.scheduler.start()
+    elif isinstance(_scheduler, str):
+        if _scheduler == 'self':
+            common['scheduler'] = Default.get_scheduler()
+            common.scheduler.start()
+        else:
+            common['scheduler'] = _scheduler
+
+
+@app.on_event('shutdown')
+async def shutdown():
+    common.supervisor.producer.close()
+    for service in Default.instances.values():
+        if hasattr(service, 'mongodb'):
+            service.mongodb.close()
+    if isinstance(common.scheduler, Scheduler):
+        if common.scheduler.is_running():
+            common.scheduler.stop()
+        if hasattr(common.scheduler, 'mongodb'):
+            common.scheduler.mongodb.close()
+
+
+@app.get('/organize')
+async def organize():
+    common.supervisor.organize()
+    return None
+
+
+@app.get('/organize/{serial_number}')
+async def organize(serial_number: str):
+    common.supervisor.organize(serial_number=serial_number)
+    return None
diff --git a/app/apiserver/models/__init__.py b/app/apiserver/models/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/apiserver/models/argpack.py b/app/apiserver/models/argpack.py
new file mode 100644
index 0000000..0e35f4f
--- /dev/null
+++ b/app/apiserver/models/argpack.py
@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class ArgPackJSON(BaseModel):
+    argpack: str
diff --git a/app/apiserver/models/code.py b/app/apiserver/models/code.py
new file mode 100644
index 0000000..f591a54
--- /dev/null
+++ b/app/apiserver/models/code.py
@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+
+
+class CodeID(BaseModel):
+    id: str
+    args: tuple = ()
+    kwargs: dict = {}
+
+
+class CodeJSON(BaseModel):
+    code: str
+    args: tuple = ()
+    kwargs: dict = {}
+
+
+class SnapshotJSON(BaseModel):
+    snapshot: str
diff --git a/app/apiserver/models/codepack.py b/app/apiserver/models/codepack.py
new file mode 100644
index 0000000..a193999
--- /dev/null
+++ b/app/apiserver/models/codepack.py
@@ -0,0 +1,20 @@
+from pydantic import BaseModel
+
+
+class CodePackID(BaseModel):
+    id: str
+    argpack: str = '{}'
+
+
+class IDPair(BaseModel):
+    codepack_id: str
+    argpack_id: str
+
+
+class CodePackJSON(BaseModel):
+    codepack: str
+    argpack: str = '{}'
+
+
+class SnapshotJSON(BaseModel):
+    snapshot: str
diff --git a/app/apiserver/models/scheduler.py b/app/apiserver/models/scheduler.py
new file mode 100644
index 0000000..ac1e7b7
--- /dev/null
+++ b/app/apiserver/models/scheduler.py
@@ -0,0 +1,26 @@
+from pydantic import BaseModel
+
+
+class Job(BaseModel):
+    job_id: str = None
+    trigger: str
+    trigger_config = dict
+
+
+class CodePackIDJob(Job):
+    id: str
+    argpack: str = '{}'
+
+
+class IDPairJob(Job):
+    codepack_id: str
+    argpack_id: str
+
+
+class CodePackJSONJob(Job):
+    codepack: str
+    argpack: str = '{}'
+
+
+class SnapshotJSONJob(Job):
+    snapshot: str
diff --git a/app/apiserver/routers/__init__.py b/app/apiserver/routers/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/apiserver/routers/argpack.py b/app/apiserver/routers/argpack.py
new file mode 100644
index 0000000..a58354f
--- /dev/null
+++ b/app/apiserver/routers/argpack.py
@@ -0,0 +1,41 @@
+from fastapi import APIRouter
+from ..models.argpack import ArgPackJSON
+from codepack.config import Default
+from codepack.argpack import ArgPack
+
+
+router = APIRouter(
+    prefix='/argpack',
+    tags=['argpack'],
+    responses={404: {'description': 'Not found'}}
+)
+
+
+@router.post('/save')
+async def save(argpack: ArgPackJSON):
+    storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    tmp = ArgPack.from_json(argpack.argpack)
+    storage_service.save(item=tmp)
+    return {'id': tmp.id}
+
+
+@router.post('/update')
+async def update(argpack: ArgPackJSON):
+    storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    tmp = ArgPack.from_json(argpack.argpack)
+    storage_service.save(item=tmp, update=True)
+    return {'id': tmp.id}
+
+
+@router.get('/remove/{id}')
+async def remove(id: str):
+    storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    storage_service.remove(id)
+    return {'id': id}
+
+
+@router.get('/load/{id}')
+async def load(id: str):
+    storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    argpack = storage_service.load(id)
+    return {'argpack': argpack.to_json()}
diff --git a/app/apiserver/routers/code.py b/app/apiserver/routers/code.py
new file mode 100644
index 0000000..21b8e6f
--- /dev/null
+++ b/app/apiserver/routers/code.py
@@ -0,0 +1,85 @@
+from fastapi import APIRouter
+from ..models.code import CodeID, CodeJSON, SnapshotJSON
+from codepack.config import Default
+from codepack import Code
+from codepack.snapshot import CodeSnapshot
+from ..dependencies import common
+
+
+router = APIRouter(
+    prefix='/code',
+    tags=['code'],
+    responses={404: {'description': 'Not found'}},
+)
+
+
+@router.post('/run')
+async def run(params: CodeJSON):
+    code = Code.from_json(params.code)
+    common.supervisor.run_code(code=code, args=params.args, kwargs=params.kwargs)
+    return {'serial_number': code.serial_number}
+
+
+@router.post('/run/id')
+async def run_by_id(params: CodeID):
+    storage_service = Default.get_storage_instance('code', 'storage_service')
+    code = storage_service.load(params.id)
+    common.supervisor.run_code(code=code, args=params.args, kwargs=params.kwargs)
+    return {'serial_number': code.serial_number}
+
+
+@router.post('/run/snapshot')
+async def run_by_snapshot(params: SnapshotJSON):
+    snapshot = CodeSnapshot.from_json(params.snapshot)
+    code = Code.from_snapshot(snapshot)
+    common.supervisor.run_code(code=code, args=snapshot.args, kwargs=snapshot.kwargs)
+    return {'serial_number': code.serial_number}
+
+
+@router.post('/save')
+async def save(code: CodeJSON):
+    tmp = Code.from_json(code.code)
+    tmp.save()
+    return {'id': tmp.id}
+
+
+@router.post('/update')
+async def update(code: CodeJSON):
+    tmp = Code.from_json(code.code)
+    tmp.save(update=True)
+    return {'id': tmp.id}
+
+
+@router.get('/remove/{id}')
+async def remove(id: str):
+    storage_service = Default.get_storage_instance('code', 'storage_service')
+    storage_service.remove(id)
+    return {'id': id}
+
+
+@router.get('/load/{id}')
+async def load(id: str):
+    code = Code(id=id)
+    return {'code': code.to_json()}
+
+
+@router.get('/state/{serial_number}')
+async def state(serial_number: str):
+    snapshot_service = Default.get_storage_instance('code_snapshot', 'snapshot_service')
+    ret = snapshot_service.load(serial_number=serial_number, projection={'state'})
+    if ret:
+        _state = ret['state']
+    else:
+        _state = 'UNKNOWN'
+    return {'seriual_number': serial_number, 'state': _state}
+
+
+@router.get('/result/{serial_number}')
+async def result(serial_number: str):
+    delivery_service = Default.get_storage_instance('delivery', 'delivery_service')
+    tmp = delivery_service.check(serial_number=serial_number)
+    if tmp:
+        _result = delivery_service.receive(serial_number=serial_number)
+    else:
+        _result = None
+    return {'serial_number': serial_number, 'result': _result}
diff --git a/app/apiserver/routers/codepack.py b/app/apiserver/routers/codepack.py
new file mode 100644
index 0000000..9a4986f
--- /dev/null
+++ b/app/apiserver/routers/codepack.py
@@ -0,0 +1,103 @@
+from fastapi import APIRouter
+from ..models.codepack import CodePackID, CodePackJSON, SnapshotJSON, IDPair
+from codepack.config import Default
+from codepack import CodePack
+from codepack.snapshot import CodePackSnapshot
+from codepack.argpack import ArgPack
+from ..dependencies import common
+
+
+router = APIRouter(
+    prefix='/codepack',
+    tags=['codepack'],
+    responses={404: {'description': 'Not found'}},
+)
+
+
+@router.post('/run')
+async def run(params: CodePackJSON):
+    codepack = CodePack.from_json(params.codepack)
+    argpack = ArgPack.from_json(params.argpack)
+    common.supervisor.run_codepack(codepack=codepack, argpack=argpack)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/run/id')
+async def run_by_id(params: CodePackID):
+    storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    codepack = storage_service.load(params.id)
+    argpack = ArgPack.from_json(params.argpack)
+    common.supervisor.run_codepack(codepack=codepack, argpack=argpack)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/run/id-pair')
+async def run_by_id_pair(params: IDPair):
+    codepack_storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    codepack = codepack_storage_service.load(params.codepack_id)
+    argpack_storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    argpack = argpack_storage_service.load(params.argpack_id)
+    common.supervisor.run_codepack(codepack=codepack, argpack=argpack)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/run/snapshot')
+async def run_by_snapshot(params: SnapshotJSON):
+    snapshot = CodePackSnapshot.from_json(params.snapshot)
+    codepack = CodePack.from_snapshot(snapshot)
+    argpack = ArgPack.from_dict(snapshot.argpack)
+    common.supervisor.run_codepack(codepack=codepack, argpack=argpack)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/save')
+async def save(codepack: CodePackJSON):
+    tmp = CodePack.from_json(codepack.codepack)
+    tmp.save()
+    return {'id': tmp.id}
+
+
+@router.post('/update')
+async def update(codepack: CodePackJSON):
+    tmp = CodePack.from_json(codepack.codepack)
+    tmp.save(update=True)
+    return {'id': tmp.id}
+
+
+@router.get('/remove/{id}')
+async def remove(id: str):
+    storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    storage_service.remove(id)
+    return {'id': id}
+
+
+@router.get('/load/{id}')
+async def load(id: str):
+    storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    codepack = storage_service.load(id)
+    return {'codepack': codepack.to_json()}
+
+
+@router.get('/state/{serial_number}')
+async def state(serial_number: str):
+    snapshot_service = Default.get_storage_instance('codepack_snapshot', 'snapshot_service')
+    tmp = snapshot_service.load(serial_number=serial_number)
+    if tmp:
+        codepack_snapshot = CodePackSnapshot.from_dict(tmp)
+        codepack = CodePack.from_snapshot(codepack_snapshot)
+        _state = codepack.get_state()
+    else:
+        _state = 'UNKNOWN'
+    return {'serial_number': serial_number, 'state': _state}
+
+
+@router.get('/result/{serial_number}')
+async def result(serial_number: str):
+    snapshot_service = Default.get_storage_instance('codepack_snapshot', 'snapshot_service')
+    snapshot = snapshot_service.load(serial_number=serial_number)
+    _result = None
+    if snapshot and snapshot['subscribe']:
+        codepack_snapshot = CodePackSnapshot.from_dict(snapshot)
+        codepack = CodePack.from_snapshot(codepack_snapshot)
+        _result = codepack.get_result()
+    return {'serial_number': serial_number, 'result': _result}
diff --git a/app/apiserver/routers/scheduler.py b/app/apiserver/routers/scheduler.py
new file mode 100644
index 0000000..0646ad1
--- /dev/null
+++ b/app/apiserver/routers/scheduler.py
@@ -0,0 +1,58 @@
+from fastapi import APIRouter
+from ..models.scheduler import CodePackIDJob, CodePackJSONJob, SnapshotJSONJob, IDPairJob
+from codepack.config import Default
+from codepack import CodePack
+from codepack.snapshot import CodePackSnapshot
+from codepack.argpack import ArgPack
+from ..dependencies import common
+
+
+router = APIRouter(
+    prefix='/scheduler',
+    tags=['scheduler'],
+    responses={404: {'description': 'Not found'}},
+)
+
+
+@router.post('/register')
+async def register(params: CodePackJSONJob):
+    codepack = CodePack.from_json(params.codepack)
+    argpack = ArgPack.from_json(params.argpack)
+    common.scheduler.add_codepack(codepack=codepack, argpack=argpack, job_id=params.job_id,
+                                  trigger=params.trigger, **params.trigger_config)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/register/id')
+async def register_by_id(params: CodePackIDJob):
+    storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    codepack = storage_service.load(params.id)
+    argpack = ArgPack.from_json(params.argpack)
+    common.scheduler.add_codepack(codepack=codepack, argpack=argpack, job_id=params.job_id,
+                                  trigger=params.trigger, **params.trigger_config)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/register/id-pair')
+async def register_by_id_pair(params: IDPairJob):
+    codepack_storage_service = Default.get_storage_instance('codepack', 'storage_service')
+    codepack = codepack_storage_service.load(params.codepack_id)
+    argpack_storage_service = Default.get_storage_instance('argpack', 'storage_service')
+    argpack = argpack_storage_service.load(params.argpack_id)
+    common.scheduler.add_codepack(codepack=codepack, argpack=argpack, job_id=params.job_id,
+                                  trigger=params.trigger, **params.trigger_config)
+    return {'serial_number': codepack.serial_number}
+
+
+@router.post('/register/snapshot')
+async def register_by_snapshot(params: SnapshotJSONJob):
+    snapshot = CodePackSnapshot.from_json(params.snapshot)
+    common.scheduler.add_codepack(snapshot=snapshot, job_id=params.job_id,
+                                  trigger=params.trigger, **params.trigger_config)
+    return {'serial_number': snapshot.serial_number}
+
+
+@router.get('/unregister/{id}')
+async def unregister(id: str):
+    common.scheduler.remove_job(job_id=id)
+    return {'id': id}
diff --git a/codepack/argpack/__init__.py b/codepack/argpack/__init__.py
new file mode 100644
index 0000000..2987f01
--- /dev/null
+++ b/codepack/argpack/__init__.py
@@ -0,0 +1,2 @@
+from codepack.argpack.arg import Arg
+from codepack.argpack.argpack import ArgPack
diff --git a/codepack/argpack/arg.py b/codepack/argpack/arg.py
new file mode 100644
index 0000000..c9b8b6a
--- /dev/null
+++ b/codepack/argpack/arg.py
@@ -0,0 +1,57 @@
+from codepack.storage import Storable
+
+
+class Arg(Storable):
+    def __init__(self, code=None, kwargs=None):
+        Storable.__init__(self)
+        self.kwargs = dict()
+        if kwargs:
+            for arg, value in kwargs.items():
+                self.kwargs[arg] = value
+        elif code:
+            self.kwargs = self.extract(code)
+
+    @staticmethod
+    def extract(code):
+        ret = dict()
+        for arg, value in code.get_args().items():
+            if arg not in code.dependency.get_args().keys():
+                ret[arg] = value
+        return ret
+
+    def __getitem__(self, item):
+        return self.kwargs[item]
+
+    def __setitem__(self, key, value):
+        self.kwargs[key] = value
+
+    def __call__(self, **kwargs):
+        for arg, value in kwargs.items():
+            if arg not in self.kwargs:
+                raise TypeError(arg)
+            self.kwargs[arg] = value
+
+    def to_dict(self):
+        return self.kwargs
+
+    @classmethod
+    def from_dict(cls, d):
+        return cls(kwargs=d)
+
+    def __getattr__(self, item):
+        return getattr(self.kwargs, item)
+
+    def __iter__(self):
+        return self.kwargs.__iter__()
+
+    def __str__(self):
+        ret = '%s(' % self.__class__.__name__
+        for i, (arg, value) in enumerate(self.kwargs.items()):
+            if i:
+                ret += ', '
+            ret += '%s=%s' % (arg, value)
+        ret += ')'
+        return ret
+
+    def __repr__(self):
+        return self.__str__()  # pragma: no cover
diff --git a/codepack/argpack/argpack.py b/codepack/argpack/argpack.py
new file mode 100644
index 0000000..665f2a9
--- /dev/null
+++ b/codepack/argpack/argpack.py
@@ -0,0 +1,76 @@
+from codepack.argpack import Arg
+from codepack.storage import Storable
+
+
+class ArgPack(Storable):
+    def __init__(self, codepack=None, id=None, args=None):
+        _id = None
+        if codepack:
+            _id = codepack.id
+        if id:
+            _id = id
+        Storable.__init__(self, id=_id)
+        if args:
+            self.args = dict()
+            for id, kwargs in args.items():
+                self.args[id] = Arg.from_dict(kwargs)
+        elif codepack:
+            self.args = self.extract(codepack)
+        else:
+            self.args = dict()
+
+    @staticmethod
+    def extract(codepack):
+        ret = dict()
+        stack = list()
+        for root in codepack.roots:
+            stack.append(root)
+            while len(stack):
+                n = stack.pop(-1)
+                if n.id not in ret:
+                    ret[n.id] = Arg(n)
+                for c in n.children.values():
+                    stack.append(c)
+        return ret
+
+    def __getitem__(self, item):
+        return self.args[item]
+
+    def __setitem__(self, key, value):
+        self.args[key] = value
+
+    def to_dict(self):
+        ret = dict()
+        for id, arg in self.args.items():
+            ret[id] = arg.to_dict()
+        ret['_id'] = self.id
+        return ret
+
+    @classmethod
+    def from_dict(cls, d):
+        args = dict()
+        id = None
+        for k, v in d.items():
+            if k == '_id':
+                id = v
+            else:
+                args[k] = v
+        return cls(id=id, args=args)
+
+    def __getattr__(self, item):
+        return getattr(self.args, item)
+
+    def __iter__(self):
+        return self.args.__iter__()
+
+    def __str__(self):
+        ret = '%s(id: %s, args: {' % (self.__class__.__name__, self.id)
+        for i, (id, arg) in enumerate(self.args.items()):
+            if i:
+                ret += ', '
+            ret += '%s%s' % (id, arg.__str__().replace('Arg(', '('))
+        ret += '})'
+        return ret
+
+    def __repr__(self):
+        return self.__str__()  # pragma: no cover
diff --git a/codepack/code.py b/codepack/code.py
index 2f6bcf3..36f93d0 100644
--- a/codepack/code.py
+++ b/codepack/code.py
@@ -1,14 +1,13 @@
 import inspect
 from collections.abc import Iterable, Callable
 import dill
-from codepack.service import DefaultService
+from codepack.config import Default
 from codepack.base import CodeBase
 import re
 import ast
 from collections import OrderedDict
 from datetime import datetime
-from codepack.utils.dependency import Dependency
-from codepack.utils.dependency_state import DependencyState
+from codepack.dependency import Dependency, DependencyManager
 from codepack.snapshot import CodeSnapshot
 
 
@@ -43,19 +42,21 @@ def init_linkage(self):
         self.children = dict()
 
     def init_dependency(self, dependency=None):
-        self.dependency = dict()
+        self.dependency = DependencyManager(code=self)
         if dependency:
             self.add_dependency(dependency=dependency)
 
     def init_service(self, delivery_service=None, snapshot_service=None, storage_service=None, config_path=None):
         self.service = dict()
-        self.service['delivery_service'] =\
-            delivery_service if delivery_service else DefaultService.get_default_delivery_service(config_path=config_path)
-        self.service['snapshot_service'] =\
-            snapshot_service if snapshot_service else DefaultService.get_default_code_snapshot_service(config_path=config_path)
-        self.service['storage_service'] =\
-            storage_service if storage_service else DefaultService.get_default_code_storage_service(obj=self.__class__,
-                                                                                                    config_path=config_path)
+        self.service['delivery'] =\
+            delivery_service if delivery_service else Default.get_storage_instance('delivery', 'delivery_service',
+                                                                                   config_path=config_path)
+        self.service['snapshot'] =\
+            snapshot_service if snapshot_service else Default.get_storage_instance('code_snapshot', 'snapshot_service',
+                                                                                   config_path=config_path)
+        self.service['storage'] =\
+            storage_service if storage_service else Default.get_storage_instance('code', 'storage_service',
+                                                                                 config_path=config_path)
 
     def register(self, callback):
         self.callback = callback
@@ -98,7 +99,7 @@ def set_function(self, function=None, source=None):
             self.function = function
             self.source = self.get_source(self.function)
         elif self.id:
-            tmp = self.service['storage_service'].load(id=self.id)
+            tmp = self.service['storage'].load(id=self.id)
             self.function = self.get_function(tmp.source)
             self.source = tmp.source
         else:
@@ -140,7 +141,8 @@ def __floordiv__(self, other):
     def update_serial_number(self, serial_number):
         for c in self.children.values():
             if self.serial_number in c.dependency:
-                d = c.dependency.pop(self.serial_number)
+                d = c.dependency[self.serial_number]
+                c.remove_dependency(self.serial_number)
                 d.serial_number = serial_number
                 c.add_dependency(d)
         self.serial_number = serial_number
@@ -174,13 +176,13 @@ def get_info(self, state=True):
             ret += ', state: %s)'
             return ret % (self.__class__.__name__, self.id, self.function.__name__,
                           self.print_args(),
-                          self.get_dependent_args(),
+                          self.dependency.get_args(),
                           self.get_state())
         else:
             ret += ')'
             return ret % (self.__class__.__name__, self.id, self.function.__name__,
                           self.print_args(),
-                          self.get_dependent_args())
+                          self.dependency.get_args())
 
     def __str__(self):
         return self.get_info(state=False)  # pragma: no cover
@@ -192,26 +194,26 @@ def update_state(self, state, timestamp=None, args=None, kwargs=None):
         if state:
             snapshot = self.to_snapshot(args=args, kwargs=kwargs, timestamp=timestamp)
             snapshot['state'] = state
-            self.service['snapshot_service'].save(snapshot)
+            self.service['snapshot'].save(snapshot)
             if self.callback:
                 self.callback({'serial_number': self.serial_number, 'state': state})
 
     def get_state(self):
-        ret = self.service['snapshot_service'].load(serial_number=self.serial_number, projection={'state'})
+        ret = self.service['snapshot'].load(serial_number=self.serial_number, projection={'state'})
         if ret:
             return ret['state']
         else:
             return 'UNKNOWN'
 
     def send_result(self, item, timestamp=None):
-        self.service['delivery_service'].send(id=self.id, serial_number=self.serial_number, item=item, timestamp=timestamp)
+        self.service['delivery'].send(id=self.id, serial_number=self.serial_number, item=item, timestamp=timestamp)
 
     def get_result(self, serial_number=None):
         serial_number = serial_number if serial_number else self.serial_number
-        return self.service['delivery_service'].receive(serial_number=serial_number)
+        return self.service['delivery'].receive(serial_number=serial_number)
 
-    def save(self):
-        self.service['storage_service'].save(item=self)
+    def save(self, update=False):
+        self.service['storage'].save(item=self, update=update)
 
     def receive(self, arg):
         self.assert_arg(arg)
@@ -221,78 +223,16 @@ def assert_arg(self, arg):
         if arg and arg not in self.get_args():
             raise AssertionError("'%s' is not an argument of %s" % (arg, self.function))
 
-    def get_dependent_args(self):
-        ret = dict()
-        for dependency in self.dependency.values():
-            if dependency.arg:
-                ret[dependency.arg] = dependency.id
-        return ret
-
     def add_dependency(self, dependency):
-        if isinstance(dependency, Dependency):
-            self.assert_arg(dependency.arg)
-            self.dependency[dependency.serial_number] = dependency
-        elif isinstance(dependency, dict):
-            self.assert_arg(dependency['arg'])
-            self.dependency[dependency['serial_number']] = Dependency.from_dict(d=dependency)
-            self.dependency[dependency['serial_number']].bind(self)
-        elif isinstance(dependency, Iterable):
-            for d in dependency:
-                self.add_dependency(d)
-        else:
-            raise TypeError(type(dependency))  # pragma: no cover
+        self.dependency.add(dependency=dependency)
 
     def remove_dependency(self, serial_number):
-        self.dependency.pop(serial_number, None)
-
-    def get_dependent_snapshots(self):
-        ret = dict()
-        snapshots = self.service['snapshot_service'].load(serial_number=list(self.dependency.keys()))
-        for snapshot in snapshots:
-            ret[snapshot['_id']] = snapshot
-        return ret
-
-    def get_dependency_cache_info(self):
-        ret = dict()
-        caches = self.service['delivery_service'].check(serial_number=[k for k, v in self.dependency.items() if v.arg])
-        for cache in caches:
-            ret[cache['_id']] = cache
-        return ret
-
-    def check_dependency_state(self, snapshots):
-        for snapshot in snapshots.values():
-            if snapshot['state'] == 'ERROR':
-                return DependencyState.ERROR
-            elif snapshot['state'] != 'TERMINATED':
-                return DependencyState.NOT_READY
-        if len(self.dependency) != len(snapshots):
-            return DependencyState.NOT_READY
-        return DependencyState.READY
-
-    def validate_dependency_result(self, snapshots, caches):
-        if len([k for k, v in self.dependency.items() if v.arg]) != len(caches):
-            return DependencyState.NOT_READY
-        for cache in caches.values():
-            if cache['_id'] not in snapshots:
-                return DependencyState.NOT_READY
-            elif 'timestamp' not in cache or 'timestamp' not in snapshots[cache['_id']]:
-                return DependencyState.NOT_READY
-            elif cache['timestamp'] != snapshots[cache['_id']]['timestamp']:
-                return DependencyState.NOT_READY
-        return DependencyState.READY
-
-    def check_dependency(self):
-        snapshots = self.get_dependent_snapshots()
-        dependency_state = self.check_dependency_state(snapshots=snapshots)
-        if dependency_state != 'READY':
-            return dependency_state
-        caches = self.get_dependency_cache_info()
-        return self.validate_dependency_result(snapshots=snapshots, caches=caches)
+        self.dependency.remove(serial_number)
 
     def __call__(self, *args, **kwargs):
         try:
-            dependency_state = self.check_dependency()
-            if dependency_state == 'READY':
+            dependency_state = self.dependency.get_state()
+            if dependency_state == 'RESOLVED':
                 for dependency in self.dependency.values():
                     if dependency.arg and dependency.arg not in kwargs:
                         kwargs[dependency.arg] = self.get_result(serial_number=dependency.serial_number)
@@ -302,7 +242,7 @@ def __call__(self, *args, **kwargs):
                 self.send_result(item=ret, timestamp=now)
                 self.update_state('TERMINATED', args=args, kwargs=kwargs, timestamp=now)
                 return ret
-            elif dependency_state == 'NOT_READY':
+            elif dependency_state == 'PENDING':
                 self.update_state('WAITING', args=args, kwargs=kwargs)
                 return None
             elif dependency_state == 'ERROR':
diff --git a/codepack/codepack.py b/codepack/codepack.py
index 6e48b81..8a94aff 100644
--- a/codepack/codepack.py
+++ b/codepack/codepack.py
@@ -1,15 +1,17 @@
 from codepack.base import CodeBase, CodePackBase
 from codepack import Code
 from queue import Queue
-from codepack.service import DefaultService
-from codepack.utils.state import State
+from codepack.config import Default
+from codepack.snapshot import State
 from parse import compile as parser
 from ast import literal_eval
 from codepack.snapshot import CodePackSnapshot
+from codepack.argpack import ArgPack
 
 
 class CodePack(CodePackBase):
-    def __init__(self, id, code, subscribe=None, serial_number=None, config_path=None, snapshot_service=None, storage_service=None):
+    def __init__(self, id, code, subscribe=None, serial_number=None, config_path=None,
+                 snapshot_service=None, storage_service=None, argpack_service=None):
         super().__init__(id=id, serial_number=serial_number)
         self.root = None
         self.roots = None
@@ -25,34 +27,32 @@ def __init__(self, id, code, subscribe=None, serial_number=None, config_path=Non
         self.init()
         self.init_service(snapshot_service=snapshot_service,
                           storage_service=storage_service,
+                          argpack_service=argpack_service,
                           config_path=config_path)
 
     def init(self):
         self.roots = self.get_roots()
 
-    def init_service(self, snapshot_service=None, storage_service=None, config_path=None):
+    def init_service(self, snapshot_service=None, storage_service=None, argpack_service=None, config_path=None):
         self.service = dict()
-        self.service['snapshot_service'] =\
-            snapshot_service if snapshot_service else DefaultService.get_default_codepack_snapshot_service(config_path=config_path)
-        self.service['storage_service'] =\
-            storage_service if storage_service else DefaultService.get_default_codepack_storage_service(obj=self.__class__,
-                                                                                                        config_path=config_path)
+        self.service['snapshot'] =\
+            snapshot_service if snapshot_service else Default.get_storage_instance('codepack_snapshot', 'snapshot_service',
+                                                                                   config_path=config_path)
+        self.service['storage'] =\
+            storage_service if storage_service else Default.get_storage_instance('codepack', 'storage_service',
+                                                                                 config_path=config_path)
+        self.service['argpack'] = \
+            argpack_service if argpack_service else Default.get_storage_instance('argpack', 'storage_service',
+                                                                                 config_path=config_path)
 
     def make_argpack(self):
-        ret = dict()
-        stack = list()
-        for root in self.roots:
-            stack.append(root)
-            while len(stack):
-                n = stack.pop(-1)
-                if n.id not in ret:
-                    ret[n.id] = dict()
-                for arg, value in n.get_args().items():
-                    if arg not in n.get_dependent_args().keys():
-                        ret[n.id][arg] = value
-                for c in n.children.values():
-                    stack.append(c)
-        return ret
+        return ArgPack(self)
+
+    def save_argpack(self, argpack):
+        self.service['argpack'].save(item=argpack)
+
+    def load_argpack(self, id):
+        return self.service['argpack'].load(id=id)
 
     def set_root(self, code):
         if not isinstance(code, CodeBase):
@@ -80,14 +80,12 @@ def __str__(self):
     def __repr__(self):
         return self.__str__()
 
-    def update_state(self, state, timestamp=None, argpack=None):
-        if state:
-            snapshot = self.to_snapshot(argpack=argpack, timestamp=timestamp)
-            snapshot['state'] = state
-            self.service['snapshot_service'].save(snapshot)
+    def save_snapshot(self, timestamp=None, argpack=None):
+        snapshot = self.to_snapshot(argpack=argpack, timestamp=timestamp)
+        self.service['snapshot'].save(snapshot)
 
     def get_state(self):
-        states = self.root.service['snapshot_service']\
+        states = self.root.service['snapshot']\
             .load(serial_number=[code.serial_number for code in self.codes.values()], projection={'state'})
         if states:
             max_state = 0
@@ -104,8 +102,8 @@ def get_state(self):
                 return State(max_state).name
         return 'UNKNOWN'
 
-    def save(self):
-        self.service['storage_service'].save(item=self)
+    def save(self, update=False):
+        self.service['storage'].save(item=self, update=update)
 
     def get_leaves(self):
         leaves = set()
@@ -156,21 +154,23 @@ def async_run(self, argpack=None):
 
     def __call__(self, argpack=None, sync=True):
         self.init_code_state(state='READY', argpack=argpack)
-        self.update_state('READY', argpack=argpack)
+        self.save_snapshot(argpack=argpack)
         if not argpack:
             argpack = self.make_argpack()
         if sync:
             self.sync_run(argpack=argpack)
-            self.update_state('TERMINATED', argpack=argpack)
         else:
             self.async_run(argpack=argpack)
-            self.update_state(self.get_state(), argpack=argpack)
         return self.get_result()
 
     def init_code_state(self, state, argpack=None):
         for id, code in self.codes.items():
             if argpack and id in argpack:
-                code.update_state(state, kwargs=argpack[id])
+                if isinstance(argpack[id], dict):
+                    _kwargs = argpack[id]
+                else:
+                    _kwargs = argpack[id].to_dict()
+                code.update_state(state, kwargs=_kwargs)
 
     def get_result(self):
         if self.subscribe and self.codes[self.subscribe].get_state() == 'TERMINATED':
diff --git a/codepack/config/__init__.py b/codepack/config/__init__.py
new file mode 100644
index 0000000..7b2739d
--- /dev/null
+++ b/codepack/config/__init__.py
@@ -0,0 +1,3 @@
+from codepack.config.config import Config
+from codepack.config.default import Default
+from codepack.config.alias import Alias
diff --git a/codepack/config/alias.py b/codepack/config/alias.py
new file mode 100644
index 0000000..2685a2a
--- /dev/null
+++ b/codepack/config/alias.py
@@ -0,0 +1,53 @@
+from importlib import import_module
+from codepack.config import Config
+import os
+
+
+class Alias:
+    PREFIX = 'ALIAS'
+
+    def __init__(self, data=None):
+        self.aliases = None
+        if isinstance(data, str):
+            _aliases = Config.parse_config(section='alias', config_path=data)
+            if len(_aliases) == 1 and 'path' in _aliases:
+                alias_path = Config.get_value(section='alias', key='path', config=_aliases)
+                aliases = Config.parse_config(section='alias', config_path=alias_path)
+            else:
+                aliases = _aliases
+            self.aliases = aliases
+        elif isinstance(data, dict):
+            self.aliases = data
+        elif data is None:
+            pass
+        else:
+            raise TypeError(type(data))  # pragma: no cover
+
+    @classmethod
+    def get_env(cls, item):
+        return ('%s_%s_%s' % (Config.PREFIX, cls.PREFIX, item)).upper()
+
+    @staticmethod
+    def get_class(module, name):
+        return getattr(import_module(module), name)
+
+    def __getitem__(self, item):
+        if self.aliases:
+            path = self.aliases[item]
+        elif self.get_env(item) in os.environ:
+            path = os.environ.get(self.get_env(item))
+        elif '%s_ALIAS_PATH' % Config.PREFIX in os.environ:
+            aliases = Config.parse_config(section='alias', config_path=os.environ['%s_ALIAS_PATH' % Config.PREFIX])
+            path = aliases[item]
+        elif Config.LABEL_CONFIG_PATH in os.environ:
+            config = Config.parse_config(section='alias', config_path=os.environ.get(Config.LABEL_CONFIG_PATH), ignore_error=True)
+            alias_path = Config.get_value(section='alias', key='path', config=config)
+            aliases = Config.parse_config(section='alias', config_path=alias_path)
+            path = aliases[item]
+        else:
+            raise AttributeError("%s not found in os.environ['%s'], os.environ['%s'], and os.environ['%s']"
+                                 % (item, self.get_env(item), '%s_ALIAS_PATH' % Config.PREFIX, Config.LABEL_CONFIG_PATH))
+        tokens = path.split('.')
+        module = '.'.join(tokens[:-1])
+        name = tokens[-1]
+        return self.get_class(module, name)
diff --git a/codepack/config/config.py b/codepack/config/config.py
new file mode 100644
index 0000000..4c8966b
--- /dev/null
+++ b/codepack/config/config.py
@@ -0,0 +1,71 @@
+from configparser import ConfigParser
+import os
+
+
+class Config:
+    PREFIX = 'CODEPACK'
+    LABEL_CONFIG_PATH = '%s_CONFIG_PATH' % PREFIX
+
+    def __init__(self, config_path: str = None):
+        self.config_path = config_path
+
+    @staticmethod
+    def parse_config(section: str, config_path: str, ignore_error: bool = False):
+        cp = ConfigParser()
+        cp.read(config_path)
+        if ignore_error and not cp.has_section(section):
+            return None
+        items = cp.items(section)
+        return {item[0]: item[1] for item in items}
+
+    def get_config(self, section: str, config_path: str = None, ignore_error: bool = False):
+        _config_path = config_path
+        if not _config_path:
+            _config_path = self.config_path
+        if not _config_path:
+            _config_path = os.environ.get(self.LABEL_CONFIG_PATH, None)
+        if _config_path:
+            return self.parse_config(section=section, config_path=_config_path)
+        elif ignore_error:
+            return None
+        else:
+            raise AttributeError("path of configuration file should be provided in either 'config_path' or os.environ['%s']"
+                                 % self.LABEL_CONFIG_PATH)
+
+    @classmethod
+    def get_value(cls, section: str, key: str, config: dict = None):
+        env = '%s_%s_%s' % (cls.PREFIX, section.upper(), key.upper())
+        if config:
+            ret = config[key]
+        elif env in os.environ:
+            ret = os.environ.get(env, None)
+        else:
+            raise AssertionError("'%s' information should be provided in os.environ['%s']" % (section, env))
+        return ret
+
+    def get_storage_config(self, section: str, config_path: str = None):
+        config = self.get_config(section=section, config_path=config_path, ignore_error=True)
+        ret = dict()
+        ret['source'] = self.get_value(section=section, key='source', config=config)
+        if ret['source'] == 'memory':
+            pass
+        elif ret['source'] == 'file':
+            ret['path'] = self.get_value(section=section, key='path', config=config)
+        elif ret['source'] == 'mongodb':
+            ret['db'] = self.get_value(section=section, key='db', config=config)
+            ret['collection'] = self.get_value(section=section, key='collection', config=config)
+            conn_config = self.get_config(section='conn', config_path=config_path, ignore_error=True)
+            conn_config_path = self.get_value(section='conn', key='path', config=conn_config)
+            ret['mongodb'] = self.parse_config(section='mongodb', config_path=conn_config_path)
+        elif ret['source'] == 'kafka':
+            ret['topic'] = self.get_value(section=section, key='topic', config=config)
+            conn_config = self.get_config(section='conn', config_path=config_path, ignore_error=True)
+            conn_config_path = self.get_value(section='conn', key='path', config=conn_config)
+            ret['kafka'] = self.parse_config(section='kafka', config_path=conn_config_path)
+        else:
+            raise NotImplementedError("'%s' is not implemented" % ret['source'])
+        if config:
+            for k in config:
+                if k not in ret:
+                    ret[k] = self.get_value(section=section, key=k, config=config)
+        return ret
diff --git a/codepack/config/default.py b/codepack/config/default.py
new file mode 100644
index 0000000..a4a3ce4
--- /dev/null
+++ b/codepack/config/default.py
@@ -0,0 +1,148 @@
+from codepack.config.config import Config
+from codepack.config.alias import Alias
+import os
+from inspect import getfullargspec
+
+
+class Default:
+    config = None
+    alias = None
+    instances = dict()
+
+    @classmethod
+    def __init__(cls, config_path: str = None, alias_path: str = None):
+        cls.init(config_path=config_path, alias_path=alias_path)
+
+    @classmethod
+    def init(cls, config_path: str = None, alias_path: str = None):
+        cls.init_config(config_path=config_path)
+        cls.init_alias(config_path=config_path, alias_path=alias_path)
+        cls.init_instances()
+
+    @classmethod
+    def init_config(cls, config_path: str = None):
+        cls.config = cls._get_config(config_path=config_path)
+
+    @staticmethod
+    def _get_config(config_path: str = None):
+        return Config(config_path=config_path)
+
+    @classmethod
+    def _get_storage_config(cls, section: str, config_path: str = None):
+        if config_path:
+            _config = cls._get_config(config_path=config_path)
+        elif not cls.config:
+            cls.init_config()
+            _config = cls.config
+        else:
+            _config = cls.config
+        return _config.get_storage_config(section=section)
+
+    @classmethod
+    def init_alias(cls, config_path: str = None, alias_path: str = None):
+        cls.alias = cls._get_alias(config_path=config_path, alias_path=alias_path)
+
+    @staticmethod
+    def _get_alias(config_path: str = None, alias_path: str = None):
+        if alias_path:
+            return Alias(data=alias_path)
+        elif config_path:
+            return Alias(data=config_path)
+        else:
+            return Alias()
+
+    @classmethod
+    def init_instances(cls):
+        cls.instances = dict()
+
+    @classmethod
+    def set_config(cls, config: Config):
+        cls.config = config
+
+    @classmethod
+    def set_alias(cls, alias: Alias):
+        cls.alias = alias
+
+    @staticmethod
+    def get_alias_from_source(source: str, prefix: str = None, suffix: str = None):
+        if source == 'mongodb':
+            ret = 'mongo'
+        else:
+            ret = source
+        if prefix:
+            ret = '%s_%s' % (prefix, ret)
+        if suffix:
+            ret = '%s_%s' % (ret, suffix)
+        return ret
+
+    @classmethod
+    def get_class_from_alias(cls, alias: str, config_path: str = None, alias_path: str = None):
+        if config_path or alias_path:
+            _alias = cls._get_alias(config_path=config_path, alias_path=alias_path)
+        elif not cls.alias:
+            cls.init_alias()
+            _alias = cls.alias
+        else:
+            _alias = cls.alias
+        return _alias[alias]
+
+    @classmethod
+    def get_storage_instance(cls, section: str, instance_type: str, config_path: str = None, alias_path: str = None):
+        key = '%s_%s' % (section, instance_type)
+        if config_path or alias_path or key not in cls.instances:
+            storage_config = cls._get_storage_config(section=section, config_path=config_path)
+            source = storage_config.pop('source')
+            alias = cls.get_alias_from_source(source=source, suffix=instance_type)
+            c = cls.get_class_from_alias(alias, config_path=config_path, alias_path=alias_path)
+            item_type = cls.get_class_from_alias(section, config_path=config_path, alias_path=alias_path)
+            _instance = c(item_type=item_type, **storage_config)
+            if config_path is None and alias_path is None:
+                cls.instances[key] = _instance
+            return _instance
+        else:
+            return cls.instances[key]
+
+    @classmethod
+    def get_scheduler(cls, section: str = 'scheduler', config_path: str = None, alias_path: str = None):
+        key = section
+        if config_path or alias_path or key not in cls.instances:
+            storage_config = cls._get_storage_config(section=section, config_path=config_path)
+            source = storage_config.pop('source')
+            alias = cls.get_alias_from_source(source=source, suffix=section)
+            c = cls.get_class_from_alias(alias, config_path=config_path, alias_path=alias_path)
+            if 'CODEPACK_SCHEDULER_SUPERVISOR' in os.environ:
+                storage_config['supervisor'] = os.environ['CODEPACK_SCHEDULER_SUPERVISOR']
+            _instance = c(**storage_config)
+            if config_path is None and alias_path is None:
+                cls.instances[key] = _instance
+            return _instance
+        else:
+            return cls.instances[key]
+
+    @classmethod
+    def get_employee(cls, section: str, config_path: str = None, alias_path: str = None):
+        key = section
+        if 'worker' in section:
+            kafka_type = 'consumer'
+        elif 'supervisor' in section:
+            kafka_type = 'producer'
+        else:
+            raise NotImplementedError("'%s' is unknown")
+        if config_path or alias_path or key not in cls.instances:
+            storage_config = cls._get_storage_config(section=section, config_path=config_path)
+            storage_config.pop('source')
+            kafka_config = storage_config.pop('kafka')
+            c = cls.get_class_from_alias(section, config_path=config_path, alias_path=alias_path)
+            employee_config = dict()
+            for k, v in storage_config.items():
+                if k in getfullargspec(c.__init__).args:
+                    employee_config[k] = v
+                else:
+                    kafka_config[k] = v
+            employee_config[kafka_type] = kafka_config
+            _instance = c(**employee_config)
+            if config_path is None and alias_path is None:
+                cls.instances[key] = _instance
+            return _instance
+        else:
+            return cls.instances[key]
diff --git a/codepack/delivery/__init__.py b/codepack/delivery/__init__.py
new file mode 100644
index 0000000..7fb78d0
--- /dev/null
+++ b/codepack/delivery/__init__.py
@@ -0,0 +1,1 @@
+from codepack.delivery.delivery import Delivery
diff --git a/codepack/utils/delivery.py b/codepack/delivery/delivery.py
similarity index 93%
rename from codepack/utils/delivery.py
rename to codepack/delivery/delivery.py
index 9e7de5a..7d2e099 100644
--- a/codepack/utils/delivery.py
+++ b/codepack/delivery/delivery.py
@@ -5,7 +5,7 @@
 
 class Delivery(Storable):
     def __init__(self, id, serial_number, item=None, timestamp=None):
-        super().__init__(id=id, serial_number=serial_number)
+        Storable.__init__(self, id=id, serial_number=serial_number)
         self.item = item
         self.timestamp = timestamp if timestamp else datetime.now().timestamp()
 
diff --git a/codepack/dependency/__init__.py b/codepack/dependency/__init__.py
new file mode 100644
index 0000000..1a760b8
--- /dev/null
+++ b/codepack/dependency/__init__.py
@@ -0,0 +1,3 @@
+from codepack.dependency.dependency import Dependency
+from codepack.dependency.dependency_manager import DependencyManager
+from codepack.dependency.dependency_state import DependencyState
diff --git a/codepack/utils/dependency.py b/codepack/dependency/dependency.py
similarity index 56%
rename from codepack/utils/dependency.py
rename to codepack/dependency/dependency.py
index b041b23..912ea6b 100644
--- a/codepack/utils/dependency.py
+++ b/codepack/dependency/dependency.py
@@ -1,4 +1,3 @@
-from codepack.base import CodeBase
 from codepack.storage import Storable
 
 
@@ -8,13 +7,13 @@ def __init__(self, code=None, id=None, serial_number=None, arg=None):
         self.code = None
         self.arg = None
         self.bind(code)
-        self.set(id=id, serial_number=serial_number, arg=arg)
+        self.depend_on(id=id, serial_number=serial_number, arg=arg)
 
     def __lshift__(self, sender):
-        if isinstance(sender, CodeBase):
-            self.set(id=sender.id, serial_number=sender.serial_number, arg=self.arg)
+        if isinstance(sender, Storable):
+            self.depend_on(id=sender.id, serial_number=sender.serial_number, arg=self.arg)
         elif isinstance(sender, dict):
-            self.set(id=sender['id'], serial_number=sender['serial_number'], arg=self.arg)
+            self.depend_on(id=sender['id'], serial_number=sender['serial_number'], arg=self.arg)
         else:
             raise TypeError(type(sender))
         self.code.add_dependency(self)
@@ -22,27 +21,32 @@ def __lshift__(self, sender):
     def bind(self, code):
         self.code = code
 
-    def set(self, id=None, serial_number=None, arg=None):
+    def depend_on(self, id=None, serial_number=None, arg=None):
         self.id = id
         self.serial_number = serial_number
         self.arg = arg
 
-    def __hash__(self):
-        return hash(self.serial_number)
-
     def __eq__(self, other):
         if isinstance(other, type(self)):
-            return self.serial_number == other.serial_number
+            ret = True
+            ret &= (self.serial_number == other.serial_number)
+            ret &= (self.id == other.id)
+            ret &= (self.arg == other.arg)
+            return ret
         elif isinstance(other, dict):
-            return self.serial_number == other['serial_number']
+            ret = True
+            ret &= (self.serial_number == other['serial_number'])
+            ret &= (self.id == other['id'])
+            ret &= (self.arg == other['arg'])
+            return ret
         else:
             return False
 
     def __str__(self):
-        return '%s(arg: %s, id: %s)' % (self.__class__.__name__, self.arg, self.id)
+        return '%s(arg: %s, id: %s)' % (self.__class__.__name__, self.arg, self.id)  # pragma: no cover
 
     def __repr__(self):
-        return self.__str__()
+        return self.__str__()  # pragma: no cover
 
     def to_dict(self):
         return {'id': self.id, 'serial_number': self.serial_number, 'arg': self.arg}
diff --git a/codepack/dependency/dependency_manager.py b/codepack/dependency/dependency_manager.py
new file mode 100644
index 0000000..dd36572
--- /dev/null
+++ b/codepack/dependency/dependency_manager.py
@@ -0,0 +1,90 @@
+from codepack.storage import MemoryStorage
+from codepack.dependency.dependency import Dependency
+from codepack.dependency.dependency_state import DependencyState
+from collections.abc import Iterable
+
+
+class DependencyManager(MemoryStorage, Iterable):
+    def __init__(self, code, item_type=Dependency):
+        MemoryStorage.__init__(self, item_type=item_type)
+        self.code = code
+
+    def add(self, dependency):
+        if isinstance(dependency, Dependency):
+            self.code.assert_arg(dependency.arg)
+            self.memory[dependency.serial_number] = dependency
+        elif isinstance(dependency, dict):
+            self.code.assert_arg(dependency['arg'])
+            self.memory[dependency['serial_number']] = Dependency.from_dict(d=dependency)
+            self.memory[dependency['serial_number']].bind(self.code)
+        elif isinstance(dependency, Iterable):
+            for d in dependency:
+                self.add(d)
+        else:
+            raise TypeError(type(dependency))  # pragma: no cover
+
+    def __getitem__(self, item):
+        return self.memory[item]
+
+    def remove(self, serial_number):
+        self.memory.pop(serial_number, None)
+
+    def get_args(self):
+        ret = dict()
+        for dependency in self.memory.values():
+            if dependency.arg:
+                ret[dependency.arg] = dependency.id
+        return ret
+
+    def load_snapshot(self):
+        return self.code.service['snapshot'].load(serial_number=list(self.memory.keys()))
+
+    def check_delivery(self):
+        return self.code.service['delivery'].check(serial_number=[k for k, v in self.memory.items() if v.arg])
+
+    def validate_snapshot(self, snapshot):
+        for s in snapshot:
+            if s['state'] == 'ERROR':
+                return DependencyState.ERROR
+            elif s['state'] != 'TERMINATED':
+                return DependencyState.PENDING
+        if len(self) != len(snapshot):
+            return DependencyState.PENDING
+        return DependencyState.RESOLVED
+
+    def validate_delivery(self, snapshot, delivery):
+        if len(self.get_args()) != len(delivery):
+            return DependencyState.PENDING
+        s = {x['_id']: x for x in snapshot}
+        for d in delivery:
+            serial_number = d['_id']
+            if serial_number not in s:
+                return DependencyState.PENDING
+            elif 'timestamp' not in d or 'timestamp' not in s[serial_number]:
+                return DependencyState.PENDING
+            elif d['timestamp'] != s[serial_number]['timestamp']:
+                return DependencyState.PENDING
+        return DependencyState.RESOLVED
+
+    def get_state(self):
+        snapshot = self.load_snapshot()
+        dependency_state = self.validate_snapshot(snapshot=snapshot)
+        if dependency_state != DependencyState.RESOLVED:
+            return dependency_state
+        delivery = self.check_delivery()
+        return self.validate_delivery(snapshot=snapshot, delivery=delivery)
+
+    def __iter__(self):
+        return self.memory.__iter__()
+
+    def keys(self):
+        return self.memory.keys()
+
+    def values(self):
+        return self.memory.values()
+
+    def items(self):
+        return self.memory.items()
+
+    def __len__(self):
+        return len(self.memory)
diff --git a/codepack/utils/dependency_state.py b/codepack/dependency/dependency_state.py
similarity index 94%
rename from codepack/utils/dependency_state.py
rename to codepack/dependency/dependency_state.py
index f3f799f..21d5ceb 100644
--- a/codepack/utils/dependency_state.py
+++ b/codepack/dependency/dependency_state.py
@@ -2,8 +2,8 @@
 
 
 class DependencyState(Enum):
-    READY = 0
-    NOT_READY = 1
+    RESOLVED = 0
+    PENDING = 1
     ERROR = 2
 
     def __str__(self):
diff --git a/codepack/docker/__init__.py b/codepack/docker/__init__.py
new file mode 100644
index 0000000..f22b9b8
--- /dev/null
+++ b/codepack/docker/__init__.py
@@ -0,0 +1,1 @@
+from codepack.docker.docker_manager import DockerManager
diff --git a/codepack/docker/docker_manager.py b/codepack/docker/docker_manager.py
new file mode 100644
index 0000000..4fda1c7
--- /dev/null
+++ b/codepack/docker/docker_manager.py
@@ -0,0 +1,130 @@
+import os
+from docker.errors import ImageNotFound
+from shutil import rmtree
+from typing import Union
+from docker import DockerClient
+from codepack.interface import Docker
+
+
+class DockerManager:
+    CONTAINER_WORKDIR = '/usr/src/codepack'
+
+    def __init__(self, docker: Union[Docker, DockerClient] = None, path: str = './'):
+        if docker is None:
+            self.docker = Docker()
+        else:
+            self.docker = docker
+        self.path = path
+
+    def get_image(self, image: str):
+        ret = None
+        try:
+            ret = self.docker.images.get(image)
+        except ImageNotFound:
+            ret = self.docker.images.pull(image)
+        finally:
+            return ret
+
+    def pull_image(self, image: str):
+        return self.docker.images.pull(image)
+
+    def push_image(self, image: str):
+        return self.docker.images.push(image)
+
+    def run(self, image: str, **kwargs):
+        return self.docker.containers.run(image=image, volumes=['%s:%s' % (os.path.abspath(self.path), self.CONTAINER_WORKDIR)],
+                                          working_dir=self.CONTAINER_WORKDIR, auto_remove=True, name=id(self), **kwargs)
+
+    @staticmethod
+    def extract_requirements_from_file(path: str):
+        with open(path, 'r') as f:
+            lines = f.readlines()
+        return [line.strip() for line in lines]
+
+    @staticmethod
+    def _collect_requirements_in_dict(r: list, d: dict):
+        for _r in r:
+            tokens = _r.split('==')
+            if len(tokens) == 2:
+                d[tokens[0]] = tokens[1]
+            elif len(tokens) == 1:
+                d[tokens[0]] = None
+            else:
+                raise IndexError(len(tokens))
+
+    @staticmethod
+    def combine_requirements(base: list, additive: list):
+        ret = list()
+        buffer = dict()
+        DockerManager._collect_requirements_in_dict(base, buffer)
+        DockerManager._collect_requirements_in_dict(additive, buffer)
+        for module, version in buffer.items():
+            tmp = module
+            if version:
+                tmp += '==%s' % version
+            ret.append(tmp)
+        return ret
+
+    @staticmethod
+    def make_dockerfile(path: str, base_image: str, args: dict = None, envs: dict = None,
+                        requirements: list = None, pip_options: dict = None):
+        ret = str()
+        if args is None:
+            args = dict()
+        if envs is None:
+            envs = dict()
+        if requirements is None:
+            requirements = list()
+        lines = list()
+        lines.append('FROM %s' % base_image)
+        for k, v in args.items():
+            lines.append('ARG %s %s' % (k, v))
+        for k, v in envs.items():
+            lines.append('ENV %s %s' % (k, v))
+        if len(requirements) > 0:
+            prefix = 'RUN python -m pip install'
+            if pip_options:
+                for k, v in pip_options.items():
+                    if v:
+                        if isinstance(v, list):
+                            for _v in v:
+                                prefix += ' --%s %s' % (k, _v)
+                        elif isinstance(v, str):
+                            prefix += ' --%s %s' % (k, v)
+                        else:
+                            raise TypeError(type(v))
+                    else:
+                        prefix += ' --%s' % k
+            lines.append('%s pip' % prefix)
+            lines.append('%s %s' % (prefix, ' '.join(requirements)))
+        with open(os.path.join(path, 'Dockerfile'), 'w') as f:
+            f.writelines(line + '\n' for line in lines)
+        for line in lines:
+            ret += line + '\n'
+        return ret
+
+    def build_image(self, tag, base_image, args=None, envs=None, requirements=None, pip_options=None, tmp_dir=None):
+        if not tmp_dir:
+            tmp_dir = str(id(self))
+        path = os.path.join(self.path, tmp_dir)
+        new_path = False
+        image = None
+        logs = None
+        try:
+            if self.get_image(base_image) is None:
+                return None, None
+            if not os.path.exists(path):
+                os.makedirs(path)
+                new_path = True
+            self.make_dockerfile(path, base_image=base_image, args=args, envs=envs,
+                                 requirements=requirements, pip_options=pip_options)
+            image, logs = self.docker.images.build(path=path, tag=tag)
+        finally:
+            if new_path:
+                rmtree(path)
+            else:
+                if os.path.exists(os.path.join(path, 'requirements.txt')):
+                    os.remove(os.path.join(path, 'requirements.txt'))
+                if os.path.exists(os.path.join(path, 'Dockerfile')):
+                    os.remove(os.path.join(path, 'Dockerfile'))
+            return image, logs
diff --git a/codepack/employee/__init__.py b/codepack/employee/__init__.py
new file mode 100644
index 0000000..8000191
--- /dev/null
+++ b/codepack/employee/__init__.py
@@ -0,0 +1,2 @@
+from codepack.employee.worker import Worker
+from codepack.employee.supervisor import Supervisor
diff --git a/codepack/employee/supervisor.py b/codepack/employee/supervisor.py
new file mode 100644
index 0000000..8be98c0
--- /dev/null
+++ b/codepack/employee/supervisor.py
@@ -0,0 +1,49 @@
+from codepack import Code, CodePack
+from codepack.config import Default
+from codepack.snapshot import CodeSnapshot
+from codepack.storage import KafkaStorage
+
+
+class Supervisor(KafkaStorage):
+    def __init__(self, producer=None, topic=None, snapshot_service=None, producer_config=None):
+        KafkaStorage.__init__(self, producer=producer, topic=topic, producer_config=producer_config)
+        self.snapshot_service =\
+            snapshot_service if snapshot_service else Default.get_storage_instance('code_snapshot',
+                                                                                   'snapshot_service')
+
+    def run_code(self, code, args=None, kwargs=None):
+        if isinstance(code, Code):
+            code.update_state('READY')
+            if kwargs and not isinstance(kwargs, dict):
+                _kwargs = kwargs.to_dict()
+            else:
+                _kwargs = kwargs
+            self.producer.produce(self.topic, value=code.to_snapshot(args=args, kwargs=_kwargs).to_dict())
+        else:
+            raise TypeError(type(code))
+
+    def run_codepack(self, codepack, argpack=None):
+        if isinstance(codepack, CodePack):
+            codepack.save_snapshot(argpack=argpack)
+            codepack.init_code_state(state='READY', argpack=argpack)
+            for id, code in codepack.codes.items():
+                _kwargs = argpack[id] if id in argpack else None
+                self.run_code(code=code, kwargs=_kwargs)
+        else:
+            raise TypeError(type(codepack))
+
+    def organize(self, serial_number=None):
+        for snapshot in self.snapshot_service.search(key='state', value='WAITING'):
+            resolved = True
+            dependent_serial_numbers = [snapshot['serial_number'] for snapshot in snapshot['dependency']]
+            dependencies = self.snapshot_service.load(dependent_serial_numbers, projection={'state'})
+            if serial_number and serial_number not in {dependency['serial_number'] for dependency in dependencies}:
+                continue
+            for dependency in dependencies:
+                if dependency['state'] != 'TERMINATED':
+                    resolved = False
+                    break
+            if resolved:
+                code_snapshot = CodeSnapshot.from_dict(snapshot)
+                code = Code.from_snapshot(code_snapshot)
+                self.run_code(code=code, args=code_snapshot.args, kwargs=code_snapshot.kwargs)
diff --git a/codepack/employee/worker.py b/codepack/employee/worker.py
new file mode 100644
index 0000000..bfad9d0
--- /dev/null
+++ b/codepack/employee/worker.py
@@ -0,0 +1,51 @@
+import requests
+from codepack import Code
+from codepack.snapshot import CodeSnapshot
+from codepack.employee.supervisor import Supervisor
+from codepack.storage import KafkaStorage
+
+
+class Worker(KafkaStorage):
+    def __init__(self, consumer=None, interval=1, callback=None, supervisor=None, consumer_config=None):
+        KafkaStorage.__init__(self, consumer=consumer, consumer_config=consumer_config)
+        self.interval = interval
+        self.supervisor = supervisor
+        self.callback = callback
+        if self.supervisor and not self.callback:
+            if isinstance(self.supervisor, str) or isinstance(self.supervisor, Supervisor):
+                self.callback = self.inform_supervisor_of_termination
+            else:
+                self.close()
+                raise TypeError(type(self.supervisor))
+        self.register(callback=self.callback)
+
+    def register(self, callback):
+        self.callback = callback
+
+    def start(self):
+        self.consumer.consume(self.work, timeout_ms=int(float(self.interval) * 1000))
+
+    def stop(self):
+        pass
+
+    def work(self, buffer):
+        for tp, msgs in buffer.items():
+            for msg in msgs:
+                code = None
+                try:
+                    snapshot = CodeSnapshot.from_dict(msg.value)
+                    code = Code.from_snapshot(snapshot)
+                    code.register(callback=self.callback)
+                    code(*snapshot.args, **snapshot.kwargs)
+                except Exception as e:
+                    print(e)  # log.error(e)
+                    if code is not None:
+                        code.update_state('ERROR')
+                    continue
+
+    def inform_supervisor_of_termination(self, x):
+        if x['state'] == 'TERMINATED':
+            if isinstance(self.supervisor, str):
+                requests.get(self.supervisor + '/organize/%s' % x['serial_number'])
+            elif isinstance(self.supervisor, Supervisor):
+                self.supervisor.organize(x['serial_number'])
diff --git a/codepack/engine/engine.py b/codepack/engine/engine.py
index a99fb8f..30a8090 100644
--- a/codepack/engine/engine.py
+++ b/codepack/engine/engine.py
@@ -1,5 +1,5 @@
 import abc
-from codepack.service import DefaultService
+from codepack.config import Default
 
 
 class Engine(metaclass=abc.ABCMeta):
@@ -7,7 +7,10 @@ def __init__(self, callback, interval=1, config_path=None, snapshot_service=None
         """initialize instance"""
         self.callback = callback
         self.interval = interval
-        self.snapshot_service = snapshot_service if snapshot_service else DefaultService.get_default_code_snapshot_service(config_path=config_path)
+        self.snapshot_service =\
+            snapshot_service if snapshot_service else Default.get_storage_instance('code_snapshot',
+                                                                                   'snapshot_service',
+                                                                                   config_path=config_path)
 
     @abc.abstractmethod
     def start(self):
diff --git a/codepack/interface/__init__.py b/codepack/interface/__init__.py
index f5a9c54..0b4ba5d 100644
--- a/codepack/interface/__init__.py
+++ b/codepack/interface/__init__.py
@@ -5,3 +5,6 @@
 from codepack.interface.mssql import MSSQL
 from codepack.interface.dynamodb import DynamoDB
 from codepack.interface.oracledb import OracleDB
+from codepack.interface.docker import Docker
+from codepack.interface.kafka_consumer import KafkaConsumer
+from codepack.interface.kafka_producer import KafkaProducer
diff --git a/codepack/interface/docker.py b/codepack/interface/docker.py
new file mode 100644
index 0000000..182d9aa
--- /dev/null
+++ b/codepack/interface/docker.py
@@ -0,0 +1,22 @@
+from docker import DockerClient
+from codepack.interface import Interface
+
+
+class Docker(Interface):
+    def __init__(self, config=None, *args, **kwargs):
+        super().__init__(config if config else dict())
+        self.connect(*args, **kwargs)
+
+    def connect(self, *args, **kwargs):
+        self.session = DockerClient(*args, **self.config, **kwargs)
+        self._closed = False
+        return self.session
+
+    def __getattr__(self, item):
+        assert not self.closed(), "connection is closed"
+        return getattr(self.session, item)
+
+    def close(self):
+        self.session.close()
+        if not self.closed():
+            self._closed = True
diff --git a/codepack/interface/dynamodb.py b/codepack/interface/dynamodb.py
index a878548..bd6d9db 100644
--- a/codepack/interface/dynamodb.py
+++ b/codepack/interface/dynamodb.py
@@ -12,6 +12,7 @@ def __init__(self, config, *args, **kwargs):
 
     def connect(self, *args, **kwargs):
         self.session = boto3.client(config=Config(retries=dict(max_attempts=3)), *args, **self.config, **kwargs)
+        self._closed = False
         return self.session
 
     def list_tables(self, name):
@@ -67,7 +68,7 @@ def array_parser(s, sep='\x7f', dtype=str):
             return s
 
     def close(self):
-        pass
+        self._closed = True
 
     @staticmethod
     def do_nothing(x, *args, **kwargs):
diff --git a/codepack/interface/interface.py b/codepack/interface/interface.py
index 163618f..d0b3653 100644
--- a/codepack/interface/interface.py
+++ b/codepack/interface/interface.py
@@ -1,6 +1,6 @@
 import abc
 from sshtunnel import SSHTunnelForwarder
-from codepack.utils.config import get_config
+from codepack.config import Config
 from copy import deepcopy
 
 
@@ -10,16 +10,16 @@ def __init__(self, config):
         self.ssh_config = None
         self.ssh = None
         self.session = None
-        self.closed = True
+        self._closed = True
         self.init_config(config)
 
     def init_config(self, config):
         _config = deepcopy(config)
-        if 'sshtunnel' in _config:
+        if _config and 'sshtunnel' in _config:
             _ssh_config = _config.pop('sshtunnel')
             if isinstance(_ssh_config, str):
                 config_path, section = _ssh_config.split(':')
-                self.ssh_config = get_config(filename=config_path, section=section)
+                self.ssh_config = Config.parse_config(section=section, config_path=config_path)
             elif isinstance(_ssh_config, dict):
                 self.ssh_config = _ssh_config
             else:
@@ -34,6 +34,9 @@ def connect(self, *args, **kwargs):
     def close(self):
         """close the connection to the server"""
 
+    def closed(self):
+        return self._closed
+
     @staticmethod
     def exclude_keys(d, keys):
         return {k: v for k, v in d.items() if k not in keys}
diff --git a/codepack/interface/kafka_consumer.py b/codepack/interface/kafka_consumer.py
new file mode 100644
index 0000000..ed3e9e1
--- /dev/null
+++ b/codepack/interface/kafka_consumer.py
@@ -0,0 +1,43 @@
+from kafka import KafkaConsumer as Kafka
+from codepack.interface import Interface
+from copy import deepcopy
+import json
+
+
+class KafkaConsumer(Interface):
+    def __init__(self, config, *args, **kwargs):
+        super().__init__(config)
+        self.connect(*args, **kwargs)
+
+    def connect(self, *args, **kwargs):
+        if 'value_deserializer' not in self.config and 'value_deserializer' not in kwargs:
+            kwargs['value_deserializer'] = lambda x: json.loads(x.decode('utf-8'))
+        if 'topic' in self.config:
+            _config = deepcopy(self.config)
+            _topic = _config.pop('topic')
+            self.session = Kafka(_topic, *args, **_config, **kwargs)
+        else:
+            self.session = Kafka(*args, **self.config, **kwargs)
+        self._closed = False
+        return self.session
+
+    def __getattr__(self, item):
+        assert not self.closed(), "connection is closed"
+        return getattr(self.session, item)
+
+    def consume(self, callback, *args, **kwargs):
+        while True:
+            try:
+                buffer = self.session.poll(*args, **kwargs)
+                if buffer:
+                    callback(buffer)
+            except KeyboardInterrupt:
+                break
+
+    def close(self):
+        self.session.close()
+        if not self.closed():
+            if self.ssh_config and self.ssh is not None:
+                self.ssh.stop()
+                self.ssh = None
+            self._closed = True
diff --git a/codepack/interface/kafka_producer.py b/codepack/interface/kafka_producer.py
new file mode 100644
index 0000000..7096530
--- /dev/null
+++ b/codepack/interface/kafka_producer.py
@@ -0,0 +1,33 @@
+from kafka import KafkaProducer as Kafka
+from codepack.interface import Interface
+from copy import deepcopy
+import json
+
+
+class KafkaProducer(Interface):
+    def __init__(self, config, *args, **kwargs):
+        super().__init__(config)
+        self.connect(*args, **kwargs)
+
+    def connect(self, *args, **kwargs):
+        if 'value_serializer' not in self.config and 'value_serializer' not in kwargs:
+            kwargs['value_serializer'] = lambda x: json.dumps(x).encode('utf-8')
+        self.session = Kafka(*args, **self.config, **kwargs)
+        self._closed = False
+        return self.session
+
+    def __getattr__(self, item):
+        assert not self.closed(), "connection is closed"
+        return getattr(self.session, item)
+
+    def produce(self, *args, **kwargs):
+        self.session.send(*args, **kwargs)
+        self.session.flush()
+
+    def close(self):
+        self.session.close()
+        if not self.closed():
+            if self.ssh_config and self.ssh is not None:
+                self.ssh.stop()
+                self.ssh = None
+            self._closed = True
diff --git a/codepack/interface/mongodb.py b/codepack/interface/mongodb.py
index ec79e9e..b79f5c2 100644
--- a/codepack/interface/mongodb.py
+++ b/codepack/interface/mongodb.py
@@ -13,21 +13,21 @@ def connect(self, *args, **kwargs):
         if 'connect' in _config:
             _config['connect'] = self.eval_bool(_config['connect'])
         self.session = MongoClient(host=host, port=port, *args, **_config, **kwargs)
-        self.closed = False
+        self._closed = False
         return self.session
 
     def __getitem__(self, item):
-        assert not self.closed, "connection is closed"
+        assert not self.closed(), "connection is closed"
         return self.session[item]
 
     def __getattr__(self, item):
-        assert not self.closed, "connection is closed"
+        assert not self.closed(), "connection is closed"
         return self.__getitem__(item)
 
     def close(self):
         self.session.close()
-        if not self.closed:
+        if not self.closed():
             if self.ssh_config and self.ssh is not None:
                 self.ssh.stop()
                 self.ssh = None
-            self.closed = True
+            self._closed = True
diff --git a/codepack/interface/mssql.py b/codepack/interface/mssql.py
index 0608c55..09a7217 100644
--- a/codepack/interface/mssql.py
+++ b/codepack/interface/mssql.py
@@ -18,11 +18,11 @@ def connect(self, *args, **kwargs):
             self.as_dict = self.eval_bool(kwargs['as_dict'])
         _config['as_dict'] = self.as_dict
         self.session = pymssql.connect(host=host, port=port, *args, **_config, **kwargs)
-        self.closed = False
+        self._closed = False
         return self.session
 
     def query(self, q, commit=False):
-        assert not self.closed, "connection is closed"
+        assert not self.closed(), "connection is closed"
         columns = None
         rows = None
         try:
@@ -120,9 +120,9 @@ def delete(self, db, table, commit=False, **kwargs):
             return None
 
     def close(self):
-        if not self.closed:
+        if not self.closed():
             self.session.close()
             if self.ssh_config and self.ssh is not None:
                 self.ssh.stop()
                 self.ssh = None
-            self.closed = True
+            self._closed = True
diff --git a/codepack/interface/mysql.py b/codepack/interface/mysql.py
index 6123823..3da708f 100644
--- a/codepack/interface/mysql.py
+++ b/codepack/interface/mysql.py
@@ -14,7 +14,7 @@ def connect(self, *args, **kwargs):
         if 'cursorclass' in _config:
             _config['cursorclass'] = self.eval_cursor_object(_config['cursorclass'])
         self.session = pymysql.connect(host=host, port=port, *args, **_config, **kwargs)
-        self.closed = False
+        self._closed = False
         return self.session
 
     @staticmethod
@@ -33,7 +33,7 @@ def rollback(self):
         self.session.rollback()
 
     def query(self, q, commit=False):
-        assert not self.closed, "connection is closed"
+        assert not self.closed(), "connection is closed"
         columns = None
         rows = None
         try:
@@ -133,9 +133,9 @@ def make_call_query(db, procedure, **kwargs):
         return q
 
     def close(self):
-        if not self.closed:
+        if not self.closed():
             self.session.close()
             if self.ssh_config and self.ssh is not None:
                 self.ssh.stop()
                 self.ssh = None
-            self.closed = True
+            self._closed = True
diff --git a/codepack/interface/oracledb.py b/codepack/interface/oracledb.py
index b7fae50..4a89de1 100644
--- a/codepack/interface/oracledb.py
+++ b/codepack/interface/oracledb.py
@@ -32,11 +32,11 @@ def connect(self, *args, **kwargs):
             kwargs = self.exclude_keys(kwargs, keys=['as_dict'])
         _config = self.exclude_keys(self.config, keys=exclude_keys)
         self.session = cx_Oracle.connect(dsn=dsn, *args, **_config, **kwargs)
-        self.closed = False
+        self._closed = False
         return self.session
 
     def query(self, q, commit=False):
-        assert not self.closed, "connection is closed"
+        assert not self.closed(), "connection is closed"
         columns = None
         rows = None
         try:
@@ -66,9 +66,9 @@ def query(self, q, commit=False):
                 return None
 
     def close(self):
-        if not self.closed:
+        if not self.closed():
             self.session.close()
             if self.ssh_config and self.ssh is not None:
                 self.ssh.stop()
                 self.ssh = None
-            self.closed = True
+            self._closed = True
diff --git a/codepack/scheduler/__init__.py b/codepack/scheduler/__init__.py
index 50639b4..9463a71 100644
--- a/codepack/scheduler/__init__.py
+++ b/codepack/scheduler/__init__.py
@@ -1,2 +1,20 @@
 from codepack.scheduler.scheduler import Scheduler
 from codepack.scheduler.mongo_scheduler import MongoScheduler
+from codepack.config import Config
+import os
+
+
+def get_default_scheduler(config_path=None):
+    config = Config(config_path=config_path)
+    storage_config = config.get_storage_config(section='scheduler')
+    source = storage_config.pop('source', None)
+    supervisor = storage_config.pop('supervisor', None)
+    if 'CODEPACK_SCHEDULER_SUPERVISOR' in os.environ:
+        supervisor = os.environ['CODEPACK_SCHEDULER_SUPERVISOR']
+    # cls = SchedulerAlias[source].value
+    scheduler = MongoScheduler(**storage_config)
+    if supervisor:
+        if 'CODEPACK_SCHEDULER_SUPERVISOR' not in os.environ:
+            os.environ['CODEPACK_SCHEDULER_SUPERVISOR'] = supervisor
+        scheduler.register(scheduler.request_to_supervisor)
+    return scheduler
diff --git a/codepack/scheduler/jobstore/mongo_jobstore.py b/codepack/scheduler/jobstore/mongo_jobstore.py
index b389e85..52ce9a7 100644
--- a/codepack/scheduler/jobstore/mongo_jobstore.py
+++ b/codepack/scheduler/jobstore/mongo_jobstore.py
@@ -21,6 +21,7 @@ def add_job(self, job):
                 '_id': job.id,
                 'trigger': job.trigger.__str__(),
                 'codepack': job.kwargs['snapshot']['id'],
+                'argpack': job.kwargs['snapshot']['argpack']['_id'],
                 'snapshot': job.kwargs['snapshot']['serial_number'],
                 'last_run_time': datetime.now(timezone.utc).timestamp(),
                 'next_run_time': utc_timestamp,
diff --git a/codepack/scheduler/mongo_scheduler.py b/codepack/scheduler/mongo_scheduler.py
index 20d2a20..0775e85 100644
--- a/codepack/scheduler/mongo_scheduler.py
+++ b/codepack/scheduler/mongo_scheduler.py
@@ -1,13 +1,12 @@
 from codepack.scheduler import Scheduler
 from codepack.scheduler.jobstore import MongoJobStore
-from codepack import CodePack
 from codepack.storage import MongoStorage
 
 
 class MongoScheduler(MongoStorage, Scheduler):
-    def __init__(self, mongodb=None, db=None, collection=None, blocking=False, *args, **kwargs):
-        MongoStorage.__init__(self, obj=CodePack, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
-        Scheduler.__init__(self, blocking=blocking)
+    def __init__(self, mongodb=None, db=None, collection=None, blocking=False, callback=None, supervisor=None, *args, **kwargs):
+        MongoStorage.__init__(self, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
+        Scheduler.__init__(self, blocking=blocking, callback=callback, supervisor=supervisor)
 
     def get_jobstore(self):
         return MongoJobStore(db=self.db, collection=self.collection, client=self.mongodb.session)
diff --git a/codepack/scheduler/scheduler.py b/codepack/scheduler/scheduler.py
index f212031..c855dea 100644
--- a/codepack/scheduler/scheduler.py
+++ b/codepack/scheduler/scheduler.py
@@ -1,15 +1,34 @@
 import abc
 from apscheduler.schedulers.background import BackgroundScheduler, BlockingScheduler
 from codepack import CodePack
+from codepack.argpack import ArgPack
 from codepack.snapshot import CodePackSnapshot
+from codepack.employee import Supervisor
+import requests
+import json
 
 
 class Scheduler(metaclass=abc.ABCMeta):
-    def __init__(self, blocking=False):
+    supervisor = None
+
+    def __init__(self, callback=None, blocking=False, supervisor=None):
         self.scheduler = None
         self.jobstores = dict()
         self.jobstores['codepack'] = self.get_jobstore()
+        self.callback = None
         self.init_scheduler(blocking=blocking)
+        self.register(callback)
+        if isinstance(supervisor, str) or isinstance(supervisor, Supervisor) or supervisor is None:
+            self.init_supervisor(supervisor=supervisor)
+        else:
+            raise TypeError(type(supervisor))
+
+    @classmethod
+    def init_supervisor(cls, supervisor):
+        Scheduler.supervisor = supervisor
+
+    def register(self, callback):
+        self.callback = callback
 
     def start(self):
         try:
@@ -23,6 +42,9 @@ def init_scheduler(self, blocking=False, **kwargs):
         else:
             self.scheduler = BackgroundScheduler(jobstores=self.jobstores, **kwargs)
 
+    def is_running(self):
+        return self.scheduler.running
+
     @abc.abstractmethod
     def get_jobstore(self):
         """get jobstore"""
@@ -33,22 +55,67 @@ def stop(self):
     def add_job(self, func, job_id, trigger, **kwargs):
         return self.scheduler.add_job(func=func, id=job_id, trigger=trigger, jobstore='codepack', **kwargs)
 
-    def add_codepack(self, codepack, trigger, job_id=None, argpack=None, **kwargs):
+    def add_codepack(self, codepack, trigger, job_id=None, argpack=None, callback=None, **kwargs):
         if job_id is None:
             job_id = codepack.id
         snapshot = codepack.to_snapshot(argpack=argpack)
-        ret = self.add_job(self.run_codepack, job_id=job_id, trigger=trigger, kwargs={'snapshot': snapshot.to_dict()}, **kwargs)
+        ret = self.add_job(self._get_callback(callback), job_id=job_id, trigger=trigger, kwargs={'snapshot': snapshot.to_dict()}, **kwargs)
+        return ret
+
+    def add_snapshot(self, snapshot, trigger, job_id=None, callback=None, **kwargs):
+        if job_id is None:
+            job_id = snapshot.id
+        ret = self.add_job(self._get_callback(callback), job_id=job_id, trigger=trigger, kwargs={'snapshot': snapshot.to_dict()}, **kwargs)
         return ret
 
     def remove_job(self, job_id, **kwargs):
         return self.scheduler.remove_job(job_id=job_id, **kwargs)
 
+    def _get_callback(self, callback=None):
+        if callback:
+            _callback = callback
+        elif self.callback:
+            _callback = self.callback
+        elif self.supervisor:
+            _callback = self.request_to_supervisor
+        else:
+            _callback = self.run_snapshot
+        return _callback
+
     @staticmethod
-    def run_codepack(snapshot):
+    def _get_snapshot(snapshot):
         if isinstance(snapshot, dict):
-            codepack = CodePack.from_snapshot(CodePackSnapshot.from_dict(snapshot))
+            _snapshot = CodePackSnapshot.from_dict(snapshot)
         elif isinstance(snapshot, CodePackSnapshot):
-            codepack = CodePack.from_snapshot(snapshot)
+            _snapshot = snapshot
+        elif isinstance(snapshot, str):
+            _snapshot = CodePackSnapshot.from_json(snapshot)
         else:
             raise TypeError(type(snapshot))
-        return codepack(argpack=snapshot['argpack'])
+        return _snapshot
+
+    @staticmethod
+    def _get_codepack(snapshot):
+        return CodePack.from_snapshot(snapshot)
+
+    @staticmethod
+    def _get_argpack(snapshot):
+        return ArgPack.from_dict(snapshot['argpack'])
+
+    @staticmethod
+    def request_to_supervisor(snapshot):
+        _snapshot = Scheduler._get_snapshot(snapshot)
+        if isinstance(Scheduler.supervisor, str):
+            return requests.post('%s/codepack/run/snapshot' % Scheduler.supervisor,
+                                 data=json.dumps({'snapshot': _snapshot.to_json()}))
+        else:
+            codepack = Scheduler._get_codepack(_snapshot)
+            argpack = Scheduler._get_argpack(_snapshot)
+            return Scheduler.supervisor.run_codepack(codepack=codepack, argpack=argpack)
+
+    @staticmethod
+    def run_snapshot(snapshot):
+        _snapshot = Scheduler._get_snapshot(snapshot)
+        codepack = Scheduler._get_codepack(_snapshot)
+        argpack = Scheduler._get_argpack(_snapshot)
+        return codepack(argpack=argpack)
diff --git a/codepack/service/__init__.py b/codepack/service/__init__.py
index ad24262..60e64c1 100644
--- a/codepack/service/__init__.py
+++ b/codepack/service/__init__.py
@@ -1,4 +1,4 @@
-from codepack.service.delivery_service import MemoryDeliveryService, FileDeliveryService, MongoDeliveryService, DeliveryServiceAlias
-from codepack.service.storage_service import MemoryStorageService, FileStorageService, MongoStorageService, StorageServiceAlias
-from codepack.service.snapshot_service import MemorySnapshotService, FileSnapshotService, MongoSnapshotService, SnapshotServiceAlias
-from codepack.service.default_service import DefaultService
+from codepack.service.delivery_service import MemoryDeliveryService, FileDeliveryService, MongoDeliveryService
+from codepack.service.storage_service import MemoryStorageService, FileStorageService, MongoStorageService
+from codepack.service.snapshot_service import MemorySnapshotService, FileSnapshotService, MongoSnapshotService
+from codepack.service.service import Service
diff --git a/codepack/service/default_service.py b/codepack/service/default_service.py
deleted file mode 100644
index 84cac00..0000000
--- a/codepack/service/default_service.py
+++ /dev/null
@@ -1,73 +0,0 @@
-from codepack.service.delivery_service import DeliveryServiceAlias
-from codepack.service.storage_service import StorageServiceAlias
-from codepack.service.snapshot_service import SnapshotServiceAlias
-from codepack.utils import Singleton
-from codepack.snapshot import CodeSnapshot, CodePackSnapshot
-from codepack.utils.delivery import Delivery
-from codepack.utils.config import get_default_service_config
-
-
-class DefaultService(Singleton):
-    delivery_service = None
-    code_storage_service = None
-    code_snapshot_service = None
-    codepack_storage_service = None
-    codepack_snapshot_service = None
-
-    @classmethod
-    def init(cls):
-        cls.delivery_service = None
-        cls.code_storage_service = None
-        cls.code_snapshot_service = None
-        cls.codepack_storage_service = None
-        cls.codepack_snapshot_service = None
-
-    @staticmethod
-    def get_delivery_service(source, *args, **kwargs):
-        cls = DeliveryServiceAlias[source].value
-        return cls(*args, **kwargs)
-
-    @staticmethod
-    def get_snapshot_service(source, *args, **kwargs):
-        cls = SnapshotServiceAlias[source].value
-        return cls(*args, **kwargs)
-
-    @staticmethod
-    def get_storage_service(source, *args, **kwargs):
-        cls = StorageServiceAlias[source].value
-        return cls(*args, **kwargs)
-
-    @classmethod
-    def get_default_delivery_service(cls, config_path=None):
-        if not cls.delivery_service:
-            config = get_default_service_config(section='cache', config_path=config_path)
-            cls.delivery_service = cls.get_delivery_service(obj=Delivery, **config)
-        return cls.delivery_service
-
-    @classmethod
-    def get_default_code_storage_service(cls, obj, config_path=None):
-        if not cls.code_storage_service:
-            config = get_default_service_config(section='code_storage', config_path=config_path)
-            cls.code_storage_service = cls.get_storage_service(obj=obj, **config)
-        return cls.code_storage_service
-
-    @classmethod
-    def get_default_code_snapshot_service(cls, config_path=None):
-        if not cls.code_snapshot_service:
-            config = get_default_service_config(section='code_snapshot', config_path=config_path)
-            cls.code_snapshot_service = cls.get_snapshot_service(obj=CodeSnapshot, **config)
-        return cls.code_snapshot_service
-
-    @classmethod
-    def get_default_codepack_storage_service(cls, obj, config_path=None):
-        if not cls.codepack_storage_service:
-            config = get_default_service_config(section='codepack_storage', config_path=config_path)
-            cls.codepack_storage_service = cls.get_storage_service(obj=obj, **config)
-        return cls.codepack_storage_service
-
-    @classmethod
-    def get_default_codepack_snapshot_service(cls, config_path=None):
-        if not cls.codepack_snapshot_service:
-            config = get_default_service_config(section='codepack_snapshot', config_path=config_path)
-            cls.codepack_snapshot_service = cls.get_snapshot_service(obj=CodePackSnapshot, **config)
-        return cls.codepack_snapshot_service
diff --git a/codepack/service/delivery_service/__init__.py b/codepack/service/delivery_service/__init__.py
index d4c0589..e1dd3e7 100644
--- a/codepack/service/delivery_service/__init__.py
+++ b/codepack/service/delivery_service/__init__.py
@@ -2,11 +2,3 @@
 from codepack.service.delivery_service.memory_delivery_service import MemoryDeliveryService
 from codepack.service.delivery_service.file_delivery_service import FileDeliveryService
 from codepack.service.delivery_service.mongo_delivery_service import MongoDeliveryService
-
-from enum import Enum
-
-
-class DeliveryServiceAlias(Enum):
-    MEMORY = MemoryDeliveryService
-    FILE = FileDeliveryService
-    MONGODB = MongoDeliveryService
diff --git a/codepack/service/delivery_service/file_delivery_service.py b/codepack/service/delivery_service/file_delivery_service.py
index 606df28..4d74f1c 100644
--- a/codepack/service/delivery_service/file_delivery_service.py
+++ b/codepack/service/delivery_service/file_delivery_service.py
@@ -5,21 +5,21 @@
 
 
 class FileDeliveryService(DeliveryService, FileStorage):
-    def __init__(self, obj=None, path='./'):
-        FileStorage.__init__(self, obj=obj, path=path)
+    def __init__(self, item_type=None, path='./'):
+        FileStorage.__init__(self, item_type=item_type, path=path)
 
     def send(self, id, serial_number, item=None, timestamp=None):
-        self.obj(id=id, serial_number=serial_number, item=item, timestamp=timestamp)\
-            .to_file(path=self.obj.get_path(serial_number=serial_number, path=self.path))
+        self.item_type(id=id, serial_number=serial_number, item=item, timestamp=timestamp)\
+            .to_file(path=self.item_type.get_path(serial_number=serial_number, path=self.path))
 
     def receive(self, serial_number):
-        return self.obj.from_file(path=self.obj.get_path(serial_number=serial_number, path=self.path)).receive()
+        return self.item_type.from_file(path=self.item_type.get_path(serial_number=serial_number, path=self.path)).receive()
 
     def check(self, serial_number):
         if isinstance(serial_number, str):
             d = None
             try:
-                d = self.obj.from_file(path=self.obj.get_path(serial_number=serial_number, path=self.path)).to_dict()
+                d = self.item_type.from_file(path=self.item_type.get_path(serial_number=serial_number, path=self.path)).to_dict()
                 d.pop('item', None)
             finally:
                 return d
@@ -34,4 +34,4 @@ def check(self, serial_number):
             raise TypeError(type(serial_number))  # pragma: no cover
 
     def cancel(self, serial_number):
-        os.remove(self.obj.get_path(serial_number=serial_number, path=self.path))
+        os.remove(self.item_type.get_path(serial_number=serial_number, path=self.path))
diff --git a/codepack/service/delivery_service/memory_delivery_service.py b/codepack/service/delivery_service/memory_delivery_service.py
index 98ad8e5..a771ae6 100644
--- a/codepack/service/delivery_service/memory_delivery_service.py
+++ b/codepack/service/delivery_service/memory_delivery_service.py
@@ -4,11 +4,11 @@
 
 
 class MemoryDeliveryService(DeliveryService, MemoryStorage):
-    def __init__(self, obj=None):
-        MemoryStorage.__init__(self, obj=obj)
+    def __init__(self, item_type=None):
+        MemoryStorage.__init__(self, item_type=item_type)
 
     def send(self, id, serial_number, item=None, timestamp=None):
-        self.memory[serial_number] = self.obj(id=id, serial_number=serial_number, item=item, timestamp=timestamp)
+        self.memory[serial_number] = self.item_type(id=id, serial_number=serial_number, item=item, timestamp=timestamp)
 
     def receive(self, serial_number):
         return self.memory[serial_number].receive()
diff --git a/codepack/service/delivery_service/mongo_delivery_service.py b/codepack/service/delivery_service/mongo_delivery_service.py
index 2290ef1..6d0844f 100644
--- a/codepack/service/delivery_service/mongo_delivery_service.py
+++ b/codepack/service/delivery_service/mongo_delivery_service.py
@@ -5,11 +5,11 @@
 
 
 class MongoDeliveryService(DeliveryService, MongoStorage):
-    def __init__(self, obj=None, db=None, collection=None, mongodb=None, *args, **kwargs):
-        MongoStorage.__init__(self, obj=obj, db=db, collection=collection, mongodb=mongodb, *args, **kwargs)
+    def __init__(self, item_type=None, db=None, collection=None, mongodb=None, *args, **kwargs):
+        MongoStorage.__init__(self, item_type=item_type, db=db, collection=collection, mongodb=mongodb, *args, **kwargs)
 
     def send(self, id, serial_number, item=None, timestamp=None):
-        d = self.obj(id=id, serial_number=serial_number, item=item, timestamp=timestamp).to_dict()
+        d = self.item_type(id=id, serial_number=serial_number, item=item, timestamp=timestamp).to_dict()
         d.pop('_id', None)
         self.mongodb[self.db][self.collection].update_one({'_id': serial_number}, {'$set': d}, upsert=True)
 
diff --git a/codepack/service/snapshot_service/__init__.py b/codepack/service/snapshot_service/__init__.py
index 9bc7e6e..93d23dc 100644
--- a/codepack/service/snapshot_service/__init__.py
+++ b/codepack/service/snapshot_service/__init__.py
@@ -2,10 +2,3 @@
 from codepack.service.snapshot_service.memory_snapshot_service import MemorySnapshotService
 from codepack.service.snapshot_service.file_snapshot_service import FileSnapshotService
 from codepack.service.snapshot_service.mongo_snapshot_service import MongoSnapshotService
-from enum import Enum
-
-
-class SnapshotServiceAlias(Enum):
-    MEMORY = MemorySnapshotService
-    FILE = FileSnapshotService
-    MONGODB = MongoSnapshotService
diff --git a/codepack/service/snapshot_service/file_snapshot_service.py b/codepack/service/snapshot_service/file_snapshot_service.py
index 3644b65..4821d35 100644
--- a/codepack/service/snapshot_service/file_snapshot_service.py
+++ b/codepack/service/snapshot_service/file_snapshot_service.py
@@ -6,19 +6,19 @@
 
 
 class FileSnapshotService(SnapshotService, FileStorage):
-    def __init__(self, obj=None, path='./'):
-        FileStorage.__init__(self, obj=obj, path=path)
+    def __init__(self, item_type=None, path='./'):
+        FileStorage.__init__(self, item_type=item_type, path=path)
 
     def save(self, snapshot):
-        if isinstance(snapshot, self.obj):
+        if isinstance(snapshot, self.item_type):
             d = self.load(snapshot.serial_number)
             if d:
-                existing_snapshot = self.obj.from_dict(d)
+                existing_snapshot = self.item_type.from_dict(d)
                 for key, value in existing_snapshot.diff(snapshot).items():
                     existing_snapshot[key] = value
-                    existing_snapshot.to_file(self.obj.get_path(serial_number=existing_snapshot.serial_number, path=self.path))
+                    existing_snapshot.to_file(self.item_type.get_path(serial_number=existing_snapshot.serial_number, path=self.path))
             else:
-                snapshot.to_file(self.obj.get_path(serial_number=snapshot.serial_number, path=self.path))
+                snapshot.to_file(self.item_type.get_path(serial_number=snapshot.serial_number, path=self.path))
         else:
             raise TypeError(type(snapshot))
 
@@ -26,7 +26,7 @@ def load(self, serial_number, projection=None):
         if isinstance(serial_number, str):
             ret = None
             try:
-                d = self.obj.from_file(self.obj.get_path(serial_number=serial_number, path=self.path)).to_dict()
+                d = self.item_type.from_file(self.item_type.get_path(serial_number=serial_number, path=self.path)).to_dict()
                 if projection:
                     ret = {k: d[k] for k in set(projection).union({'serial_number'})}
                 else:
@@ -44,12 +44,12 @@ def load(self, serial_number, projection=None):
             raise TypeError(type(serial_number))  # pragma: no cover
 
     def remove(self, serial_number):
-        os.remove(self.obj.get_path(serial_number=serial_number, path=self.path))
+        os.remove(self.item_type.get_path(serial_number=serial_number, path=self.path))
 
     def search(self, key, value, projection=None):
         ret = list()
         for filename in glob(self.path + '*.json'):
-            snapshot = self.obj.from_file(filename)
+            snapshot = self.item_type.from_file(filename)
             if snapshot[key] != value:
                 continue
             d = snapshot.to_dict()
diff --git a/codepack/service/snapshot_service/memory_snapshot_service.py b/codepack/service/snapshot_service/memory_snapshot_service.py
index 1ce1321..89c1b13 100644
--- a/codepack/service/snapshot_service/memory_snapshot_service.py
+++ b/codepack/service/snapshot_service/memory_snapshot_service.py
@@ -4,14 +4,14 @@
 
 
 class MemorySnapshotService(SnapshotService, MemoryStorage):
-    def __init__(self, obj=None):
-        MemoryStorage.__init__(self, obj=obj)
+    def __init__(self, item_type=None):
+        MemoryStorage.__init__(self, item_type=item_type)
 
     def save(self, snapshot):
-        if isinstance(snapshot, self.obj):
+        if isinstance(snapshot, self.item_type):
             d = self.load(snapshot.serial_number)
             if d:
-                for key, value in self.obj.from_dict(d).diff(snapshot).items():
+                for key, value in self.item_type.from_dict(d).diff(snapshot).items():
                     self.memory[snapshot.serial_number][key] = value
             else:
                 self.memory[snapshot.serial_number] = snapshot
diff --git a/codepack/service/snapshot_service/mongo_snapshot_service.py b/codepack/service/snapshot_service/mongo_snapshot_service.py
index 65d3cf9..e5054da 100644
--- a/codepack/service/snapshot_service/mongo_snapshot_service.py
+++ b/codepack/service/snapshot_service/mongo_snapshot_service.py
@@ -1,18 +1,18 @@
 from codepack.storage import MongoStorage
 from codepack.service.snapshot_service import SnapshotService
 from collections.abc import Iterable
-from codepack.utils.state import State
+from codepack.snapshot import State
 
 
 class MongoSnapshotService(SnapshotService, MongoStorage):
-    def __init__(self, obj=None, mongodb=None, db=None, collection=None, *args, **kwargs):
-        MongoStorage.__init__(self, obj=obj, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
+    def __init__(self, item_type=None, mongodb=None, db=None, collection=None, *args, **kwargs):
+        MongoStorage.__init__(self, item_type=item_type, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
 
     def save(self, snapshot):
-        if isinstance(snapshot, self.obj):
+        if isinstance(snapshot, self.item_type):
             d = self.load(snapshot.serial_number)
             if d:
-                existing_snapshot = self.obj.from_dict(d)
+                existing_snapshot = self.item_type.from_dict(d)
                 diff = existing_snapshot.diff(snapshot)
                 for key, value in diff.items():
                     existing_snapshot[key] = value
diff --git a/codepack/service/storage_service/__init__.py b/codepack/service/storage_service/__init__.py
index 2466e21..6076b6c 100644
--- a/codepack/service/storage_service/__init__.py
+++ b/codepack/service/storage_service/__init__.py
@@ -2,10 +2,3 @@
 from codepack.service.storage_service.memory_storage_service import MemoryStorageService
 from codepack.service.storage_service.file_storage_service import FileStorageService
 from codepack.service.storage_service.mongo_storage_service import MongoStorageService
-from enum import Enum
-
-
-class StorageServiceAlias(Enum):
-    MEMORY = MemoryStorageService
-    FILE = FileStorageService
-    MONGODB = MongoStorageService
diff --git a/codepack/service/storage_service/file_storage_service.py b/codepack/service/storage_service/file_storage_service.py
index 43ee07d..18478ed 100644
--- a/codepack/service/storage_service/file_storage_service.py
+++ b/codepack/service/storage_service/file_storage_service.py
@@ -5,26 +5,31 @@
 
 
 class FileStorageService(StorageService, FileStorage):
-    def __init__(self, obj=None, path='./'):
-        FileStorage.__init__(self, obj=obj, path=path)
+    def __init__(self, item_type=None, path='./'):
+        FileStorage.__init__(self, item_type=item_type, path=path)
 
-    def save(self, item):
-        if isinstance(item, self.obj):
-            item.to_file(self.obj.get_path(serial_number=item.id, path=self.path))
+    def save(self, item, update=False):
+        if isinstance(item, self.item_type):
+            if update:
+                item.to_file(self.item_type.get_path(serial_number=item.id, path=self.path))
+            elif self.check(item.id):
+                raise ValueError('%s already exists' % item.id)
+            else:
+                item.to_file(self.item_type.get_path(serial_number=item.id, path=self.path))
         else:
             raise TypeError(type(item))
 
     def load(self, id):
-        return self.obj.from_file(path=self.obj.get_path(serial_number=id, path=self.path))
+        return self.item_type.from_file(path=self.item_type.get_path(serial_number=id, path=self.path))
 
     def remove(self, id):
-        os.remove(path=self.obj.get_path(serial_number=id, path=self.path))
+        os.remove(path=self.item_type.get_path(serial_number=id, path=self.path))
 
     def check(self, id):
         if isinstance(id, str):
             ret = None
             try:
-                ret = self.obj.from_file(path=self.obj.get_path(serial_number=id, path=self.path)).id
+                ret = self.item_type.from_file(path=self.item_type.get_path(serial_number=id, path=self.path)).id
             finally:
                 return ret
         elif isinstance(id, Iterable):
diff --git a/codepack/service/storage_service/memory_storage_service.py b/codepack/service/storage_service/memory_storage_service.py
index 8d2d1b2..045a75b 100644
--- a/codepack/service/storage_service/memory_storage_service.py
+++ b/codepack/service/storage_service/memory_storage_service.py
@@ -4,12 +4,17 @@
 
 
 class MemoryStorageService(StorageService, MemoryStorage):
-    def __init__(self, obj=None):
-        MemoryStorage.__init__(self, obj=obj)
+    def __init__(self, item_type=None):
+        MemoryStorage.__init__(self, item_type=item_type)
 
-    def save(self, item):
-        if isinstance(item, self.obj):
-            self.memory[item.id] = item
+    def save(self, item, update=False):
+        if isinstance(item, self.item_type):
+            if update:
+                self.memory[item.id] = item
+            elif self.check(item.id):
+                raise ValueError('%s already exists' % item.id)
+            else:
+                self.memory[item.id] = item
         else:
             raise TypeError(type(item))
 
diff --git a/codepack/service/storage_service/mongo_storage_service.py b/codepack/service/storage_service/mongo_storage_service.py
index a31aae4..863a46a 100644
--- a/codepack/service/storage_service/mongo_storage_service.py
+++ b/codepack/service/storage_service/mongo_storage_service.py
@@ -4,17 +4,24 @@
 
 
 class MongoStorageService(StorageService, MongoStorage):
-    def __init__(self, obj=None, mongodb=None, db=None, collection=None, *args, **kwargs):
-        MongoStorage.__init__(self, obj=obj, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
+    def __init__(self, item_type=None, mongodb=None, db=None, collection=None, *args, **kwargs):
+        MongoStorage.__init__(self, item_type=item_type, mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
 
-    def save(self, item):
-        if isinstance(item, self.obj):
-            item.to_db(mongodb=self.mongodb, db=self.db, collection=self.collection)
+    def save(self, item, update=False):
+        if isinstance(item, self.item_type):
+            if update:
+                d = item.to_dict()
+                _id = d.pop('_id')
+                self.mongodb[self.db][self.collection].update_one({'_id': _id}, {'$set': d}, upsert=True)
+            elif self.check(item.id):
+                raise ValueError('%s already exists' % item.id)
+            else:
+                item.to_db(mongodb=self.mongodb, db=self.db, collection=self.collection)
         else:
             raise TypeError(type(item))
 
     def load(self, id):
-        return self.obj.from_db(id=id, mongodb=self.mongodb, db=self.db, collection=self.collection)
+        return self.item_type.from_db(id=id, mongodb=self.mongodb, db=self.db, collection=self.collection)
 
     def remove(self, id):
         self.mongodb[self.db][self.collection].delete_one({'_id': id})
@@ -22,7 +29,7 @@ def remove(self, id):
     def check(self, id):
         if isinstance(id, str):
             ret = None
-            item = self.obj.from_db(id=id, mongodb=self.mongodb, db=self.db, collection=self.collection)
+            item = self.item_type.from_db(id=id, mongodb=self.mongodb, db=self.db, collection=self.collection)
             if item:
                 ret = item.id
             return ret
diff --git a/codepack/service/storage_service/storage_service.py b/codepack/service/storage_service/storage_service.py
index fa731f1..a75f5ce 100644
--- a/codepack/service/storage_service/storage_service.py
+++ b/codepack/service/storage_service/storage_service.py
@@ -4,7 +4,7 @@
 
 class StorageService(Service, metaclass=abc.ABCMeta):
     @abc.abstractmethod
-    def save(self, item):
+    def save(self, item, update=False):
         """save item"""
 
     @abc.abstractmethod
diff --git a/codepack/snapshot/__init__.py b/codepack/snapshot/__init__.py
index 21e00ea..6e8488c 100644
--- a/codepack/snapshot/__init__.py
+++ b/codepack/snapshot/__init__.py
@@ -2,3 +2,4 @@
 from codepack.snapshot.snapshotable import Snapshotable
 from codepack.snapshot.code_snapshot import CodeSnapshot
 from codepack.snapshot.codepack_snapshot import CodePackSnapshot
+from codepack.snapshot.state import State
diff --git a/codepack/snapshot/code_snapshot.py b/codepack/snapshot/code_snapshot.py
index 26718d3..9d3d0a7 100644
--- a/codepack/snapshot/code_snapshot.py
+++ b/codepack/snapshot/code_snapshot.py
@@ -1,5 +1,5 @@
 from codepack.snapshot import Snapshot
-from codepack.utils.dependency import Dependency
+from codepack.dependency import Dependency
 
 
 class CodeSnapshot(Snapshot):
diff --git a/codepack/snapshot/codepack_snapshot.py b/codepack/snapshot/codepack_snapshot.py
index bf5325b..8c29d87 100644
--- a/codepack/snapshot/codepack_snapshot.py
+++ b/codepack/snapshot/codepack_snapshot.py
@@ -1,4 +1,6 @@
 from codepack.snapshot import Snapshot
+from codepack.argpack import ArgPack
+from copy import deepcopy
 
 
 class CodePackSnapshot(Snapshot):
@@ -6,7 +8,7 @@ def __init__(self, codepack=None, argpack=None, timestamp=None):
         if codepack:
             _id = codepack.id
             _serial_number = codepack.serial_number
-            _state = codepack.get_state()
+            _state = None
             _codes = {k: v.serial_number for k, v in codepack.codes.items()}
             _source = codepack.get_source()
             _structure = codepack.get_structure()
@@ -27,7 +29,17 @@ def __init__(self, codepack=None, argpack=None, timestamp=None):
         self.set_argpack(argpack=argpack)
 
     def set_argpack(self, argpack=None):
-        self.__setitem__('argpack', argpack if argpack else dict())
+        if isinstance(argpack, ArgPack):
+            tmp = argpack.to_dict()
+        elif isinstance(argpack, dict):
+            tmp = deepcopy(argpack)
+            if '_id' not in tmp:
+                tmp['_id'] = None
+        elif argpack is None:
+            tmp = {'_id': None}
+        else:
+            tmp = dict()
+        self.__setitem__('argpack', tmp)
 
     @classmethod
     def from_dict(cls, d):
diff --git a/codepack/snapshot/snapshot.py b/codepack/snapshot/snapshot.py
index c782cec..605930d 100644
--- a/codepack/snapshot/snapshot.py
+++ b/codepack/snapshot/snapshot.py
@@ -1,6 +1,6 @@
 from codepack.storage import Storable
-from datetime import datetime
-from codepack.utils.state import State
+from datetime import datetime, timezone
+from codepack.snapshot.state import State
 from copy import deepcopy
 
 
@@ -11,7 +11,7 @@ def __init__(self, id, serial_number, state=None, timestamp=None, **kwargs):
         self.__setitem__('id', self.id)
         self.__setitem__('serial_number', self.serial_number)
         self.__setitem__('state', State.get(state))
-        self.__setitem__('timestamp', timestamp if timestamp else datetime.now().timestamp())
+        self.__setitem__('timestamp', timestamp if timestamp else datetime.now(timezone.utc).timestamp())
         for k, v in kwargs.items():
             self.__setitem__(k, v)
 
@@ -43,7 +43,9 @@ def diff(self, snapshot):
         return ret
 
     def to_dict(self):
-        ret = deepcopy(self.attr)
+        ret = dict()
+        for k, v in self.attr.items():
+            ret[k] = v
         ret['state'] = ret['state'].name
         ret['_id'] = ret['serial_number']
         return ret
diff --git a/codepack/utils/state.py b/codepack/snapshot/state.py
similarity index 100%
rename from codepack/utils/state.py
rename to codepack/snapshot/state.py
diff --git a/codepack/storage/__init__.py b/codepack/storage/__init__.py
index 7ce299b..81a9d3b 100644
--- a/codepack/storage/__init__.py
+++ b/codepack/storage/__init__.py
@@ -3,3 +3,4 @@
 from codepack.storage.memory_storage import MemoryStorage
 from codepack.storage.file_storage import FileStorage
 from codepack.storage.mongo_storage import MongoStorage
+from codepack.storage.kafka_storage import KafkaStorage
diff --git a/codepack/storage/file_storage.py b/codepack/storage/file_storage.py
index 2b14976..9a55d7a 100644
--- a/codepack/storage/file_storage.py
+++ b/codepack/storage/file_storage.py
@@ -1,11 +1,45 @@
-from codepack.storage import Storage
+from codepack.storage import Storage, Storable
+from shutil import rmtree
+from glob import glob
+import os
+from typing import Type
 
 
 class FileStorage(Storage):
-    def __init__(self, obj=None, path='./'):
-        super().__init__(obj=obj)
+    def __init__(self, item_type: Type[Storable] = None, path: str = '.'):
+        super().__init__(item_type=item_type)
         self.path = None
+        self.new_path = None
         self.init(path=path)
 
-    def init(self, path='./'):
+    def init(self, path: str = '.'):
         self.path = path
+        if os.path.exists(path):
+            self.new_path = False
+        else:
+            self.new_path = True
+            self.mkdir(path)
+
+    def close(self):
+        if self.new_path:
+            self.rmdir(self.path)
+
+    @staticmethod
+    def mkdir(path: str):
+        if not os.path.exists(path):
+            os.makedirs(path)
+
+    @staticmethod
+    def rmdir(path: str):
+        if os.path.exists(path):
+            rmtree(path)
+
+    @staticmethod
+    def empty_dir(path: str):
+        for item in glob(os.path.join(path, '*')):
+            if os.path.isfile(item):
+                os.remove(item)
+            elif os.path.isdir(item):
+                rmtree(item)
+            else:
+                raise NotImplementedError('%s is unknown' % item)  # pragma: no cover
diff --git a/codepack/storage/kafka_storage.py b/codepack/storage/kafka_storage.py
new file mode 100644
index 0000000..8932186
--- /dev/null
+++ b/codepack/storage/kafka_storage.py
@@ -0,0 +1,52 @@
+from codepack.interface import KafkaConsumer, KafkaProducer
+from codepack.storage import Storage, Storable
+from typing import Union
+from typing import Type
+
+
+class KafkaStorage(Storage):
+    def __init__(self, item_type: Type[Storable] = None,
+                 producer: Union[KafkaProducer, dict] = None, consumer: Union[KafkaConsumer, dict] = None,
+                 topic: str = None, producer_config: dict = None, consumer_config: dict = None):
+        super().__init__(item_type=item_type)
+        self.producer = None
+        self.consumer = None
+        self.topic = None
+        self.new_producer_connection = None
+        self.new_consumer_connection = None
+        self.init(producer=producer, consumer=consumer,
+                  topic=topic, producer_config=producer_config, consumer_config=consumer_config)
+
+    def init(self, producer: Union[KafkaProducer, dict] = None, consumer: Union[KafkaConsumer, dict] = None,
+             topic: str = None, producer_config: dict = None, consumer_config: dict = None):
+        self.topic = topic
+        if isinstance(producer, KafkaProducer) or producer is None:
+            self.producer = producer
+            self.new_producer_connection = False
+        elif isinstance(producer, dict):
+            if not producer_config:
+                producer_config = dict()
+            self.producer = KafkaProducer(producer, **producer_config)
+            self.new_producer_connection = True
+        else:
+            raise TypeError(type(producer))
+        if isinstance(consumer, KafkaConsumer) or consumer is None:
+            self.consumer = consumer
+            self.new_consumer_connection = False
+        elif isinstance(consumer, dict):
+            if not consumer_config:
+                consumer_config = dict()
+            self.consumer = KafkaConsumer(consumer, **consumer_config)
+            if 'topic' in consumer:
+                self.topic = consumer['topic']
+            self.new_consumer_connection = True
+        else:
+            raise TypeError(type(consumer))
+
+    def close(self):
+        if self.new_producer_connection:
+            self.producer.close()
+        self.producer = None
+        if self.new_consumer_connection:
+            self.consumer.close()
+        self.consumer = None
diff --git a/codepack/storage/memory_storage.py b/codepack/storage/memory_storage.py
index df2b144..c58849f 100644
--- a/codepack/storage/memory_storage.py
+++ b/codepack/storage/memory_storage.py
@@ -1,11 +1,16 @@
-from codepack.storage import Storage
+from codepack.storage import Storage, Storable
+from typing import Type
 
 
 class MemoryStorage(Storage):
-    def __init__(self, obj=None):
-        super().__init__(obj=obj)
+    def __init__(self, item_type: Type[Storable] = None):
+        super().__init__(item_type=item_type)
         self.memory = None
         self.init()
 
     def init(self):
         self.memory = dict()
+
+    def close(self):
+        self.memory.clear()
+        self.memory = None
diff --git a/codepack/storage/mongo_storage.py b/codepack/storage/mongo_storage.py
index 4e10c32..6049152 100644
--- a/codepack/storage/mongo_storage.py
+++ b/codepack/storage/mongo_storage.py
@@ -1,17 +1,20 @@
 from codepack.interface import MongoDB
-from codepack.storage import Storage
+from codepack.storage import Storage, Storable
+from typing import Union
+from typing import Type
 
 
 class MongoStorage(Storage):
-    def __init__(self, obj=None, mongodb=None, db=None, collection=None, *args, **kwargs):
-        super().__init__(obj=obj)
+    def __init__(self, item_type: Type[Storable] = None,
+                 mongodb: Union[MongoDB, dict] = None, db: str = None, collection: str = None, *args, **kwargs):
+        super().__init__(item_type=item_type)
         self.mongodb = None
         self.db = None
         self.collection = None
         self.new_connection = None
         self.init(mongodb=mongodb, db=db, collection=collection, *args, **kwargs)
 
-    def init(self, mongodb=None, db=None, collection=None, *args, **kwargs):
+    def init(self, mongodb: Union[MongoDB, dict] = None, db: str = None, collection: str = None, *args, **kwargs):
         self.db = db
         self.collection = collection
         if isinstance(mongodb, MongoDB):
@@ -22,3 +25,8 @@ def init(self, mongodb=None, db=None, collection=None, *args, **kwargs):
             self.new_connection = True
         else:
             raise TypeError(type(mongodb))
+
+    def close(self):
+        if self.new_connection:
+            self.mongodb.close()
+        self.mongodb = None
diff --git a/codepack/storage/storage.py b/codepack/storage/storage.py
index bb0e0c0..6ade636 100644
--- a/codepack/storage/storage.py
+++ b/codepack/storage/storage.py
@@ -1,13 +1,18 @@
 import abc
 from codepack.storage import Storable
+from typing import Type
 
 
 class Storage(metaclass=abc.ABCMeta):
-    def __init__(self, obj=None):
-        if isinstance(obj, Storable):
-            raise TypeError(type(obj))
-        self.obj = obj
+    def __init__(self, item_type: Type[Storable] = None):
+        if item_type and not issubclass(item_type, Storable):
+            raise TypeError(type(item_type))
+        self.item_type = item_type
 
     @abc.abstractmethod
     def init(self, *args, **kwargs):
         """initialize storage"""
+
+    @abc.abstractmethod
+    def close(self):
+        """close storage"""
diff --git a/codepack/utils/config.py b/codepack/utils/config.py
deleted file mode 100644
index a7be140..0000000
--- a/codepack/utils/config.py
+++ /dev/null
@@ -1,55 +0,0 @@
-from configparser import ConfigParser
-import os
-
-
-def get_config(filename, section):
-    cp = ConfigParser()
-    cp.read(filename)
-    items = cp.items(section)
-    return {item[0]: item[1] for item in items}
-
-
-def get_config_assertion_error_message(x):
-    return "'%s' should be given as an argument or defined in a configuration file in 'config_path' " \
-           "or os.environ['CODEPACK_CONFIG_PATH']" % x
-
-
-def get_default_config(section, config_path=None):
-    if not config_path:
-        config_path = os.environ.get('CODEPACK_CONFIG_PATH', None)
-    if config_path:
-        config = get_config(config_path, section=section)
-        return config
-    else:
-        return None
-
-
-def get_default_value(section, key, config=None):
-    env = 'CODEPACK_%s_%s' % (section.upper(), key.upper())
-    if config:
-        ret = config[key]
-    elif env in os.environ:
-        ret = os.environ.get(env, None)
-    else:
-        raise AssertionError("'%s' information should be provided in a configuration file in 'config_path' "
-                             "or os.environ['%s']" % (section, env))
-    return ret
-
-
-def get_default_service_config(section, config_path=None):
-    config = get_default_config(section=section, config_path=config_path)
-    ret = dict()
-    ret['source'] = get_default_value(section=section, key='source', config=config).upper()
-    if ret['source'] == 'MEMORY':
-        pass
-    elif ret['source'] == 'FILE':
-        ret['path'] = get_default_value(section=section, key='path', config=config)
-    elif ret['source'] == 'MONGODB':
-        ret['db'] = get_default_value(section=section, key='db', config=config)
-        ret['collection'] = get_default_value(section=section, key='collection', config=config)
-        conn_config = get_default_config(section='conn', config_path=config_path)
-        conn_config_path = get_default_value(section='conn', key='path', config=conn_config)
-        ret['mongodb'] = get_config(conn_config_path, 'mongodb')
-    else:
-        raise NotImplementedError("'%s' is not implemented" % ret['source'])
-    return ret
diff --git a/config/alias.ini b/config/alias.ini
new file mode 100644
index 0000000..8696c29
--- /dev/null
+++ b/config/alias.ini
@@ -0,0 +1,19 @@
+[alias]
+code = codepack.code.Code
+codepack = codepack.codepack.CodePack
+argpack = codepack.argpack.argpack.ArgPack
+code_snapshot = codepack.snapshot.code_snapshot.CodeSnapshot
+codepack_snapshot = codepack.snapshot.codepack_snapshot.CodePackSnapshot
+delivery = codepack.delivery.delivery.Delivery
+memory_storage_service = codepack.service.storage_service.memory_storage_service.MemoryStorageService
+file_storage_service = codepack.service.storage_service.file_storage_service.FileStorageService
+mongo_storage_service = codepack.service.storage_service.mongo_storage_service.MongoStorageService
+memory_delivery_service = codepack.service.delivery_service.memory_delivery_service.MemoryDeliveryService
+file_delivery_service = codepack.service.delivery_service.file_delivery_service.FileDeliveryService
+mongo_delivery_service = codepack.service.delivery_service.mongo_delivery_service.MongoDeliveryService
+memory_snapshot_service = codepack.service.snapshot_service.memory_snapshot_service.MemorySnapshotService
+file_snapshot_service = codepack.service.snapshot_service.file_snapshot_service.FileSnapshotService
+mongo_snapshot_service = codepack.service.snapshot_service.mongo_snapshot_service.MongoSnapshotService
+mongo_scheduler = codepack.scheduler.mongo_scheduler.MongoScheduler
+supervisor = codepack.employee.supervisor.Supervisor
+worker = codepack.employee.worker.Worker
diff --git a/config/codepack.ini b/config/codepack.ini
index 6d3ee71..ff2ca91 100644
--- a/config/codepack.ini
+++ b/config/codepack.ini
@@ -1,22 +1,25 @@
 [conn]
 path = config/conn.ini
 
-[scheduler]
-source = mongodb
-db = codepack
-collection = scheduler
+[alias]
+path = config/alias.ini
 
-[code_storage]
+[code]
 source = mongodb
 db = codepack
 collection = code
 
-[codepack_storage]
+[codepack]
 source = mongodb
 db = codepack
 collection = codepack
 
-[cache]
+[argpack]
+source = mongodb
+db = codepack
+collection = argpack
+
+[delivery]
 source = mongodb
 db = codepack
 collection = cache
@@ -30,3 +33,24 @@ collection = code_snapshot
 source = mongodb
 db = codepack
 collection = codepack_snapshot
+
+[apiserver]
+scheduler = self
+supervisor = self
+
+[scheduler]
+source = mongodb
+db = codepack
+collection = scheduler
+supervisor = http://localhost:8000
+
+[worker]
+source = kafka
+topic = codepack
+group_id = codepack_worker
+interval = 1
+supervisor = http://localhost:8000
+
+[supervisor]
+source = kafka
+topic = codepack
diff --git a/requirements.txt b/requirements.txt
index 9895431..03c26f3 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,4 +8,6 @@ boto3==1.19.6
 cx-Oracle==8.2.1
 parse==1.19.0
 APScheduler==3.8.1
-mongomock==3.23.0
+kafka-python==2.0.2
+docker==5.0.3
+requests==2.26.0


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 8114226..5881dc8 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -20,7 +20,7 @@ jobs:
     - name: Install Dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pytest pytest-cov
+        pip install pytest pytest-cov mongomock
         pip install -r requirements.txt
     - name: Run Tests
       run: |
diff --git a/config/test.ini b/config/test.ini
index 62743c5..081b1c2 100644
--- a/config/test.ini
+++ b/config/test.ini
@@ -1,18 +1,19 @@
 [conn]
 path = config/test_conn.ini
 
-[scheduler]
-source = mongodb
-db = codepack
-collection = scheduler
+[alias]
+path = config/alias.ini
 
-[code_storage]
+[code]
 source = memory
 
-[codepack_storage]
+[codepack]
 source = memory
 
-[cache]
+[argpack]
+source = memory
+
+[delivery]
 source = memory
 
 [code_snapshot]
@@ -20,3 +21,24 @@ source = memory
 
 [codepack_snapshot]
 source = memory
+
+[apiserver]
+scheduler = self
+supervisor = self
+
+[scheduler]
+source = mongodb
+db = codepack
+collection = scheduler
+supervisor = http://localhost:8000
+
+[worker]
+source = kafka
+topic = test
+group_id = codepack_worker
+interval = 1
+supervisor = http://localhost:8000
+
+[supervisor]
+source = kafka
+topic = test
diff --git a/config/test_conn.ini b/config/test_conn.ini
index 827d819..1221196 100644
--- a/config/test_conn.ini
+++ b/config/test_conn.ini
@@ -1,3 +1,6 @@
 [mongodb]
 host = localhost
 port = 27017
+
+[kafka]
+bootstrap_servers = localhost:9092
diff --git a/tests/conftest.py b/tests/conftest.py
index 3195de3..5a8bcb7 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2,8 +2,7 @@
 from codepack.interface import MongoDB
 import os
 import mongomock
-from codepack.service import DefaultService
-from codepack import Code, CodePack
+from codepack.config import Default
 from shutil import rmtree
 from glob import glob
 
@@ -31,10 +30,10 @@ def empty_dir(directory):
 @pytest.fixture(scope='function', autouse=False)
 def default_os_env():
     os.environ['CODEPACK_CONFIG_PATH'] = 'config/test.ini'
-    DefaultService.get_default_delivery_service().init()
-    DefaultService.get_default_code_snapshot_service().init()
-    DefaultService.get_default_code_storage_service(obj=Code).init()
-    DefaultService.get_default_codepack_storage_service(obj=CodePack).init()
+    Default.get_storage_instance('delivery', 'delivery_service').init()
+    Default.get_storage_instance('code_snapshot', 'snapshot_service').init()
+    Default.get_storage_instance('code', 'storage_service').init()
+    Default.get_storage_instance('codepack', 'storage_service').init()
     yield
     os.environ.pop('CODEPACK_CONFIG_PATH', None)
 
@@ -55,6 +54,7 @@ def testdir():
     mkdir(rootdir + 'state_manager/')
     mkdir(rootdir + 'storage_service/')
     mkdir(rootdir + 'snapshot_service/')
+    mkdir(rootdir + 'docker_test/')
     yield
     rmdir(rootdir)
 
@@ -83,13 +83,23 @@ def testdir_snapshot_service():
     empty_dir(directory)
 
 
+@pytest.fixture(scope='function', autouse=False)
+def testdir_docker_manager():
+    directory = 'testdir/docker_test/'
+    empty_dir(directory)
+    yield directory
+    empty_dir(directory)
+
+
 @pytest.fixture(scope='function', autouse=True)
-def init_default_service():
-    DefaultService.init()
+def init_default():
+    Default.init()
 
 
 @pytest.fixture(scope='function', autouse=False)
-def default_services():
-    DefaultService.init()
-    yield DefaultService
-    DefaultService.init()
+def default():
+    os.environ['CODEPACK_ALIAS_PATH'] = 'config/alias.ini'
+    Default.init()
+    yield Default
+    Default.init()
+    os.environ.pop('CODEPACK_ALIAS_PATH', None)
diff --git a/tests/test_alias.py b/tests/test_alias.py
new file mode 100644
index 0000000..7fb62ce
--- /dev/null
+++ b/tests/test_alias.py
@@ -0,0 +1,81 @@
+from codepack.config import Alias
+import os
+import pytest
+from codepack.service import MemoryStorageService, FileStorageService
+
+
+def test_alias_from_individual_os_env():
+    a = Alias()
+    assert a.aliases is None
+    with pytest.raises(AttributeError):
+        tmp = a['memory_storage_service']
+    os_env = 'CODEPACK_ALIAS_MEMORY_STORAGE_SERVICE'
+    os.environ[os_env] = 'codepack.service.storage_service.memory_storage_service.MemoryStorageService'
+    assert a['memory_storage_service'] == MemoryStorageService
+    os.environ.pop(os_env)
+
+
+def test_alias_from_alias_path_os_env():
+    a = Alias()
+    assert a.aliases is None
+    with pytest.raises(AttributeError):
+        tmp = a['memory_storage_service']
+    os_env = 'CODEPACK_ALIAS_PATH'
+    os.environ[os_env] = 'config/alias.ini'
+    assert a['memory_storage_service'] == MemoryStorageService
+    os.environ.pop(os_env)
+
+
+def test_alias_from_config_path_os_env():
+    a = Alias()
+    assert a.aliases is None
+    with pytest.raises(AttributeError):
+        tmp = a['memory_storage_service']
+    os_env = 'CODEPACK_CONFIG_PATH'
+    os.environ[os_env] = 'config/test.ini'
+    assert a['memory_storage_service'] == MemoryStorageService
+    os.environ.pop(os_env)
+
+
+def test_alias_priority1():
+    a = Alias(data={'memory_storage_service': 'codepack.service.storage_service.file_storage_service.FileStorageService'})
+    assert a.aliases is not None
+    assert a['memory_storage_service'] == FileStorageService
+    os_env = 'CODEPACK_ALIAS_MEMORY_STORAGE_SERVICE'
+    os.environ[os_env] = 'codepack.service.storage_service.memory_storage_service.MemoryStorageService'
+    assert a['memory_storage_service'] == FileStorageService
+    os.environ.pop(os_env)
+    os_env = 'CODEPACK_ALIAS_PATH'
+    os.environ[os_env] = 'config/alias.ini'
+    assert a['memory_storage_service'] == FileStorageService
+    os.environ.pop(os_env)
+    os_env = 'CODEPACK_CONFIG_PATH'
+    os.environ[os_env] = 'config/test.ini'
+    assert a['memory_storage_service'] == FileStorageService
+    os.environ.pop(os_env)
+
+
+def test_alias_priority2():
+    a = Alias()
+    assert a.aliases is None
+    with pytest.raises(AttributeError):
+        tmp = a['memory_storage_service']
+    first_os_env = 'CODEPACK_ALIAS_MEMORY_STORAGE_SERVICE'
+    os.environ[first_os_env] = 'codepack.service.storage_service.file_storage_service.FileStorageService'
+    assert a['memory_storage_service'] == FileStorageService
+    os_env = 'CODEPACK_ALIAS_PATH'
+    os.environ[os_env] = 'config/alias.ini'
+    assert a['memory_storage_service'] == FileStorageService
+    os.environ.pop(os_env)
+    os_env = 'CODEPACK_CONFIG_PATH'
+    os.environ[os_env] = 'config/test.ini'
+    assert a['memory_storage_service'] == FileStorageService
+    os.environ.pop(os_env)
+    os.environ.pop(first_os_env)
+
+
+def test_alias_path_argument():
+    a = Alias(data='config/alias.ini')
+    assert a['memory_storage_service'] == MemoryStorageService
+    b = Alias(data='config/test.ini')
+    assert b['memory_storage_service'] == MemoryStorageService
diff --git a/tests/test_argpack.py b/tests/test_argpack.py
new file mode 100644
index 0000000..37d8451
--- /dev/null
+++ b/tests/test_argpack.py
@@ -0,0 +1,73 @@
+from codepack import Code, CodePack
+from codepack.snapshot import CodePackSnapshot
+from tests import *
+import pytest
+
+
+def test_argpack_input(default_os_env):
+    code1 = Code(add2)
+    code2 = Code(mul2)
+    code1 >> code2
+    code2.receive('b') << code1
+    codepack = CodePack(id='argpack_test', code=code1, subscribe=code2)
+    argpack = codepack.make_argpack()
+    argpack['add2'](a=3, b=2)
+    with pytest.raises(TypeError):
+        argpack['mul2'](c=5)
+    argpack['mul2'](a=2)
+    assert argpack['add2']['a'] == 3
+    assert argpack['add2']['b'] == 2
+    assert argpack['mul2']['a'] == 2
+    assert codepack(argpack=argpack) == 10
+
+
+def test_argpack_codepack_execution(default_os_env):
+    code1 = Code(add2)
+    code2 = Code(mul2)
+    code1 >> code2
+    code2.receive('b') << code1
+    codepack = CodePack(id='argpack_test', code=code1, subscribe=code2)
+    argpack = codepack.make_argpack()
+    snapshot = codepack.to_snapshot(argpack=argpack)
+    assert isinstance(snapshot.argpack, dict)
+    assert snapshot.argpack['add2']['a'] is None
+    assert snapshot.argpack['add2']['b'] is None
+    assert snapshot.argpack['mul2']['a'] is None
+    assert 'b' not in snapshot.argpack['mul2']
+    snapshot2 = CodePackSnapshot.from_dict(snapshot.to_dict())
+    assert isinstance(snapshot2.argpack, dict)
+    assert snapshot2.argpack['add2']['a'] is None
+    assert snapshot2.argpack['add2']['b'] is None
+    assert snapshot2.argpack['mul2']['a'] is None
+    assert 'b' not in snapshot2.argpack['mul2']
+    argpack['add2'](a=3, b=2)
+    argpack['mul2'](a=2)
+    snapshot3 = codepack.to_snapshot(argpack=argpack)
+    assert snapshot3.argpack['add2']['a'] == 3
+    assert snapshot3.argpack['add2']['b'] == 2
+    assert snapshot3.argpack['mul2']['a'] == 2
+    snapshot4 = CodePackSnapshot.from_dict(snapshot3.to_dict())
+    assert snapshot4.argpack['add2']['a'] == 3
+    assert snapshot4.argpack['add2']['b'] == 2
+    assert snapshot4.argpack['mul2']['a'] == 2
+    codepack2 = CodePack.from_snapshot(snapshot4)
+    assert codepack2(argpack=snapshot4.argpack) == 10
+
+
+def test_argpack_str(default_os_env):
+    code1 = Code(add2)
+    code2 = Code(mul2)
+    code1 >> code2
+    code2.receive('b') << code1
+    codepack = CodePack(id='argpack_test', code=code1, subscribe=code2)
+    argpack = codepack.make_argpack()
+    assert argpack.__str__() == 'ArgPack(id: argpack_test, args: {add2(a=None, b=None), mul2(a=None)})'
+    argpack['add2'](a=3, b=2)
+    with pytest.raises(TypeError):
+        argpack['mul2'](c=5)
+    argpack['mul2'](a=2)
+    assert argpack['add2']['a'] == 3
+    assert argpack['add2']['b'] == 2
+    assert argpack['mul2']['a'] == 2
+    assert argpack.__str__() == 'ArgPack(id: argpack_test, args: {add2(a=3, b=2), mul2(a=2)})'
+    assert codepack(argpack=argpack) == 10
diff --git a/tests/test_code.py b/tests/test_code.py
index 404a7f6..8d9c9e1 100644
--- a/tests/test_code.py
+++ b/tests/test_code.py
@@ -76,7 +76,7 @@ def test_add_dependency(default_os_env):
     assert '1234' in code.dependency and '5678' in code.dependency
     assert code.dependency['5678'].arg == 'a'
     assert not code.dependency['1234'].arg
-    tmp = code.get_dependent_args()
+    tmp = code.dependency.get_args()
     assert len(tmp) == 1 and 'a' in tmp
 
 
@@ -162,7 +162,7 @@ def test_validate_dependency_result(default_os_env):
     code3.receive('c') << code2
     code1(1, 2)
     code2(3, 4)
-    code2.service['delivery_service'].cancel(code2.serial_number)
+    code2.service['delivery'].cancel(code2.serial_number)
     ret = code3(a=3)
     assert ret is None
     assert code3.get_state() == 'WAITING'
@@ -172,20 +172,22 @@ def test_validate_dependency_result(default_os_env):
     assert ret is None
     assert code3.get_state() == 'WAITING'
     code2(3, 4)
-    snapshots = code3.get_dependent_snapshots()
-    caches = code3.get_dependency_cache_info()
+    snapshots = code3.dependency.load_snapshot()
+    deliveries = code3.dependency.check_delivery()
     assert len(snapshots) == 2
-    assert len(caches) == 2
-    code2_state_info = snapshots.pop(code2.serial_number)
-    assert code3.validate_dependency_result(snapshots=snapshots, caches=caches) == 'NOT_READY'
+    assert len(deliveries) == 2
+    snapshot_dict = {x['_id']: x for x in snapshots}
+    code2_state_info = snapshot_dict.pop(code2.serial_number)
+    assert code3.dependency.validate_delivery(snapshot=snapshot_dict.values(), delivery=deliveries) == 'PENDING'
     update_time = code2_state_info.pop('timestamp')
-    snapshots[code2.serial_number] = code2_state_info
-    assert code3.validate_dependency_result(snapshots=snapshots, caches=caches) == 'NOT_READY'
-    snapshots[code2.serial_number]['timestamp'] = update_time
-    send_time = caches[code2.serial_number].pop('timestamp')
-    assert code3.validate_dependency_result(snapshots=snapshots, caches=caches) == 'NOT_READY'
-    caches[code2.serial_number]['timestamp'] = send_time
-    assert code3.validate_dependency_result(snapshots=snapshots, caches=caches) == 'READY'
+    snapshot_dict[code2.serial_number] = code2_state_info
+    assert code3.dependency.validate_delivery(snapshot=snapshot_dict.values(), delivery=deliveries) == 'PENDING'
+    snapshot_dict[code2.serial_number]['timestamp'] = update_time
+    delivery_dict = {x['_id']: x for x in deliveries}
+    send_time = delivery_dict[code2.serial_number].pop('timestamp')
+    assert code3.dependency.validate_delivery(snapshot=snapshot_dict.values(), delivery=delivery_dict.values()) == 'PENDING'
+    delivery_dict[code2.serial_number]['timestamp'] = send_time
+    assert code3.dependency.validate_delivery(snapshot=snapshot_dict.values(), delivery=delivery_dict.values()) == 'RESOLVED'
 
 
 def test_default_arg(default_os_env):
diff --git a/tests/test_utils_config.py b/tests/test_config.py
similarity index 51%
rename from tests/test_utils_config.py
rename to tests/test_config.py
index d02c571..a499da8 100644
--- a/tests/test_utils_config.py
+++ b/tests/test_config.py
@@ -1,6 +1,6 @@
 from codepack import Code
 from tests import *
-from codepack.utils.config import get_default_service_config
+from codepack.config import Config
 import pytest
 import os
 from codepack.service import MemoryDeliveryService, FileSnapshotService, MongoStorageService
@@ -21,28 +21,30 @@ def test_config_path():
     assert code(1, 2) == 3
 
 
-def test_default_memory_code_snapshot_service_with_os_env(default_services):
+def test_default_memory_code_snapshot_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('code_snapshot')
+        config.get_storage_config('code_snapshot')
     env_source = 'CODEPACK_CODE_SNAPSHOT_SOURCE'
     try:
-        os.environ[env_source] = 'MEMORY'
-        mss = default_services.get_default_code_snapshot_service()
+        os.environ[env_source] = 'memory'
+        mss = default.get_storage_instance('code_snapshot', 'snapshot_service')
         assert hasattr(mss, 'memory')
     finally:
         if env_source in os.environ:
             os.environ.pop(env_source, None)
 
 
-def test_default_file_code_snapshot_service_with_os_env(default_services):
+def test_default_file_code_snapshot_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('code_snapshot')
+        config.get_storage_config('code_snapshot')
     env_source = 'CODEPACK_CODE_SNAPSHOT_SOURCE'
     env_path = 'CODEPACK_CODE_SNAPSHOT_PATH'
     try:
-        os.environ[env_source] = 'FILE'
+        os.environ[env_source] = 'file'
         os.environ[env_path] = 'tmp/'
-        fss = default_services.get_default_code_snapshot_service()
+        fss = default.get_storage_instance('code_snapshot', 'snapshot_service')
         assert hasattr(fss, 'path')
         assert fss.path == 'tmp/'
     finally:
@@ -51,20 +53,21 @@ def test_default_file_code_snapshot_service_with_os_env(default_services):
                 os.environ.pop(env, None)
 
 
-def test_default_mongo_code_snapshot_service_with_os_env(default_services):
+def test_default_mongo_code_snapshot_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('code_snapshot')
+        config.get_storage_config('code_snapshot')
     env_source = 'CODEPACK_CODE_SNAPSHOT_SOURCE'
     env_db = 'CODEPACK_CODE_SNAPSHOT_DB'
     env_collection = 'CODEPACK_CODE_SNAPSHOT_COLLECTION'
     env_config_path = 'CODEPACK_CONN_PATH'
     mss = None
     try:
-        os.environ[env_source] = 'MONGODB'
+        os.environ[env_source] = 'mongodb'
         os.environ[env_db] = 'test'
         os.environ[env_collection] = 'snapshot'
         os.environ[env_config_path] = 'config/test_conn.ini'
-        mss = default_services.get_default_code_snapshot_service()
+        mss = default.get_storage_instance('code_snapshot', 'snapshot_service')
         assert hasattr(mss, 'mongodb')
     finally:
         for env in [env_source, env_db, env_collection, env_config_path]:
@@ -74,28 +77,30 @@ def test_default_mongo_code_snapshot_service_with_os_env(default_services):
             mss.mongodb.close()
 
 
-def test_default_memory_delivery_service_with_os_env(default_services):
+def test_default_memory_delivery_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('cache')
-    env_source = 'CODEPACK_CACHE_SOURCE'
+        config.get_storage_config('delivery')
+    env_source = 'CODEPACK_DELIVERY_SOURCE'
     try:
-        os.environ[env_source] = 'MEMORY'
-        mds = default_services.get_default_delivery_service()
+        os.environ[env_source] = 'memory'
+        mds = default.get_storage_instance('delivery', 'delivery_service')
         assert hasattr(mds, 'memory')
     finally:
         if env_source in os.environ:
             os.environ.pop(env_source, None)
 
 
-def test_default_file_delivery_service_with_os_env(default_services):
+def test_default_file_delivery_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('cache')
-    env_source = 'CODEPACK_CACHE_SOURCE'
-    env_path = 'CODEPACK_CACHE_PATH'
+        config.get_storage_config('delivery')
+    env_source = 'CODEPACK_DELIVERY_SOURCE'
+    env_path = 'CODEPACK_DELIVERY_PATH'
     try:
-        os.environ[env_source] = 'FILE'
+        os.environ[env_source] = 'file'
         os.environ[env_path] = 'tmp/'
-        fds = default_services.get_default_delivery_service()
+        fds = default.get_storage_instance('delivery', 'delivery_service')
         assert hasattr(fds, 'path')
         assert fds.path == 'tmp/'
     finally:
@@ -104,20 +109,21 @@ def test_default_file_delivery_service_with_os_env(default_services):
                 os.environ.pop(env, None)
 
 
-def test_default_mongo_delivery_service_with_os_env(default_services):
+def test_default_mongo_delivery_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('cache')
-    env_source = 'CODEPACK_CACHE_SOURCE'
-    env_db = 'CODEPACK_CACHE_DB'
-    env_collection = 'CODEPACK_CACHE_COLLECTION'
+        config.get_storage_config('delivery')
+    env_source = 'CODEPACK_DELIVERY_SOURCE'
+    env_db = 'CODEPACK_DELIVERY_DB'
+    env_collection = 'CODEPACK_DELIVERY_COLLECTION'
     env_config_path = 'CODEPACK_CONN_PATH'
     mds = None
     try:
-        os.environ[env_source] = 'MONGODB'
+        os.environ[env_source] = 'mongodb'
         os.environ[env_db] = 'test'
-        os.environ[env_collection] = 'cache'
+        os.environ[env_collection] = 'delivery'
         os.environ[env_config_path] = 'config/test_conn.ini'
-        mds = default_services.get_default_delivery_service()
+        mds = default.get_storage_instance('delivery', 'delivery_service')
         assert hasattr(mds, 'mongodb')
     finally:
         for env in [env_source, env_db, env_collection, env_config_path]:
@@ -127,54 +133,57 @@ def test_default_mongo_delivery_service_with_os_env(default_services):
             mds.mongodb.close()
 
 
-def test_default_memory_code_storage_service_with_os_env(default_services):
+def test_default_memory_code_storage_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('code')
-    env_source = 'CODEPACK_CODE_STORAGE_SOURCE'
+        config.get_storage_config('code')
+    env_source = 'CODEPACK_CODE_SOURCE'
     try:
-        os.environ[env_source] = 'MEMORY'
-        mss = default_services.get_default_code_storage_service(obj=Code)
+        os.environ[env_source] = 'memory'
+        mss = default.get_storage_instance('code', 'storage_service')
         assert hasattr(mss, 'memory')
-        assert mss.obj == Code
+        assert mss.item_type == Code
     finally:
         if env_source in os.environ:
             os.environ.pop(env_source, None)
 
 
-def test_default_file_code_storage_service_with_os_env(default_services):
+def test_default_file_code_storage_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('cache')
-    env_source = 'CODEPACK_CODE_STORAGE_SOURCE'
-    env_path = 'CODEPACK_CODE_STORAGE_PATH'
+        config.get_storage_config('code')
+    env_source = 'CODEPACK_CODE_SOURCE'
+    env_path = 'CODEPACK_CODE_PATH'
     try:
-        os.environ[env_source] = 'FILE'
+        os.environ[env_source] = 'file'
         os.environ[env_path] = 'tmp/'
-        fss = default_services.get_default_code_storage_service(obj=Code)
+        fss = default.get_storage_instance('code', 'storage_service')
         assert hasattr(fss, 'path')
         assert fss.path == 'tmp/'
-        assert fss.obj == Code
+        assert fss.item_type == Code
     finally:
         for env in [env_source, env_path]:
             if env in os.environ:
                 os.environ.pop(env, None)
 
 
-def test_default_mongo_code_storage_service_with_os_env(default_services):
+def test_default_mongo_code_storage_service_with_os_env(default):
+    config = Config()
     with pytest.raises(AssertionError):
-        get_default_service_config('cache')
-    env_source = 'CODEPACK_CODE_STORAGE_SOURCE'
-    env_db = 'CODEPACK_CODE_STORAGE_DB'
-    env_collection = 'CODEPACK_CODE_STORAGE_COLLECTION'
+        config.get_storage_config('code')
+    env_source = 'CODEPACK_CODE_SOURCE'
+    env_db = 'CODEPACK_CODE_DB'
+    env_collection = 'CODEPACK_CODE_COLLECTION'
     env_config_path = 'CODEPACK_CONN_PATH'
     mss = None
     try:
-        os.environ[env_source] = 'MONGODB'
+        os.environ[env_source] = 'mongodb'
         os.environ[env_db] = 'test'
         os.environ[env_collection] = 'codes'
         os.environ[env_config_path] = 'config/test_conn.ini'
-        mss = default_services.get_default_code_storage_service(obj=Code)
+        mss = default.get_storage_instance('code', 'storage_service')
         assert hasattr(mss, 'mongodb')
-        assert mss.obj == Code
+        assert mss.item_type == Code
     finally:
         for env in [env_source, env_db, env_collection, env_config_path]:
             if env in os.environ:
@@ -183,34 +192,34 @@ def test_default_mongo_code_storage_service_with_os_env(default_services):
             mss.mongodb.close()
 
 
-def test_if_default_services_have_single_instance_for_each_service(default_services, testdir_snapshot_service):
-    os.environ['CODEPACK_CACHE_SOURCE'] = 'MEMORY'
-    os.environ['CODEPACK_CODE_SNAPSHOT_SOURCE'] = 'FILE'
+def test_if_default_services_have_single_instance_for_each_service(default, testdir_snapshot_service):
+    os.environ['CODEPACK_DELIVERY_SOURCE'] = 'memory'
+    os.environ['CODEPACK_CODE_SNAPSHOT_SOURCE'] = 'file'
     os.environ['CODEPACK_CODE_SNAPSHOT_PATH'] = testdir_snapshot_service
-    os.environ['CODEPACK_CODE_STORAGE_SOURCE'] = 'MONGODB'
-    os.environ['CODEPACK_CODE_STORAGE_DB'] = 'test'
-    os.environ['CODEPACK_CODE_STORAGE_COLLECTION'] = 'codes'
+    os.environ['CODEPACK_CODE_SOURCE'] = 'mongodb'
+    os.environ['CODEPACK_CODE_DB'] = 'test'
+    os.environ['CODEPACK_CODE_COLLECTION'] = 'codes'
     os.environ['CODEPACK_CONN_PATH'] = 'config/test_conn.ini'
     try:
         code1 = Code(add2)
         code2 = Code(add3)
-        assert isinstance(code1.service['delivery_service'], MemoryDeliveryService)
-        assert isinstance(code1.service['snapshot_service'], FileSnapshotService)
-        assert isinstance(code1.service['storage_service'], MongoStorageService)
-        assert isinstance(code2.service['delivery_service'], MemoryDeliveryService)
-        assert isinstance(code2.service['snapshot_service'], FileSnapshotService)
-        assert isinstance(code2.service['storage_service'], MongoStorageService)
-
-        assert (code1.service['delivery_service']) == (code2.service['delivery_service'])
-        assert (code1.service['snapshot_service']) == (code2.service['snapshot_service'])
-        assert (code1.service['storage_service']) == (code2.service['storage_service'])
-
-        assert id(code1.service['delivery_service']) == id(code2.service['delivery_service'])
-        assert id(code1.service['snapshot_service']) == id(code2.service['snapshot_service'])
-        assert id(code1.service['storage_service']) == id(code2.service['storage_service'])
+        assert isinstance(code1.service['delivery'], MemoryDeliveryService)
+        assert isinstance(code1.service['snapshot'], FileSnapshotService)
+        assert isinstance(code1.service['storage'], MongoStorageService)
+        assert isinstance(code2.service['delivery'], MemoryDeliveryService)
+        assert isinstance(code2.service['snapshot'], FileSnapshotService)
+        assert isinstance(code2.service['storage'], MongoStorageService)
+
+        assert (code1.service['delivery']) == (code2.service['delivery'])
+        assert (code1.service['snapshot']) == (code2.service['snapshot'])
+        assert (code1.service['storage']) == (code2.service['storage'])
+
+        assert id(code1.service['delivery']) == id(code2.service['delivery'])
+        assert id(code1.service['snapshot']) == id(code2.service['snapshot'])
+        assert id(code1.service['storage']) == id(code2.service['storage'])
     finally:
-        for env in ['CODEPACK_CACHE_SOURCE',
+        for env in ['CODEPACK_DELIVERY_SOURCE',
                     'CODEPACK_CODE_SNAPSHOT_SOURCE', 'CODEPACK_CODE_SNAPSHOT_PATH',
-                    'CODEPACK_CODE_STORAGE_SOURCE', 'CODEPACK_CODE_STORAGE_DB', 'CODEPACK_CODE_STORAGE_COLLECTION',
+                    'CODEPACK_CODE_SOURCE', 'CODEPACK_CODE_DB', 'CODEPACK_CODE_COLLECTION',
                     'CODEPACK_CONN_PATH']:
             os.environ.pop(env, None)
diff --git a/tests/test_delivery_service.py b/tests/test_delivery_service.py
index 1592d03..94bb9ee 100644
--- a/tests/test_delivery_service.py
+++ b/tests/test_delivery_service.py
@@ -1,10 +1,10 @@
 from codepack.service import MemoryDeliveryService, FileDeliveryService, MongoDeliveryService
-from codepack.utils.delivery import Delivery
+from codepack.delivery import Delivery
 import os
 
 
 def test_memory_delivery_service_check():
-    mds = MemoryDeliveryService(obj=Delivery)
+    mds = MemoryDeliveryService(item_type=Delivery)
     mds.init()
     sender = 'sender'
     serial_number = '1234'
@@ -21,7 +21,7 @@ def test_memory_delivery_service_check():
 
 
 def test_file_delivery_service_check(testdir_delivery_service):
-    fds = FileDeliveryService(obj=Delivery, path=testdir_delivery_service)
+    fds = FileDeliveryService(item_type=Delivery, path=testdir_delivery_service)
     sender = 'sender'
     serial_number = '1234'
     serial_numbers = ['123', '456', '789']
@@ -48,7 +48,7 @@ def test_file_delivery_service_check(testdir_delivery_service):
 def test_mongo_delivery_service_check(fake_mongodb):
     db = 'test'
     collection = 'cache'
-    mds = MongoDeliveryService(obj=Delivery, mongodb=fake_mongodb, db=db, collection=collection)
+    mds = MongoDeliveryService(item_type=Delivery, mongodb=fake_mongodb, db=db, collection=collection)
     sender = 'sender'
     serial_number = '1234'
     serial_numbers = ['123', '456', '789']
@@ -67,7 +67,7 @@ def test_mongo_delivery_service_check(fake_mongodb):
 
 
 def test_memory_delivery_service():
-    mds = MemoryDeliveryService(obj=Delivery)
+    mds = MemoryDeliveryService(item_type=Delivery)
     mds.init()
     sender = 'sender'
     serial_number = '1234'
@@ -84,7 +84,7 @@ def test_memory_delivery_service():
 
 
 def test_file_delivery_service(testdir_delivery_service):
-    fds = FileDeliveryService(obj=Delivery, path=testdir_delivery_service)
+    fds = FileDeliveryService(item_type=Delivery, path=testdir_delivery_service)
     sender = 'sender'
     serial_number = '1234'
     item = 'Hello, World!'
@@ -102,7 +102,7 @@ def test_file_delivery_service(testdir_delivery_service):
 def test_mongo_delivery_service(fake_mongodb):
     db = 'test'
     collection = 'cache'
-    mds = MongoDeliveryService(obj=Delivery, mongodb=fake_mongodb, db=db, collection=collection)
+    mds = MongoDeliveryService(item_type=Delivery, mongodb=fake_mongodb, db=db, collection=collection)
     sender = 'sender'
     serial_number = '1234'
     item = 'Hello, World!'
diff --git a/tests/test_dependency.py b/tests/test_dependency.py
new file mode 100644
index 0000000..8b038e3
--- /dev/null
+++ b/tests/test_dependency.py
@@ -0,0 +1,12 @@
+from codepack.dependency import Dependency
+
+
+def test_eq():
+    dependency1 = Dependency(arg='a', id='add3', serial_number='1234')
+    dependency2 = Dependency(arg='a', id='add3', serial_number='1234')
+    assert dependency1 == dependency2
+    dependency_dict2 = dependency2.to_dict()
+    assert dependency1 == dependency_dict2
+    dependency2.arg = 'b'
+    assert dependency1 != dependency2
+    assert dependency2 != dependency_dict2
diff --git a/tests/test_dependency_state.py b/tests/test_dependency_state.py
index cbf940d..702b273 100644
--- a/tests/test_dependency_state.py
+++ b/tests/test_dependency_state.py
@@ -1,7 +1,7 @@
-from codepack.utils.dependency_state import DependencyState
+from codepack.dependency import DependencyState
 
 
-def test_eq(default_os_env):
-    assert DependencyState.READY == 'READY'
-    assert DependencyState.READY != DependencyState.NOT_READY
-    assert DependencyState.READY == 0
+def test_eq():
+    assert DependencyState.RESOLVED == 'RESOLVED'
+    assert DependencyState.RESOLVED != DependencyState.PENDING
+    assert DependencyState.RESOLVED == 0
diff --git a/tests/test_docker_manager.py b/tests/test_docker_manager.py
new file mode 100644
index 0000000..33ab556
--- /dev/null
+++ b/tests/test_docker_manager.py
@@ -0,0 +1,28 @@
+from codepack.docker import DockerManager
+
+
+def test_make_dockerfile(testdir_docker_manager):
+    ret = DockerManager.make_dockerfile(path='testdir/docker_test', base_image='python:3.7-slim',
+                                        args={'http_proxy': "http://1.2.3.4:5678", 'https_proxy': "''"},
+                                        requirements=['pymongo==3.12.1', 'numpy'],
+                                        pip_options={'upgrade': None,
+                                                     'trusted-host': ['pypi.python.org', 'pypi.org', 'files.pythonhosted.org']})
+    assert ret == "FROM python:3.7-slim\nARG http_proxy http://1.2.3.4:5678\nARG https_proxy ''\nRUN python -m " \
+                  "pip install --upgrade --trusted-host pypi.python.org --trusted-host pypi.org " \
+                  "--trusted-host files.pythonhosted.org pip\nRUN python -m pip install --upgrade " \
+                  "--trusted-host pypi.python.org --trusted-host pypi.org --trusted-host files.pythonhosted.org " \
+                  "pymongo==3.12.1 numpy\n"
+
+
+def test_combine_requirements():
+    ret = DockerManager.combine_requirements(base=['aaa', 'bbb==2.3', 'ccc'], additive=['bbb', 'ccc==2.5'])
+    assert sorted(ret) == sorted(['aaa', 'bbb', 'ccc==2.5'])
+
+
+def test_extract_requirements_from_file():
+    requirements = DockerManager.extract_requirements_from_file('requirements.txt')
+    assert requirements == ['dill==0.3.4', 'pymongo==3.12.1', 'numpy',
+                            'sshtunnel==0.4.0', 'PyMySQL==1.0.2', 'pymssql==2.2.2',
+                            'boto3==1.19.6', 'cx-Oracle==8.2.1', 'parse==1.19.0',
+                            'APScheduler==3.8.1', 'kafka-python==2.0.2',
+                            'docker==5.0.3', 'requests==2.26.0']
diff --git a/tests/test_snapshot_service.py b/tests/test_snapshot_service.py
index bf471d2..bbbbd4a 100644
--- a/tests/test_snapshot_service.py
+++ b/tests/test_snapshot_service.py
@@ -8,7 +8,7 @@
 
 
 def test_memory_code_snapshot_service_save_and_load(default_os_env):
-    mss = MemorySnapshotService(obj=CodeSnapshot)
+    mss = MemorySnapshotService(item_type=CodeSnapshot)
     mss.init()
     code = Code(add2)
     snapshot = CodeSnapshot(code)
@@ -24,7 +24,7 @@ def test_memory_code_snapshot_service_save_and_load(default_os_env):
 
 
 def test_memory_code_snapshot_service_search_and_remove(default_os_env):
-    mss = MemorySnapshotService(obj=CodeSnapshot)
+    mss = MemorySnapshotService(item_type=CodeSnapshot)
     mss.init()
     code1 = Code(add2)
     code2 = Code(add3)
@@ -58,7 +58,7 @@ def test_memory_code_snapshot_service_search_and_remove(default_os_env):
 
 
 def test_memory_snapshot_service_update():
-    mss = MemorySnapshotService(obj=Snapshot)
+    mss = MemorySnapshotService(item_type=Snapshot)
     mss.init()
     timestamp = datetime.now().timestamp()
     snapshot1 = Snapshot(id='1234', serial_number='5678', timestamp=timestamp)
@@ -72,7 +72,7 @@ def test_memory_snapshot_service_update():
 
 
 def test_file_code_snapshot_service_save_and_load(default_os_env, testdir_snapshot_service):
-    fss = FileSnapshotService(obj=CodeSnapshot, path=testdir_snapshot_service)
+    fss = FileSnapshotService(item_type=CodeSnapshot, path=testdir_snapshot_service)
     code = Code(add2)
     snapshot = CodeSnapshot(code)
     fss.save(snapshot=snapshot)
@@ -87,7 +87,7 @@ def test_file_code_snapshot_service_save_and_load(default_os_env, testdir_snapsh
 
 
 def test_file_code_snapshot_service_search_and_remove(default_os_env, testdir_snapshot_service):
-    fss = FileSnapshotService(obj=CodeSnapshot, path=testdir_snapshot_service)
+    fss = FileSnapshotService(item_type=CodeSnapshot, path=testdir_snapshot_service)
     code1 = Code(add2)
     code2 = Code(add3)
     code1 >> code2
@@ -120,7 +120,7 @@ def test_file_code_snapshot_service_search_and_remove(default_os_env, testdir_sn
 
 
 def test_file_snapshot_service_update(testdir_snapshot_service):
-    fss = FileSnapshotService(obj=Snapshot, path=testdir_snapshot_service)
+    fss = FileSnapshotService(item_type=Snapshot, path=testdir_snapshot_service)
     timestamp = datetime.now().timestamp()
     snapshot1 = Snapshot(id='1234', serial_number='5678', timestamp=timestamp)
     snapshot2 = Snapshot(id='1234', serial_number='5678', timestamp=timestamp + 1)
@@ -135,7 +135,7 @@ def test_file_snapshot_service_update(testdir_snapshot_service):
 def test_mongo_code_snapshot_service_save_and_load(default_os_env, fake_mongodb):
     db = 'test'
     collection = 'snapshot'
-    mss = MongoSnapshotService(obj=CodeSnapshot, mongodb=fake_mongodb, db=db, collection=collection)
+    mss = MongoSnapshotService(item_type=CodeSnapshot, mongodb=fake_mongodb, db=db, collection=collection)
     code = Code(add2)
     snapshot = CodeSnapshot(code)
     mss.save(snapshot=snapshot)
@@ -152,7 +152,7 @@ def test_mongo_code_snapshot_service_save_and_load(default_os_env, fake_mongodb)
 def test_mongo_code_snapshot_service_search_and_remove(default_os_env, fake_mongodb):
     db = 'test'
     collection = 'snapshot'
-    mss = MongoSnapshotService(obj=CodeSnapshot, mongodb=fake_mongodb, db=db, collection=collection)
+    mss = MongoSnapshotService(item_type=CodeSnapshot, mongodb=fake_mongodb, db=db, collection=collection)
     code1 = Code(add2)
     code2 = Code(add3)
     code1 >> code2
@@ -187,7 +187,7 @@ def test_mongo_code_snapshot_service_search_and_remove(default_os_env, fake_mong
 def test_mongo_snapshot_service_update(fake_mongodb):
     db = 'test'
     collection = 'snapshot'
-    mss = MongoSnapshotService(obj=Snapshot, mongodb=fake_mongodb, db=db, collection=collection)
+    mss = MongoSnapshotService(item_type=Snapshot, mongodb=fake_mongodb, db=db, collection=collection)
     timestamp = datetime.now().timestamp()
     snapshot1 = Snapshot(id='1234', serial_number='5678', timestamp=timestamp)
     snapshot2 = Snapshot(id='1234', serial_number='5678', timestamp=timestamp + 1)
diff --git a/tests/test_state.py b/tests/test_state.py
index be43241..b3f680a 100644
--- a/tests/test_state.py
+++ b/tests/test_state.py
@@ -1,7 +1,6 @@
-from codepack.utils.state import State
 from codepack import Code
 from tests import *
-from codepack.snapshot import Snapshot
+from codepack.snapshot import Snapshot, State
 
 
 def test_state_eq():
@@ -32,7 +31,7 @@ def test_code_waiting(default_os_env):
     code2.receive('b') << code1
     code2(3, c=2)
     assert code2.get_state() == 'WAITING'
-    snapshot = code2.service['snapshot_service'].load(serial_number=code2.serial_number)
+    snapshot = code2.service['snapshot'].load(serial_number=code2.serial_number)
     assert 'args' in snapshot
     assert isinstance(snapshot['args'], list)
     assert len(snapshot['args']) == 1
diff --git a/tests/test_storage_service.py b/tests/test_storage_service.py
index ea6f718..12f2561 100644
--- a/tests/test_storage_service.py
+++ b/tests/test_storage_service.py
@@ -5,7 +5,7 @@
 
 
 def test_memory_storage_service_check(default_os_env):
-    mss = MemoryStorageService(obj=Code)
+    mss = MemoryStorageService(item_type=Code)
     mss.init()
     code1 = Code(hello, id='hello1', storage_service=mss)
     code2 = Code(hello, id='hello2', storage_service=mss)
@@ -23,7 +23,7 @@ def test_memory_storage_service_check(default_os_env):
 
 
 def test_file_storage_service_check(default_os_env, testdir_storage_service):
-    fss = FileStorageService(obj=Code, path=testdir_storage_service)
+    fss = FileStorageService(item_type=Code, path=testdir_storage_service)
     code1 = Code(hello, id='hello1', storage_service=fss)
     code2 = Code(hello, id='hello2', storage_service=fss)
     code3 = Code(hello, id='hello3', storage_service=fss)
@@ -48,7 +48,7 @@ def test_file_storage_service_check(default_os_env, testdir_storage_service):
 def test_mongo_storage_service_check(default_os_env, fake_mongodb):
     db = 'test'
     collection = 'codes'
-    mss = MongoStorageService(obj=Code, mongodb=fake_mongodb, db=db, collection=collection)
+    mss = MongoStorageService(item_type=Code, mongodb=fake_mongodb, db=db, collection=collection)
     code1 = Code(hello, id='hello1', storage_service=mss)
     code2 = Code(hello, id='hello2', storage_service=mss)
     code3 = Code(hello, id='hello3', storage_service=mss)
@@ -71,7 +71,7 @@ def test_mongo_storage_service_check(default_os_env, fake_mongodb):
 
 
 def test_memory_storage_service(default_os_env):
-    mss = MemoryStorageService(obj=Code)
+    mss = MemoryStorageService(item_type=Code)
     mss.init()
     code1 = Code(hello, storage_service=mss)
     code2 = Code(add2, storage_service=mss)
@@ -89,7 +89,7 @@ def test_memory_storage_service(default_os_env):
 
 def test_file_storage_service(default_os_env, testdir_storage_service):
     filepath = testdir_storage_service
-    fss = FileStorageService(obj=Code, path=filepath)
+    fss = FileStorageService(item_type=Code, path=filepath)
     assert fss.path == filepath
     code1 = Code(hello, storage_service=fss)
     code2 = Code(add2, storage_service=fss)
@@ -108,14 +108,14 @@ def test_file_storage_service(default_os_env, testdir_storage_service):
 def test_mongo_storage_service(default_os_env, fake_mongodb):
     db = 'test'
     collection = 'codes'
-    mss = MongoStorageService(obj=Code, mongodb=fake_mongodb, db=db, collection=collection)
-    assert mss.obj == Code
+    mss = MongoStorageService(item_type=Code, mongodb=fake_mongodb, db=db, collection=collection)
+    assert mss.item_type == Code
     code1 = Code(hello, storage_service=mss)
     code2 = Code(add2, storage_service=mss)
     code1.save()
     assert mss.mongodb[db][collection].count_documents({'_id': code1.id}) == 1
     assert mss.mongodb[db][collection].count_documents({'_id': code2.id}) == 0
-    assert mss.obj == Code
+    assert mss.item_type == Code
     code3 = mss.load(code1.id)
     assert code1.id == code3.id
     assert code1.source.strip() == code3.source.strip()