owner/RSE
1
1
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Create runtime variables and remove MessagePack.

Browse Source
This commit is contained in:
owner
2024-11-26 19:40:27 +01:00
parent 28c5918a83
commit be07cdb13e
28 changed files with 336 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Backend/Handler/Communication.cs
+59 -14
View File
@@ -1,7 +1,6 @@
using System.Diagnostics.CodeAnalysis;
using System.Text.Json;
using Backend.Helper;
using MessagePack;
using Models.Handler;
using Models.Model.Backend;
using Models.Model.External;
@@ -14,12 +13,18 @@ public class Communication
{
private readonly DbHandler _dbHandler;
private readonly ThreadHandler _threadHandler;
private readonly IpScanner _ipScanner;
private readonly ContentFilter _contentFilter;
private bool _isRunning = true;
private string _basePath;
public Communication(DbHandler dbHandler, ThreadHandler threadHandler)
public Communication(DbHandler dbHandler, ThreadHandler threadHandler, IpScanner ipScanner, ContentFilter contentFilter, string basePath)
{
_dbHandler = dbHandler;
_threadHandler = threadHandler;
_ipScanner = ipScanner;
_contentFilter = contentFilter;
_basePath = basePath;
}
public WaitHandle[] Start()
@@ -43,15 +48,20 @@ public class Communication
while (_isRunning)
{
byte[] message = rep.ReceiveFrameBytes();
CommunicationObject? communicationObject = JsonSerializer.Deserialize<CommunicationObject>(message);
CommunicationObject communicationObject = MessagePackSerializer.Deserialize<CommunicationObject>(message);
if (communicationObject is null)
{
continue;
}
OnServerOnReceiveReady(communicationObject, rep);
}
((EventWaitHandle) obj).Set();
}
[RequiresDynamicCode("Calls System.Text.Json.JsonSerializer.Serialize<TValue>(TValue, JsonSerializerOptions)")]
[RequiresUnreferencedCode("Calls System.Text.Json.JsonSerializer.Serialize<TValue>(TValue, JsonSerializerOptions)")]
private void OnServerOnReceiveReady(CommunicationObject communicationObject, ResponseSocket rep)
@@ -60,8 +70,8 @@ public class Communication
{
case CommunicationCommand.GetScanningProgress:
{
DatabaseSizes databaseSizes = FilesystemHelper.GetDatabaseSizes();
DatabaseSizes databaseSizes = FilesystemHelper.GetDatabaseSizes(_basePath);
long discardedIndexes = _dbHandler.GetDiscardedIndexes();
ScanningStatus status = new();
@@ -77,13 +87,13 @@ public class Communication
}
status.AmountOfIpv4Left = 4294967296 - discardedIndexes;
status.TotalFiltered = DbHandler.GetFilteredIndexes();
status.TotalFiltered = _dbHandler.GetFilteredIndexes();
status.TotalDiscarded = discardedIndexes;
status.MyDbSize = databaseSizes.MyDbSize;
status.FilteredDbSize = databaseSizes.FilteredDbSize;
status.DiscardedDbSize = databaseSizes.DiscardedDbSize;
byte[] serializedResult = MessagePackSerializer.Serialize(status, MessagePackSerializerOptions.Standard.WithCompression(MessagePackCompression.Lz4BlockArray));
byte[] serializedResult = JsonSerializer.SerializeToUtf8Bytes(status);
rep.SendFrame(serializedResult);
@@ -93,7 +103,7 @@ public class Communication
case CommunicationCommand.DbReindex:
{
_dbHandler.ReIndex();
SendStringResponse(rep, "All Dbs have been reindexed.");
break;
}
@@ -101,7 +111,7 @@ public class Communication
case CommunicationCommand.DbVacuum:
{
_dbHandler.Vacuum();
SendStringResponse(rep, "All Dbs have been vacuumed.");
break;
}
@@ -121,13 +131,48 @@ public class Communication
_isRunning = false;
break;
}
case CommunicationCommand.ChangeRuntimeVariable:
{
Console.WriteLine("lmao");
if (string.IsNullOrWhiteSpace(communicationObject.VariableValue))
{
break;
}
if (communicationObject.Variable == RuntimeVariable.DbContent.ToString())
{
int value = int.Parse(communicationObject.VariableValue);
_dbHandler.SetContentWaitTime(value);
}
if (communicationObject.Variable == RuntimeVariable.DbDiscarded.ToString())
{
int value = int.Parse(communicationObject.VariableValue);
_dbHandler.SetDiscardedWaitTime(value);
}
if (communicationObject.Variable == RuntimeVariable.ScannerTimeout.ToString())
{
int value = int.Parse(communicationObject.VariableValue);
_ipScanner.SetTimeout(value);
}
if (communicationObject.Variable == RuntimeVariable.ContentFilter.ToString())
{
int value = int.Parse(communicationObject.VariableValue);
_contentFilter.SetTimeout(value);
}
break;
}
}
}
private static void SendStringResponse(ResponseSocket rep, string response)
{
MessagePackSerializerOptions withCompression = MessagePackSerializerOptions.Standard.WithCompression(MessagePackCompression.Lz4BlockArray);
byte[] serializedResult = MessagePackSerializer.Serialize(response, withCompression);
byte[] serializedResult = JsonSerializer.SerializeToUtf8Bytes(response);
rep.SendFrame(serializedResult);
}
Backend/Handler/ContentFilter.cs
+55 -39
View File
@@ -11,14 +11,25 @@ public class ContentFilter
{
private readonly ConcurrentQueue<QueueItem> _queue;
private readonly DbHandler _dbHandler;
private const string GetDomainPort80 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/Scripts/GetDomainNamePort80.sh";
private const string GetDomainPort443 = "/home/skingging/Documents/Projects/CSharp/RSE/Backend/Scripts/GetDomainNamePort443.sh";
private readonly string _getDomainPort80;
private readonly string _getDomainPort443;
private bool _stop;
private int _timeOut;
private string _basePath;
public ContentFilter(ConcurrentQueue<QueueItem> queue, DbHandler dbHandler)
public ContentFilter(ConcurrentQueue<QueueItem> queue, DbHandler dbHandler, string basePath)
{
_queue = queue;
_dbHandler = dbHandler;
_basePath = basePath;
_getDomainPort80 = $"{basePath}/Backend/Scripts/GetDomainNamePort80.sh";
_getDomainPort443 = $"{basePath}/Backend/Scripts/GetDomainNamePort443.sh";
}
public void SetTimeout(int timeOut)
{
_timeOut = timeOut;
}
public WaitHandle[] Start()
@@ -34,49 +45,54 @@ public class ContentFilter
private void Filter(object obj)
{
long indexes = DbHandler.GetUnfilteredIndexes();
for (long i = 0; i < indexes; i++)
while (!_stop)
{
if (_stop) break;
long indexes = _dbHandler.GetUnfilteredIndexes();
Unfiltered? unfiltered = DbHandler.ReadUnfilteredWithId(i);
if (unfiltered is null || unfiltered.Filtered == 1) continue;
unfiltered.Filtered = 1;
QueueItem superUnfilteredObject = new()
for (long i = 0; i < indexes; i++)
{
Unfiltered = unfiltered,
Operations = Operations.Update
};
_queue.Enqueue(superUnfilteredObject);
if (_dbHandler.GetFilteredIp(unfiltered.Ip))
{
continue;
if (_stop) break;
Unfiltered? unfiltered = _dbHandler.ReadUnfilteredWithId(i);
if (unfiltered is null || unfiltered.Filtered == 1) continue;
unfiltered.Filtered = 1;
QueueItem superUnfilteredObject = new()
{
Unfiltered = unfiltered,
Operations = Operations.Update
};
_queue.Enqueue(superUnfilteredObject);
if (_dbHandler.GetFilteredIp(unfiltered.Ip))
{
continue;
}
Filtered filtered = GetSiteData(unfiltered.Ip);
filtered.Port1 = unfiltered.Port1;
filtered.Port2 = unfiltered.Port2;
QueueItem superFilteredObject = new()
{
Filtered = filtered,
Operations = Operations.Insert
};
_queue.Enqueue(superFilteredObject);
}
Filtered filtered = GetSiteData(unfiltered.Ip);
filtered.Port1 = unfiltered.Port1;
filtered.Port2 = unfiltered.Port2;
QueueItem superFilteredObject = new()
{
Filtered = filtered,
Operations = Operations.Insert
};
_queue.Enqueue(superFilteredObject);
Thread.Sleep(_timeOut);
}
((EventWaitHandle) obj).Set();
}
private static Filtered GetSiteData(string ip)
private Filtered GetSiteData(string ip)
{
StartProcess(ip, 80);
StartProcess(ip, 443);
@@ -112,7 +128,7 @@ public class ContentFilter
for (int i = 0; i < ports.Length; i++)
{
using StreamReader streamReader = new($"{ports[i]}Header.txt");
using StreamReader streamReader = new($"{_basePath}/Backend/Scripts/{ports[i]}Header.txt");
while (streamReader.Peek() != -1)
{
@@ -204,9 +220,9 @@ public class ContentFilter
return siteData;
}
private static void StartProcess(string ip, int port)
private void StartProcess(string ip, int port)
{
string fileName = port == 80 ? GetDomainPort80 : GetDomainPort443;
string fileName = port == 80 ? _getDomainPort80 : _getDomainPort443;
Process proc = new();
proc.StartInfo = new()
Backend/Handler/IpScanner.cs
+10 -1
View File
@@ -22,12 +22,21 @@ public class IpScanner
private readonly ConcurrentQueue<Discarded> _discardedQueue;
private readonly DbHandler _dbHandler;
private bool _stop;
private int _timeout;
public IpScanner(ConcurrentQueue<QueueItem> queue, DbHandler dbHandler, ConcurrentQueue<Discarded> discardedQueue)
{
_queue = queue;
_dbHandler = dbHandler;
_discardedQueue = discardedQueue;
SetTimeout(128);
}
public void SetTimeout(int milliseconds)
{
Console.WriteLine($"Setting timeout to {milliseconds}ms");
_timeout = milliseconds;
}
public WaitHandle[] Start(int threads)
@@ -133,7 +142,7 @@ public class IpScanner
_ = IPAddress.TryParse(ip, out IPAddress? address);
if (address is not null)
{
responseCode = ping.Send(address, 512, buf, null).Status;
responseCode = ping.Send(address, _timeout, buf, null).Status;
}
}
catch (Exception e)
Backend/Handler/ThreadHandler.cs
+11 -25
View File
@@ -14,23 +14,22 @@ public class ThreadHandler
private bool _communicationStopped;
private bool _ipScannerStopped;
private bool _contentFilterStopped;
private bool _stopSignal;
public ThreadHandler()
public ThreadHandler(string path)
{
ConcurrentQueue<QueueItem> contentQueue = new();
ConcurrentQueue<Discarded> discardedQueue = new();
_dbHandler = new(contentQueue, discardedQueue);
_communication = new(_dbHandler, this);
_dbHandler = new(contentQueue, discardedQueue, path);
_ipScanner = new(contentQueue, _dbHandler, discardedQueue);
_contentFilter = new(contentQueue, _dbHandler);
_contentFilter = new(contentQueue, _dbHandler, path);
_communication = new(_dbHandler, this, _ipScanner, _contentFilter, path);
}
public void Start()
{
Thread scanner = new(StartScanner);
Thread indexer = new(StartIndexer);
Thread indexer = new(StartContentFilter);
Thread database = new(StartDbHandler);
Thread discarded = new(StartDiscardedDbHandler);
Thread communication = new(StartCommunicationHandler);
@@ -48,18 +47,11 @@ public class ThreadHandler
communication.Join();
}
public static void ManualGc()
{
GC.Collect();
GC.WaitForPendingFinalizers();
GC.Collect();
}
private void StartScanner()
{
Thread.Sleep(10000); // Let the database handler instantiate and warm up first.
WaitHandle[] wait = _ipScanner.Start(4);
WaitHandle[] wait = _ipScanner.Start(2);
WaitHandle.WaitAll(wait);
@@ -68,18 +60,13 @@ public class ThreadHandler
_ipScannerStopped = true;
}
private void StartIndexer()
private void StartContentFilter()
{
while (!_stopSignal)
{
WaitHandle[] wait = _contentFilter.Start();
WaitHandle.WaitAll(wait);
Thread.Sleep(300000); // 5 minutes
}
WaitHandle[] wait = _contentFilter.Start();
Console.WriteLine("Indexer finished");
WaitHandle.WaitAll(wait);
Console.WriteLine("Content filter finished");
_contentFilterStopped = true;
}
@@ -113,7 +100,6 @@ public class ThreadHandler
private void Stop()
{
_stopSignal = true;
_ipScanner.Stop();
_contentFilter.Stop();