server.py 5.34 KB
Newer Older
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
1
import asyncio
Olivier Dehaene's avatar
Olivier Dehaene committed
2
import os
3
import torch
Olivier Dehaene's avatar
Olivier Dehaene committed
4

Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
5
from grpc import aio
6
from loguru import logger
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
7
8
9

from grpc_reflection.v1alpha import reflection
from pathlib import Path
10
from typing import List, Optional
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
11

12
13
14
15
16
from text_generation_server.cache import Cache
from text_generation_server.interceptor import ExceptionInterceptor
from text_generation_server.models import Model, get_model
from text_generation_server.pb import generate_pb2_grpc, generate_pb2
from text_generation_server.tracing import UDSOpenTelemetryAioServerInterceptor
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
17
18


Olivier Dehaene's avatar
Olivier Dehaene committed
19
class TextGenerationService(generate_pb2_grpc.TextGenerationServiceServicer):
20
    def __init__(self, model: Model, cache: Cache, server_urls: List[str]):
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
21
22
23
        self.cache = cache
        self.model = model
        self.server_urls = server_urls
24
25
26
27
        # For some reason, inference_mode does not work well with GLOO which we use on CPU
        if model.device.type == "cuda":
            # Force inference mode for the lifetime of TextGenerationService
            self._inference_mode_raii_guard = torch._C._InferenceMode(True)
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
28

29
30
31
    async def Info(self, request, context):
        return self.model.info

32
33
34
35
36
    async def Health(self, request, context):
        if self.model.device.type == "cuda":
            torch.zeros((2, 2)).cuda()
        return generate_pb2.HealthResponse()

Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
37
38
39
40
    async def ServiceDiscovery(self, request, context):
        return generate_pb2.ServiceDiscoveryResponse(urls=self.server_urls)

    async def ClearCache(self, request, context):
41
42
43
44
45
46
        if request.HasField("id"):
            self.cache.delete(request.id)
        else:
            self.cache.clear()
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
Olivier Dehaene's avatar
Olivier Dehaene committed
47
        return generate_pb2.ClearCacheResponse()
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
48

49
50
51
52
53
54
55
56
57
    async def FilterBatch(self, request, context):
        batch = self.cache.pop(request.batch_id)
        if batch is None:
            raise ValueError(f"Batch ID {request.batch_id} not found in cache.")
        filtered_batch = batch.filter(request.keep_requests)
        self.cache.set(filtered_batch)

        return generate_pb2.FilterBatchResponse(batch=filtered_batch.to_pb())

58
    async def Prefill(self, request, context):
59
60
61
        batch = self.model.batch_type.from_pb(
            request.batch, self.model.tokenizer, self.model.device
        )
Olivier Dehaene's avatar
Olivier Dehaene committed
62

63
        generations, next_batch = self.model.generate_token(batch)
Olivier Dehaene's avatar
Olivier Dehaene committed
64
65
        self.cache.set(next_batch)

66
67
        return generate_pb2.PrefillResponse(
            generations=[generation.to_pb() for generation in generations],
Olivier Dehaene's avatar
Olivier Dehaene committed
68
            batch=next_batch.to_pb() if next_batch else None,
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
69
70
        )

71
    async def Decode(self, request, context):
Olivier Dehaene's avatar
Olivier Dehaene committed
72
73
74
75
76
77
78
79
        if len(request.batches) == 0:
            raise ValueError("Must provide at least one batch")

        batches = []
        for batch_pb in request.batches:
            batch = self.cache.pop(batch_pb.id)
            if batch is None:
                raise ValueError(f"Batch ID {batch_pb.id} not found in cache.")
80
            batches.append(batch)
81
82
83

        if len(batches) == 0:
            raise ValueError("All batches are empty")
Olivier Dehaene's avatar
Olivier Dehaene committed
84
85

        if len(batches) > 1:
86
            batch = self.model.batch_type.concatenate(batches)
Olivier Dehaene's avatar
Olivier Dehaene committed
87
88
89
        else:
            batch = batches[0]

90
        generations, next_batch = self.model.generate_token(batch)
Olivier Dehaene's avatar
Olivier Dehaene committed
91
92
        self.cache.set(next_batch)

93
94
        return generate_pb2.DecodeResponse(
            generations=[generation.to_pb() for generation in generations],
Olivier Dehaene's avatar
Olivier Dehaene committed
95
96
97
            batch=next_batch.to_pb() if next_batch else None,
        )

Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
98

Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
99
def serve(
100
    model_id: str,
101
    revision: Optional[str],
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
102
    sharded: bool,
103
    quantize: Optional[str],
104
    trust_remote_code: bool,
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
105
106
    uds_path: Path,
):
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
107
    async def serve_inner(
108
        model_id: str,
109
        revision: Optional[str],
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
110
        sharded: bool = False,
111
        quantize: Optional[str] = None,
112
        trust_remote_code: bool = False,
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
113
    ):
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
114
        unix_socket_template = "unix://{}-{}"
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
115
116
        if sharded:
            server_urls = [
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
117
                unix_socket_template.format(uds_path, rank)
118
                for rank in range(int(os.environ["WORLD_SIZE"]))
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
119
            ]
120
            local_url = server_urls[int(os.environ["RANK"])]
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
121
        else:
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
122
            local_url = unix_socket_template.format(uds_path, 0)
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
123
124
            server_urls = [local_url]

125
        try:
126
            model = get_model(model_id, revision, sharded, quantize, trust_remote_code)
127
128
129
        except Exception:
            logger.exception("Error when initializing model")
            raise
130

131
132
133
134
135
136
        server = aio.server(
            interceptors=[
                ExceptionInterceptor(),
                UDSOpenTelemetryAioServerInterceptor(),
            ]
        )
Olivier Dehaene's avatar
Olivier Dehaene committed
137
138
        generate_pb2_grpc.add_TextGenerationServiceServicer_to_server(
            TextGenerationService(model, Cache(), server_urls), server
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
139
140
        )
        SERVICE_NAMES = (
Olivier Dehaene's avatar
Olivier Dehaene committed
141
            generate_pb2.DESCRIPTOR.services_by_name["TextGenerationService"].full_name,
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
142
143
144
145
            reflection.SERVICE_NAME,
        )
        reflection.enable_server_reflection(SERVICE_NAMES, server)
        server.add_insecure_port(local_url)
146

Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
147
        await server.start()
148

149
        logger.info("Server started at {}".format(local_url))
150

Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
151
152
153
        try:
            await server.wait_for_termination()
        except KeyboardInterrupt:
154
            logger.info("Signal received. Shutting down")
Olivier Dehaene's avatar
v0.1.0  
Olivier Dehaene committed
155
            await server.stop(0)
Olivier Dehaene's avatar
Init  
Olivier Dehaene committed
156

157
    asyncio.run(serve_inner(model_id, revision, sharded, quantize, trust_remote_code))