DBRaw.cs 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410
  1. using System;
  2. using System.Collections.Generic;
  3. using System.Collections.Concurrent;
  4. using System.Data;
  5. using System.Data.Common;
  6. using System.Linq;
  7. using System.Runtime.Versioning;
  8. using System.Text;
  9. using System.Threading.Tasks;
  10. using System.Runtime.InteropServices.ComTypes;
  11. using BeetleX.EventArgs;
  12. using Microsoft.Extensions.Caching.Memory;
  13. using Npgsql;
  14. namespace PlatformBenchmarks
  15. {
  16. public class RawDb
  17. {
  18. private readonly ConcurrentRandom _random;
  19. private readonly DbProviderFactory _dbProviderFactory;
  20. private readonly static MemoryCache _cache = new MemoryCache(
  21. new MemoryCacheOptions()
  22. {
  23. ExpirationScanFrequency = TimeSpan.FromMinutes(60)
  24. });
  25. private static readonly object[] _cacheKeys = Enumerable.Range(0, 10001).Select((i) => new CacheKey(i)).ToArray();
  26. public static string _connectionString = null;
  27. public RawDb(ConcurrentRandom random, DbProviderFactory dbProviderFactory)
  28. {
  29. _random = random;
  30. _dbProviderFactory = dbProviderFactory;
  31. OnCreateCommand();
  32. }
  33. private void OnCreateCommand()
  34. {
  35. SingleCommand = new Npgsql.NpgsqlCommand();
  36. SingleCommand.CommandText = "SELECT id, randomnumber FROM world WHERE id = @Id";
  37. mID = new Npgsql.NpgsqlParameter<int>("@Id", _random.Next(1, 10001));
  38. SingleCommand.Parameters.Add(mID);
  39. FortuneCommand = new Npgsql.NpgsqlCommand();
  40. FortuneCommand.CommandText = "SELECT id, message FROM fortune";
  41. }
  42. private DbCommand SingleCommand;
  43. private DbCommand FortuneCommand;
  44. private Npgsql.NpgsqlParameter<int> mID;
  45. public async Task<World> LoadSingleQueryRow()
  46. {
  47. using (var db = new NpgsqlConnection(_connectionString))
  48. {
  49. await db.OpenAsync();
  50. SingleCommand.Connection = db;
  51. mID.TypedValue = _random.Next(1, 10001);
  52. return await ReadSingleRow(db, SingleCommand);
  53. }
  54. }
  55. async Task<World> ReadSingleRow(DbConnection connection, DbCommand cmd)
  56. {
  57. using (var rdr = await cmd.ExecuteReaderAsync(CommandBehavior.SingleRow))
  58. {
  59. await rdr.ReadAsync();
  60. return new World
  61. {
  62. Id = rdr.GetInt32(0),
  63. RandomNumber = rdr.GetInt32(1)
  64. };
  65. }
  66. }
  67. public async Task<World[]> LoadMultipleQueriesRows(int count)
  68. {
  69. using (var db = new NpgsqlConnection(_connectionString))
  70. {
  71. await db.OpenAsync();
  72. return await LoadMultipleRows(count, db);
  73. }
  74. }
  75. public Task<World[]> LoadCachedQueries(int count)
  76. {
  77. var result = new World[count];
  78. var cacheKeys = _cacheKeys;
  79. var cache = _cache;
  80. var random = _random;
  81. for (var i = 0; i < result.Length; i++)
  82. {
  83. var id = random.Next(1, 10001);
  84. var key = cacheKeys[id];
  85. var data = cache.Get<CachedWorld>(key);
  86. if (data != null)
  87. {
  88. result[i] = data;
  89. }
  90. else
  91. {
  92. return LoadUncachedQueries(id, i, count, this, result);
  93. }
  94. }
  95. return Task.FromResult(result);
  96. static async Task<World[]> LoadUncachedQueries(int id, int i, int count, RawDb rawdb, World[] result)
  97. {
  98. using (var db = new NpgsqlConnection(_connectionString))
  99. {
  100. await db.OpenAsync();
  101. Func<ICacheEntry, Task<CachedWorld>> create = async (entry) =>
  102. {
  103. return await rawdb.ReadSingleRow(db, rawdb.SingleCommand);
  104. };
  105. var cacheKeys = _cacheKeys;
  106. var key = cacheKeys[id];
  107. rawdb.SingleCommand.Connection = db;
  108. rawdb.mID.TypedValue = id;
  109. for (; i < result.Length; i++)
  110. {
  111. var data = await _cache.GetOrCreateAsync<CachedWorld>(key, create);
  112. result[i] = data;
  113. id = rawdb._random.Next(1, 10001);
  114. rawdb.SingleCommand.Connection = db;
  115. rawdb.mID.TypedValue = id;
  116. key = cacheKeys[id];
  117. }
  118. }
  119. return result;
  120. }
  121. }
  122. private async Task<World[]> LoadMultipleRows(int count, DbConnection db)
  123. {
  124. SingleCommand.Connection = db;
  125. SingleCommand.Parameters[0].Value = _random.Next(1, 10001);
  126. var result = new World[count];
  127. for (int i = 0; i < result.Length; i++)
  128. {
  129. result[i] = await ReadSingleRow(db, SingleCommand);
  130. SingleCommand.Parameters[0].Value = _random.Next(1, 10001);
  131. }
  132. return result;
  133. }
  134. public async Task<List<Fortune>> LoadFortunesRows()
  135. {
  136. var result = new List<Fortune>();
  137. using (var db = new NpgsqlConnection(_connectionString))
  138. {
  139. await db.OpenAsync();
  140. FortuneCommand.Connection = db;
  141. using (var rdr = await FortuneCommand.ExecuteReaderAsync(CommandBehavior.Default))
  142. {
  143. while (await rdr.ReadAsync())
  144. {
  145. result.Add(new Fortune
  146. {
  147. Id = rdr.GetInt32(0),
  148. Message = rdr.GetString(1)
  149. });
  150. }
  151. }
  152. }
  153. result.Add(new Fortune { Message = "Additional fortune added at request time." });
  154. result.Sort();
  155. return result;
  156. }
  157. public async Task<World[]> LoadMultipleUpdatesRows(int count)
  158. {
  159. using (var db = new NpgsqlConnection(_connectionString))
  160. {
  161. await db.OpenAsync();
  162. var updateCmd = UpdateCommandsCached.PopCommand(count);
  163. try
  164. {
  165. var command = updateCmd.Command;
  166. command.Connection = db;
  167. SingleCommand.Connection = db;
  168. mID.TypedValue = _random.Next(1, int.MaxValue) % 10000 + 1;
  169. var results = new World[count];
  170. for (int i = 0; i < count; i++)
  171. {
  172. results[i] = await ReadSingleRow(db, SingleCommand);
  173. mID.TypedValue = _random.Next(1, int.MaxValue) % 10000 + 1;
  174. }
  175. for (int i = 0; i < count; i++)
  176. {
  177. var randomNumber = _random.Next(1, int.MaxValue) % 10000 + 1;
  178. updateCmd.Parameters[i * 2].TypedValue = results[i].Id;
  179. updateCmd.Parameters[i * 2 + 1].TypedValue = randomNumber;
  180. //updateCmd.Parameters[i * 2].Value = results[i].Id;
  181. //updateCmd.Parameters[i * 2 + 1].Value = randomNumber;
  182. results[i].RandomNumber = randomNumber;
  183. }
  184. await command.ExecuteNonQueryAsync();
  185. return results;
  186. }
  187. catch (Exception e_)
  188. {
  189. throw e_;
  190. }
  191. finally
  192. {
  193. UpdateCommandsCached.PushCommand(count, updateCmd);
  194. }
  195. }
  196. }
  197. }
  198. public sealed class CacheKey : IEquatable<CacheKey>
  199. {
  200. private readonly int _value;
  201. public CacheKey(int value)
  202. => _value = value;
  203. public bool Equals(CacheKey key)
  204. => key._value == _value;
  205. public override bool Equals(object obj)
  206. => ReferenceEquals(obj, this);
  207. public override int GetHashCode()
  208. => _value;
  209. public override string ToString()
  210. => _value.ToString();
  211. }
  212. internal class UpdateCommandsCached
  213. {
  214. private static System.Collections.Concurrent.ConcurrentStack<CommandCacheItem>[] mCacheTable
  215. = new System.Collections.Concurrent.ConcurrentStack<CommandCacheItem>[1024];
  216. public static string[] IDParamereNames = new string[1024];
  217. public static string[] RandomParamereNames = new string[1024];
  218. static UpdateCommandsCached()
  219. {
  220. for (int i = 0; i < 1024; i++)
  221. {
  222. IDParamereNames[i] = $"@Id_{i}";
  223. RandomParamereNames[i] = $"@Random_{i}";
  224. mCacheTable[i] = new System.Collections.Concurrent.ConcurrentStack<CommandCacheItem>();
  225. }
  226. }
  227. private static CommandCacheItem CreatCommand(int count)
  228. {
  229. CommandCacheItem item = new CommandCacheItem();
  230. NpgsqlCommand cmd = new Npgsql.NpgsqlCommand();
  231. cmd.CommandText = BatchUpdateString.Query(count);
  232. for (int i = 0; i < count; i++)
  233. {
  234. var id = new NpgsqlParameter<int>();
  235. id.ParameterName = IDParamereNames[i];
  236. cmd.Parameters.Add(id);
  237. item.Parameters.Add(id);
  238. var random = new NpgsqlParameter<int>();
  239. random.ParameterName = RandomParamereNames[i];
  240. cmd.Parameters.Add(random);
  241. item.Parameters.Add(random);
  242. }
  243. item.Command = cmd;
  244. return item;
  245. }
  246. public static void PushCommand(int count, CommandCacheItem cmd)
  247. {
  248. mCacheTable[count].Push(cmd);
  249. }
  250. public static CommandCacheItem PopCommand(int count)
  251. {
  252. if (mCacheTable[count].TryPop(out CommandCacheItem cmd))
  253. return cmd;
  254. return CreatCommand(count);
  255. }
  256. private static bool mInited = false;
  257. public static void Init()
  258. {
  259. if (mInited)
  260. return;
  261. lock (typeof(UpdateCommandsCached))
  262. {
  263. if (mInited)
  264. return;
  265. for (int i = 1; i <= 500; i++)
  266. {
  267. for (int k = 0; k < 10; k++)
  268. {
  269. var cmd = CreatCommand(i);
  270. mCacheTable[i].Push(cmd);
  271. }
  272. }
  273. mInited = true;
  274. HttpServer.ApiServer.Log(LogType.Info, null, $"Init update commands cached");
  275. return;
  276. }
  277. }
  278. }
  279. class CommandCacheItem
  280. {
  281. public Npgsql.NpgsqlCommand Command { get; set; }
  282. public List<Npgsql.NpgsqlParameter<int>> Parameters { get; private set; } = new List<Npgsql.NpgsqlParameter<int>>(1024);
  283. }
  284. internal class BatchUpdateString
  285. {
  286. private const int MaxBatch = 500;
  287. private static string[] _queries = new string[MaxBatch + 1];
  288. public static string Query(int batchSize)
  289. {
  290. if (_queries[batchSize] != null)
  291. {
  292. return _queries[batchSize];
  293. }
  294. var lastIndex = batchSize - 1;
  295. var sb = StringBuilderCache.Acquire();
  296. sb.Append("UPDATE world SET randomNumber = temp.randomNumber FROM (VALUES ");
  297. Enumerable.Range(0, lastIndex).ToList().ForEach(i => sb.Append($"(@Id_{i}, @Random_{i}), "));
  298. sb.Append($"(@Id_{lastIndex}, @Random_{lastIndex}) ORDER BY 1) AS temp(id, randomNumber) WHERE temp.id = world.id");
  299. return _queries[batchSize] = StringBuilderCache.GetStringAndRelease(sb);
  300. }
  301. }
  302. internal static class StringBuilderCache
  303. {
  304. private const int DefaultCapacity = 1386;
  305. private const int MaxBuilderSize = DefaultCapacity * 3;
  306. [ThreadStatic]
  307. private static StringBuilder t_cachedInstance;
  308. /// <summary>Get a StringBuilder for the specified capacity.</summary>
  309. /// <remarks>If a StringBuilder of an appropriate size is cached, it will be returned and the cache emptied.</remarks>
  310. public static StringBuilder Acquire(int capacity = DefaultCapacity)
  311. {
  312. if (capacity <= MaxBuilderSize)
  313. {
  314. StringBuilder sb = t_cachedInstance;
  315. if (capacity < DefaultCapacity)
  316. {
  317. capacity = DefaultCapacity;
  318. }
  319. if (sb != null)
  320. {
  321. // Avoid stringbuilder block fragmentation by getting a new StringBuilder
  322. // when the requested size is larger than the current capacity
  323. if (capacity <= sb.Capacity)
  324. {
  325. t_cachedInstance = null;
  326. sb.Clear();
  327. return sb;
  328. }
  329. }
  330. }
  331. return new StringBuilder(capacity);
  332. }
  333. public static void Release(StringBuilder sb)
  334. {
  335. if (sb.Capacity <= MaxBuilderSize)
  336. {
  337. t_cachedInstance = sb;
  338. }
  339. }
  340. public static string GetStringAndRelease(StringBuilder sb)
  341. {
  342. string result = sb.ToString();
  343. Release(sb);
  344. return result;
  345. }
  346. }
  347. }