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

Reworked the queue items.

Browse Source
This commit is contained in:
owner
2024-11-29 12:59:07 +01:00
parent 3034e66126
commit f2ace6f571
8 changed files with 152 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Models/Handler/DbHandler.cs
+95 -32
View File
@@ -1,4 +1,3 @@
using System.Collections;
using System.Collections.Concurrent;
using Microsoft.Data.Sqlite;
using Models.Model.Backend;
@@ -8,8 +7,10 @@ namespace Models.Handler;
public class DbHandler
{
private readonly ConcurrentQueue<QueueItem> _contentQueue;
private readonly ConcurrentQueue<Filtered> _filteredQueue;
private readonly ConcurrentQueue<UnfilteredQueueItem> _unfilteredQueue;
private readonly ConcurrentQueue<Discarded> _discardedQueue;
private readonly ConcurrentQueue<ScannerResumeObject> _resumeQueue;
private readonly string _unfilteredConnectionString;
private readonly string _discardedConnectionString;
@@ -17,10 +18,34 @@ public class DbHandler
private readonly string _resumeConnectionString;
private readonly List<string> _discardedConnectionStrings = [];
private const string InsertStatement = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; INSERT INTO Unfiltered (Ip1, Ip2, Ip3, Ip4, Port1, Port2, Filtered) VALUES (@ip1, @ip2, @ip3, @ip4, @port1, @port2, @filtered)";
private const string InsertIntoFiltered = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; INSERT INTO Filtered (Ip1, Ip2, Ip3, Ip4, Port1, Port2, Title1, Title2, Description1, Description2, Url1, Url2, ServerType1, ServerType2, RobotsTXT1, RobotsTXT2, HttpVersion1, HttpVersion2, CertificateIssuerCountry, CertificateOrganizationName, IpV6, TlsVersion, CipherSuite, KeyExchangeAlgorithm, PublicKeyType1, PublicKeyType2, PublicKeyType3, AcceptEncoding1, AcceptEncoding2, ALPN, Connection1, Connection2) VALUES (@ip1, @ip2, @ip3, @ip4, @port1, @port2, @title1, @title2, @description1, @description2, @url1, @url2, @serverType1, @serverType2, @robotsTXT1, @robotsTXT2, @httpVersion1, @httpVersion2, @certificateIssuerCountry, @certificateOrganizationName, @ipV6, @tlsVersion, @cipherSuite, @keyExchangeAlgorithm, @publicKeyType1, @publicKeyType2, @publicKeyType3, @acceptEncoding1, @acceptEncoding2, @aLPN, @connection1, @connection2)";
private const string InsertIntoDiscarded = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; INSERT INTO Discarded (Ip1, Ip2, Ip3, Ip4, ResponseCode) VALUES (@ip1, @ip2, @ip3, @ip4, @responseCode)";
private const string InsertIntoResume = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; INSERT INTO Resume (ThreadNumber, StartRange, EndRange, FirstByte, SecondByte, ThirdByte, FourthByte) VALUES (@threadNumber, @startRange, @endRange, @firstByte, @secondByte, @thirdByte, @fourthByte);";
private const string InsertStatement = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off;" +
" INSERT INTO Unfiltered (Ip1, Ip2, Ip3, Ip4, Port1, Port2, Filtered)" +
" VALUES (@ip1, @ip2, @ip3, @ip4, @port1, @port2, @filtered)";
private const string InsertIntoFiltered = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off;" +
" INSERT INTO Filtered (Ip1, Ip2, Ip3, Ip4, Port1, Port2, Title1, Title2," +
" Description1, Description2, Url1, Url2, ServerType1, ServerType2," +
" RobotsTXT1, RobotsTXT2, HttpVersion1, HttpVersion2, CertificateIssuerCountry," +
" CertificateOrganizationName, IpV6, TlsVersion, CipherSuite, KeyExchangeAlgorithm," +
" PublicKeyType1, PublicKeyType2, PublicKeyType3, AcceptEncoding1, AcceptEncoding2," +
" ALPN, Connection1, Connection2) VALUES (@ip1, @ip2, @ip3, @ip4, @port1, @port2," +
" @title1, @title2, @description1, @description2, @url1, @url2, @serverType1," +
" @serverType2, @robotsTXT1, @robotsTXT2, @httpVersion1, @httpVersion2," +
" @certificateIssuerCountry, @certificateOrganizationName, @ipV6, @tlsVersion," +
" @cipherSuite, @keyExchangeAlgorithm, @publicKeyType1, @publicKeyType2," +
" @publicKeyType3, @acceptEncoding1, @acceptEncoding2, @aLPN, @connection1, @connection2)";
private const string InsertIntoDiscarded = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off;" +
" INSERT INTO Discarded (Ip1, Ip2, Ip3, Ip4, ResponseCode)" +
" VALUES (@ip1, @ip2, @ip3, @ip4, @responseCode)";
private const string InsertIntoResume = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off;" +
" INSERT INTO Resume (ThreadNumber, StartRange, EndRange, FirstByte, SecondByte, ThirdByte, FourthByte)" +
" VALUES (@threadNumber, @startRange, @endRange, @firstByte, @secondByte, @thirdByte, @fourthByte);";
private const string ReadUnfilteredStatement = "SELECT * FROM Unfiltered WHERE Id = @id;";
private const string ReadUnfilteredIdsStatement = "SELECT Id FROM Unfiltered WHERE Id != 0 ORDER BY Id DESC LIMIT 1;";
@@ -48,12 +73,17 @@ public class DbHandler
private readonly string _basePath;
public DbHandler(ConcurrentQueue<QueueItem> contentQueue, ConcurrentQueue<Discarded> discardedQueue, string basePath)
public DbHandler(ConcurrentQueue<Filtered> filteredQueue,
ConcurrentQueue<Discarded> discardedQueue,
ConcurrentQueue<UnfilteredQueueItem> unfilteredQueue,
ConcurrentQueue<ScannerResumeObject> resumeQueue, string basePath)
{
_contentQueue = contentQueue;
_filteredQueue = filteredQueue;
_discardedQueue = discardedQueue;
SetContentWaitTime(10);
_unfilteredQueue = unfilteredQueue;
_resumeQueue = resumeQueue;
SetContentWaitTime(100);
SetDiscardedWaitTime(10);
_basePath = basePath;
@@ -74,45 +104,78 @@ public class DbHandler
_discardedWaitTime = waitTime;
}
public void StartContent()
public void UnfilteredDbHandler()
{
Console.WriteLine("Content DbHandler started");
Console.WriteLine("Unfiltered DbHandler started");
while (!_stop)
{
if (_contentQueue.IsEmpty || _pause)
if (_unfilteredQueue.IsEmpty || _pause)
{
Thread.Sleep(_contentWaitTime);
_paused = true;
continue;
}
_contentQueue.TryDequeue(out QueueItem? queueItem);
_unfilteredQueue.TryDequeue(out UnfilteredQueueItem queueItem);
if (queueItem is null) { continue; }
if (queueItem.Operations == Operations.Insert && queueItem.DbType == DbType.Unfiltered)
if (queueItem.Operations == Operations.Insert)
{
InsertUnfiltered(queueItem.Unfiltered);
}
else if (queueItem.Operations == Operations.Insert && queueItem.DbType == DbType.Filtered)
{
InsertFiltered(queueItem.Filtered!);
}
else if (queueItem.Operations == Operations.Insert && queueItem.ResumeObject is not null)
{
InsertResumeObject(queueItem.ResumeObject);
}
else if (queueItem.Operations == Operations.Update && queueItem.DbType == DbType.Unfiltered)
else if (queueItem.Operations == Operations.Update)
{
UpdateUnfiltered(queueItem.Unfiltered);
}
}
Console.WriteLine("Content DbHandler stopped.");
Console.WriteLine("Unfiltered DbHandler stopped.");
}
public void FilteredDbHandler()
{
Console.WriteLine("Filtered DB handler started");
while (!_stop)
{
if (_filteredQueue.IsEmpty || _pause)
{
Thread.Sleep(_contentWaitTime);
_paused = true;
continue;
}
_filteredQueue.TryDequeue(out Filtered? queueItem);
InsertFiltered(queueItem!);
}
Console.WriteLine("Filtered DbHandler stopped.");
}
public void ResumeDbHandler()
{
Console.WriteLine("Resume DB handler started");
while (!_stop)
{
if (_resumeQueue.IsEmpty || _pause)
{
Thread.Sleep(_contentWaitTime);
_paused = true;
continue;
}
_resumeQueue.TryDequeue(out ScannerResumeObject? queueItem);
if (queueItem is not null)
{
InsertResumeObject(queueItem);
}
}
Console.WriteLine("Resume DbHandler stopped.");
}
public WaitHandle[] Start(int threads)
@@ -131,7 +194,7 @@ public class DbHandler
waitHandles[i] = handle;
Thread f = new (RunDiscarded!);
Thread f = new (DiscardedDbHandler!);
f.Start(discardedDbHandlerSetting);
Thread.Sleep(1000);
@@ -140,7 +203,7 @@ public class DbHandler
return waitHandles;
}
private void RunDiscarded(object obj)
private void DiscardedDbHandler(object obj)
{
DiscardedDbHandlerSetting discardedDbHandlerSetting = (DiscardedDbHandlerSetting)obj;
Console.WriteLine($"Discarded DbHandler started with thread: ({discardedDbHandlerSetting.ThreadId})");
@@ -163,7 +226,7 @@ public class DbHandler
discardedDbHandlerSetting.Handle!.Set();
Console.WriteLine("Content DbHandler stopped.");
Console.WriteLine("Discarded DbHandler stopped.");
}
private void InsertUnfiltered(Unfiltered unfiltered)
Models/Model/Backend/DbType.cs
-7
View File
@@ -1,7 +0,0 @@
namespace Models.Model.Backend;
public enum DbType
{
Unfiltered,
Filtered,
}
Models/Model/Backend/Operations.cs
-1
View File
@@ -4,5 +4,4 @@ public enum Operations
{
Insert,
Update,
Optimize,
}
Models/Model/Backend/QueueItem.cs
-10
View File
@@ -1,10 +0,0 @@
namespace Models.Model.Backend;
public class QueueItem
{
public Unfiltered Unfiltered { get; init; }
public Filtered? Filtered { get; init; }
public ScannerResumeObject? ResumeObject { get; init; }
public Operations Operations { get; init; }
public DbType DbType { get; init; }
}
Models/Model/Backend/UnfilteredQueueItem.cs
+7
View File
@@ -0,0 +1,7 @@
namespace Models.Model.Backend;
public struct UnfilteredQueueItem
{
public Unfiltered Unfiltered { get; init; }
public Operations Operations { get; init; }
}