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

Made a lot of changes. Enhanced memory usage.

Browse Source
This commit is contained in:
owner
2025-02-16 09:38:00 +01:00
parent 8d659d4261
commit c70b42c9d7
169 changed files with 3628 additions and 1304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Models/Handler/DbHandler.cs
+73 -63
View File
@@ -13,8 +13,10 @@ public class DbHandler
private readonly ConcurrentQueue<UnfilteredQueueItem> _unfilteredQueue;
private readonly ConcurrentQueue<Discarded> _discardedQueue;
private readonly ConcurrentQueue<ScannerResumeObject> _resumeQueue;
private readonly ConcurrentQueue<FilterQueueItem> _preFilteredQueue;
private readonly string _unfilteredConnectionString;
private readonly string _preFilteredConnectionString;
private readonly string _filteredConnectionString;
private readonly string _resumeConnectionString;
private readonly string _compressedConnectionString;
@@ -25,6 +27,11 @@ public class DbHandler
" INSERT INTO Unfiltered (Ip1, Ip2, Ip3, Ip4, Port1, Port2, Filtered)" +
" VALUES (@ip1, @ip2, @ip3, @ip4, @port1, @port2, @filtered)";
private const string InsertPreFilteredStatement = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off;" +
" INSERT INTO PreFiltered (Ip1, Ip2, Ip3, Ip4, ResponseCode, Filtered)" +
" VALUES (@ip1, @ip2, @ip3, @ip4, @responseCode, @filtered)";
private const string InsertIntoFiltered = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY;" +
" PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = on;" +
" INSERT INTO Filtered (Ip1, Ip2, Ip3, Ip4, Port1, Port2," +
@@ -88,6 +95,9 @@ public class DbHandler
private const string ReadDiscardedSeqIdsStatement = "SELECT seq FROM sqlite_sequence;";
private const string ReadResumeStatement = "SELECT * FROM Resume WHERE ThreadNumber == @threadNumber;";
private const string ReadCompressedDbRowsStatement = "SELECT Rows FROM CompressedDatabases;";
private const string ReadPreFilteredIdsStatement = "SELECT Id FROM PreFiltered WHERE Filtered == 0;";
private const string ReadPreFilteredStatement = "SELECT Ip1, Ip2, Ip3, Ip4, ResponseCode, Id FROM PreFiltered WHERE Filtered == 0 ORDER BY Ip1 ASC LIMIT 1;";
private const string UpdatePreFilteredStatement = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; UPDATE PreFiltered SET Filtered = 1 WHERE Id == @id;";
private const string UpdateUnfilteredStatement = "PRAGMA synchronous = OFF; PRAGMA temp_store = MEMORY; PRAGMA journal_mode = MEMORY; PRAGMA foreign_keys = off; UPDATE Unfiltered SET Filtered = 1 WHERE Id == @id;";
@@ -113,12 +123,15 @@ public class DbHandler
public DbHandler(ConcurrentQueue<Filtered> filteredQueue,
ConcurrentQueue<Discarded> discardedQueue,
ConcurrentQueue<UnfilteredQueueItem> unfilteredQueue,
ConcurrentQueue<ScannerResumeObject> resumeQueue, string basePath)
ConcurrentQueue<ScannerResumeObject> resumeQueue,
ConcurrentQueue<FilterQueueItem> preFilteredQueue,
string basePath)
{
_filteredQueue = filteredQueue;
_discardedQueue = discardedQueue;
_unfilteredQueue = unfilteredQueue;
_resumeQueue = resumeQueue;
_preFilteredQueue = preFilteredQueue;
SetContentWaitTime(100);
SetDiscardedWaitTime(10);
@@ -129,6 +142,7 @@ public class DbHandler
_filteredConnectionString = $"Data Source={basePath}/Models/Filtered.db";
_resumeConnectionString = $"Data Source={basePath}/Models/ScannerResume.db";
_compressedConnectionString = $"Data Source={basePath}/Models/CompressedDatabases.db";
_preFilteredConnectionString = $"Data Source={basePath}/Models/PreFiltered.db";
}
public void SetContentWaitTime(int waitTime)
@@ -190,6 +204,24 @@ public class DbHandler
Console.WriteLine("Filtered DbHandler stopped.");
}
public void PrefilteredDbHandler()
{
Console.WriteLine("PreFiltered Db handler started.");
while (!_stop)
{
if (_preFilteredQueue.IsEmpty)
{
Thread.Sleep(4);
continue;
}
_preFilteredQueue.TryDequeue(out FilterQueueItem queueItem);
InsertPrefiltered(queueItem);
}
}
public void ResumeDbHandler()
{
@@ -204,12 +236,9 @@ public class DbHandler
continue;
}
_resumeQueue.TryDequeue(out ScannerResumeObject? queueItem);
_resumeQueue.TryDequeue(out ScannerResumeObject queueItem);
if (queueItem is not null)
{
InsertResumeObject(queueItem);
}
InsertResumeObject(queueItem);
}
Console.WriteLine("Resume DbHandler stopped.");
@@ -499,6 +528,24 @@ public class DbHandler
_ = command.ExecuteNonQuery();
connection.Close();
}
private void InsertPrefiltered(FilterQueueItem filterQueueItem)
{
using SqliteConnection connection = new(_preFilteredConnectionString);
connection.Open();
using SqliteCommand command = new(InsertPreFilteredStatement, connection);
command.Parameters.AddWithValue("@ip1", filterQueueItem.Ip.Ip1);
command.Parameters.AddWithValue("@ip2", filterQueueItem.Ip.Ip2);
command.Parameters.AddWithValue("@ip3", filterQueueItem.Ip.Ip3);
command.Parameters.AddWithValue("@ip4", filterQueueItem.Ip.Ip4);
command.Parameters.AddWithValue("@responseCode", filterQueueItem.ResponseCode);
command.Parameters.AddWithValue("@filtered", 0);
_ = command.ExecuteNonQuery();
connection.Close();
}
private void UpdateUnfiltered(Unfiltered unfiltered)
{
@@ -567,79 +614,42 @@ public class DbHandler
return ids;
}
public long GetFilteredIndexes()
public bool GetPreFilterQueueItem(out FilterQueueItem filterQueueItem)
{
long rowId = 0;
using SqliteConnection connection = new(_filteredConnectionString);
using SqliteConnection connection = new(_preFilteredConnectionString);
connection.Open();
using SqliteCommand command = new(ReadFilteredIdsStatement, connection);
SqliteCommand command = new(ReadPreFilteredStatement, connection);
using SqliteDataReader reader = command.ExecuteReader();
filterQueueItem = new();
Ip ip = new();
long id = 0;
if (!reader.HasRows)
{
return 0;
return false;
}
while (reader.Read())
{
rowId = reader.GetInt64(0);
ip.Ip1 = reader.GetInt32(0);
ip.Ip2 = reader.GetInt32(1);
ip.Ip3 = reader.GetInt32(2);
ip.Ip4 = reader.GetInt32(3);
filterQueueItem.ResponseCode = reader.GetInt32(4);
id = reader.GetInt64(5);
}
return rowId;
}
public long GetDiscardedIndexes()
{
long rowId = 0;
filterQueueItem.Ip = ip;
command = new(UpdatePreFilteredStatement, connection);
command.Parameters.AddWithValue("@id", id);
SqliteConnection connection;
SqliteCommand command;
SqliteDataReader reader;
for (int i = 0; i < _discardedConnectionStrings.Count; i++)
{
connection = new(_discardedConnectionStrings[i]);
connection.Open();
command = new(ReadDiscardedSeqIdsStatement, connection);
reader = command.ExecuteReader();
if (!reader.HasRows)
{
return rowId;
}
while (reader.Read())
{
rowId += reader.GetInt64(0);
}
connection.Close();
}
connection = new(_compressedConnectionString);
connection.Open();
command = new(ReadCompressedDbRowsStatement, connection);
reader = command.ExecuteReader();
if (!reader.HasRows)
{
return rowId;
}
while (reader.Read())
{
rowId += reader.GetInt64(0);
}
connection.Close();
connection.Dispose();
command.ExecuteNonQuery();
command.Dispose();
reader.Dispose();
return rowId;
return true;
}
private static long GetDiscardedIndexesForSpecificDb(string connectionString)