DBRaw.cs 14 KB

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