|
3 | 3 | // Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. |
4 | 4 |
|
5 | 5 | using System.Diagnostics; |
| 6 | +using PerformanceTest; |
6 | 7 | using RabbitMQ.AMQP.Client; |
7 | 8 | using RabbitMQ.AMQP.Client.Impl; |
8 | 9 | using Trace = Amqp.Trace; |
9 | 10 | using TraceLevel = Amqp.TraceLevel; |
10 | 11 |
|
| 12 | +// ---- Configuration ---- |
| 13 | +const int total = 5_000_000; |
| 14 | +const int tasksSize = 200; |
| 15 | +bool enableConsumer = true; |
| 16 | +// ----------------------- |
| 17 | + |
| 18 | + |
11 | 19 | Trace.TraceLevel = TraceLevel.Verbose; |
12 | 20 |
|
13 | 21 | ConsoleTraceListener consoleListener = new(); |
14 | 22 | Trace.TraceListener = (l, f, a) => |
15 | 23 | consoleListener.WriteLine($"[{DateTime.Now}] [{l}] - {f}"); |
16 | 24 |
|
17 | | -Trace.WriteLine(TraceLevel.Information, "Starting"); |
| 25 | +Trace.WriteLine(TraceLevel.Information, "Starting performance test..."); |
18 | 26 | const string containerId = "performance-test-connection"; |
19 | 27 |
|
20 | 28 | IEnvironment environment = await AmqpEnvironment |
|
30 | 38 | IQueueSpecification queueSpec = management.Queue(queueName).Type(QueueType.QUORUM); |
31 | 39 | await queueSpec.DeleteAsync(); |
32 | 40 | await queueSpec.DeclareAsync(); |
33 | | -Trace.WriteLine(TraceLevel.Information, "Queue Created"); |
34 | | - |
35 | | -IPublisher publisher = await connection.PublisherBuilder().Queue(queueName).MaxInflightMessages(5000).BuildAsync(); |
36 | | - |
37 | | -int received = 0; |
38 | | -DateTime start = DateTime.Now; |
| 41 | +Trace.WriteLine(TraceLevel.Information, $"Queue {queueName} recreated"); |
| 42 | +Stats stats = new(); |
| 43 | +IPublisher publisher = await connection.PublisherBuilder().Queue(queueName).BuildAsync(); |
39 | 44 |
|
40 | 45 | async Task MessageHandler(IContext context, IMessage message) |
41 | 46 | { |
42 | 47 | await context.AcceptAsync(); |
| 48 | + stats.IncrementConsumed(); |
| 49 | +} |
| 50 | + |
| 51 | +IConsumer? consumer = null; |
| 52 | + |
| 53 | +if (enableConsumer) |
| 54 | +{ |
| 55 | + consumer = await connection.ConsumerBuilder() |
| 56 | + .Queue(queueName) |
| 57 | + .InitialCredits(1000) |
| 58 | + .MessageHandler(MessageHandler) |
| 59 | + .BuildAsync(); |
| 60 | +} |
| 61 | + |
43 | 62 |
|
44 | | - if (Interlocked.Increment(ref received) % 200_000 == 0) |
| 63 | +stats.Start(); |
| 64 | +_ = Task.Run(async () => |
| 65 | +{ |
| 66 | + while (stats.IsRunning()) |
45 | 67 | { |
46 | | - DateTime end = DateTime.Now; |
47 | | - Console.WriteLine($"Received Time: {end - start} {received}"); |
| 68 | + await Task.Delay(1000); |
| 69 | + Trace.WriteLine(TraceLevel.Information, $"{stats.Report()}"); |
48 | 70 | } |
49 | | -}; |
| 71 | +}); |
50 | 72 |
|
51 | | -IConsumer consumer = await connection.ConsumerBuilder() |
52 | | - .Queue(queueName) |
53 | | - .InitialCredits(1000) |
54 | | - .MessageHandler(MessageHandler) |
55 | | - .Stream().Offset(1).Builder().BuildAsync(); |
| 73 | +List<Task<PublishResult>> tasks = []; |
56 | 74 |
|
57 | 75 | try |
58 | 76 | { |
59 | | - int confirmed = 0; |
60 | | - |
61 | | - const int total = 1_000_000; |
62 | | - for (int i = 0; i < total; i++) |
| 77 | + for (int i = 0; i < (total / tasksSize); i++) |
63 | 78 | { |
64 | 79 | try |
65 | 80 | { |
66 | | - if (i % 200_000 == 0) |
| 81 | + for (int j = 0; j < tasksSize; j++) |
67 | 82 | { |
68 | | - DateTime endp = DateTime.Now; |
69 | | - Console.WriteLine($"Sending Time: {endp - start} - messages {i}"); |
| 83 | + var message = new AmqpMessage(new byte[10]); |
| 84 | + tasks.Add(publisher.PublishAsync(message)); |
| 85 | + stats.IncrementPublished(); |
70 | 86 | } |
71 | 87 |
|
72 | | - var message = new AmqpMessage(new byte[10]); |
73 | | - PublishResult pr = await publisher.PublishAsync(message); |
74 | | - if (pr.Outcome.State == OutcomeState.Accepted) |
| 88 | + foreach (var result in await Task.WhenAll(tasks)) |
75 | 89 | { |
76 | | - if (Interlocked.Increment(ref confirmed) % 200_000 != 0) |
| 90 | + if (result.Outcome.State == OutcomeState.Accepted) |
77 | 91 | { |
78 | | - return; |
| 92 | + stats.IncrementAccepted(); |
| 93 | + } |
| 94 | + else |
| 95 | + { |
| 96 | + stats.IncrementFailed(); |
79 | 97 | } |
80 | | - |
81 | | - DateTime confirmEnd = DateTime.Now; |
82 | | - Console.WriteLine($"Confirmed Time: {confirmEnd - start} {confirmed}"); |
83 | | - } |
84 | | - else |
85 | | - { |
86 | | - Console.WriteLine( |
87 | | - $"outcome result, state: {pr.Outcome.State}, message_id: " + |
88 | | - $"{message.MessageId()}, error: {pr.Outcome.Error}"); |
89 | 98 | } |
| 99 | + |
| 100 | + tasks.Clear(); |
90 | 101 | } |
91 | 102 | catch (Exception e) |
92 | 103 | { |
93 | 104 | Trace.WriteLine(TraceLevel.Error, $"{e.Message}"); |
94 | 105 | } |
95 | 106 | } |
96 | 107 |
|
97 | | - DateTime end = DateTime.Now; |
98 | | - Console.WriteLine($"Total Sent Time: {end - start}"); |
| 108 | + stats.Stop(); |
| 109 | + await Task.Delay(1000); |
| 110 | + Trace.WriteLine(TraceLevel.Information, $"Consumer: {enableConsumer} - TaskSize: {tasksSize} - {stats.Report(true)}"); |
99 | 111 | } |
100 | 112 | catch (Exception e) |
101 | 113 | { |
102 | 114 | Trace.WriteLine(TraceLevel.Error, $"{e.Message}"); |
103 | 115 | } |
| 116 | +finally |
| 117 | +{ |
| 118 | + Console.WriteLine("Press any key to delete the queue and close the connection."); |
| 119 | + Console.ReadKey(); |
104 | 120 |
|
105 | | -Console.WriteLine("Press any key to delete the queue and close the connection."); |
106 | | -Console.ReadKey(); |
| 121 | + try |
| 122 | + { |
| 123 | + await publisher.CloseAsync(); |
| 124 | + publisher.Dispose(); |
| 125 | + } |
| 126 | + catch (Exception ex) |
| 127 | + { |
| 128 | + Console.WriteLine("[ERROR] unexpected exception while closing publisher: {0}", ex); |
| 129 | + } |
107 | 130 |
|
108 | | -await publisher.CloseAsync(); |
109 | | -publisher.Dispose(); |
| 131 | + try |
| 132 | + { |
| 133 | + if (consumer != null) |
| 134 | + { |
| 135 | + await consumer.CloseAsync(); |
| 136 | + consumer.Dispose(); |
| 137 | + } |
| 138 | + } |
| 139 | + catch (Exception ex) |
| 140 | + { |
| 141 | + Console.WriteLine("[ERROR] unexpected exception while closing consumer: {0}", ex); |
| 142 | + } |
110 | 143 |
|
111 | | -await consumer.CloseAsync(); |
112 | | -consumer.Dispose(); |
| 144 | + try |
| 145 | + { |
| 146 | + await queueSpec.DeleteAsync(); |
| 147 | + } |
| 148 | + catch (Exception ex) |
| 149 | + { |
| 150 | + Console.WriteLine("[ERROR] unexpected exception while deleting queue: {0}", ex); |
| 151 | + } |
113 | 152 |
|
114 | | -await queueSpec.DeleteAsync(); |
115 | | -await environment.CloseAsync(); |
| 153 | + try |
| 154 | + { |
| 155 | + await environment.CloseAsync(); |
| 156 | + } |
| 157 | + catch (Exception ex) |
| 158 | + { |
| 159 | + Console.WriteLine("[ERROR] unexpected exception while closing environment: {0}", ex); |
| 160 | + } |
| 161 | +} |
0 commit comments