fix(client): fixes installer and inferencing issues
Build & Deploy / build (push) Successful in 1m45s

This commit is contained in:
2026-07-30 19:33:44 +02:00
parent 2ca5ca2da2
commit 5b7ae55cc8
6 changed files with 881 additions and 31 deletions
+9
View File
@@ -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 <dir>` | Path to Ollama models directory (`manifests/blobs`); required with the flag above |
| `--llama-cpp-docker-image <img>` | Docker image; defaults to auto-detected (rocm/cuda/cpu) |
| `--llama-cpp-base-port <num>` | 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
+8 -5
View File
@@ -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" <<EOF
[Unit]
Description=Ngino Tunnel Client
+66 -6
View File
@@ -109,6 +109,17 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable
var port = FindAvailablePort();
var containerName = SanitizeContainerName($"ngino-llamacpp-{ollamaName}");
var existingPort = await FindExistingContainerPortAsync(containerName);
if (existingPort.HasValue)
{
_logger.LogInformation(
"Reusing existing container for {Model} on port {Port}", ollamaName, existingPort.Value);
_modelPorts[ollamaName] = existingPort.Value;
return true;
}
await RunDockerAsync(["rm", "-f", containerName], CancellationToken.None);
var args = BuildDockerRunArgs(containerName, model, port);
_logger.LogInformation(
"Starting llama.cpp container for {Model} on port {Port}: docker {Args}",
@@ -334,6 +345,7 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable
}
args.Add(_dockerImage);
args.Add("--embeddings");
args.Add("-m");
args.Add($"/models/blobs/{blobFile}");
args.Add("-ngl");
@@ -348,6 +360,56 @@ internal sealed partial class LlamaCppManager : IAsyncDisposable
return [.. args];
}
private async Task<int?> 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<int>(_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()
{
@@ -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<string, object?>();
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<string, object?>();
if (root.TryGetProperty("messages", out var messages))
result["messages"] = messages.Deserialize<object>(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<string, object?>
{
["model"] = _modelName
};
if (root.TryGetProperty("input", out var input))
result["input"] = input.Deserialize<object>(JsonOptions);
return JsonSerializer.SerializeToUtf8Bytes(result, JsonOptions);
}
public Func<HttpResponseMessage, CancellationToken, Task> CreateResponseHandler(
Func<TunnelMessage, CancellationToken, Task> sendAsync,
string requestId,
string originalPath,
Func<bool> originalRequestedStream) =>
async (response, ct) =>
{
await TranslateAndSendResponse(response, sendAsync, requestId, originalPath, originalRequestedStream(), ct);
};
private async Task TranslateAndSendResponse(
HttpResponseMessage httpResponse,
Func<TunnelMessage, CancellationToken, Task> 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<TunnelMessage, CancellationToken, Task> 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<TunnelMessage, CancellationToken, Task> 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<TunnelMessage, CancellationToken, Task> sendAsync,
string requestId,
CancellationToken cancellationToken)
{
var body = JsonSerializer.SerializeToUtf8Bytes(new { models = Array.Empty<object>() }, 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<TunnelMessage, CancellationToken, Task> 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<TunnelMessage, CancellationToken, Task> 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<string, object?>
{
["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<string, object?>
{
["model"] = _modelName,
["created_at"] = DateTime.UtcNow.ToString("o"),
["message"] = new Dictionary<string, object?>
{
["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<string, object?>
{
["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<string, object?>
{
["model"] = _modelName,
["created_at"] = DateTime.UtcNow.ToString("o"),
["message"] = new Dictionary<string, object?>
{
["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<string, object?> 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<string, object?> 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<object>(JsonOptions)
};
}
private static string EscapeJson(string value) =>
value.Replace("\\", "\\\\").Replace("\"", "\\\"");
}
+129 -7
View File
@@ -16,6 +16,7 @@ internal sealed class TunnelClient
private const string EmbeddingWarmupInput = "Ngino warmup";
private readonly ConcurrentDictionary<string, UpstreamRequest> _activeRequests = new();
private readonly ConcurrentDictionary<string, PendingRequestBody> _pendingRequestBodies = new();
private readonly HttpClient _httpClient;
private readonly ClientOptions _options;
private readonly ILogger<TunnelClient> _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>(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<string, string?>? pathTransform = null;
Func<byte[], byte[]>? bodyTransform = null;
Func<HttpResponseMessage, CancellationToken, Task>? 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<byte[]> _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();
}
}
}
}
+64 -13
View File
@@ -36,6 +36,10 @@ internal sealed class UpstreamRequest
private readonly Action<string> _onComplete;
private readonly Uri _upstream;
private readonly Func<TunnelMessage, CancellationToken, Task> _sendAsync;
private readonly Func<HttpResponseMessage, CancellationToken, Task>? _responseHandler;
private readonly Func<string, string?>? _pathTransform;
private readonly Func<byte[], byte[]>? _bodyTransform;
private readonly List<byte[]> _bufferedBody = [];
public UpstreamRequest(
ClientOptions options,
@@ -44,13 +48,19 @@ internal sealed class UpstreamRequest
Func<TunnelMessage, CancellationToken, Task> sendAsync,
Action<string> onComplete,
CancellationToken cancellationToken,
Uri? effectiveUpstream = null)
Uri? effectiveUpstream = null,
Func<HttpResponseMessage, CancellationToken, Task>? responseHandler = null,
Func<string, string?>? pathTransform = null,
Func<byte[], byte[]>? 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)
{