Code Samples

Code Samples

This is just a short note, that the code samples have a first sample ;-).

What Are Samples?

I have deiced to create a google page to post my code samples, that are to small to be called projects and posted on codeplex or google code, but are to big to put the whole code in a post because this would obscure the content and make the post super long.

So the first sample that's there is the DataStream, that I wrote about here, its basically a decorator on a stream that can do reads and writes in a type safe way, and also has some pretty neat methods to do advanced stuff with streaming.

Stream ByteLock

Stream ByteLock

Having created an improved stream class with various much needed methods, to suit my needs for creating DataStores, I needed a specific locking mechanism that can do only section locking on a stream or array, or anything that can be divided into set of elements, so I googled and found nothing of use.

What Is It for?


Usually for problems when many threads share a stream or an array, I'm intending to use it in a scenario where there will be a FileStream with no lock on a file handle, so multiple threads can write to it, and block needed sections of the file as the file structure will be a BinaryTree, so threads will be blocking branches.

Also it's a lot faster than using a waiting queue on the whole file.

Win32 Solution

A partial solution is to use Win API function "LockFileEx" that has the following interface:

BOOL WINAPI LockFileEx(
  __in        HANDLE hFile,
  __in        DWORD dwFlags,
  __reserved  DWORD dwReserved,
  __in        DWORD nNumberOfBytesToLockLow,
  __in        DWORD nNumberOfBytesToLockHigh,
  __inout     LPOVERLAPPED lpOverlapped
);

but this needs unmanaged calls and working with the API code that has many params and it's hard to use, but on the plus side it's very fast and it's more or less a guaranteed to work as planned.

But I want to be OS independent (Database Engines are and i'm creating one and need such functionality), and besides I want 100% managed code. So having such requirements I didn't find any out of the box solution to this problem, so I had to write my own from the ground up.

My Own ByteLock

My own version is just a simple static class that uses the most basic synchronization mechanism, nothing fancy to keep the complexity of the locking code as simple as possible because writing such code in a proper manner is very VERY hard, and deadlocks, race conditions and other surprises can happen, and they are hard to find.

So let's look at the code:

public class Range
    {
        public long StartIndex { get; set; }
        public long EndIndex { get; set; }
     
        public Range(long startIndex, long endIndex)
        {
            this.StartIndex = startIndex;
            this.EndIndex = endIndex;
        }
    }

    public class RangeLock : Range
    {
        public RangeLock() : base(-1, -1)
        {
            this.ThreadId = Thread.CurrentThread.ManagedThreadId;
        }

        public RangeLock(Range range) : base(range.StartIndex, range.EndIndex)
        {
            this.ThreadId = Thread.CurrentThread.ManagedThreadId;
        }

        public int ThreadId {get; set;}
    }


    /// <summary>
    /// Byte Lock class that performs, locking on specified bytes in the stream,
    /// but can be used to perform element locking in arrays.
    /// </summary>
    public static class ByteLock
    {    
        private static List<RangeLock> sections; 
        private static object _lock = new object();

        static ByteLock()
        {
            sections = new List<RangeLock>();
        }

        /// <summary>
        /// Locks the section.
        /// </summary>
        /// <param name="range">The range.</param>
        public static void LockSection(Range range)
        {
            List<RangeLock> results = new List<RangeLock>();

            Monitor.Enter(_lock);
            {

                while (IsInSections(range.StartIndex, range.EndIndex))
                {
                    Monitor.Wait(_lock);
                }

                sections.Add(new RangeLock(range));

            }
            Monitor.Exit(_lock);
        }

        /// <summary>
        /// Writes to some stream/or array.
        /// </summary>
        /// <param name="range">The range.</param>
        /// <param name="action">The action.</param>
        public static void WriteTo(Range range, Action<Range> action)
        {
            RangeLock intersectingRange = null;

            Monitor.Enter(_lock);
            {
                intersectingRange = FindSection(range.StartIndex, range.EndIndex);
            }
            Monitor.Exit(_lock);

            if (intersectingRange != null &&
                intersectingRange.ThreadId != Thread.CurrentThread.ManagedThreadId)
            {
                Monitor.Enter(_lock);
                {
                    Monitor.Wait(_lock);
                }
                Monitor.Exit(_lock);
            }

            if (sections.
                Find(x => x.ThreadId == Thread.CurrentThread.ManagedThreadId) == null)
            {
                Monitor.Enter(_lock);
                {
                    Monitor.Pulse(_lock);
                }
                Monitor.Exit(_lock);
            }

            action(range);
        }

        /// <summary>
        /// Unlocks the section.
        /// </summary>
        /// <param name="range">The range.</param>
        public static void UnLockSection(Range range)
        {
            Monitor.Enter(_lock);
            {
                RemoveSections(range.StartIndex, range.EndIndex);
                Monitor.Pulse(_lock);
            }
            Monitor.Exit(_lock);
        }

        /// <summary>
        /// Determines whether [is in sections] [the specified start].
        /// </summary>
        /// <param name="start">The start.</param>
        /// <param name="end">The end.</param>
        /// <returns>
        ///     <c>true</c> if [is in sections] [the specified start]; otherwise, <c>false</c>.
        /// </returns>
        private static bool IsInSections(long start, long end)
        {
            foreach (var section in sections)
            {
                if (IsRangeInSection(start, end, section))
                    return true;
            }

            return false;
        }

        /// <summary>
        /// Finds the section.
        /// </summary>
        /// <param name="start">The start.</param>
        /// <param name="end">The end.</param>
        /// <returns></returns>
        private static RangeLock FindSection(long start, long end)
        {
            foreach (var section in sections)
            {
                if (IsRangeInSection(start, end, section))
                    return section;
            }

            return null;
        }

        /// <summary>
        /// Removes the sections.
        /// </summary>
        /// <param name="start">The start.</param>
        /// <param name="end">The end.</param>
        private static void RemoveSections(long start, long end)
        {
            List<RangeLock> results = new List<RangeLock>();

            foreach (var section in sections)
            {
                if (IsRangeInSection(start, end, section))
                {
                    if (section.ThreadId == Thread.CurrentThread.ManagedThreadId)
                        results.Add(section);
                }
            }
            foreach (var x in results)
            {
                sections.Remove(x);
            }
        }

        /// <summary>
        /// Determines whether [is range in section] [the specified start].
        /// </summary>
        /// <param name="start">The start.</param>
        /// <param name="end">The end.</param>
        /// <param name="section">The section.</param>
        /// <returns>
        ///     <c>true</c> if [is range in section] [the specified start]; otherwise, <c>false</c>.
        /// </returns>
        private static bool IsRangeInSection(long start, long end, RangeLock section)
        {
            if (start < section.StartIndex && end > section.StartIndex && end <= section.EndIndex)
                return true;
            if (start >= section.StartIndex && start < section.EndIndex && end > section.EndIndex)
                return true;
            if (start <= section.StartIndex && end >= section.EndIndex) return true;
            if (start >= section.StartIndex && end <= section.EndIndex) return true;

            return false;
        }
    }

As you can see from the code, the class is using an internal single lock and it's using the Wait and Pulse pattern, that's implemented in the Monitor class. The ByteLock hold a private list of RangeLocks to know who is holding what, and checks is thread wanting to write or lock a given range doesn't intersect the list of sections if he is then that thread is put into a waiting queue.

The most usual case of use will be for a thread to lock a section or wait for a lock and then execute writing or reading code on the own section. And this will not block the other threads that will use different parts of the stream. Now to do a full FileStream locking code one should also consider a situation when other processes would want to write to that file and multiple threads, so mutex locking would be desired to perform a full locking byteLock.

Sample Usage

private static DataStream dataStream = new DataStream(new MemoryStream());

        public void MethodOne(object nothing)
        {
            Console.WriteLine("Thread: " + Thread.CurrentThread.ManagedThreadId);
            Range range = new Range(0,50);

            Thread.Sleep(100);

            ByteLock.LockSection(range);
            ByteLock.WriteTo(range, (x) =>
            {
                dataStream.Position = x.StartIndex;
                dataStream.WriteBytes((int)(x.EndIndex - x.StartIndex), 100);
            });
            ByteLock.UnLockSection(range);
        }

        public void MethodTwo(object nothing)
        {
            Console.WriteLine("Thread: " + Thread.CurrentThread.ManagedThreadId);

            Thread.Sleep(50);

            Range range = new Range(50,100);
            ByteLock.WriteTo(range, (x) =>
            {
                dataStream.Position = x.StartIndex;
                dataStream.WriteBytes((int)(x.EndIndex - x.StartIndex), 200);
            });
        }

        public void MethodThree(object nothing)
        {
            Console.WriteLine("Thread: " + Thread.CurrentThread.ManagedThreadId);
            Range range = new Range(0, 50);

            ByteLock.LockSection(range);
            Thread.Sleep(100);
            ByteLock.WriteTo(range, (x) =>
            {
                dataStream.Position = x.StartIndex;
                dataStream.WriteBytes((int)(x.EndIndex - x.StartIndex), 250);
            });
            ByteLock.UnLockSection(range);
        }

        static void Main(string[] args)
        {
            Program p = new Program();

            Thread t1 = new Thread(new ParameterizedThreadStart(p.MethodOne));
            Thread t2 = new Thread(new ParameterizedThreadStart(p.MethodTwo));
            Thread t3 = new Thread(new ParameterizedThreadStart(p.MethodThree));

            t1.Start();
            t2.Start();
            t3.Start();
        }

There is nothing to explain here ;-), some basic stupid code just to show how to use.

Summing Up


This is a simple implementation of a class that is very needed but missing in Net so I thought that I should fill the gap. I will be releasing this along with the DataStream class and some other support classes as a codeplex project, but I need more classes to form a 'real' project. So for now full source code can be found in this post for ByteLock, but only partial code samples are present for DataStream as the class is super big in therms of lines of code, so either I will put it on codeplex in time or find another place to post it..

Stream Improvements

Stream Improvements

When dealing with a lot of file or memory writes and reads, it becomes apparent that the default stream implementation is a rather not a flexible one, because it lacks individual writes and reads of value types, and it's even more apparent when building a custom serialization when there is a need for writing value types of objects.

So I made my own stream wrapper to help in some more advanced stream handling, now the code for that class is simply to big to fit into the class so the whole thing including something additional that I write will land in the samples. Note that comments are removed from the code, because the code is self explanatory, and in the post it would obscure the implementation details.

I use this class for my database engine project, to do some custom object storing and header and index reading and writing, normal streams and serializators just couldn't be used in such scenario.

The Class

The class is just a wrapper on a simple stream, and it's inheriting from it, to be a more like decorator pattern style to give a possibility to further decorate it if needed.

Each method has it's write and read counterpart and it's for different value type, so this is nothing new really but rather it's very helpful in a class such as stream, the methods all return sizes for written value types (string isn't really a value type, but it handles like one so this is a special case besides it wasn't really needed to give it special treatment).

        public int Write(string data)
        {
            int length = 0;

            if (data == null)
            {
                return WriteNullString();
            }

            int byteCount = encoding.GetByteCount(data);
            length += WriteStringLength(byteCount);

            byte[] bytes = encoding.GetBytes(data);
            stream.Write(bytes, 0, bytes.Length);

            length += bytes.Length;

            return length;
        }

        public int Write(UInt32 value)
        {
            stream.Write(BitConverter.GetBytes(value), 0, 4);
            return intSize;
        }

        public int Write(UInt64 value)
        {
            stream.Write(BitConverter.GetBytes(value), 0, 8);
            return longSize;
        }

        public int Write(long value)
        {
            stream.Write(BitConverter.GetBytes(value), 0, 8);
            return longSize;
        }

By looking at the code one might wonder why those methods return read only int fields that define the size of the valueType. The reason for this is that VMs aren't strictly bound to processor architecture and implementations may differ, now I know that this may by unlikely but it is possible, and also practice shows that in NET event when something seams guaranteed it's not actually take for e.g typeof(T) and T.GetType() I will not go into details of this but know that these methods aren't the same.

this is just an example of the few of many many methods, but perhaps the most flexible and important are write and read value type, that use the underlying methods to perform type safe reads and writes.

        public int WriteValue<T>(T data)
        {
            return WriteValueType(data);
        }

        public int WriteValueType(object data)
        {
            Type dataType = data.GetType();
            int length = 0;

            if (dataType == typeof(int))
                length = Write((int)data);
            else if (dataType == typeof(long))
                length = Write((long)data);
            else if (dataType == typeof(double))
                length = Write((double)data);
            else if (dataType == typeof(float))
                length = Write((float)data);
            else if (dataType == typeof(byte))
                length = Write((byte)data);
            else if (dataType == typeof(decimal))
                length = Write((decimal)data);
            else if (dataType == typeof(uint))
                length = Write((uint)data);
            else if (dataType == typeof(char))
                length = Write((char)data);
            else if (dataType == typeof(TimeSpan))
                length = Write((TimeSpan)data);
            else if (dataType == typeof(Guid))
                length = Write((Guid)data);
            else if (dataType == typeof(DateTime))
                length = Write((DateTime)data);
            else if (dataType == typeof(byte[]))
                length = Write((byte[])data);
            else if (dataType == typeof(string))
                length = Write((string)data);
            else
                throw new DataStreamException("Error: element is not a value type.");

            return length;
        }

        public T ReadValueType<T>()
        {
            Type elemType = typeof(T);
            return (T)ReadValueType(elemType);
        }

        public object ReadValueType(Type element)
        {
            if (element == typeof(string))
                return ReadString();
            else if (element == typeof(Int32))
                return ReadInt32();
            else if (element == typeof(long))
                return ReadLong();
            else if (element == typeof(Guid))
                return ReadGuid();
            else if (element == typeof(DateTime))
                return ReadDateTime();
            else if (element == typeof(TimeSpan))
                return ReadTimeSpan();
            else if (element == typeof(float))
                return ReadFloat();
            else if (element == typeof(double))
                return ReadDouble();
            else if (element == typeof(decimal))
                return ReadDecimal();
            else if (element == typeof(char))
                return ReadChar();
            else if (element == typeof(uint))
                return ReadUInt16();

            throw new DataStreamException("Error: element is not a value type.");
        }

Another section of the class is reading by sections in the stream and stream copy (present in C# 4.0 by default).

        public int Write(Stream data)
        {
            data.Position = 0;
            byte[] streamData = new byte[data.Length];
            stream.Write(streamData, 0, streamData.Length);
            return (int)data.Length;
        }

        public IEnumerable<int> ReadSectionBytes(int sectionSize)
        {
            while(stream.Position < stream.Length)
                yield return stream.Read(new byte[sectionSize], 0, sectionSize);
        }

        public IEnumerable<byte[]> ReadSectionData(int sectionSize)
        {
            byte[] data = new byte[sectionSize];

            while (stream.Position < stream.Length)
            {
                stream.Read(data, 0, sectionSize);

                yield return data;
            }
        }

        public IEnumerable<T> ReadValueTypeSection<T>()
        {
            while (stream.Position < stream.Length)
            {
                T obj = ReadValueType<T>();
                yield return obj;
            }
        }

Reading data by section could be useful if the file has indexes or structures of given size, or to read section in a type safe manner, this is most useful when dealing with serialized objects that have only int fields etc, a common example would be reading a header index in a file that defines the layout of objects.

Improvements


This class could use some more improvements like:

More types handling.
Custom Value types writes and reads.
Creating non linear tree access to data but making it transparent to the consumer. (in progress).
Functional access, and extension methods.
Move/Append Method - this one requires additional word, such method is in progress and will definitely be a part of this class, it's purpose is to append data at a specified index of the stream, resulting in moving the data after the index, the hard part is how to do it efficiently.


Expanding upon this class


With such stream class many cool scenarios are possible, as it gives the user much grater control over the stream and it's operations, and type safety writing and reading, I felt that those operations were missing from the original implementation. The stream collection could be expanded even further by providing an indexed stream that will provide more dictionary Like access to files and in fact I'm in the process of writing one :-)

To Sum Up

I will be expanding this stream classes and after I'm done I will probably releasing it on codeplex or something.