Repository: insight-technology/restful-functions

Issue:
Set Task Timeout

Discussion:

Patch:
diff --git a/README.rst b/README.rst
index 188f40a..3c1c1b6 100644
--- a/README.rst
+++ b/README.rst
@@ -97,6 +97,7 @@ Example Usage::
         A value of X
         y INTEGER Requiered
         A value of Y
+    Timeout: 86400 sec
 
 
     multi
@@ -107,6 +108,7 @@ Example Usage::
     Description:
             Heavy Function
     No Args
+    Timeout: 86400 sec
 
 
     # Call Asynchronous
@@ -115,7 +117,7 @@ Example Usage::
     {"success": true, "message": "", "task_id": "c3a6a0ef-b19e-4e6f-bce3-8d0e5a9046aa"}
 
     # Obtain the result by task_id
-    $ curl http://localhost:8888/task/info/3a6a0ef-b19e-4e6f-bce3-8d0e5a9046aa
+    $ curl http://localhost:8888/task/info/c3a6a0ef-b19e-4e6f-bce3-8d0e5a9046aa
     {"task_id": "c3a6a0ef-b19e-4e6f-bce3-8d0e5a9046aa", "function_name": "addition", "status": "DONE", "result": 9}
 
     $ curl http://localhost:8888/task/result/c3a6a0ef-b19e-4e6f-bce3-8d0e5a9046aa
@@ -137,6 +139,83 @@ Example Usage::
     {"success": false, "message": "Over Max Concurrency 2", "task_id": ""}
 
 
+TIMEOUT
+-------
+
+You can set timeout to the functions.
+
+FunctionServer checks timeout to all runninng processes every 60 seconds (default).
+The interval can be changed with an argument `polling_timeout_process_interval`.
+
+Example Code::
+
+    import time
+
+    from restful_functions import FunctionServer
+
+    def long_process():
+        time.sleep(100)
+
+
+    if __name__ == '__main__':
+        # Make polling_timeout_process_interval a small value to confirm a terminated process due to timeout
+        server = FunctionServer(polling_timeout_process_interval=1.0)
+        server.add_function(
+            long_process,
+            [],
+            1,
+            'long_process'
+        )
+        server.start()
+
+Example Usage::
+
+    $ curl localhost:8888/
+    /function/list/data
+    /function/list/text
+    /function/definition/{function_name}
+    /function/running-count/{function_name}
+    /task/info/{task_id}
+    /task/done/{task_id}
+    /task/result/{task_id}
+    /task/list/{function_name}
+    /terminate/function/{function_name}
+    /terminate/task/{task_id}
+
+    $ curl localhost:8888/function/list/text
+    long_process
+      URL:
+        async api: /long_process
+        block api: /long_process/keep-connection
+      Max Concurrency: 1
+      Description:
+        timeout test
+      No Args
+      Timeout: 10 sec
+
+
+    # Call Asynchronous
+    # Obtain task_id
+    $ curl -X POST http://localhost:8888/long_process
+    {"success": true, "message": "", "task_id": "d6eb93f9-a8e7-4bba-817c-0fd975b3e41a"}
+
+    # Obtain the result by task_id
+    ## Before 10 seconds at the function called.
+    $ curl http://localhost:8888/task/info/d6eb93f9-a8e7-4bba-817c-0fd975b3e41a
+    {"task_id": "d6eb93f9-a8e7-4bba-817c-0fd975b3e41a", "function_name": "long_process", "status": "RUNNING", "result": "timeout"}
+
+    ## After 10 seconds at the function called.
+    $ curl http://localhost:8888/task/info/d6eb93f9-a8e7-4bba-817c-0fd975b3e41a
+    {"task_id": "d6eb93f9-a8e7-4bba-817c-0fd975b3e41a", "function_name": "long_process", "status": "TIMEOUT", "result": "timeout"}
+
+
+    # Call synchronous
+    # Keeping the connection until the process ends.
+    $ curl -X POST http://localhost:8888/long_process/keep-connection
+    "timeout"
+    # curl request ends after 10 seconds at the function called
+
+
 DEVELOPMENT
 -----------
 recommend: VSCode Remote Container Extension
diff --git a/examples/example_multi_pattern.py b/examples/example_multi_pattern.py
index d61b086..544472c 100755
--- a/examples/example_multi_pattern.py
+++ b/examples/example_multi_pattern.py
@@ -2,7 +2,6 @@ import concurrent.futures
 import math
 import os
 import sys
-import time
 
 # for develpment
 try:
@@ -15,8 +14,7 @@ except Exception:
 from restful_functions import ArgDefinition, ArgType, FunctionServer
 
 
-def awesome(x: int, y: int, wait: int):
-    time.sleep(wait)
+def addition(x: int, y: int):
     print(os.getpid(), os.getppid(), f'{x}+{y}={x+y}')
     return x+y
 
@@ -25,7 +23,7 @@ def no_arg_job():
     print(os.getpid(), os.getppid(), 'No args')
 
 
-def is_prime(n):
+def is_prime(n: int):
     if n % 2 == 0:
         return False
 
@@ -56,11 +54,10 @@ if __name__ == '__main__':
     server = FunctionServer(debug=True)
 
     server.add_function(
-        awesome,
+        addition,
         [
             ArgDefinition('x', ArgType.INTEGER, True, 'x'),
             ArgDefinition('y', ArgType.INTEGER, True, 'y'),
-            ArgDefinition('wait', ArgType.INTEGER, True, 'Waiting Seconds'),
         ],
         10,
         'Simple Addition')
diff --git a/examples/example_timeout.py b/examples/example_timeout.py
new file mode 100755
index 0000000..45cd7f7
--- /dev/null
+++ b/examples/example_timeout.py
@@ -0,0 +1,33 @@
+import os
+import sys
+
+# for develpment
+try:
+    import restful_functions  # NOQA
+except Exception:
+    print('using local version')
+    sys.path.append(os.path.join(os.path.dirname(__file__), '..'))
+# ------------ #
+
+
+import time
+
+from restful_functions import FunctionServer
+
+
+def long_process():
+    time.sleep(100)
+
+
+if __name__ == '__main__':
+    # Make polling_timeout_process_interval a smaller value than defaults to confirm a terminated process due to timeout
+    server = FunctionServer(polling_timeout_process_interval=1.0, debug=True)
+    server.add_function(
+        long_process,
+        [],
+        1,  # concurency
+        'timeout test',
+        None,
+        10  # timeout
+    )
+    server.start()
diff --git a/restful_functions/manager.py b/restful_functions/manager.py
index bf20386..7eee8f5 100644
--- a/restful_functions/manager.py
+++ b/restful_functions/manager.py
@@ -1,3 +1,4 @@
+import asyncio
 import functools
 import multiprocessing
 from typing import Any, Callable, Dict, List, Optional
@@ -33,6 +34,7 @@ class FunctionManager:
     def __init__(
             self,
             task_store_settings: TaskStoreSettings = TaskStoreSettings(),
+            polling_timeout_process_interval: float = 60.0,
             debug: bool = False):
 
         self._task_store_settings = task_store_settings
@@ -42,6 +44,8 @@ class FunctionManager:
         self._process_manager = ProcessManager()
         self._task_store = task_store_factory(task_store_settings, True, logger=get_logger('TaskStore'))
 
+        self._polling_timeout_process_interval = polling_timeout_process_interval
+
         self._logger = get_logger(self.__class__.__name__, debug=debug)
 
     @property
@@ -54,7 +58,8 @@ class FunctionManager:
             function_name: str,
             arg_definitions: List[ArgDefinition],
             max_concurrency: int = 0,
-            description: str = ''):
+            description: str = '',
+            timeout: int = 60 * 60 * 24):
         """Add Job to FunctionServer.
 
         Parameters
@@ -70,6 +75,8 @@ class FunctionManager:
             (the default is 0, which tells there is No Limitation.)
         description
             A Description for The Function. (the default is '', which [default_description])
+        timeout
+            Function timeout seconds for running. (the default is 86400, which means 1 day.)
 
         Raises
         ------
@@ -85,7 +92,8 @@ class FunctionManager:
             arg_definitions,
             max_concurrency,
             description,
-            function_name
+            function_name,
+            timeout
         )
 
     def _job_decorator(
@@ -120,7 +128,7 @@ class FunctionManager:
                 ''
             )
 
-        self._task_store.initialize_task(task_id, function_name)
+        self._task_store.initialize_task(task_id, func_def.function_name, func_def.timeout)
         self._process_manager.fork_process(jobnized_func, func_args, task_id)
 
         return TryForkResult(
@@ -157,3 +165,14 @@ class FunctionManager:
         for name in self._function_definitions:
             self._task_store.terminate_function(name)
         self._process_manager.terminate_processes()
+
+    def terminate_timeout_process_impl(self):
+        for task in self._task_store.list_timeout_task_info():
+            self._process_manager.terminate_task(task.task_id)
+            self._task_store.terminate_timeout_task(task.task_id)
+
+    async def terminate_timeout_processes_coro(self):
+        """Terminate Timeout Processes. coroutine."""
+        while True:
+            self.terminate_timeout_process_impl()
+            await asyncio.sleep(self._polling_timeout_process_interval)
diff --git a/restful_functions/modules/function.py b/restful_functions/modules/function.py
index ab4968c..2ac1251 100755
--- a/restful_functions/modules/function.py
+++ b/restful_functions/modules/function.py
@@ -93,6 +93,7 @@ class FunctionDefinition:
         'max_concurrency',
         'description',
         'function_name',
+        'timeout'
     ]
 
     def __init__(
@@ -101,7 +102,8 @@ class FunctionDefinition:
             arg_definitions: List[ArgDefinition],
             max_concurrency: int,
             description: str,
-            function_name: str):
+            function_name: str,
+            timeout: int):
         """.
 
         Parameters
@@ -116,13 +118,15 @@ class FunctionDefinition:
             A Description for this Function.
         function_name
             Function Name. It is not necessary to be same with func.__name__
-
+        timeout
+            Function timeout for running
         """
         self.func = func
         self.arg_definitions = arg_definitions
         self.max_concurrency = max_concurrency
         self.description = description
         self.function_name = function_name
+        self.timeout = timeout
 
     def to_dict(self) -> Dict[str, Any]:
         return {
@@ -130,4 +134,5 @@ class FunctionDefinition:
             'arg_definitions': [elm.to_dict() for elm in self.arg_definitions],
             'max_concurrency': self.max_concurrency,
             'description': self.description,
+            'timeout': self.timeout,
         }
diff --git a/restful_functions/modules/task.py b/restful_functions/modules/task.py
index 5b9ad9d..84ad7a4 100755
--- a/restful_functions/modules/task.py
+++ b/restful_functions/modules/task.py
@@ -29,7 +29,6 @@ class TaskStoreSettings:
         expired
             Expiration Time for storing a task result.
             The time is measured in seconds.
-
         """
         default_sqlite_dsn = os.path.join(gettempdir(), 'restful-functions.db')
 
@@ -41,6 +40,7 @@ class TaskStoreSettings:
 class TaskStatus(Enum):
     RUNNING = 'RUNNING'
     FAILED = 'FAILED'
+    TIMEOUT = 'TIMEOUT'
     DONE = 'DONE'
 
 
@@ -67,7 +67,7 @@ class TaskInfo:
         return self.status == TaskStatus.DONE
 
     def is_failed(self):
-        return self.status == TaskStatus.FAILED
+        return self.status == TaskStatus.FAILED or self.status == TaskStatus.TIMEOUT
 
     def to_dict(self):
         return {
@@ -84,7 +84,7 @@ class TaskStore:
     処理状態、結果格納
     """
 
-    def initialize_task(self, task_id: str, function_name: str):
+    def initialize_task(self, task_id: str, function_name: str, timeout: int):
         raise NotImplementedError
 
     def finish_task(self, task_id: str, result: Any):
@@ -99,6 +99,12 @@ class TaskStore:
     def list_task_info(self, function_name: str) -> List[TaskInfo]:
         raise NotImplementedError
 
+    def list_timeout_task_info(self) -> List[TaskInfo]:
+        raise NotImplementedError
+
+    def terminate_timeout_task(self, task_id: str):
+        raise NotImplementedError
+
     def terminate_task(self, task_id: str):
         self.finish_task(task_id, Exception('Manual Termination'))
 
@@ -115,13 +121,15 @@ class SQLiteTaskStore(TaskStore):
     Use SQLite to store tasks.
     """
 
-    _INITIALIZE_TASK_SQL = 'INSERT INTO task(task_id, function_name, status, result, expired) values(?, ?, ?, ?, ?)'
+    _INITIALIZE_TASK_SQL = 'INSERT INTO task(task_id, function_name, status, result, expired, timeout) values(?, ?, ?, ?, ?, ?)'
     _FINISH_TASK_SQL = 'UPDATE task SET status = ?, result = ?, expired = ? WHERE task_id = ?'
     _GET_TASK_INFO_SQL = 'SELECT * FROM task WHERE task_id = ? LIMIT 1'
-    _CURRENT_COUNT_SQL = f'SELECT count(*) FROM task WHERE function_name = ? AND status = "{TaskStatus.RUNNING.name}"'
+    _CURRENT_COUNT_SQL = f"SELECT count(*) FROM task WHERE function_name = ? AND status = '{TaskStatus.RUNNING.name}'"
     _LIST_TASK_INFO_SQL = 'SELECT * FROM task WHERE function_name = ?'
 
-    _DELETE_OLD_STATUS_SQL = 'DELETE FROM task WHERE expired < ?'
+    _LIST_TIMEOUT_TASK_SQL = f"SELECT * FROM task WHERE status == '{TaskStatus.RUNNING.name}' AND timeout IS NOT NULL AND timeout < ?"
+
+    _DELETE_OLD_STATUS_SQL = 'DELETE FROM task WHERE expired IS NOT NULL AND expired < ?'
 
     def __init__(
             self,
@@ -137,8 +145,9 @@ class SQLiteTaskStore(TaskStore):
             cur = conn.cursor()
 
             cur.execute('DROP TABLE IF EXISTS task')
-            cur.execute('CREATE TABLE IF NOT EXISTS task (task_id TEXT PRIMARY KEY, function_name TEXT, status TEXT, result TEXT, expired INTEGER)')
+            cur.execute('CREATE TABLE IF NOT EXISTS task (task_id TEXT PRIMARY KEY, function_name TEXT, status TEXT, result TEXT, expired INTEGER, timeout INTEGER)')
             cur.execute('CREATE INDEX expired_idx ON task(expired)')
+            cur.execute('CREATE INDEX timeout_idx ON task(timeout)')
             conn.commit()
 
             conn.close()
@@ -150,11 +159,11 @@ class SQLiteTaskStore(TaskStore):
     def _get_db(self):
         return sqlite3.connect(self._dsn, 30.0)
 
-    def initialize_task(self, task_id: str, function_name: str):
+    def initialize_task(self, task_id: str, function_name: str, timeout: int):
         current_unix_time = int(datetime.datetime
                                 .now(datetime.timezone.utc)
                                 .timestamp())
-        expired = current_unix_time + self._expired
+        timeout = current_unix_time + timeout
 
         conn = self._get_db()
         cur = conn.cursor()
@@ -163,7 +172,8 @@ class SQLiteTaskStore(TaskStore):
             function_name,
             TaskStatus.RUNNING.name,
             json.dumps({}),
-            expired,
+            None,
+            timeout
         ))
         conn.commit()
         conn.close()
@@ -198,6 +208,22 @@ class SQLiteTaskStore(TaskStore):
         conn.commit()
         conn.close()
 
+    def terminate_timeout_task(self, task_id: str):
+        current_unix_time = int(datetime.datetime
+                                .now(datetime.timezone.utc)
+                                .timestamp())
+        expired = current_unix_time + self._expired
+        conn = self._get_db()
+        cur = conn.cursor()
+        cur.execute(SQLiteTaskStore._FINISH_TASK_SQL, (
+            TaskStatus.TIMEOUT.name,
+            json.dumps('timeout'),
+            expired,
+            task_id,
+        ))
+        conn.commit()
+        conn.close()
+
     def get_task_info(self, task_id: str) -> Optional[TaskInfo]:
         self._refresh_tasks()
 
@@ -248,6 +274,31 @@ class SQLiteTaskStore(TaskStore):
 
         return info_list
 
+    def list_timeout_task_info(self) -> List[TaskInfo]:
+        self._refresh_tasks()
+
+        current_unix_time = int(datetime.datetime
+                                .now(datetime.timezone.utc)
+                                .timestamp())
+
+        conn = self._get_db()
+        cur = conn.cursor()
+        ret = cur.execute(
+            SQLiteTaskStore._LIST_TIMEOUT_TASK_SQL,
+            (current_unix_time,)).fetchall()
+        conn.close()
+
+        info_list = []
+        for row in ret:
+            info_list.append(TaskInfo(
+                row[0],
+                row[1],
+                TaskStatus[row[2]],
+                json.loads(row[3]),
+            ))
+
+        return info_list
+
     def _refresh_tasks(self):
         current_unix_time = int(datetime.datetime
                                 .now(datetime.timezone.utc)
diff --git a/restful_functions/server.py b/restful_functions/server.py
index b8acf4d..430f11f 100755
--- a/restful_functions/server.py
+++ b/restful_functions/server.py
@@ -21,8 +21,8 @@ class FunctionServer:
             shutdown_mode: str = 'join',
             host: str = '0.0.0.0',
             port: int = 8888,
-            timeout: int = 60*60*24,
-            polling_interval: float = 1.0,
+            polling_process_status_interval: float = 1.0,
+            polling_timeout_process_interval: float = 60.0,
             debug: bool = False,
             register_sys_signals: bool = True,
             task_store_settings: TaskStoreSettings = TaskStoreSettings(),
@@ -40,14 +40,13 @@ class FunctionServer:
             RESTful APIs host. (the default is '0.0.0.0')
         port
             RESTful APIs port. (the default is 8888)
-        timeout
-            Timeout of keeping the connection after a connection established.
-            This is measured in Second.
-            (the default is 60*60*24)
-        polling_interval
+        polling_process_status_interval
             An interval of confirming processes runnning.
-            FuncitonServer repeats checking a process status when you use blocking API.
+            FunctionServer repeats checking the process status for blocking API while the function is running.
             (the default is 1.0)
+        polling_timeout_process_interval
+            An interval of searching for timeout processes in all runninng processes.
+            (the default is 60.0)
         debug
             Is Debug Mode or Not. (the default is False)
         register_sys_signals
@@ -66,13 +65,13 @@ class FunctionServer:
         if shutdown_mode not in ['terminate', 'join']:
             raise ValueError('"shutdown_mode" should be "terminate" or "join"')
 
-        if timeout <= 0:
-            raise ValueError('"timeout" should be greater than 0')
+        if polling_process_status_interval <= 0.0:
+            raise ValueError('"polling_process_status_interval" should be greater than 0.0')
 
-        if polling_interval <= 0.0:
-            raise ValueError('"polling_interval" should be greater than 0.0')
+        if polling_timeout_process_interval <= 0.0:
+            raise ValueError('"polling_timeout_process_interval" should be greater than 0.0')
 
-        self._function_manager = FunctionManager(task_store_settings, debug)
+        self._function_manager = FunctionManager(task_store_settings, polling_timeout_process_interval, debug)
 
         self._app = web.Application(
             middlewares=(web.normalize_path_middleware(append_slash=False, remove_slash=True),)
@@ -82,7 +81,7 @@ class FunctionServer:
         self._host = host
         self._port = port
 
-        self._polling_interval = polling_interval
+        self._polling_timeout_process_interval = polling_process_status_interval
 
         self._debug = debug
 
@@ -128,7 +127,9 @@ class FunctionServer:
             while True:
                 await asyncio.sleep(3600)
 
-        self._loop.run_until_complete(server_coro())
+        self._loop.create_task(server_coro())
+        self._loop.create_task(self._function_manager.terminate_timeout_processes_coro())
+        self._loop.run_forever()
 
     def _construct_endpoints(self):
         """Splited to Unit Test with no server running."""
@@ -155,6 +156,7 @@ class FunctionServer:
                         rows.append(f'    {arg.name} {arg.type.name} {"Requiered" if arg.is_required else "NOT-Required"}')
                         if arg.description != '':
                             rows.append(f'      {arg.description}')
+                rows.append(f'  Timeout: {definition.timeout} sec')
                 rows.append('\n')
 
             return web.Response(text='\n'.join(rows))
@@ -327,7 +329,8 @@ class FunctionServer:
             arg_definitions: List[ArgDefinition],
             max_concurrency: int = 0,
             description: str = '',
-            function_name: str = ''):
+            function_name: str = None,
+            timeout: int = 60 * 60 * 24):
         """Add Job to FunctionServer.
 
         Parameters
@@ -345,11 +348,15 @@ class FunctionServer:
             Function Name. It is not necessary to same with func.__name__.
             This values is used as REST API's endpoint.
             (the default is None, which uses the func.__name__.)
+        timeout
+            Function timeout seconds for running. (the default is 86400, which means 1 day.)
+            timeout works properly if this value is more than polling_timeout_process_interval (this class's property).
 
         Raises
         ------
         ValueError
             For max_concurrency.
+            For timeout.
 
         """
         if max_concurrency < 0:
@@ -361,12 +368,16 @@ class FunctionServer:
         if function_name is None or function_name == '':
             function_name = func.__name__
 
+        if timeout < 0:
+            raise ValueError
+
         self._function_manager.add_function(
             func,
             function_name,
             arg_definitions,
             max_concurrency,
-            description
+            description,
+            timeout
         )
 
         api_async_endpoint = f'/{function_name}'
@@ -386,7 +397,7 @@ class FunctionServer:
                 self._logger.info(e)
                 return web.json_response({'error': str(e)}, status=400)
 
-            result = self._function_manager.launch_function(function_name, func_args)
+            result = self._function_manager.launch_function(function_name, func_args)  # type: ignore
             return web.json_response(result.to_dict())
 
         async def post_sync_function(request: web.Request):
@@ -403,7 +414,7 @@ class FunctionServer:
                 self._logger.info(e)
                 return web.json_response({'error': str(e)}, status=400)
 
-            result = self._function_manager.launch_function(function_name, func_args)
+            result = self._function_manager.launch_function(function_name, func_args)  # type: ignore
 
             if not result.success:
                 return web.json_response(result.to_dict())
@@ -420,7 +431,7 @@ class FunctionServer:
                 if task_info.is_done():
                     break
 
-                await asyncio.sleep(self._polling_interval)
+                await asyncio.sleep(self._polling_timeout_process_interval)
 
             return web.json_response(task_info.result)
 


Test patch:
diff --git a/tests/modules/test_task.py b/tests/modules/test_task.py
index 25687fa..e77f58b 100644
--- a/tests/modules/test_task.py
+++ b/tests/modules/test_task.py
@@ -41,13 +41,14 @@ def test_task_store_factory(
 ])
 def test_task_store_count_operation(settings: TaskStoreSettings):
     FUNC_NAME = 'job1'
+    TIMEOUT = 60
 
     store = task_store_factory(settings, True)
 
     assert store.get_current_count(FUNC_NAME) == 0
     assert store.get_task_info('fake_id_1') is None
 
-    store.initialize_task('fake_id_1', FUNC_NAME)
+    store.initialize_task('fake_id_1', FUNC_NAME, TIMEOUT)
     assert store.get_current_count(FUNC_NAME) == 1
     assert store.get_task_info('fake_id_1').is_running()
 
@@ -56,8 +57,8 @@ def test_task_store_count_operation(settings: TaskStoreSettings):
     assert store.get_task_info('fake_id_1').is_done()
     assert store.get_task_info('fake_id_1').is_success()
 
-    store.initialize_task('fake_id_2', FUNC_NAME)
-    store.initialize_task('fake_id_3', FUNC_NAME)
+    store.initialize_task('fake_id_2', FUNC_NAME, TIMEOUT)
+    store.initialize_task('fake_id_3', FUNC_NAME, TIMEOUT)
     assert store.get_current_count(FUNC_NAME) == 2
 
     store.finish_task('fake_id_3', 'ret')
@@ -82,10 +83,11 @@ def test_task_store_task_result_expiration_operation(settings: TaskStoreSettings
     TASK_ID = 'fake_id'
     TASK_RESULT = 'artifact'
     SLEEP_TIME = 2
+    TIMEOUT = 60
 
     store = task_store_factory(settings, True)
 
-    store.initialize_task(TASK_ID, FUNC_NAME)
+    store.initialize_task(TASK_ID, FUNC_NAME, TIMEOUT)
     store.finish_task(TASK_ID, TASK_RESULT)
     sleep(SLEEP_TIME)
 
@@ -104,26 +106,25 @@ def test_task_store_task_result_expiration_operation(settings: TaskStoreSettings
     ),
 ])
 def test_task_store_termination_task_operation(settings: TaskStoreSettings):
+    TIMEOUT = 60
+
     store = task_store_factory(settings, True)
 
     assert store.get_current_count('func1') == 0
-
     assert store.get_task_info('task_id_1') is None
 
-    store.initialize_task('task_id_1', 'func1')
+    store.initialize_task('task_id_1', 'func1', TIMEOUT)
 
     assert store.get_task_info('task_id_1').is_running()
-
     assert store.get_current_count('func1') == 1
 
     store.terminate_task('task_id_1')
 
     assert store.get_current_count('func1') == 0
-
     assert store.get_task_info('task_id_1').is_failed()
 
-    store.initialize_task('task_id_2', 'func1')
-    store.initialize_task('task_id_3', 'func1')
+    store.initialize_task('task_id_2', 'func1', TIMEOUT)
+    store.initialize_task('task_id_3', 'func1', TIMEOUT)
 
     assert store.get_current_count('func1') == 2
 
@@ -135,24 +136,65 @@ def test_task_store_termination_task_operation(settings: TaskStoreSettings):
     assert store.get_task_info('task_id_3').is_failed()
 
 
+@pytest.mark.parametrize('settings', [
+    TaskStoreSettings(
+        store_type='sqlite',
+    ),
+])
+def test_task_store_termination_timeout_functions_operation(settings: TaskStoreSettings):
+    TIMEOUT = 60
+
+    store = task_store_factory(settings, True)
+
+    assert store.get_current_count('func1') == 0
+    assert store.get_task_info('task_id_1') is None
+
+    store.initialize_task('task_id_1', 'func1', TIMEOUT)
+
+    assert store.get_task_info('task_id_1').is_running()
+    assert store.get_current_count('func1') == 1
+
+    store.terminate_timeout_task('task_id_1')
+
+    assert store.get_current_count('func1') == 0
+    assert store.get_task_info('task_id_1').is_failed()
+    assert store.get_task_info('task_id_1').status.name == 'TIMEOUT'
+
+    store.initialize_task('task_id_2', 'func1', TIMEOUT)
+    store.initialize_task('task_id_3', 'func1', TIMEOUT)
+
+    assert store.get_current_count('func1') == 2
+
+    store.terminate_timeout_task('task_id_2')
+    store.terminate_timeout_task('task_id_3')
+
+    assert store.get_current_count('func1') == 0
+    assert store.get_task_info('task_id_2').is_failed()
+    assert store.get_task_info('task_id_2').status.name == 'TIMEOUT'
+    assert store.get_task_info('task_id_3').is_failed()
+    assert store.get_task_info('task_id_3').status.name == 'TIMEOUT'
+
+
 @pytest.mark.parametrize('settings', [
     TaskStoreSettings(
         store_type='sqlite',
     ),
 ])
 def test_task_store_termination_functions_operation(settings: TaskStoreSettings):
+    TIMEOUT = 60
+
     store = task_store_factory(settings, True)
 
     assert store.get_current_count('func1') == 0
     store.terminate_function('func1')
     assert store.get_current_count('func1') == 0
 
-    store.initialize_task('task_id_1', 'func1')
-    store.initialize_task('task_id_2', 'func1')
-    store.initialize_task('task_id_3', 'func1')
+    store.initialize_task('task_id_1', 'func1', TIMEOUT)
+    store.initialize_task('task_id_2', 'func1', TIMEOUT)
+    store.initialize_task('task_id_3', 'func1', TIMEOUT)
 
-    store.initialize_task('task_id_4', 'func2')
-    store.initialize_task('task_id_5', 'func2')
+    store.initialize_task('task_id_4', 'func2', TIMEOUT)
+    store.initialize_task('task_id_5', 'func2', TIMEOUT)
 
     assert store.get_current_count('func1') == 3
     assert store.get_current_count('func2') == 2
diff --git a/tests/test_manager.py b/tests/test_manager.py
index bff46f0..e516071 100644
--- a/tests/test_manager.py
+++ b/tests/test_manager.py
@@ -4,7 +4,14 @@ import pytest
 
 from restful_functions.manager import FunctionManager
 from restful_functions.modules.function import ArgDefinition, ArgType
-from restful_functions.modules.task import TaskStoreSettings
+from restful_functions.modules.task import TaskInfo, TaskStatus, TaskStoreSettings
+
+
+def get_task_info_with_waiting_done(manager: FunctionManager, task_id: str) -> TaskInfo:
+    task_info = manager.get_task_info(task_id)
+    while task_info is None or task_info.is_running():
+        task_info = manager.get_task_info(task_id)
+    return task_info
 
 
 @pytest.mark.parametrize('settings', [
@@ -63,18 +70,14 @@ def test_task_manager_launch_function(settings: TaskStoreSettings):
     ret = manager.launch_function(FAST_FUNC.__name__, {})
     assert ret.success
 
-    task_info = manager.get_task_info(ret.task_id)
-    while task_info is None or task_info.is_running():
-        task_info = manager.get_task_info(ret.task_id)
+    task_info = get_task_info_with_waiting_done(manager, ret.task_id)
     assert task_info.is_failed()
 
     # Valid Args
     ret = manager.launch_function(FAST_FUNC.__name__, {'x': 3, 'y': 4})
     assert ret.success
 
-    task_info = manager.get_task_info(ret.task_id)
-    while task_info is None or task_info.is_running():
-        task_info = manager.get_task_info(ret.task_id)
+    task_info = get_task_info_with_waiting_done(manager, ret.task_id)
     assert task_info.is_done()
     assert task_info.result == 7
 
@@ -111,10 +114,7 @@ def test_task_manager_launch_function_slow_func(settings: TaskStoreSettings):
         SLOW_FUNC.__name__,
         {'x': 3, 'y': 4, 'wait': -1})
 
-    task_info = manager.get_task_info(ret.task_id)
-    while task_info is None or task_info.is_running():
-        task_info = manager.get_task_info(ret.task_id)
-
+    task_info = get_task_info_with_waiting_done(manager, ret.task_id)
     assert task_info.is_failed()
 
     # concurrency
@@ -139,3 +139,63 @@ def test_task_manager_launch_function_slow_func(settings: TaskStoreSettings):
     assert not ret.success
 
     manager.terminate_processes()
+
+
+@pytest.mark.parametrize('settings', [
+    TaskStoreSettings(
+        store_type='sqlite',
+    ),
+])
+def test_task_manager_timeout(settings: TaskStoreSettings):
+    manager = FunctionManager(settings)
+
+    def SLOW_FUNC(x: int, y: int, wait: int):
+        if wait < 1:
+            raise ValueError
+
+        sleep(wait)
+        return x + y
+
+    manager.add_function(
+        SLOW_FUNC,
+        SLOW_FUNC.__name__,
+        [
+            ArgDefinition('x', ArgType.INTEGER, True, '1st'),
+            ArgDefinition('y', ArgType.INTEGER, True, '2nd'),
+            ArgDefinition('wait', ArgType.INTEGER, True, 'wait sec')
+        ],
+        10,
+        'slow',
+        3  # timeout
+    )
+
+    # short waiting time
+    ret = manager.launch_function(
+        SLOW_FUNC.__name__,
+        {'x': 3, 'y': 4, 'wait': 1})
+    assert ret.success
+
+    sleep(2)
+
+    manager.terminate_timeout_process_impl()  # manual calling
+
+    task_info = get_task_info_with_waiting_done(manager, ret.task_id)
+    assert task_info.is_success()
+    assert task_info.result == 7
+
+    # long waiting time (timeout)
+    ret = manager.launch_function(
+        SLOW_FUNC.__name__,
+        {'x': 3, 'y': 4, 'wait': 10.0})
+    assert ret.success
+
+    sleep(4)
+
+    manager.terminate_timeout_process_impl()  # manual calling
+
+    task_info = task_info = get_task_info_with_waiting_done(manager, ret.task_id)
+    assert task_info.is_failed()
+    assert task_info.status == TaskStatus.TIMEOUT
+    assert task_info.result == 'timeout'
+
+    manager.terminate_processes()