Recipe 18.11. Synchronizing Multiple Processes with the Mutex Problem You have two processes or appdomains that are running code with actions that you need to coordinate. Solution Use a named Mutex as a common signaling mechanism to do the coordination. A named Mutex can be accessed from both pieces of code even when running in different processes or appdomains. One situation in which this can be useful is when you are using shared memory to communicate between processes. The SharedMemoryManager class presented in this recipe will show the named Mutex in action by setting up a section of shared memory that can be used to pass serializable objects between processes. The "server" process creates a SharedMemoryManager instance which sets up the shared memory and then creates the Mutex as the initial owner. The "client" process then also creates a SharedMemoryManager instance that finds the shared memory and hooks up to it. Once this connection is established, the "client" process then sets up to receive the serialized objects and waits until one is sent by waiting on the Mutex the "server" process created. The "server" process then takes a serializable object, serializes it into the shared memory, and releases the Mutex. It then waits on it again so that when the "client" is done receiving the object, it can release the Mutex and give control back to the "server." The "client" process that was waiting on the Mutex then deserializes the object from the shared memory and releases the Mutex. In the example, you will send the Contact structure, which looks like this: [StructLayout(LayoutKind.Sequential)] [Serializable()] public struct Contact { public string _name; public int _age; }
The "server" process code to send the Contact looks like this: // Create the initial shared memory manager to get things set up. using(SharedMemoryManager<Contact> sm = new SharedMemoryManager<Contact>("Contacts",8092)) { // This is the sender process. // Launch the second process to get going. string processName = Process.GetCurrentProcess().MainModule.FileName; int index = processName.IndexOf("vshost"); if (index != -1) { string first = processName.Substring(0, index); int numChars = processName.Length - (index + 7); string second = processName.Substring(index + 7, numChars); processName = first + second; } Process receiver = Process.Start( new ProcessStartInfo( processName, "Receiver")); // Give it 5 seconds to spin up. Thread.Sleep(5000); // Make up a contact. Contact man; man._age = 23; man._name = "Dirk Daring"; // Send it to the other process via shared memory. sm.SendObject(man); }
The "client" process code to receive the Contact looks like this: // Create the initial shared memory manager to get things set up. using(SharedMemoryManager<Contact> sm = new SharedMemoryManager<Contact>("Contacts",8092)) { // Get the contact once it has been sent. Contact c = (Contact)sm.ReceiveObject(); // Write it out (or to a database…) Console.WriteLine("Contact {0} is {1} years old.", c._name, c._age); // Show for 5 seconds. Thread.Sleep(5000); }
The way this usually works is that one process creates a section of shared memory backed by the paging file using the unmanaged Win32 APIs CreateFileMapping and MapViewOfFile. Currently there is no purely managed way to do this, so you have to use P/Invoke, as you can see in Example 18-9 in the constructor code for the SharedMemoryManager and the private SetupSharedMemory method. The constructor takes a name to use as part of the name of the shared memory and the base size of the shared memory block to allocate. It is the base size because the SharedMemoryManager has to allocate a bit extra for keeping track of the data moving through the buffer. Example 18-9. Constructor and SetupSharedMemory private method public SharedMemoryManager(string name,int sharedMemoryBaseSize) { if (string.IsNullOrEmpty(name)) throw new ArgumentNullException("name"); if (sharedMemoryBaseSize <= 0) throw new ArgumentOutOfRangeException("sharedMemoryBaseSize", "Shared Memory Base Size must be a value greater than zero"); // Set name of the region. _memoryRegionName = name; // Save base size. _sharedMemoryBaseSize = sharedMemoryBaseSize; // Set up the memory region size. _memRegionSize = (uint)(_sharedMemoryBaseSize + sizeof(int)); // Set up the shared memory section. SetupSharedMemory( ); } private void SetupSharedMemory( ) { // Grab some storage from the page file. _handleFileMapping = PInvoke.CreateFileMapping((IntPtr)INVALID_HANDLE_VALUE, IntPtr.Zero, PInvoke.PageProtection.ReadWrite, 0, _memRegionSize, _memoryRegionName); if (_handleFileMapping == IntPtr.Zero) { throw new Win32Exception( "Could not create file mapping"); } // Check the error status. int retVal = Marshal.GetLastWin32Error( ); if (retVal == ERROR_ALREADY_EXISTS) { // We opened one that already existed. // Make the mutex not the initial owner // of the mutex since we are connecting // to an existing one. _mtxSharedMem = new Mutex(false, string.Format("{0}mtx{1}", typeof(TransferItemType), _memoryRegionName)); } else if (retVal == 0) { // We opened a new one. // Make the mutex the initial owner. _mtxSharedMem = new Mutex(true, string.Format("{0}mtx{1}", typeof(TransferItemType), _memoryRegionName)); } else { // Something else went wrong. throw new Win32Exception(retVal, "Error creating file mapping"); } // Map the shared memory. _ptrToMemory = PInvoke.MapViewOfFile(_handleFileMapping, FILE_MAP_WRITE, 0, 0, IntPtr.Zero); if (_ptrToMemory == IntPtr.Zero) { retVal = Marshal.GetLastWin32Error( ); throw new Win32Exception(retVal, "Could not map file view"); } retVal = Marshal.GetLastWin32Error( ); if (retVal != 0 && retVal != ERROR_ALREADY_EXISTS) { // Something else went wrong. throw new Win32Exception(retVal, "Error mapping file view"); } }
|
The code to send an object through the shared memory is contained in the SendObject method, shown in Example 18-10. First it checks to see if the object being sent is indeed serializable by checking the IsSerializable property on the type of the object. If the object is serializable, an integer with the size of the serialized object and the serialized object content are written out to the shared memory section. Then the Mutex is released to indicate that there is an object in the shared memory. It then waits on the Mutex again to wait until the "client" has received the object. Example 18-10. SendObject method public void SendObject(TransferItemType transferObject) { // Can send only Seralizable objects. if (!transferObject.GetType( ).IsSerializable) throw new ArgumentException( string.Format("Object {0} is not serializeable.", transferObject)); // Create a memory stream, initialize size. using (MemoryStream ms = new MemoryStream( )) { // Get a formatter to serialize with. BinaryFormatter formatter = new BinaryFormatter( ); try { // Serialize the object to the stream. formatter.Serialize(ms, transferObject); // Get the bytes for the serialized object. byte[] bytes = ms.GetBuffer( ); // Check that this object will fit. if(bytes.Length + sizeof(int) > _memRegionSize) { string fmt = "{0} object instance serialized to {1} bytes " + "which is too large for the shared memory region"; string msg = string.Format(fmt, typeof(TransferItemType),bytes.Length); throw new ArgumentException(msg, "transferObject"); } // Write out how long this object is. Marshal.WriteInt32(this._ptrToMemory, bytes.Length); // Write out the bytes. Marshal.Copy(bytes, 0, this._ptrToMemory, bytes.Length); } finally { // Signal the other process using the mutex to tell it // to do receive processing. _mtxSharedMem.ReleaseMutex( ); // Wait for the other process to signal it has received // and we can move on. _mtxSharedMem.WaitOne( ); } } }
|
The ReceiveObject method shown in Example 18-11 allows the client to wait until there is an object in the shared memory section, then read the size of the serialized object and deserialize it to a managed object. It then releases the Mutex to let the sender know to continue. Example 18-11. ReceiveObject method public TransferItemType ReceiveObject() { // Wait on the mutex for an object to be queued by the sender. _mtxSharedMem.WaitOne(); // Get the count of what is in the shared memory. int count = Marshal.ReadInt32(_ptrToMemory); if (count <= 0) { throw new InvalidDataException("No object to read"); } // Make an array to hold the bytes. byte[] bytes = new byte[count]; // Read out the bytes for the object. Marshal.Copy(_ptrToMemory, bytes, 0, count); // Set up the memory stream with the object bytes. using (MemoryStream ms = new MemoryStream(bytes)) { // Set up a binary formatter. BinaryFormatter formatter = new BinaryFormatter(); // Get the object to return. TransferItemType item; try { item = (TransferItemType)formatter.Deserialize(ms); } finally { // Signal that we received the object using the mutex. _mtxSharedMem.ReleaseMutex(); } // Give them the object. return item; } }
|
Discussion A Mutex is designed to give mutually exclusive (thus the name) access to a single resource. A Mutex can be thought of as a cross-process, named Monitor where the Mutex is "entered" by waiting on it and becoming the owner, then "exited" by releasing the Mutex for the next thread that is waiting on it. If a thread that owns a Mutex ends, the Mutex is released automatically. Using a Mutex is slower than using a Monitor as a Monitor is a purely managed construct whereas a Mutex is based on the Mutex kernel object. A Mutex cannot be "pulsed" as can a Monitor, but it can be used across processes which a Monitor cannot. Finally, the Mutex is based on WaitHandle, so it can be waited on with other objects derived from WaitHandle, like Semaphore and the event classes. The SharedMemoryManager and PInvoke classes are listed in their entirety in Example 18-12. Example 18-12. SharedMemoryManager and PInvoke classes /// <summary> /// Class for sending objects through shared memory using a mutex /// to synchronize access to the shared memory /// </summary> public class SharedMemoryManager<TransferItemType> : IDisposable { #region Consts const int INVALID_HANDLE_VALUE = -1; const int FILE_MAP_WRITE = 0x0002; /// <summary> /// Define from Win32 API. /// </summary> const int ERROR_ALREADY_EXISTS = 183; #endregion #region Private members IntPtr _handleFileMapping = IntPtr.Zero; IntPtr _ptrToMemory = IntPtr.Zero; uint _memRegionSize = 0; string _memoryRegionName; bool disposed = false; int _sharedMemoryBaseSize = 0; Mutex _mtxSharedMem = null; #endregion #region Construction / Cleanup public SharedMemoryManager(string name,int sharedMemoryBaseSize) { // Can be built for only Seralizable objects if (!typeof(TransferItemType).IsSerializable) throw new ArgumentException( string.Format("Object {0} is not serializeable.", typeof(TransferItemType))); if (string.IsNullOrEmpty(name)) throw new ArgumentNullException("name"); if (sharedMemoryBaseSize <= 0) throw new ArgumentOutOfRangeException("sharedMemoryBaseSize", "Shared Memory Base Size must be a value greater than zero") // Set name of the region. _memoryRegionName = name; // Save base size. _sharedMemoryBaseSize = sharedMemoryBaseSize; // Set up the memory region size. _memRegionSize = (uint)(_sharedMemoryBaseSize + sizeof(int)); // Set up the shared memory section. SetupSharedMemory(); } private void SetupSharedMemory() { // Grab some storage from the page file. _handleFileMapping = PInvoke.CreateFileMapping((IntPtr)INVALID_HANDLE_VALUE, IntPtr.Zero, PInvoke.PageProtection.ReadWrite, 0, _memRegionSize, _memoryRegionName); if (_handleFileMapping == IntPtr.Zero) { throw new Win32Exception( "Could not create file mapping"); } // Check the error status. int retVal = Marshal.GetLastWin32Error(); if (retVal == ERROR_ALREADY_EXISTS) { // We opened one that already existed. // Make the mutex not the initial owner // of the mutex since we are connecting // to an existing one. _mtxSharedMem = new Mutex(false, string.Format("{0}mtx{1}", typeof(TransferItemType), _memoryRegionName)); } else if (retVal == 0) { // We opened a new one. // Make the mutex the initial owner. _mtxSharedMem = new Mutex(true, string.Format("{0}mtx{1}", typeof(TransferItemType), _memoryRegionName)); } else { // Something else went wrong. throw new Win32Exception(retVal, "Error creating file mapping"); } // Map the shared memory. _ptrToMemory = PInvoke.MapViewOfFile(_handleFileMapping, FILE_MAP_WRITE, 0, 0, IntPtr.Zero); if (_ptrToMemory == IntPtr.Zero) { retVal = Marshal.GetLastWin32Error(); throw new Win32Exception(retVal, "Could not map file view"); } retVal = Marshal.GetLastWin32Error(); if (retVal != 0 && retVal != ERROR_ALREADY_EXISTS) { // Something else went wrong. throw new Win32Exception(retVal, "Error mapping file view"); } } ~SharedMemoryManager() { // Make sure we close. Dispose(false); } public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } private void Dispose(bool disposing) { // Check to see if Dispose has already been called. if (!this.disposed) { CloseSharedMemory(); } disposed = true; } private void CloseSharedMemory() { if (_ptrToMemory != IntPtr.Zero) { // Close map for shared memory. PInvoke.UnmapViewOfFile(_ptrToMemory); _ptrToMemory = IntPtr.Zero; } if (_handleFileMapping != IntPtr.Zero) { // Close handle. PInvoke.CloseHandle(_handleFileMapping); _handleFileMapping = IntPtr.Zero; } } public void Close() { CloseSharedMemory(); } #endregion #region Properties public int SharedMemoryBaseSize { get { return _sharedMemoryBaseSize; } } #endregion #region Public Methods /// <summary> /// Send a serializable object through the shared memory /// and wait for it to be picked up. /// </summary> /// <param name="transferObject"></param> public void SendObject(TransferItemType transferObject) { // Create a memory stream, initialize size. using (MemoryStream ms = new MemoryStream()) { // Get a formatter to serialize with. BinaryFormatter formatter = new BinaryFormatter(); try { // Serialize the object to the stream. formatter.Serialize(ms, transferObject); // Get the bytes for the serialized object. byte[] bytes = ms.ToArray(); // Check that this object will fit. if(bytes.Length + sizeof(int) > _memRegionSize) { string fmt = " "{0} object instance serialized to {1} bytes " + "which is too large for the shared memory region"; string msg = string.Format(fmt, typeof(TransferItemType),bytes.Length); throw new ArgumentException(msg, "transferObject"); } // Write out how long this object is. Marshal.WriteInt32(this._ptrToMemory, bytes.Length); // Write out the bytes. Marshal.Copy(bytes, 0, this._ptrToMemory, bytes.Length); } finally { // Signal the other process using the mutex to tell it // to do receive processing. _mtxSharedMem.ReleaseMutex(); // Wait for the other process to signal it has received // and we can move on. _mtxSharedMem.WaitOne(); } } } /// <summary> /// Wait for an object to hit the shared memory and then deserialize it. /// </summar> /// <returns>object passed</returns> public TransferItemType ReceiveObject() { // Wait on the mutex for an object to be queued by the sender. _mtxSharedMem.WaitOne(); // Get the count of what is in the shared memory. int count = Marshal.ReadInt32(_ptrToMemory); if (count <= 0) { throw new InvalidDataException("No object to read"); } // Make an array to hold the bytes. byte[] bytes = new byte[count]; // Read out the bytes for the object. Marshal.Copy(_ptrToMemory, bytes, 0, count); // Set up the memory stream with the object bytes. using (MemoryStream ms = new MemoryStream(bytes)) { // Set up a binary formatter. BinaryFormatter formatter = new BinaryFormatter(); // Get the object to return. TransferItemType item; try { item = (TransferItemType)formatter.Deserialize(ms); } finally { // Signal that we received the object using the mutex. _mtxSharedMem.ReleaseMutex(); } // Give them the object. return item; } } #endregion } public class PInvoke { #region PInvoke defines [Flags] public enum PageProtection : uint { NoAccess = 0x01, Readonly = 0x02, ReadWrite = 0x04, WriteCopy = 0x08, Execute = 0x10, ExecuteRead = 0x20, ExecuteReadWrite = 0x40, ExecuteWriteCopy = 0x80, Guard = 0x100, NoCache = 0x200, WriteCombine = 0x400, } [DllImport("kernel32.dll", SetLastError = true)] public static extern IntPtr CreateFileMapping(IntPtr hFile, IntPtr lpFileMappingAttributes, PageProtection flProtect, uint dwMaximumSizeHigh, uint dwMaximumSizeLow, string lpName); [DllImport("kernel32.dll", SetLastError = true)] public static extern IntPtr MapViewOfFile(IntPtr hFileMappingObject, uint dwDesiredAccess, uint dwFileOffsetHigh, uint dwFileOffsetLow, IntPtr dwNumberOfBytesToMap); [DllImport("kernel32.dll", SetLastError = true)] public static extern bool UnmapViewOfFile(IntPtr lpBaseAddress); [DllImport("kernel32.dll", SetLastError = true)] public static extern bool CloseHandle(IntPtr hObject); #endregion }
|
See Also See the "Mutex" and "Mutex Class" topics in the MSDN documentation and Programming Applications for Microsoft Windows, Fourth Edition. |
