Use BlockingCollection

pull/9643/head
Bond_009 1 year ago
parent 858dadcdd1
commit 33f97045f9

@ -122,18 +122,17 @@ namespace Emby.Server.Implementations.Data
{
WriteConnections = new ConnectionPool(WriteConnectionsCount, CreateWriteConnection);
ReadConnections = new ConnectionPool(ReadConnectionsCount, CreateReadConnection);
}
protected ManagedConnection GetConnection(bool readOnly = false)
{
if (readOnly)
// Configuration and pragmas can affect VACUUM so it needs to be last.
using (var connection = GetConnection(true))
{
return ReadConnections.GetConnection();
connection.Execute("VACUUM");
}
return WriteConnections.GetConnection();
}
protected ManagedConnection GetConnection(bool readOnly = false)
=> readOnly ? ReadConnections.GetConnection() : WriteConnections.GetConnection();
protected SQLiteDatabaseConnection CreateWriteConnection()
{
var writeConnection = SQLite3.Open(
@ -173,52 +172,44 @@ namespace Emby.Server.Implementations.Data
writeConnection.Execute("PRAGMA temp_store=" + (int)TempStore);
// Configuration and pragmas can affect VACUUM so it needs to be last.
writeConnection.Execute("VACUUM");
return writeConnection;
}
protected SQLiteDatabaseConnection CreateReadConnection()
{
var writeConnection = SQLite3.Open(
var connection = SQLite3.Open(
DbFilePath,
DefaultConnectionFlags | ConnectionFlags.ReadOnly,
null);
if (CacheSize.HasValue)
{
writeConnection.Execute("PRAGMA cache_size=" + CacheSize.Value);
connection.Execute("PRAGMA cache_size=" + CacheSize.Value);
}
if (!string.IsNullOrWhiteSpace(LockingMode))
{
writeConnection.Execute("PRAGMA locking_mode=" + LockingMode);
connection.Execute("PRAGMA locking_mode=" + LockingMode);
}
if (!string.IsNullOrWhiteSpace(JournalMode))
{
writeConnection.Execute("PRAGMA journal_mode=" + JournalMode);
connection.Execute("PRAGMA journal_mode=" + JournalMode);
}
if (JournalSizeLimit.HasValue)
{
writeConnection.Execute("PRAGMA journal_size_limit=" + JournalSizeLimit.Value);
connection.Execute("PRAGMA journal_size_limit=" + JournalSizeLimit.Value);
}
if (Synchronous.HasValue)
{
writeConnection.Execute("PRAGMA synchronous=" + (int)Synchronous.Value);
connection.Execute("PRAGMA synchronous=" + (int)Synchronous.Value);
}
if (PageSize.HasValue)
{
writeConnection.Execute("PRAGMA page_size=" + PageSize.Value);
}
connection.Execute("PRAGMA temp_store=" + (int)TempStore);
writeConnection.Execute("PRAGMA temp_store=" + (int)TempStore);
return writeConnection;
return connection;
}
public IStatement PrepareStatement(ManagedConnection connection, string sql)

@ -2,44 +2,47 @@
using System;
using System.Collections.Concurrent;
using System.Threading;
using SQLitePCL.pretty;
namespace Emby.Server.Implementations.Data;
public sealed class ConnectionPool : IDisposable
{
private readonly int _count;
private readonly SemaphoreSlim _lock;
private readonly ConcurrentQueue<SQLiteDatabaseConnection> _connections = new ConcurrentQueue<SQLiteDatabaseConnection>();
private readonly BlockingCollection<SQLiteDatabaseConnection> _connections = new();
private bool _disposed;
public ConnectionPool(int count, Func<SQLiteDatabaseConnection> factory)
{
_count = count;
_lock = new SemaphoreSlim(count, count);
for (int i = 0; i < count; i++)
{
_connections.Enqueue(factory.Invoke());
_connections.Add(factory.Invoke());
}
}
public ManagedConnection GetConnection()
{
_lock.Wait();
if (!_connections.TryDequeue(out var connection))
if (_disposed)
{
_lock.Release();
throw new InvalidOperationException();
ThrowObjectDisposedException();
}
return new ManagedConnection(connection, this);
return new ManagedConnection(_connections.Take(), this);
void ThrowObjectDisposedException()
{
throw new ObjectDisposedException(GetType().Name);
}
}
public void Return(SQLiteDatabaseConnection connection)
{
_connections.Enqueue(connection);
_lock.Release();
if (_disposed)
{
connection.Dispose();
return;
}
_connections.Add(connection);
}
public void Dispose()
@ -49,20 +52,11 @@ public sealed class ConnectionPool : IDisposable
return;
}
for (int i = 0; i < _count; i++)
foreach (var connection in _connections)
{
_lock.Wait();
if (!_connections.TryDequeue(out var connection))
{
_lock.Release();
throw new InvalidOperationException();
}
connection.Dispose();
}
_lock.Dispose();
_disposed = true;
}
}

Loading…
Cancel
Save