October 2005 - Posts
The BCL in .NET 2.0 has a plethora of small additions, which are not that visible, but are very useful. I’ll try to cover many of these in my subsequent posts.
String.IsNullOrEmpty is one such method. As the name would indicate, this method returns true if the string argument passed to this method is either null (nothing in VB) or empty.
VB 2005 has some good additions too. Among others, I like the IsNot operator. How much more did you want this operator after writing code like this:
If Not obj Is Nothing
..
Then
Now,
If obj IsNot Nothing …looks much better.
MSMQ simply provides wonderful capabilities when it comes to building asynchronous messaging based applications. You typically run into situations where two applications exchange messages through a queue – one application sends messages to a known MSMQ destination and the other application “listens” to this queue and picks messages up as and when they arrive. The application which listens usually is a windows service which either has a dedicated thread making a blocking call on the Receive method or uses a timer of sorts to periodically check the queue for new messages. Either ways, to maximize throughput, you need to write the queue-listener code in such a way that you can have multiple threads simultaneously processing different messages that arrive at the queue.
MSMQ triggers provide one such solution where we can associate incoming messages to a COM component, based on certain rules, which can be application specific. MSMQ triggers are pretty powerful and also easy to administer. But there is one constraint, which has to do with the fact that only COM components are supported in this scheme, or in other words, message processors should be COM components. In a way, it isn’t that bad a choice after all. You still can create a CCW for a managed class and register the trigger. But, somehow, I didn’t like doing this.
To circumvent this problem, I decided to write a C# helper class, which doesn’t really do anything much, but helps realize a simple multithreaded job (message processing) dispatcher. The helper itself doesn’t do any message processing as such; whenever a message arrives at a queue, it just fires an event (on a new thread pool thread) with the message contents being in the event arguments.
public delegate void MessageReceivedEventHandler(object sender, MessageEventArgs args);
public class MSMQListener
{
private bool _listen;
private Type[] _types;
private MessageQueue _queue;
public event MessageReceivedEventHandler MessageReceived;
public Type[] FormatterTypes
{
get { return _types; }
set { _types = value; }
}
public MSMQListener(string queuePath)
{
_queue = new MessageQueue(queuePath);
}
public void Start()
{
_listen = true;
if (_types != null && _types.Length > 0)
{
// Using only the XmlMessageFormatter. You can use other formatters as well
_queue.Formatter = new XmlMessageFormatter(_types);
}
_queue.PeekCompleted += new PeekCompletedEventHandler(OnPeekCompleted);
_queue.ReceiveCompleted += new ReceiveCompletedEventHandler(OnReceiveCompleted);
StartListening();
}
public void Stop()
{
_listen = false;
_queue.PeekCompleted -= new PeekCompletedEventHandler(OnPeekCompleted);
_queue.ReceiveCompleted -= new ReceiveCompletedEventHandler(OnReceiveCompleted);
}
private void StartListening()
{
if (!_listen)
{
return;
}
// The MSMQ class does not have a BeginRecieve method that can take in a
// MSMQ transaction object. This is a workaround - we do a BeginPeek and then
// recieve the message synchronously in a transaction.
// Check documentation for more details
if (_queue.Transactional)
{
_queue.BeginPeek();
}
else
{
_queue.BeginReceive();
}
}
private void OnPeekCompleted(object sender, PeekCompletedEventArgs e)
{
_queue.EndPeek(e.AsyncResult);
MessageQueueTransaction trans = new MessageQueueTransaction();
Message msg = null;
try
{
trans.Begin();
msg = _queue.Receive(trans);
trans.Commit();
StartListening();
FireRecieveEvent(msg.Body);
}
catch
{
trans.Abort();
}
}
private void FireRecieveEvent(object body)
{
if (MessageReceived != null)
{
MessageReceived(this, new MessageEventArgs(body));
}
}
private void OnReceiveCompleted(object sender, ReceiveCompletedEventArgs e)
{
Message msg = _queue.EndReceive(e.AsyncResult);
StartListening();
FireRecieveEvent(msg.Body);
}
}
public class MessageEventArgs : EventArgs
{
private object _messageBody;
public object MessageBody
{
get { return _messageBody; }
}
public MessageEventArgs(object body)
{
_messageBody = body;
}
}
As you can see, the helper class is indeed simple. One thing to notice though is that, I switch to BeginPeek if the queue is transactional. The reason behind this is there is no overload of the BeginRecieve method that takes a MessageQueueTransaction object and the MSDN documentation also explains the same. The workaround is to use BeginPeek and then do call Receive (synchronous/blocking operation) in the EndPeek method.
At this point of time, this class does not provide much control to the consumer. For instance, it does not allow you to set the maximum number of concurrent threads that can be processing messages - This class queues as many jobs as that supported by the .NET Thread pool. Also, I haven’t tested this code in all scenarios – I have not seen if this works well in the context of a distributed (DTC based) transaction. I shall update this version every now and then to make it more powerful and to work seamlessly in all scenarios.
The ThreadPool class is very handy to developers. You could queue a method to run on a separate thread and forget about it, without worrying about managing the thread object or about a possible thread explosion scenario. That’s because the .NET ThreadPool keeps a tab on the number of executing threads, which I believe is a maximum of 25 per processor. This might sound just fine, but recently, I came across a situation where even 25 threads per processor looked expensive – I needed to limit the count to something like 10. Unfortunately, the ThreadPool class didn’t give me a method/property where I could set a thread count in the pool to something smaller. I wasn’t adept enough to host the CLR, where I could tweak this count. Instead, I decided to write a simplistic wrapper over the ThreadPool class, which basically did these things:
- Keep a count on the number of active threads (which I wanted to limit in the first place)
- Maintain a queue to store requests is threads aren’t available
- A timer implementation, where a thread periodically checks whether queued jobs can now be pushed to the thread pool
Quite obviously, this is a logical replica of the ThreadPool implementation itself, which is pretty wasteful. But that was until .NET Framework 1.1. In .NET 2.0, the ThreadPool class comes with this new method called SetMaxThreads – just what I craved for. You can use this to set the maximum number threads in the thread pool. There are two values to set here – one for the worker threads (which is more relevant in my case), and one for IO Port completion (this is useful for asynchronous IO operations).
Again, it is one's responsibility not to set an arbitrary high value of worker threads; Do I need to explain why?
One of the biggest cribs about the DataSet in .NET 1.0 and 1.1 was its serialization performance. The DataSet, albeit providing a rich object model to work with data in a disconnected way, was a bane when it came to passing it across tiers of the application, typically in a remoting scenario. It turned out that in this scenario, the DataSet serialization glutted on bandwidth, memory and CPU cycles. Developers who were hell bent on using the DataSet eventually became the undisputed champions of object serialization, using wonky ways to serialize the dataset in an effort to make it optimal. Less nerdy developers adopted not so adventurous means like using custom collections and arrays, but sacrificing the high-fidelity object model of the DataSet in the process.
Someone said that every complex problem has a simple solution, and .NET 2.0 is turning out to be an angel to this end. As you might have already guessed, the voice from the sky is in form of a new property of the DataSet (and surprise, DataTable too) called RemotingFormat, which is an enumeration of type SerializationFormat having two members – Binary and Xml. Quite obviously, the default is Xml, to preserve backward compatibility.
Okay, I am digressing slightly here to note something interesting. The name of the property in discussion is RemotingFormat, not something like SerializationFormat or SerializedAs. I suppose, it is to enunciate two things:
DataSets (and now, even DataTable) implements IXmlSerializable, which is what the XmlSerializer looks for when serializing objects in the web service scenario. So, setting the RemotingFormat property there would be of no consequence, and it does not make sense there either.
Perhaps, DataSets are more apt for remoting scenarios, as compared to that of Web Services. This is not to say DataSets are deemed not usable in web services (as the implementation of IXmlSerializable would clarify). It is just that all the performance issues discussed earlier in this post would still apply in this case.
Finally, I would urge you to read this excellent article by Dino Esposito to get a better insight into this new feature and other improvements that have been added to the DataTable class as well.