Skip to content

Commit 595ac87

Browse files
michaelklishinmichaelklishin
committed
Merge pull request #844 from stebet/frameOptimizations
Frame optimizations (cherry picked from commit ed3b780)
1 parent ce9b039 commit 595ac87

File tree

7 files changed

+48
-95
lines changed

7 files changed

+48
-95
lines changed

‎_site

Submodule _site updated 161 files

‎projects/RabbitMQ.Client/client/impl/Command.cs

Lines changed: 2 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -98,64 +98,20 @@ public static void CheckEmptyFrameSize()
9898
}
9999

100100
internal void Transmit(int channelNumber, Connection connection)
101-
{
102-
if (Method.HasContent)
103-
{
104-
TransmitAsFrameSet(channelNumber, connection);
105-
}
106-
else
107-
{
108-
TransmitAsSingleFrame(channelNumber, connection);
109-
}
110-
}
111-
112-
internal void TransmitAsSingleFrame(int channelNumber, Connection connection)
113101
{
114102
connection.WriteFrame(new MethodOutboundFrame(channelNumber, Method));
115-
}
116-
117-
internal void TransmitAsFrameSet(int channelNumber, Connection connection)
118-
{
119-
var frames = new List<OutboundFrame> { new MethodOutboundFrame(channelNumber, Method) };
120103
if (Method.HasContent)
121104
{
122-
frames.Add(new HeaderOutboundFrame(channelNumber, Header, Body.Length));
105+
connection.WriteFrame(new HeaderOutboundFrame(channelNumber, Header, Body.Length));
123106
int frameMax = (int)Math.Min(int.MaxValue, connection.FrameMax);
124107
int bodyPayloadMax = (frameMax == 0) ? Body.Length : frameMax - EmptyFrameSize;
125108
for (int offset = 0; offset < Body.Length; offset += bodyPayloadMax)
126109
{
127110
int remaining = Body.Length - offset;
128111
int count = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
129-
frames.Add(new BodySegmentOutboundFrame(channelNumber, Body.Slice(offset, count)));
112+
connection.WriteFrame(new BodySegmentOutboundFrame(channelNumber, Body.Slice(offset, count)));
130113
}
131114
}
132-
133-
connection.WriteFrameSet(frames);
134-
}
135-
136-
137-
internal static List<OutboundFrame> CalculateFrames(int channelNumber, Connection connection, IList<Command> commands)
138-
{
139-
var frames = new List<OutboundFrame>();
140-
141-
foreach (Command cmd in commands)
142-
{
143-
frames.Add(new MethodOutboundFrame(channelNumber, cmd.Method));
144-
if (cmd.Method.HasContent)
145-
{
146-
frames.Add(new HeaderOutboundFrame(channelNumber, cmd.Header, cmd.Body.Length));
147-
int frameMax = (int)Math.Min(int.MaxValue, connection.FrameMax);
148-
int bodyPayloadMax = (frameMax == 0) ? cmd.Body.Length : frameMax - EmptyFrameSize;
149-
for (int offset = 0; offset < cmd.Body.Length; offset += bodyPayloadMax)
150-
{
151-
int remaining = cmd.Body.Length - offset;
152-
int count = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
153-
frames.Add(new BodySegmentOutboundFrame(channelNumber, cmd.Body.Slice(offset, count)));
154-
}
155-
}
156-
}
157-
158-
return frames;
159115
}
160116

161117
public void Dispose()

‎projects/RabbitMQ.Client/client/impl/Connection.cs

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -948,11 +948,6 @@ public void WriteFrame(OutboundFrame f)
948948
_frameHandler.WriteFrame(f);
949949
}
950950

951-
public void WriteFrameSet(IList<OutboundFrame> f)
952-
{
953-
_frameHandler.WriteFrameSet(f);
954-
}
955-
956951
public void UpdateSecret(string newSecret, string reason)
957952
{
958953
_model0.UpdateSecret(newSecret, reason);

‎projects/RabbitMQ.Client/client/impl/IFrameHandler.cs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -71,8 +71,6 @@ interface IFrameHandler
7171

7272
void SendHeader();
7373

74-
void WriteFrame(OutboundFrame frame, bool flush = true);
75-
76-
void WriteFrameSet(IList<OutboundFrame> frames);
74+
void WriteFrame(OutboundFrame frame);
7775
}
7876
}

‎projects/RabbitMQ.Client/client/impl/SessionBase.cs

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,9 +191,14 @@ public virtual void Transmit(Command cmd)
191191
// of frames within a channel. But that is fixed in socket frame handler instead, so no need to lock.
192192
cmd.Transmit(ChannelNumber, Connection);
193193
}
194+
194195
public virtual void Transmit(IList<Command> commands)
195196
{
196-
Connection.WriteFrameSet(Command.CalculateFrames(ChannelNumber, Connection, commands));
197+
for (int i = 0; i < commands.Count; i++)
198+
{
199+
Command command = commands[i];
200+
command.Transmit(ChannelNumber, Connection);
201+
}
197202
}
198203
}
199204
}

‎projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs

Lines changed: 29 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
using System.Net.Sockets;
4747
using System.Runtime.InteropServices;
4848
using System.Text;
49+
using System.Threading.Channels;
4950
using System.Threading.Tasks;
5051

5152
using RabbitMQ.Client.Exceptions;
@@ -79,7 +80,8 @@ class SocketFrameHandler : IFrameHandler
7980
private readonly ITcpClient _socket;
8081
private readonly Stream _writer;
8182
private readonly object _semaphore = new object();
82-
private readonly object _streamLock = new object();
83+
private readonly Channel<OutboundFrame> _frameChannel = Channel.CreateUnbounded<OutboundFrame>(new UnboundedChannelOptions { AllowSynchronousContinuations = false, SingleReader = true, SingleWriter = false });
84+
private Task _frameWriter;
8385
private bool _closed;
8486
public SocketFrameHandler(AmqpTcpEndpoint endpoint,
8587
Func<AddressFamily, ITcpClient> socketFactory,
@@ -124,6 +126,7 @@ public SocketFrameHandler(AmqpTcpEndpoint endpoint,
124126
_writer = new BufferedStream(netstream, _socket.Client.SendBufferSize);
125127

126128
WriteTimeout = writeTimeout;
129+
_frameWriter = Task.Run(WriteFrameImpl);
127130
}
128131
public AmqpTcpEndpoint Endpoint { get; set; }
129132

@@ -181,6 +184,15 @@ public void Close()
181184
{
182185
if (!_closed)
183186
{
187+
try
188+
{
189+
_frameChannel.Writer.Complete();
190+
_frameWriter.Wait();
191+
}
192+
catch(Exception)
193+
{
194+
}
195+
184196
try
185197
{
186198
_socket.Close();
@@ -222,46 +234,31 @@ public void SendHeader()
222234
headerBytes[7] = (byte)Endpoint.Protocol.MinorVersion;
223235
}
224236

225-
Write(new ArraySegment<byte>(headerBytes), true);
237+
_writer.Write(headerBytes, 0, 8);
238+
_writer.Flush();
226239
}
227240

228-
public void WriteFrame(OutboundFrame frame, bool flush = true)
241+
public void WriteFrame(OutboundFrame frame)
229242
{
230-
int bufferSize = frame.GetMinimumBufferSize();
231-
byte[] memoryArray = ArrayPool<byte>.Shared.Rent(bufferSize);
232-
Memory<byte> slice = new Memory<byte>(memoryArray, 0, bufferSize);
233-
frame.WriteTo(slice);
234-
_socket.Client.Poll(_writeableStateTimeoutMicroSeconds, SelectMode.SelectWrite);
235-
Write(slice.Slice(0, frame.ByteCount), flush);
236-
ArrayPool<byte>.Shared.Return(memoryArray);
237-
}
238-
239-
public void WriteFrameSet(IList<OutboundFrame> frames)
240-
{
241-
for (int i = 0; i < frames.Count; i++)
242-
{
243-
WriteFrame(frames[i], false);
244-
}
245-
246-
lock (_streamLock)
247-
{
248-
_writer.Flush();
249-
}
243+
_frameChannel.Writer.TryWrite(frame);
250244
}
251245

252-
private void Write(ReadOnlyMemory<byte> buffer, bool flush)
246+
public async Task WriteFrameImpl()
253247
{
254-
lock (_streamLock)
248+
while (await _frameChannel.Reader.WaitToReadAsync().ConfigureAwait(false))
255249
{
256-
if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> segment))
250+
_socket.Client.Poll(_writeableStateTimeoutMicroSeconds, SelectMode.SelectWrite);
251+
while (_frameChannel.Reader.TryRead(out OutboundFrame frame))
257252
{
258-
_writer.Write(segment.Array, segment.Offset, segment.Count);
259-
260-
if (flush)
261-
{
262-
_writer.Flush();
263-
}
253+
int bufferSize = frame.GetMinimumBufferSize();
254+
byte[] memoryArray = ArrayPool<byte>.Shared.Rent(bufferSize);
255+
Memory<byte> slice = new Memory<byte>(memoryArray, 0, bufferSize);
256+
frame.WriteTo(slice);
257+
_writer.Write(memoryArray, 0, bufferSize);
258+
ArrayPool<byte>.Shared.Return(memoryArray);
264259
}
260+
261+
_writer.Flush();
265262
}
266263
}
267264

‎projects/Unit/TestRecoverAfterCancel.cs

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -82,21 +82,23 @@ public void TestRecoverAfterCancel_()
8282
UTF8Encoding enc = new UTF8Encoding();
8383
Channel.BasicPublish("", Queue, null, enc.GetBytes("message"));
8484
EventingBasicConsumer Consumer = new EventingBasicConsumer(Channel);
85-
SharedQueue<BasicDeliverEventArgs> EventQueue = new SharedQueue<BasicDeliverEventArgs>();
86-
Consumer.Received += (_, e) => EventQueue.Enqueue(e);
85+
SharedQueue<(bool Redelivered, byte[] Body)> EventQueue = new SharedQueue<(bool Redelivered, byte[] Body)>();
86+
// Making sure we copy the delivery body since it could be disposed at any time.
87+
Consumer.Received += (_, e) => EventQueue.Enqueue((e.Redelivered, e.Body.ToArray()));
8788

8889
string CTag = Channel.BasicConsume(Queue, false, Consumer);
89-
BasicDeliverEventArgs Event = EventQueue.Dequeue();
90+
(bool Redelivered, byte[] Body) Event = EventQueue.Dequeue();
9091
Channel.BasicCancel(CTag);
9192
Channel.BasicRecover(true);
9293

9394
EventingBasicConsumer Consumer2 = new EventingBasicConsumer(Channel);
94-
SharedQueue<BasicDeliverEventArgs> EventQueue2 = new SharedQueue<BasicDeliverEventArgs>();
95-
Consumer2.Received += (_, e) => EventQueue2.Enqueue(e);
95+
SharedQueue<(bool Redelivered, byte[] Body)> EventQueue2 = new SharedQueue<(bool Redelivered, byte[] Body)>();
96+
// Making sure we copy the delivery body since it could be disposed at any time.
97+
Consumer2.Received += (_, e) => EventQueue2.Enqueue((e.Redelivered, e.Body.ToArray()));
9698
Channel.BasicConsume(Queue, false, Consumer2);
97-
BasicDeliverEventArgs Event2 = EventQueue2.Dequeue();
99+
(bool Redelivered, byte[] Body) Event2 = EventQueue2.Dequeue();
98100

99-
CollectionAssert.AreEqual(Event.Body.ToArray(), Event2.Body.ToArray());
101+
CollectionAssert.AreEqual(Event.Body, Event2.Body);
100102
Assert.IsFalse(Event.Redelivered);
101103
Assert.IsTrue(Event2.Redelivered);
102104
}

0 commit comments

Comments
 (0)