Skip to content

Commit

Permalink
fix tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@s3w3nofficial
s3w3nofficial committed Jan 22, 2025
1 parent 4b55f59 commit 9239518
Show file tree
Hide file tree
Showing 49 changed files with 531 additions and 554 deletions.
2 changes: 1 addition & 1 deletion hosting/Windows/Garnet.worker/Worker.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ public class Worker : BackgroundService
private bool _isDisposed = false;
private readonly string[] args;

private GarnetServer server;
private GarnetApplication server;

public Worker(string[] args)
{
Expand Down
92 changes: 32 additions & 60 deletions libs/host/GarnetApplication.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,12 @@
// Licensed under the MIT license.

using System;
using System.Diagnostics;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Garnet.common;
using Garnet.server;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

namespace Garnet.host;
Expand All @@ -20,77 +17,58 @@ namespace Garnet.host;
/// </summary>
public class GarnetApplication : IHost
{
/// <summary>
/// Metrics API
/// </summary>
public MetricsApi Metrics;

/// <summary>
/// Command registration API
/// </summary>
public RegisterApi Register;

/// <summary>
/// Store API
/// </summary>
public StoreApi Store;

internal GarnetProvider Provider;

private readonly IHost host;

public GarnetApplication(IHost host)
{
this.host = host;

Metrics = host.Services.GetRequiredService<MetricsApi>();
Register = host.Services.GetService<RegisterApi>();
Store = host.Services.GetService<StoreApi>();

Provider = host.Services.GetRequiredService<GarnetProvider>();
}

public IServiceProvider Services => host.Services;

public Task StartAsync(CancellationToken cancellationToken = default)
{
var opts = host.Services.GetRequiredService<IOptions<GarnetServerOptions>>();
var logger = host.Services.GetRequiredService<ILogger<GarnetServer>>();

Debug.Assert(opts != null);

var version = GetVersion();

if (!opts.Value.QuietMode)
{
var red = "\u001b[31m";
var magenta = "\u001b[35m";
var normal = "\u001b[0m";

Console.WriteLine($@"{red} _________
/_||___||_\ {normal}Garnet {version} {(IntPtr.Size == 8 ? "64" : "32")} bit; {(opts.Value.EnableCluster ? "cluster" : "standalone")} mode{red}
'. \ / .' {normal}Port: {opts.Value.Port}{red}
'.\ /.' {magenta}https://aka.ms/GetGarnet{red}
'.'
{normal}");
}

Trace.Listeners.Add(new ConsoleTraceListener());

// Set up an initial memory logger to log messages from configuration parser into memory.
using var memLogProvider = new MemoryLoggerProvider();

var initLogger = (MemoryLogger)memLogProvider.CreateLogger("ArgParser");

logger?.LogInformation("Garnet {version} {bits} bit; {clusterMode} mode; Port: {port}", GetVersion(), IntPtr.Size == 8 ? "64" : "32", opts.Value.EnableCluster ? "cluster" : "standalone", opts.Value.Port);

// Flush initialization logs from memory logger
initLogger.FlushLogger(logger);

var setMax = opts.Value.ThreadPoolMaxThreads <= 0 || ThreadPool.SetMaxThreads(opts.Value.ThreadPoolMaxThreads, opts.Value.ThreadPoolMaxThreads);

if (opts.Value.ThreadPoolMinThreads > 0 && !ThreadPool.SetMinThreads(opts.Value.ThreadPoolMinThreads, opts.Value.ThreadPoolMinThreads))
throw new Exception($"Unable to call ThreadPool.SetMinThreads with {opts.Value.ThreadPoolMinThreads}");

// Retry to set max threads if it wasn't set in the previous step
if (!setMax && !ThreadPool.SetMaxThreads(opts.Value.ThreadPoolMaxThreads, opts.Value.ThreadPoolMaxThreads))
throw new Exception($"Unable to call ThreadPool.SetMaxThreads with {opts.Value.ThreadPoolMaxThreads}");

logger?.LogTrace("TLS is {tlsEnabled}", opts.Value.TlsOptions == null ? "disabled" : "enabled");

return host.StartAsync(cancellationToken);
}
=> host.StartAsync(cancellationToken);

public Task StopAsync(CancellationToken cancellationToken = default)
=> host.StopAsync(cancellationToken);

public void Dispose() => host.Dispose();

public void Dispose()
{
host.Dispose();
}

public void Run()
{
HostingAbstractionsHostExtensions.Run(this);
}

public async Task RunAsync(CancellationToken cancellationToken = default)
public Task RunAsync(CancellationToken cancellationToken = default)
{
await HostingAbstractionsHostExtensions.RunAsync(this, cancellationToken);
HostingAbstractionsHostExtensions.RunAsync(this, cancellationToken);
return Task.CompletedTask;
}

public static GarnetApplicationBuilder CreateHostBuilder(string[] args)
Expand Down Expand Up @@ -125,10 +103,4 @@ public static GarnetApplicationBuilder CreateHostBuilder(string[] args, GarnetSe
{
return new (new GarnetApplicationOptions {Args = args}, options);
}

private static string GetVersion()
{
var Version = Assembly.GetExecutingAssembly().GetName().Version;
return $"{Version.Major}.{Version.Minor}.{Version.Build}";
}
}
6 changes: 3 additions & 3 deletions libs/host/GarnetApplicationBuilder.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@ internal GarnetApplicationBuilder(GarnetApplicationOptions options, GarnetServer
hostApplicationBuilder.Services.AddSingleton(garnetServerOptionsWrapped);

hostApplicationBuilder.Services.AddSingleton<IGarnetServer, GarnetServerTcp>();
hostApplicationBuilder.Services.AddHostedService<GarnetServer>();

hostApplicationBuilder.Services.AddSingleton<StoreFactory>();
hostApplicationBuilder.Services.AddSingleton<CustomCommandManager>();

Expand All @@ -73,7 +75,7 @@ internal GarnetApplicationBuilder(GarnetApplicationOptions options, GarnetServer

var store = storeFactory.CreateMainStore(out var checkpointDir);
var objectStore = storeFactory.CreateObjectStore(checkpointDir, out var objectStoreSizeTracker);

TsavoriteLog appendOnlyFile = null;

if (opts.Value.EnableAOF)
Expand Down Expand Up @@ -134,8 +136,6 @@ internal GarnetApplicationBuilder(GarnetApplicationOptions options, GarnetServer
hostApplicationBuilder.Services.AddSingleton<MetricsApi>();
hostApplicationBuilder.Services.AddSingleton<RegisterApi>();
hostApplicationBuilder.Services.AddSingleton<StoreApi>();

hostApplicationBuilder.Services.AddHostedService<GarnetServer>();
}

public GarnetApplication Build()
Expand Down
169 changes: 63 additions & 106 deletions libs/host/GarnetServer.cs
View file
Original file line number Diff line number Diff line change
@@ -1,145 +1,102 @@
// Copyright (c) Microsoft Corporation.
// Copyright (c) Microsoft Corporation.
// Licensed under the MIT license.

using System;
using System.Diagnostics;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Garnet.common;
using Garnet.networking;
using Garnet.server;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Tsavorite.core;

namespace Garnet;
namespace Garnet.host;

/// <summary>
/// Implementation Garnet server
/// </summary>
public class GarnetServer : IHostedService, IDisposable
public class GarnetServer : IHostedService
{
internal GarnetProvider Provider;

private readonly GarnetServerOptions opts;
private IGarnetServer server;
private SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> subscribeBroker;
private readonly IGarnetServer garnetServerTcp;
private readonly GarnetProvider garnetProvider;
private readonly StoreWrapper store;
private readonly IOptions<GarnetServerOptions> opts;
private readonly ILogger<GarnetServer> logger;

/// <summary>
/// Store and associated information used by this Garnet server
/// </summary>
protected StoreWrapper storeWrapper;

/// <summary>
/// Metrics API
/// </summary>
public MetricsApi Metrics;

/// <summary>
/// Command registration API
/// </summary>
public RegisterApi Register;

/// <summary>
/// Store API
/// </summary>
public StoreApi Store;


public GarnetServer(
IOptions<GarnetServerOptions> opts,
ILogger<GarnetServer> logger,
IGarnetServer garnetServerTcp,
StoreWrapper storeWrapper)
IGarnetServer garnetServerTcp,
GarnetProvider garnetProvider,
StoreWrapper store,
IOptions<GarnetServerOptions> options,
ILogger<GarnetServer> logger)
{
this.garnetServerTcp = garnetServerTcp;
this.garnetProvider = garnetProvider;
this.store = store;
this.opts = options;
this.logger = logger;
this.server = garnetServerTcp;
this.storeWrapper = storeWrapper;

Trace.Listeners.Add(new ConsoleTraceListener());

// Assign values to GarnetServerOptions
this.opts = opts.Value;
this.opts.AuthSettings = this.opts.AuthSettings;
garnetServerTcp.Register(WireFormat.ASCII, garnetProvider);

this.InitializeServer();
}

private void InitializeServer()
{
// Create session provider for Garnet
Provider = new GarnetProvider(storeWrapper, subscribeBroker);
public Task StartAsync(CancellationToken cancellationToken)
{
var version = GetVersion();

if (!opts.Value.QuietMode)
{
var red = "\u001b[31m";
var magenta = "\u001b[35m";
var normal = "\u001b[0m";

Console.WriteLine($@"{red} _________
/_||___||_\ {normal}Garnet {version} {(IntPtr.Size == 8 ? "64" : "32")} bit; {(opts.Value.EnableCluster ? "cluster" : "standalone")} mode{red}
'. \ / .' {normal}Port: {opts.Value.Port}{red}
'.\ /.' {magenta}https://aka.ms/GetGarnet{red}
'.'
{normal}");
}

Trace.Listeners.Add(new ConsoleTraceListener());

// Create user facing API endpoints
Metrics = new MetricsApi(Provider);
Register = new RegisterApi(Provider);
Store = new StoreApi(storeWrapper);
// Set up an initial memory logger to log messages from configuration parser into memory.
using var memLogProvider = new MemoryLoggerProvider();

var initLogger = (MemoryLogger)memLogProvider.CreateLogger("ArgParser");

logger?.LogInformation("Garnet {version} {bits} bit; {clusterMode} mode; Port: {port}", GetVersion(), IntPtr.Size == 8 ? "64" : "32", opts.Value.EnableCluster ? "cluster" : "standalone", opts.Value.Port);

server.Register(WireFormat.ASCII, Provider);
}
// Flush initialization logs from memory logger
initLogger.FlushLogger(logger);

/// <summary>
/// Start server instance
/// </summary>
public void Start()
{
Provider.Recover();
server.Start();
Provider.Start();
if (!opts.QuietMode)
this.logger.LogInformation("* Ready to accept connections");
}
var setMax = opts.Value.ThreadPoolMaxThreads <= 0 || ThreadPool.SetMaxThreads(opts.Value.ThreadPoolMaxThreads, opts.Value.ThreadPoolMaxThreads);

/// <summary>
/// Dispose store (including log and checkpoint directory)
/// </summary>
public void Dispose()
{
Dispose(false);
}
if (opts.Value.ThreadPoolMinThreads > 0 && !ThreadPool.SetMinThreads(opts.Value.ThreadPoolMinThreads, opts.Value.ThreadPoolMinThreads))
throw new Exception($"Unable to call ThreadPool.SetMinThreads with {opts.Value.ThreadPoolMinThreads}");

/// <summary>
/// Dispose, optionally deleting logs and checkpoints
/// </summary>
/// <param name="deleteDir">Whether to delete logs and checkpoints</param>
public void Dispose(bool deleteDir = true)
{
InternalDispose();
if (deleteDir)
{
if (opts.CheckpointDir != opts.LogDir && !string.IsNullOrEmpty(opts.CheckpointDir))
{
var ckptdir = opts.DeviceFactoryCreator();
ckptdir.Initialize(opts.CheckpointDir);
ckptdir.Delete(new FileDescriptor { directoryName = "" });
}
}
}
// Retry to set max threads if it wasn't set in the previous step
if (!setMax && !ThreadPool.SetMaxThreads(opts.Value.ThreadPoolMaxThreads, opts.Value.ThreadPoolMaxThreads))
throw new Exception($"Unable to call ThreadPool.SetMaxThreads with {opts.Value.ThreadPoolMaxThreads}");

private void InternalDispose()
{
// Provider.Dispose will get stuck
//Provider?.Dispose();
logger?.LogTrace("TLS is {tlsEnabled}", opts.Value.TlsOptions == null ? "disabled" : "enabled");

garnetProvider.Recover();
garnetServerTcp.Start();
garnetProvider.Start();

if (!opts.Value.QuietMode)
logger?.LogInformation("* Ready to accept connections");

server.Dispose();
subscribeBroker?.Dispose();
opts.AuthSettings?.Dispose();
}

public Task StartAsync(CancellationToken cancellationToken)
{
Start();

return Task.CompletedTask;
}

public Task StopAsync(CancellationToken cancellationToken)
{
Dispose();
garnetProvider?.Dispose();
garnetServerTcp?.Dispose();
//store?.Dispose();

return Task.CompletedTask;
return Task.CompletedTask;
}

private static string GetVersion()
Expand Down
2 changes: 1 addition & 1 deletion libs/server/StoreWrapper.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ namespace Garnet.server
/// <summary>
/// Wrapper for store and store-specific information
/// </summary>
public sealed class StoreWrapper
public sealed class StoreWrapper
{
internal readonly string version;
internal readonly string redisProtocolVersion;
Expand Down
Loading

0 comments on commit 9239518

Please sign in to comment.