diff --git a/README.md b/README.md index 33c9393..4d33d83 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,15 @@ sudo bash deploy/install-client.sh --server http://your-server:5050 --token "cha Options: `--server`, `--token` (required); `--client-id`, `--upstream`, `--install-dir`, `--service-name`, `--no-ollama` (optional). Missing required values are prompted interactively. +llama.cpp via Docker options (replaces Ollama for inferencing): + +| Option | Description | +|--------|-------------| +| `--use-llama-cpp-via-docker` | Use llama.cpp Docker containers instead of Ollama | +| `--use-ollama-models-path ` | Path to Ollama models directory (`manifests/blobs`); required with the flag above | +| `--llama-cpp-docker-image ` | Docker image; defaults to auto-detected (rocm/cuda/cpu) | +| `--llama-cpp-base-port ` | Base port for containers; defaults to `8081` | + The script ensures .NET 10 and Ollama are installed, builds the client self-contained, installs it to `/opt/Ngino-client`, and creates a systemd service (`Ngino-client`). Logs: `journalctl -u Ngino-client -f`. ## Notes diff --git a/deploy/install-client.sh b/deploy/install-client.sh index 23f757b..dd5d7ec 100755 --- a/deploy/install-client.sh +++ b/deploy/install-client.sh @@ -279,6 +279,14 @@ fi info "Build successful." +# ── Stop existing service before overwriting binary ───────────────────────── +SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service" + +if systemctl list-unit-files "$SERVICE_NAME.service" &>/dev/null 2>&1 || systemctl is-active --quiet "$SERVICE_NAME" 2>/dev/null; then + info "Stopping existing service $SERVICE_NAME..." + systemctl stop "$SERVICE_NAME" 2>/dev/null || true +fi + # ── Install ─────────────────────────────────────────────────────────────────── info "Installing to $INSTALL_DIR..." mkdir -p "$INSTALL_DIR" @@ -311,11 +319,6 @@ info "Environment file written to $ENV_DIR/env (mode 0600)." # ── Create systemd service ─────────────────────────────────────────────────── SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service" -if systemctl list-unit-files "$SERVICE_NAME.service" &>/dev/null 2>&1; then - info "Stopping existing service $SERVICE_NAME..." - systemctl stop "$SERVICE_NAME" 2>/dev/null || true -fi - cat > "$SERVICE_FILE" < FindExistingContainerPortAsync(string containerName) + { + var (exitCode, output) = await RunDockerWithOutputAsync( + ["ps", "--filter", $"name=^{containerName}$", "--format", "{{.Ports}}"], + CancellationToken.None); + + if (exitCode != 0 || string.IsNullOrWhiteSpace(output)) + { + return null; + } + + foreach (var line in output.Split('\n', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)) + { + var port = ParseHostPort(line); + if (port.HasValue) + { + return port; + } + } + + return null; + } + + private static int? ParseHostPort(string ports) + { + foreach (var mapping in ports.Split(',')) + { + var trimmed = mapping.Trim(); + var arrowIndex = trimmed.IndexOf("->", StringComparison.Ordinal); + if (arrowIndex < 0) + { + continue; + } + + var hostPart = trimmed[..arrowIndex].Trim(); + var colonIndex = hostPart.LastIndexOf(':'); + if (colonIndex < 0) + { + continue; + } + + if (int.TryParse(hostPart[(colonIndex + 1)..], out var port)) + { + return port; + } + } + + return null; + } + private int FindAvailablePort() { var usedPorts = new HashSet(_modelPorts.Values); @@ -393,10 +455,9 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable var readOutput = process.StandardOutput.ReadToEndAsync(cancellationToken); var readError = process.StandardError.ReadToEndAsync(cancellationToken); + var waitTask = process.WaitForExitAsync(cancellationToken); - var completed = await Task.WhenAny( - process.WaitForExitAsync(cancellationToken), - Task.Delay(DockerTimeout, cancellationToken)); + var completed = await Task.WhenAny(waitTask, Task.Delay(DockerTimeout, cancellationToken)); string output; string error; @@ -412,8 +473,7 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable error = "timed out"; } - if (completed is Task { IsCompleted: true } delayTask - && delayTask != process.WaitForExitAsync(cancellationToken)) + if (completed != waitTask) { _logger.LogWarning("Docker command timed out: docker {Args}", string.Join(" ", args)); try { process.Kill(entireProcessTree: true); } catch { } @@ -443,7 +503,7 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable return "ghcr.io/ggml-org/llama.cpp:server"; } - private static bool HasRocmDevices() => File.Exists("/dev/kfd") && File.Exists("/dev/dri"); + private static bool HasRocmDevices() => File.Exists("/dev/kfd") && Directory.Exists("/dev/dri"); private static bool HasNvidiaGpu() { diff --git a/src/Ngino.Client/OllamaToLlamaCppTranslator.cs b/src/Ngino.Client/OllamaToLlamaCppTranslator.cs new file mode 100644 index 0000000..0950e2e --- /dev/null +++ b/src/Ngino.Client/OllamaToLlamaCppTranslator.cs @@ -0,0 +1,605 @@ +using System.Text; +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Ngino.Protocol; + +namespace Ngino.Client; + +internal sealed class OllamaToLlamaCppTranslator +{ + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + + private readonly string _modelName; + private readonly ILogger _logger; + + public OllamaToLlamaCppTranslator(string modelName, ILogger logger) + { + _modelName = modelName; + _logger = logger; + } + + public bool TryTranslatePath(string method, string pathAndQuery, out string newPath) + { + var path = pathAndQuery.Split('?')[0]; + newPath = path switch + { + "/api/generate" => "/completion", + "/api/chat" => "/v1/chat/completions", + "/api/embed" or "/api/embeddings" => "/v1/embeddings", + _ => null! + }; + return newPath is not null; + } + + public byte[] TranslateBody(string pathAndQuery, byte[] body) + { + if (body is null || body.Length == 0) + return body; + + var path = pathAndQuery.Split('?')[0]; + + try + { + return path switch + { + "/api/generate" => TranslateGenerateBody(body), + "/api/chat" => TranslateChatBody(body), + "/api/embed" or "/api/embeddings" => TranslateEmbedBody(body), + _ => body + }; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to translate request body for {Path}", path); + return body; + } + } + + private byte[] TranslateGenerateBody(byte[] body) + { + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + var result = new Dictionary(); + + if (root.TryGetProperty("prompt", out var prompt)) + result["prompt"] = prompt.GetString() ?? ""; + + result["stream"] = true; + + CopyOptions(root, result); + + if (!result.ContainsKey("n_predict")) + result["n_predict"] = 2048; + + return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + } + + private byte[] TranslateChatBody(byte[] body) + { + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + var result = new Dictionary(); + + if (root.TryGetProperty("messages", out var messages)) + result["messages"] = messages.Deserialize(JsonOptions); + + result["stream"] = true; + + if (root.TryGetProperty("model", out var model)) + result["model"] = model.GetString(); + + CopyOptions(root, result, chat: true); + + if (!result.ContainsKey("max_tokens")) + result["max_tokens"] = 2048; + + return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + } + + public static bool ExtractOriginalStream(byte[] originalBody, string pathAndQuery) + { + var path = pathAndQuery.Split('?')[0]; + if (path is not "/api/generate" and not "/api/chat") + return true; + + try + { + using var doc = JsonDocument.Parse(originalBody); + var root = doc.RootElement; + if (root.TryGetProperty("stream", out var stream)) + return stream.ValueKind != JsonValueKind.False; + } + catch + { + } + + return true; + } + + private byte[] TranslateEmbedBody(byte[] body) + { + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + var result = new Dictionary + { + ["model"] = _modelName + }; + + if (root.TryGetProperty("input", out var input)) + result["input"] = input.Deserialize(JsonOptions); + + return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + } + + public Func CreateResponseHandler( + Func sendAsync, + string requestId, + string originalPath, + Func originalRequestedStream) => + async (response, ct) => + { + await TranslateAndSendResponse(response, sendAsync, requestId, originalPath, originalRequestedStream(), ct); + }; + + private async Task TranslateAndSendResponse( + HttpResponseMessage httpResponse, + Func sendAsync, + string requestId, + string originalPath, + bool originalRequestedStream, + CancellationToken cancellationToken) + { + var path = originalPath.Split('?')[0]; + + switch (path) + { + case "/api/tags": + await SynthesizeTagsResponse(sendAsync, requestId, cancellationToken); + return; + + case "/api/ps": + await SynthesizePsResponse(sendAsync, requestId, cancellationToken); + return; + } + + if (!httpResponse.IsSuccessStatusCode) + { + await ForwardRawResponse(httpResponse, sendAsync, requestId, cancellationToken); + return; + } + + if (path is "/api/generate" or "/api/chat") + { + if (originalRequestedStream) + { + await TranslateStreaming(path, httpResponse, sendAsync, requestId, cancellationToken); + } + else + { + await TranslateNonStreaming(path, httpResponse, sendAsync, requestId, cancellationToken); + } + } + else + { + await ForwardRawResponse(httpResponse, sendAsync, requestId, cancellationToken); + } + } + + private async Task ForwardRawResponse( + HttpResponseMessage httpResponse, + Func sendAsync, + string requestId, + CancellationToken cancellationToken) + { + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseHeaders, + RequestId = requestId, + StatusCode = (int)httpResponse.StatusCode, + ReasonPhrase = httpResponse.ReasonPhrase + }, cancellationToken); + + await using var stream = await httpResponse.Content.ReadAsStreamAsync(cancellationToken); + var buffer = new byte[64 * 1024]; + while (true) + { + var bytesRead = await stream.ReadAsync(buffer, cancellationToken); + if (bytesRead == 0) + break; + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = buffer.AsSpan(0, bytesRead).ToArray() + }, cancellationToken); + } + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = requestId + }, cancellationToken); + } + + private async Task SynthesizeTagsResponse( + Func sendAsync, + string requestId, + CancellationToken cancellationToken) + { + var modelList = new + { + models = new[] + { + new + { + name = _modelName, + model = _modelName, + modified_at = DateTime.UtcNow.ToString("o"), + size = 0L, + digest = "sha256:" + _modelName, + details = new + { + format = "gguf", + family = "llama", + parameter_size = "", + quantization_level = "" + } + } + } + }; + + var body = JsonSerializer.SerializeToUtf8Bytes(modelList, JsonOptions); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseHeaders, + RequestId = requestId, + StatusCode = 200, + ReasonPhrase = "OK" + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = body + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = requestId + }, cancellationToken); + } + + private async Task SynthesizePsResponse( + Func sendAsync, + string requestId, + CancellationToken cancellationToken) + { + var body = JsonSerializer.SerializeToUtf8Bytes(new { models = Array.Empty() }, JsonOptions); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseHeaders, + RequestId = requestId, + StatusCode = 200, + ReasonPhrase = "OK" + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = body + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = requestId + }, cancellationToken); + } + + private async Task TranslateNonStreaming( + string path, + HttpResponseMessage httpResponse, + Func sendAsync, + string requestId, + CancellationToken cancellationToken) + { + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseHeaders, + RequestId = requestId, + StatusCode = (int)httpResponse.StatusCode, + ReasonPhrase = httpResponse.ReasonPhrase + }, cancellationToken); + + var body = await httpResponse.Content.ReadAsByteArrayAsync(cancellationToken); + byte[] translatedBody; + + try + { + translatedBody = path switch + { + "/api/generate" => TranslateNonStreamingGenerate(body), + "/api/chat" => TranslateNonStreamingChat(body), + _ => body + }; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to translate non-streaming response"); + translatedBody = body; + } + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = translatedBody + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = requestId + }, cancellationToken); + } + + private async Task TranslateStreaming( + string path, + HttpResponseMessage httpResponse, + Func sendAsync, + string requestId, + CancellationToken cancellationToken) + { + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseHeaders, + RequestId = requestId, + StatusCode = 200, + ReasonPhrase = "OK" + }, cancellationToken); + + await using var stream = await httpResponse.Content.ReadAsStreamAsync(cancellationToken); + using var reader = new StreamReader(stream, Encoding.UTF8); + + while (!cancellationToken.IsCancellationRequested) + { + var line = await reader.ReadLineAsync(cancellationToken); + if (line is null) + break; + + if (!line.StartsWith("data: ", StringComparison.Ordinal)) + continue; + + var jsonStr = line[6..]; + if (jsonStr == "[DONE]") + continue; + + try + { + using var doc = JsonDocument.Parse(jsonStr); + var root = doc.RootElement; + + byte[]? chunk = path switch + { + "/api/generate" => TranslateGenerateStreamChunk(root), + "/api/chat" => TranslateChatStreamChunk(root), + _ => null + }; + + if (chunk is not null) + { + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = chunk + }, cancellationToken); + } + } + catch (JsonException) + { + } + } + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseBody, + RequestId = requestId, + Body = Encoding.UTF8.GetBytes( + $"{{\"model\":\"{EscapeJson(_modelName)}\",\"response\":\"\",\"done\":true}}\n") + }, cancellationToken); + + await sendAsync(new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = requestId + }, cancellationToken); + } + + private byte[] TranslateNonStreamingGenerate(byte[] body) + { + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + var result = new Dictionary + { + ["model"] = _modelName, + ["created_at"] = DateTime.UtcNow.ToString("o"), + ["response"] = root.TryGetProperty("content", out var content) ? content.GetString() : "", + ["done"] = root.TryGetProperty("stop", out var stop) && stop.GetBoolean() + }; + + CopyTimings(root, result); + return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + } + + private byte[] TranslateNonStreamingChat(byte[] body) + { + using var doc = JsonDocument.Parse(body); + var root = doc.RootElement; + + var message = root.GetProperty("choices")[0].GetProperty("message"); + var content = message.GetProperty("content").GetString() ?? ""; + + var result = new Dictionary + { + ["model"] = _modelName, + ["created_at"] = DateTime.UtcNow.ToString("o"), + ["message"] = new Dictionary + { + ["role"] = "assistant", + ["content"] = content + }, + ["done"] = true + }; + + if (root.TryGetProperty("usage", out var usage)) + { + if (usage.TryGetProperty("completion_tokens", out var comp)) + result["eval_count"] = comp.GetInt32(); + if (usage.TryGetProperty("prompt_tokens", out var prompt)) + result["prompt_eval_count"] = prompt.GetInt32(); + } + + CopyTimings(root, result); + return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + } + + private byte[]? TranslateGenerateStreamChunk(JsonElement root) + { + var done = root.TryGetProperty("stop", out var stop) && stop.GetBoolean(); + var text = root.TryGetProperty("content", out var content) ? content.GetString() ?? "" : ""; + + var result = new Dictionary + { + ["model"] = _modelName, + ["created_at"] = DateTime.UtcNow.ToString("o"), + ["response"] = text, + ["done"] = done + }; + + if (done) + CopyTimings(root, result); + + var bytes = JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + var withNewline = new byte[bytes.Length + 1]; + bytes.CopyTo(withNewline, 0); + withNewline[^1] = (byte)'\n'; + return withNewline; + } + + private byte[]? TranslateChatStreamChunk(JsonElement root) + { + if (!root.TryGetProperty("choices", out var choices) || choices.GetArrayLength() == 0) + return null; + + var choice = choices[0]; + var delta = choice.GetProperty("delta"); + var finishReason = choice.TryGetProperty("finish_reason", out var fr) ? fr.GetString() : null; + var done = finishReason is not null && finishReason != "null" && finishReason != ""; + + var content = delta.TryGetProperty("content", out var c) ? c.GetString() ?? "" : ""; + var role = delta.TryGetProperty("role", out var r) ? r.GetString() : null; + + var result = new Dictionary + { + ["model"] = _modelName, + ["created_at"] = DateTime.UtcNow.ToString("o"), + ["message"] = new Dictionary + { + ["role"] = role ?? "assistant", + ["content"] = content + }, + ["done"] = done + }; + + var bytes = JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions); + var withNewline = new byte[bytes.Length + 1]; + bytes.CopyTo(withNewline, 0); + withNewline[^1] = (byte)'\n'; + return withNewline; + } + + private static void CopyOptions(JsonElement root, Dictionary target, bool chat = false) + { + if (!root.TryGetProperty("options", out var options) || options.ValueKind != JsonValueKind.Object) + return; + + foreach (var opt in options.EnumerateObject()) + { + var key = (chat ? MapChatOptionName(opt.Name) : MapGenerateOptionName(opt.Name)) ?? opt.Name; + target[key] = ValueToObject(opt.Value); + } + } + + private static string? MapGenerateOptionName(string name) => name switch + { + "num_predict" => "n_predict", + "temperature" => "temperature", + "top_p" => "top_p", + "top_k" => "top_k", + "seed" => "seed", + "stop" => "stop", + "repeat_penalty" => "repeat_penalty", + "repeat_last_n" => "repeat_last_n", + "frequency_penalty" => "frequency_penalty", + "presence_penalty" => "presence_penalty", + "mirostat" => "mirostat", + "mirostat_tau" => "mirostat_tau", + "mirostat_eta" => "mirostat_eta", + "num_ctx" => "n_ctx", + "num_batch" => "n_batch", + _ => null + }; + + private static string? MapChatOptionName(string name) => name switch + { + "num_predict" => "max_tokens", + "temperature" => "temperature", + "top_p" => "top_p", + "seed" => "seed", + "stop" => "stop", + "frequency_penalty" => "frequency_penalty", + "presence_penalty" => "presence_penalty", + _ => null + }; + + private static void CopyTimings(JsonElement root, Dictionary target) + { + if (!root.TryGetProperty("timings", out var timings)) + return; + + if (timings.TryGetProperty("predicted_n", out var predN)) + target["eval_count"] = predN.GetInt32(); + if (timings.TryGetProperty("predicted_ms", out var predMs)) + target["eval_duration"] = (long)(predMs.GetDouble() * 1_000_000); + if (timings.TryGetProperty("prompt_n", out var promptN)) + target["prompt_eval_count"] = promptN.GetInt32(); + if (timings.TryGetProperty("prompt_ms", out var promptMs)) + target["prompt_eval_duration"] = (long)(promptMs.GetDouble() * 1_000_000); + } + + private static object? ValueToObject(JsonElement element) + { + return element.ValueKind switch + { + JsonValueKind.String => element.GetString(), + JsonValueKind.Number => element.TryGetInt64(out var l) ? (object)l : element.GetDouble(), + JsonValueKind.True => true, + JsonValueKind.False => false, + JsonValueKind.Null => null, + _ => element.Deserialize(JsonOptions) + }; + } + + private static string EscapeJson(string value) => + value.Replace("\\", "\\\\").Replace("\"", "\\\""); +} diff --git a/src/Ngino.Client/TunnelClient.cs b/src/Ngino.Client/TunnelClient.cs index f2dd5f3..528295e 100644 --- a/src/Ngino.Client/TunnelClient.cs +++ b/src/Ngino.Client/TunnelClient.cs @@ -16,6 +16,7 @@ internal sealed class TunnelClient private const string EmbeddingWarmupInput = "Ngino warmup"; private readonly ConcurrentDictionary _activeRequests = new(); + private readonly ConcurrentDictionary _pendingRequestBodies = new(); private readonly HttpClient _httpClient; private readonly ClientOptions _options; private readonly ILogger _logger; @@ -341,7 +342,8 @@ internal sealed class TunnelClient switch (message.Type) { case TunnelMessageTypes.HttpRequest: - StartRequest(socket, message, cancellationToken); + _pendingRequestBodies[message.RequestId] = new PendingRequestBody(); + _ = Task.Run(() => StartRequest(socket, message, cancellationToken), cancellationToken); break; case TunnelMessageTypes.HttpRequestBody: @@ -349,6 +351,10 @@ internal sealed class TunnelClient { requestWithBody.AddBody(message.Body ?? []); } + else if (_pendingRequestBodies.TryGetValue(message.RequestId, out var pendingBody)) + { + pendingBody.AddBody(message.Body ?? []); + } break; case TunnelMessageTypes.HttpRequestComplete: @@ -356,6 +362,10 @@ internal sealed class TunnelClient { completedRequest.CompleteBody(); } + else if (_pendingRequestBodies.TryGetValue(message.RequestId, out var pendingBody)) + { + pendingBody.Complete(); + } break; case TunnelMessageTypes.Cancel: @@ -363,6 +373,10 @@ internal sealed class TunnelClient { cancelledRequest.Cancel(); } + else + { + _pendingRequestBodies.TryRemove(message.RequestId, out _); + } break; case TunnelMessageTypes.ModelCommand: @@ -670,25 +684,78 @@ internal sealed class TunnelClient private static StringContent JsonContent(T value) => new(JsonSerializer.Serialize(value, JsonOptions), Encoding.UTF8, "application/json"); - private void StartRequest(ClientWebSocket socket, TunnelMessage message, CancellationToken cancellationToken) + private async Task StartRequest(ClientWebSocket socket, TunnelMessage message, CancellationToken cancellationToken) { Uri? effectiveUpstream = null; + var modelName = UpstreamRequest.ExtractModelName(message); if (_llamaCppManager is not null) { - var modelName = UpstreamRequest.ExtractModelName(message); if (modelName is not null) { effectiveUpstream = _llamaCppManager.GetUpstream(modelName); if (effectiveUpstream is null) { - _logger.LogWarning( - "Request for model '{Model}' but no llama.cpp container is running for it. Falling back to default upstream.", + _logger.LogInformation( + "Request for model '{Model}' but no llama.cpp container is running. Starting one on demand...", modelName); + + var model = _llamaCppManager.DiscoverModelsWithBlob() + .FirstOrDefault(m => string.Equals(m.OllamaName, modelName, StringComparison.OrdinalIgnoreCase)); + + if (model is not null) + { + var started = await _llamaCppManager.StartModelContainerAsync(model, cancellationToken); + if (started) + { + effectiveUpstream = _llamaCppManager.GetUpstream(modelName); + } + } + + if (effectiveUpstream is null) + { + _logger.LogWarning( + "Failed to start llama.cpp container for model '{Model}'. Falling back to default upstream.", + modelName); + } } } } + // When llama.cpp backend is active, set up request/response translation + Func? pathTransform = null; + Func? bodyTransform = null; + Func? responseHandler = null; + string? translatorModelName = null; + + if (_llamaCppManager is not null && effectiveUpstream is not null && modelName is not null) + { + var translator = new OllamaToLlamaCppTranslator(modelName, _logger); + translatorModelName = modelName; + bool originalRequestedStream = true; + + pathTransform = path => + { + if (translator.TryTranslatePath(message.Method ?? "GET", path, out var newPath)) + { + return newPath; + } + return null; + }; + + bodyTransform = body => + { + originalRequestedStream = OllamaToLlamaCppTranslator.ExtractOriginalStream(body, message.PathAndQuery ?? "/"); + return translator.TranslateBody(message.PathAndQuery ?? "/", body); + }; + + responseHandler = translator.CreateResponseHandler( + (response, token) => SendAsync(socket, response, token), + message.RequestId, + message.PathAndQuery ?? "/", + () => originalRequestedStream); + } + var request = new UpstreamRequest( _options, _httpClient, @@ -696,7 +763,10 @@ internal sealed class TunnelClient (response, token) => SendAsync(socket, response, token), requestId => _activeRequests.TryRemove(requestId, out _), cancellationToken, - effectiveUpstream: effectiveUpstream); + effectiveUpstream: effectiveUpstream, + responseHandler: responseHandler, + pathTransform: pathTransform, + bodyTransform: bodyTransform); if (!_activeRequests.TryAdd(message.RequestId, request)) { @@ -709,10 +779,16 @@ internal sealed class TunnelClient Error = "Duplicate request id." }, cancellationToken); + _pendingRequestBodies.TryRemove(message.RequestId, out _); return; } - _ = Task.Run(request.RunAsync, cancellationToken); + if (_pendingRequestBodies.TryRemove(message.RequestId, out var pendingBody)) + { + pendingBody.TransferTo(request); + } + + await request.RunAsync(); } private async Task SendAsync(ClientWebSocket socket, TunnelMessage message, CancellationToken cancellationToken) @@ -740,5 +816,51 @@ internal sealed class TunnelClient request.Cancel(); } } + + _pendingRequestBodies.Clear(); + } +} + +internal sealed class PendingRequestBody +{ + private readonly List _chunks = []; + private bool _completed; + private readonly object _lock = new(); + + public void AddBody(byte[] chunk) + { + if (chunk.Length == 0) + return; + + lock (_lock) + { + _chunks.Add(chunk); + } + } + + public void Complete() + { + lock (_lock) + { + _completed = true; + } + } + + public void TransferTo(UpstreamRequest request) + { + lock (_lock) + { + foreach (var chunk in _chunks) + { + request.AddBody(chunk); + } + + _chunks.Clear(); + + if (_completed) + { + request.CompleteBody(); + } + } } } diff --git a/src/Ngino.Client/UpstreamRequest.cs b/src/Ngino.Client/UpstreamRequest.cs index 0a94970..e3bd13d 100644 --- a/src/Ngino.Client/UpstreamRequest.cs +++ b/src/Ngino.Client/UpstreamRequest.cs @@ -36,6 +36,10 @@ internal sealed class UpstreamRequest private readonly Action _onComplete; private readonly Uri _upstream; private readonly Func _sendAsync; + private readonly Func? _responseHandler; + private readonly Func? _pathTransform; + private readonly Func? _bodyTransform; + private readonly List _bufferedBody = []; public UpstreamRequest( ClientOptions options, @@ -44,13 +48,19 @@ internal sealed class UpstreamRequest Func sendAsync, Action onComplete, CancellationToken cancellationToken, - Uri? effectiveUpstream = null) + Uri? effectiveUpstream = null, + Func? responseHandler = null, + Func? pathTransform = null, + Func? bodyTransform = null) { _httpClient = httpClient; _initialMessage = initialMessage; _sendAsync = sendAsync; _onComplete = onComplete; _upstream = effectiveUpstream ?? options.Upstream; + _responseHandler = responseHandler; + _pathTransform = pathTransform; + _bodyTransform = bodyTransform; _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); if (!initialMessage.HasBody) @@ -63,12 +73,36 @@ internal sealed class UpstreamRequest { if (body.Length > 0) { - _requestBody.Writer.TryWrite(body); + if (_bodyTransform is not null) + { + _bufferedBody.Add(body); + } + else + { + _requestBody.Writer.TryWrite(body); + } } } - public void CompleteBody() => + public void CompleteBody() + { + if (_bodyTransform is not null && _bufferedBody.Count > 0) + { + var totalLength = _bufferedBody.Sum(b => b.Length); + var concatenated = new byte[totalLength]; + var offset = 0; + foreach (var chunk in _bufferedBody) + { + chunk.CopyTo(concatenated, offset); + offset += chunk.Length; + } + + var transformed = _bodyTransform(concatenated); + _requestBody.Writer.TryWrite(transformed); + } + _requestBody.Writer.TryComplete(); + } public void Cancel() { @@ -86,16 +120,23 @@ internal sealed class UpstreamRequest HttpCompletionOption.ResponseHeadersRead, _cancellationTokenSource.Token); - await SendResponseHeadersAsync(response); - await SendResponseBodyAsync(response); + if (_responseHandler is not null) + { + await _responseHandler(response, _cancellationTokenSource.Token); + } + else + { + await SendResponseHeadersAsync(response); + await SendResponseBodyAsync(response); - await _sendAsync( - new TunnelMessage - { - Type = TunnelMessageTypes.HttpResponseComplete, - RequestId = _initialMessage.RequestId - }, - _cancellationTokenSource.Token); + await _sendAsync( + new TunnelMessage + { + Type = TunnelMessageTypes.HttpResponseComplete, + RequestId = _initialMessage.RequestId + }, + _cancellationTokenSource.Token); + } } catch (OperationCanceledException) when (_cancellationTokenSource.IsCancellationRequested) { @@ -133,8 +174,18 @@ internal sealed class UpstreamRequest private HttpRequestMessage BuildHttpRequest() { + var path = _initialMessage.PathAndQuery ?? "/"; + if (_pathTransform is not null) + { + var transformed = _pathTransform(path); + if (transformed is not null) + { + path = transformed; + } + } + var method = new HttpMethod(_initialMessage.Method ?? HttpMethod.Get.Method); - var request = new HttpRequestMessage(method, BuildUpstreamUri(_upstream, _initialMessage.PathAndQuery)); + var request = new HttpRequestMessage(method, BuildUpstreamUri(_upstream, path)); if (_initialMessage.HasBody) {