#758

Committed 04 Sep 2023 08:37PM UTC coverage: 0.0% (-78.3%) from 78.333%

Build # #758

Build Type

push

circle-ci

Committed by

hershd23

Commit Message

Increased underline length in at line 75 in text_summarization.rst
	modified:   docs/source/benchmarks/text_summarization.rst

Run Details

0 of 11303 relevant lines covered (0.0%)

0.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/evadb/server/server.py

# coding=utf-8
# Copyright 2018-2023 EvaDB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import string
from asyncio import StreamReader, StreamWriter

from evadb.database import init_evadb_instance
from evadb.udfs.udf_bootstrap_queries import init_builtin_udfs
from evadb.utils.logging_manager import logger


class EvaServer:
    """
    Receives messages and offloads them to another task for processing them.
    """

    def __init__(self):
        self._server = None
        self._clients = {}  # client -> (reader, writer)
        self._evadb = None

    async def start_evadb_server(
        self, db_dir: str, host: string, port: int, custom_db_uri: str = None
    ):
        """
        Start the server
        Server objects are asynchronous context managers.

        hostname: hostname of the server
        port: port of the server
        """
        from pprint import pprint

        pprint(f"EvaDB server started at host {host} and port {port}")
        self._evadb = init_evadb_instance(db_dir, host, port, custom_db_uri)

        self._server = await asyncio.start_server(self.accept_client, host, port)

        # load built-in udfs
        mode = self._evadb.config.get_value("core", "mode")
        init_builtin_udfs(self._evadb, mode=mode)

        async with self._server:
            await self._server.serve_forever()

        logger.warn("EvaDB server stopped")

    async def stop_evadb_server(self):
        logger.warn("EvaDB server stopped")
        if self._server is not None:
            await self._server.close()

    async def accept_client(
        self, client_reader: StreamReader, client_writer: StreamWriter
    ):
        task = asyncio.Task(self.handle_client(client_reader, client_writer))
        self._clients[task] = (client_reader, client_writer)

        def close_client(task):
            del self._clients[task]
            client_writer.close()
            logger.info("End connection")

        logger.info("New client connection")
        task.add_done_callback(close_client)

    async def handle_client(
        self, client_reader: StreamReader, client_writer: StreamWriter
    ):
        try:
            while True:
                data = await asyncio.wait_for(client_reader.readline(), timeout=None)
                if data == b"":
                    break

                message = data.decode().rstrip()
                logger.debug("Received --|%s|--", message)

                if message.upper() in ["EXIT;", "QUIT;"]:
                    logger.info("Close client")
                    return

                logger.debug("Handle request")
                from evadb.server.command_handler import handle_request

                asyncio.create_task(handle_request(self._evadb, client_writer, message))

        except Exception as e:
            logger.critical("Error reading from client.", exc_info=e)

1	# coding=utf-8
2	# Copyright 2018-2023 EvaDB
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15	import asyncio	×
16	import string	×
17	from asyncio import StreamReader, StreamWriter	×
18
19	from evadb.database import init_evadb_instance	×
20	from evadb.udfs.udf_bootstrap_queries import init_builtin_udfs	×
21	from evadb.utils.logging_manager import logger	×
22
23
24	class EvaServer:	×
25	"""
26	Receives messages and offloads them to another task for processing them.
27	"""
28
29	def __init__(self):	×
30	self._server = None	×
31	self._clients = {} # client -> (reader, writer)	×
32	self._evadb = None	×
33
34	async def start_evadb_server(	×
35	self, db_dir: str, host: string, port: int, custom_db_uri: str = None
36	):
37	"""
38	Start the server
39	Server objects are asynchronous context managers.
40
41	hostname: hostname of the server
42	port: port of the server
43	"""
44	from pprint import pprint	×
45
46	pprint(f"EvaDB server started at host {host} and port {port}")	×
47	self._evadb = init_evadb_instance(db_dir, host, port, custom_db_uri)	×
48
49	self._server = await asyncio.start_server(self.accept_client, host, port)	×
50
51	# load built-in udfs
52	mode = self._evadb.config.get_value("core", "mode")	×
53	init_builtin_udfs(self._evadb, mode=mode)	×
54
55	async with self._server:	×
56	await self._server.serve_forever()	×
57
58	logger.warn("EvaDB server stopped")	×
59
60	async def stop_evadb_server(self):	×
61	logger.warn("EvaDB server stopped")	×
62	if self._server is not None:	×
63	await self._server.close()	×
64
65	async def accept_client(	×
66	self, client_reader: StreamReader, client_writer: StreamWriter
67	):
68	task = asyncio.Task(self.handle_client(client_reader, client_writer))	×
69	self._clients[task] = (client_reader, client_writer)	×
70
71	def close_client(task):	×
72	del self._clients[task]	×
73	client_writer.close()	×
74	logger.info("End connection")	×
75
76	logger.info("New client connection")	×
77	task.add_done_callback(close_client)	×
78
79	async def handle_client(	×
80	self, client_reader: StreamReader, client_writer: StreamWriter
81	):
82	try:	×
83	while True:	×
84	data = await asyncio.wait_for(client_reader.readline(), timeout=None)	×
85	if data == b"":	×
86	break	×
87
88	message = data.decode().rstrip()	×
89	logger.debug("Received --\|%s\|--", message)	×
90
91	if message.upper() in ["EXIT;", "QUIT;"]:	×
92	logger.info("Close client")	×
93	return	×
94
95	logger.debug("Handle request")	×
96	from evadb.server.command_handler import handle_request	×
97
98	asyncio.create_task(handle_request(self._evadb, client_writer, message))	×
99
100	except Exception as e:
101	logger.critical("Error reading from client.", exc_info=e)

georgia-tech-db / eva / #758

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous