IBus.OutgoingHeaders updates can cause InvalidOperationException

[johnsimons]

Change IBus.OutgoingHeaders to a ConcurrentDictionary

Because the framework allows a user to modify IBus.OutgoingHeaders during runtime and those headers are global, an InvalidOperationException (Collection was modified; enumeration operation may not execute.) can be thrown during enumeration of those headers dictionary.

So a possible solution is to make a copy of the dictionary every time before enumerating it.

The code in question is https://github.com/Particular/NServiceBus/blob/develop/src/NServiceBus.Core/Unicast/Behaviors/CreatePhysicalMessageBehavior.cs#L37

[chrisbednarski]

Will this still be a problem when #1975 is fixed (ie the headers will no longer change after config time)?

[johnsimons]

@chrisbednarski we are still discussing whether the global headers should be changed after configuration or not, please comment with your opinion

[chrisbednarski]

I currently don't have any use-cases for modifying the global OutgoingHeaders after the bus has been started and all initialization interfaces have completed. But, my experience doesn't represent what others may be doing with headers.

All I was saying, I don't see any benefit in making the global header access thread safe, if headers cannot be modified after Bus has been started. It would make sense to do either one or the other.

On another note, it's feasible to atomically replace the entire collection using Interlocked.CompareExchange instead of taking a copy each time a message needs to be sent. If outgoing headers are not modified often, this may be a lot cheaper than a copy during each send.

pseudo code

public void SetOutgoingHeader(string key, string value)
{
 IDictionary<string, string> initial, changed;
 do {
    initial = bus.OutgoingHeaders;
    changed = initial.ToDictionary(); //make a copy
    changed[key] = value;

    // CompareExchange compares bus.OutgoingHeaders to the initial value. If they are not
    // equal, then another thread has updated the outgoing headers since this loop started.
 }while(initial != Interlocked.CompareExchange(ref bus.OutgoingHeaders, changed, initial));
}

[MaggiePlusPlus]

We set the outgoing headers on every message.

[johnsimons]

Hi @MaggiePlusPlus

There are 2 use cases:

1. You need to set global headers for all outgoing messages
2. You need to set headers on specific outgoing messages (not all of them)

To do case 1, you should set the global headers in a IWantToRunWhenBusStartsAndStops eg:

class Foo : IWantToRunWhenBusStartsAndStops
{
    public IBus Bus { get; set; }

    public void Start()
    {
        Bus.OutgoingHeaders.Add("MyGlobalKey", "MyGlobalValue");
    }

    public void Stop()
    {
    }
}

To do case 2, you set those headers in a IHandleMessage, eg:

public class MyMessageHandler : IHandleMessages<MyMessage>
{
    public IBus Bus { get; set; }

    public void Handle(MyMessage msg)
    {
        MyOutgoingMessage message = new MyOutgoingMessage();

        Bus.SetMessageHeader(message, "MyKeyForThisMessage", "My special value for this message");
        Bus.Send(message);

        //More code
    }
}

[MaggiePlusPlus]

Thanks. I misunderstood in #1975 when someone said they couldn't think of a use case where changing the headers would be a good idea. I was saying that we did change them. I thought I had put the comment on the other ticket.

I missed the distinction between global outgoing headers and outgoing headers. So I think I understand that now.

In your example you say to use Bus.OutgoingHeaders.Add("MyGlobalKey", "MyGlobalValue"); I thought we were supposed to use the Set method here as well because of the bug.

[SimonCropp]

related to #1975

[SimonCropp]

so the short term fix for the actually issue outline here to change the storage of these headers to a ConcurrentDictionary.

Any objections?

[andreasohlund]

go ahead

On Sun, Oct 19, 2014 at 1:27 AM, Simon Cropp notifications@github.com
wrote:

so the short term fix for the actually issue outline here to change the
storage of these headers to a ConcurrentDictionary.

Any objections?

—
Reply to this email directly or view it on GitHub
#1976 (comment)
.