Fibers
Note: Due to enormous length of this article I decided to split it into parts.When working with applications that utilize lots of threads and those threads arent particularry long running we can stumble upon a problem that we may lose most of the processing time for thread context switches, so it would be desirable to do all of the processing on a limited number of threads, usualy this can be done using a queue of delegates that each thread will process, but sometimes we cannot have that and still we would want to spawn units of work that behave like threads, enter Fibers.
A fiber logically is a unit of execution that behaves exacly like a thread but it's executing in the thread itself so one thread can "simulate" the execution of multiple threads, in most circumstancess fibers are not very beneficial as the cost of running then can be the same as threads and sometimes even higher, but this is not always true.
Fibers in actuality are different then threads as by design the user controls where and when to switch the context, and this must be done explicitly otherwise the context will never switch and other fibers will be stuck, manual switiching has some advantages thouh as the user can have a grater knowlede how to switch more effectivly.
Unmanaged Fibers
.NET hasn't got fiber functionality, but Windows so we can eaisy do a wrapper class arround the unmanaged fiber api like so:
/// <summary>
/// A container for all of the extern methods containing fiber logic.
/// </summary>
internal static class UnmanagedFiberAPI
{
public delegate uint LPFIBER_START_ROUTINE(uint param);
[DllImport("Kernel32.dll")]
public static extern uint ConvertThreadToFiber(uint lpParameter);
[DllImport("Kernel32.dll")]
public static extern void SwitchToFiber(uint lpFiber);
[DllImport("Kernel32.dll")]
public static extern void DeleteFiber(uint lpFiber);
[DllImport("Kernel32.dll")]
public static extern uint CreateFiber(uint dwStackSize, LPFIBER_START_ROUTINE lpStartAddress, uint lpParameter);
}
/// <summary>
/// A wrapper that provides Fiber (lightweight thread) functionality to managed code.
/// </summary>
public class Fiber
{
/// <summary>
/// The fiber action delegate.
/// </summary>
private Action action;
/// <summary>
/// Gets the fiber identifier.
/// </summary>
public uint Id { get; private set; }
/// <summary>
/// Gets the id of the primary fiber.
/// </summary>
/// <remarks>If the Id is 0 then this means that there is no primary Id on the fiber.</remarks>
public uint PrimaryId { get; private set; }
/// <summary>
/// Gets the flag identifing the primary fiber (a fiber that can run other fibers).
/// </summary>
public bool IsPrimary { get; private set; }
/// <summary>
/// Initializes a new instance of the <see cref="Fiber"/> class.
/// </summary>
/// <param name='action'>Action.</param>
public Fiber( Action action )
{
InnerCreate( action );
}
/// <summary>
/// Deletes the current fiber.
/// </summary>
/// <remarks>This method should only be used in the fiber action that's executing.</remarks>
public void Delete()
{
UnmanagedFiberAPI.DeleteFiber(Id);
}
/// <summary>
/// Deletes the fiber with the specified fiber id.
/// </summary>
/// <param name='fiberId'>fiber id.</param>
public static void Delete(uint fiberId)
{
UnmanagedFiberAPI.DeleteFiber(fiberId);
}
/// <summary>
/// Switches the execution context to the next fiber.
/// </summary>
/// <param name='fiberId'>Fiber id.</param>
public static void Switch (uint fiberId)
{
// for debug only and to show that indeed it works! Remove this line!!!
Console.WriteLine (string.Format ("Fiber [{0}] Switch", fiberId));
UnmanagedFiberAPI.SwitchToFiber(fiberId);
}
/// <summary>
/// Creates the fiber.
/// </summary>
/// <remarks>This method is responsible for the *actual* fiber creation.</remarks>
/// <param name='action'>Fiber action.</param>
private void InnerCreate (Action action)
{
this.action = action;
PrimaryId = UnmanagedFiberAPI.ConvertThreadToFiber(0);
if (PrimaryId != 0)
IsPrimary = true;
UnmanagedFiberAPI.LPFIBER_START_ROUTINE lpFiber = new UnmanagedFiberAPI.LPFIBER_START_ROUTINE(FiberRunnerProc);
Id = UnmanagedFiberAPI.CreateFiber(0, lpFiber, 0);
}
/// <summary>
/// Fiber method that executes the fiber action.
/// </summary>
/// <param name='lpParam'>Lp parameter.</param>
/// <returns>fiber status code.</returns>
private uint FiberRunnerProc(uint lpParam)
{
uint status = 0;
try
{
action();
}
catch (Exception)
{
status = 1;
throw;
}
finally
{
if (status == 1)
UnmanagedFiberAPI.DeleteFiber((uint)Id);
}
return status;
}
}
/// <summary>
/// A container for all of the extern methods containing fiber logic.
/// </summary>
internal static class UnmanagedFiberAPI
{
public delegate uint LPFIBER_START_ROUTINE(uint param);
[DllImport("Kernel32.dll")]
public static extern uint ConvertThreadToFiber(uint lpParameter);
[DllImport("Kernel32.dll")]
public static extern void SwitchToFiber(uint lpFiber);
[DllImport("Kernel32.dll")]
public static extern void DeleteFiber(uint lpFiber);
[DllImport("Kernel32.dll")]
public static extern uint CreateFiber(uint dwStackSize, LPFIBER_START_ROUTINE lpStartAddress, uint lpParameter);
}
/// <summary>
/// Fiber Scheduler that is reponsible for propper context
/// switching between the fibers, as well as the propper disposal
/// of fibers that have finished work.
/// </summary>
public static class FiberScheduler
{
private const int waitSpinLmit = 5;
private const int waitSpinTime = 30;
private static bool isSignalled;
private static readonly List<Fiber> fibers;
private static readonly object locker = new object();
private static readonly ManualResetEvent wait;
/// <summary>
/// The primary fiber Id.
/// </summary>
private static uint primaryFiberId;
/// <summary>
/// A worker thread that will be converted to the primary
/// fiber.
/// </summary>
private static readonly Thread primaryFiber;
/// <summary>
/// Initializes the FiberScheduler, which creates the primary fiber.
/// </summary>
static FiberScheduler()
{
fibers = new List<Fiber>();
wait = new ManualResetEvent(false);
primaryFiber = new Thread(ProcessFibers);
primaryFiber.Start();
}
/// <summary>
/// Adds a fiber to the fiber processing list.
/// </summary>
/// <param name="fiber">fiber.</param>
internal static void AddFiber(Fiber fiber)
{
lock (locker)
{
fibers.Add(fiber);
isSignalled = wait.Set();
}
}
/// <summary>
/// The main processing loop for the primary fiber thread, that schedules work.
/// </summary>
private static void ProcessFibers()
{
// Convert the current thread to fiber, this enables creation
// of other fibers.
primaryFiberId = UnmanagedFiberAPI.ConvertThreadToFiber(0);
int waitSpinCount = 0;
int fid = 0;
while (true)
{
if (isSignalled == true)
isSignalled = wait.Reset();
// This lock may look like a very expensive thing but in reality
// it's not as it will only lock operations from another threads and that's
// only creating or adding fibers which is not very often.
Monitor.Enter(locker);
{
if (fibers.Count > 0)
{
Fiber fiber = fibers[fid++ % fibers.Count];
if (fiber != null && (fiber.Status == FiberStatus.Started))
// switch to child fiber.
UnmanagedFiberAPI.SwitchToFiber(fiber.Id);
else if (fiber.Status == FiberStatus.Deleted)
{
/*
* Delete the fiber and free resources.
* the reason why it's here is a rule that all fibers
* must be freed in the context of the thread that they are
* running on.
*/
Fiber local = fiber;
fibers.Remove(local);
UnmanagedFiberAPI.DeleteFiber(local.Id);
}
Monitor.Exit(locker);
}
else
{
// no need to hold the lock at this point.
Monitor.Exit(locker);
if (waitSpinCount++ < waitSpinLmit)
{
//wait and spin, we wake up the thread as this is way more effective when executing
//multiple actions.
isSignalled = wait.WaitOne(waitSpinTime);
}
else
{
waitSpinCount = 0;
isSignalled = wait.WaitOne();
}
}
}
}
}
/// <summary>
/// Passes the execution to anotother fiber.
/// </summary>
/// <remarks>
/// In reality passes the execution back to the primary fiber (scheduler)
/// that schedules another fiber for execution.
/// </remarks>
public static void Yield()
{
UnmanagedFiberAPI.SwitchToFiber(primaryFiberId);
}
}
/// <summary>
/// Provides the information about the status
/// of the fiber.
/// </summary>
public enum FiberStatus
{
Error,
Started,
Done,
Deleted
}
/// <summary>
/// A wrapper that provides Fiber (lightweight thread) functionality to managed code.
/// </summary>
public class Fiber
{
/// <summary>
/// The fiber action delegate.
/// </summary>
private Action action;
/// <summary>
/// Gets the status that the current fiber is curently in.
/// </summary>
public FiberStatus Status { get; internal set; }
/// <summary>
/// Gets the fiber identifier.
/// </summary>
public uint Id { get; internal set; }
/// <summary>
/// Initializes a new instance of the <see cref="Fiber"/> class.
/// </summary>
/// <param name="action">Action.</param>
public Fiber(Action action)
{
InnerCreate(action);
}
/// <summary>
/// Schedules the current fiber for execution.
/// </summary>
public void Run()
{
Status = FiberStatus.Started;
FiberScheduler.AddFiber(this);
}
/// <summary>
/// Creates the fiber.
/// </summary>
/// <remarks>This method is responsible for the *actual* fiber creation.</remarks>
/// <param name="action">Fiber action.</param>
private void InnerCreate(Action action)
{
this.action = action;
UnmanagedFiberAPI.LPFIBER_START_ROUTINE fiberProcedure = new UnmanagedFiberAPI.LPFIBER_START_ROUTINE(FiberRunnerProc);
Id = UnmanagedFiberAPI.CreateFiber(0, fiberProcedure, 0);
}
/// <summary>
/// Fiber method that executes the fiber action.
/// </summary>
/// <param name="lpParam">Lp parameter.</param>
/// <returns>fiber status code.</returns>
private uint FiberRunnerProc(uint lpParam)
{
uint status = 0;
try
{
action();
Status = FiberStatus.Done;
}
catch (Exception)
{
status = 1;
Status = FiberStatus.Error;
throw;
}
finally
{
// before we exit we must yield to the main fiber
// and free resources.
this.Status = FiberStatus.Deleted;
FiberScheduler.Yield();
}
return status;
}
}
To prove that it actually works let's set up some sample code together:
private const int numThreads = 5;
public static void F()
{
string s = string.Empty;
for (int i = 0; i < 100; i++)
{
s = i.ToString();
Console.WriteLine("F: " + s);
FiberScheduler.Yield();
}
}
public static void F2()
{
string s = string.Empty;
for ( int i = 0; i < 100; i++ )
{
s = i.ToString( );
Console.WriteLine("F2:" + s);
FiberScheduler.Yield();
}
}
public static void Main( string[] args )
{
Fiber[] fibers = new Fiber[numThreads];
for ( int i = 0; i < numThreads; i++ )
{
if (i % 2 == 0)
fibers[i] = new Fiber(F);
else
fibers[i] = new Fiber(F2);
}
for ( int i = 0; i < numThreads; i++ )
{
fibers [i].Run( );
}
}
The downside of creating unamanaged fiber api over .NET code is non trival error handling. Ask yourself this, what will happen when the code will throw an unmanaged exception in some point? It can bring the entire process down. The next potential problem is that freeing resources in Fibers needs to be performed in the main fiber thread, thus the code for fiber deletion sits in the scheduler, attempt's to do it from another thread results in MemoryAccessViolation (at least that was my experience and that's what MSDN sorta states). The final problem with such wrapper on fiber API is that a fiber must run in a context of an unmanaged thread and .NET threads aren't simple unamanged wrappers as one managed thread can run on a single thread or many, moreover the GC collection thread can interrupt other threads, this may lead to very nasty or unexpected behaviors so while the solution works and it's usable it can also lead to some unexpected behaviors ( I have yet to confirm that ).
Managed Fibers
Instead we can create our own Fiber system using .NET, the main advantage is that the code is managed therefore we have grater control over the flow and error handling. To be able to create such system in a managed language without any low level api we eiter need to be able to delegate a switch upon some code execution in a method that will be executed in the context of a fiber, or we can use the yeild generators which actually are very simillar to fibers in a sence that they do most of the work like switching and keeping track of the execution context.
We will create the fiber model based arround yeild generators as it's easier to manage as using some context switch delegates to wrapp arrount existing code in a method can obscure the code quite a bit instead it's more desirable to slightly change the methods to be fiber friendly ( although the first option can also be exposed with yeilds to get best of both worlds ).
/// <summary>
/// Represents a fiber thread (a thread that runs multiple fibers).
/// </summary>
internal class FiberSchedulingThread
{
private readonly Thread fiberWorker;
private readonly List<Fiber> fibers;
private readonly object locker = new object();
private const int waitSpinLmit = 5;
private const int waitSpinTime = 30;
private bool isSignalled;
private readonly ManualResetEvent wait = new ManualResetEvent(false);
/// <summary>
/// Gets the Fiber Count.
/// </summary>
public int FiberCount
{
get { return fibers.Count; }
}
/// <summary>
/// Initializes a fiber thread and starts the thread loop.
/// </summary>
public FiberSchedulingThread()
{
fibers = new List<Fiber>(FiberScheduler.GetFibersPerThreadThreshold());
fiberWorker = new Thread(ProcessFibers);
fiberWorker.Start();
}
public void Schedule(Fiber fiber)
{
lock (locker)
{
fibers.Add(fiber);
isSignalled = wait.Set();
}
}
/// <summary>
/// Processes the list of fibers that are assigned to this thread.
/// </summary>
private void ProcessFibers()
{
int fiberIndex = 0;
int waitSpinCount = 0;
while (true)
{
if (isSignalled == false)
isSignalled = wait.Reset();
Monitor.Enter(locker);
{
if (fibers.Count > 0)
{
if (fiberIndex >= Int16.MaxValue)
fiberIndex = 0;
//get next fiber.
Fiber fiber = fibers[fiberIndex++ % fibers.Count];
IEnumerator<FiberStatus> fiberContext = fiber.FiberContext;
// do we need to wait a cycle ?
if (fiberContext.Current != null && fiberContext.Current.Status == ContextStatus.Wait)
{
fiberContext.Current.Status = ContextStatus.Switch;
}
else if (fiberContext.MoveNext() == false)
{
fibers.Remove(fiber);
}
// no need to hold the lock at this point.
Monitor.Exit(locker);
}
else
{
// no need to hold the lock at this point.
Monitor.Exit(locker);
if (waitSpinCount++ < waitSpinLmit)
{
//wait and spin, we wake up the thread as this is way more effective when executing
//multiple actions.
isSignalled = wait.WaitOne(waitSpinTime);
}
else
{
waitSpinCount = 0;
isSignalled = wait.WaitOne();
}
}
}
}
}
}
/// <summary>
/// A fiber scheduler that schedules fibers and assigns then
/// to fiber threads.
/// </summary>
public static class FiberScheduler
{
private static readonly object locker = new object();
private static FiberSchedulingThread currentFiber;
private static readonly FiberSchedulingThread[] fibers;
private static int fiberThreadCnt = 5;
/// <summary>
/// Initializes the fiber scheduler, creating scheduling threads.
/// </summary>
static FiberScheduler()
{
fibers = new FiberSchedulingThread[Environment.ProcessorCount];
}
public static void Schedule(Fiber fiber)
{
lock (locker)
{
// we first check if the fiber is not null and if it's less then the threashold
// and then we negate that to get the inverted logic, this is due that we need to
// do null check first.
if (!(currentFiber != null && currentFiber.FiberCount < fiberThreadCnt - 1))
{
currentFiber = new FiberSchedulingThread();
fibers[fiberThreadCnt++ % Environment.ProcessorCount] = currentFiber;
}
currentFiber.Schedule(fiber);
}
}
/// <summary>
/// Sets the amount of fibers that can run on a single scheduling thread (5 by default).
/// </summary>
/// <param name="number">number of fibers.</param>
public static void SetFibersPerThreadThreshold(int value)
{
lock (locker)
{
fiberThreadCnt = value;
}
}
/// <summary>
/// Gets the amount of fibers that can run on a single scheduling thread (5 by default).
/// </summary>
/// <returns></returns>
public static int GetFibersPerThreadThreshold()
{
return fiberThreadCnt;
}
}
/// <summary>
/// Represents a context switch status of a fiber.
/// </summary>
public enum ContextStatus
{
Switch,
Wait
}
/// <summary>
/// Represents a Fiber Status upon making a context switch.
/// </summary>
public class FiberStatus
{
/// <summary>
/// Gets the fiber context switch status.
/// </summary>
public ContextStatus Status { get; internal set; }
private FiberStatus(ContextStatus status)
{
this.Status = status;
}
/// <summary>
/// Invokes a context switch to another fiber.
/// </summary>
/// <returns>Fiber Status.</returns>
public static FiberStatus Yield()
{
return new FiberStatus(ContextStatus.Switch);
}
/// <summary>
/// Invokes a context switch with a full cycle wait.
/// </summary>
/// <remarks>
/// This is usefull for controling the priority of certain
/// fiber tasks.
/// </remarks>
/// <returns></returns>
public static FiberStatus Wait()
{
return new FiberStatus(ContextStatus.Wait);
}
}
/// <summary>
/// Represents a Fiber that's a lightweigh isolated unit of concurent
/// execution.
/// </summary>
public class Fiber
{
internal IEnumerator<FiberStatus> FiberContext;
private static int internalId = 0;
/// <summary>
/// Gets the unique fiber Id.
/// </summary>
public int Id { get; private set; }
/// <summary>
/// Initializes a new instance of a Fiber.
/// </summary>
/// <param name="fiber">fiber funcion.</param>
public Fiber(Func<IEnumerable<FiberStatus>> fiber)
{
Id = ++internalId;
this.FiberContext = fiber().GetEnumerator();
}
/// <summary>
/// Schedules fiber for execution.
/// </summary>
public void Run()
{
FiberScheduler.Schedule(this);
}
}
The managed implementation does not make use of multiple threads so it's using a mix of fiber code and thread code. When a single thread that is a primary fiber reaches a certain threshold of fibers the scheduler creates a new thread and starts to assign new fibers there, this is useful as a single thread running huge number of fibers will get slow, thus it is advised to run only five to ten fibers on a single thread (this depends largely on the case that we are in). The managed implementation introduces a second fiber state as opposed to simply Yield which is a context switch, now fibers can also wait, this may come in handy in situations where some fiber can be delayed a little in order to speed some other critical fiber tasks that are running.
Performance
In order to prove that fibers are a handy thing that has potential benefits we need to do some simple performance tests and compare them to running the same code on threads. We will not be testing it against a .NET thread pool as it uses some advanced techniques that this solution isn't so the performance will be the same or a bit slower (not much however), in part 2 of this article FiberPool will be made (not confirmed yet) and then those two pools will be compared. I will not compare the performance of the umanaged solution just yet their correctness in therms of CLR needs to be concluded first.
In order to compare finished subroutines I decided to implement a simple class called countdown, now before you will say that identical solution is present in .NET 4.0 let me stop you there, the code is written in 3.5 so custom implementation was necessary.
//very simple countdown for test purposes.
public class Countdown
{
private ManualResetEvent wait = new ManualResetEvent(false);
private int count;
private int initialCount;
public Countdown(int count) { this.initialCount = count; this.count = count; }
public void StartCounting() { wait.WaitOne(); }
public void Decrement()
{
if (Interlocked.Decrement(ref count) == 0)
wait.Set();
}
public void Reset() { this.count = initialCount; wait.Reset(); }
}
static Countdown cnt = new Countdown(100);
static int smallLoop = 100000;
static int largeLoop = 900000;
static IEnumerable<FiberStatus> F()
{
string s = string.Empty;
for (int i = 0; i < smallLoop ; i++)
{
s = i.ToString();
yield return FiberStatus.Yield();
}
cnt.Decrement();
}
static IEnumerable<FiberStatus> F2()
{
string s = string.Empty;
for (int i = 0; i < largeLoop; i++)
{
s = i.ToString();
yield return FiberStatus.Yield();
}
cnt.Decrement();
}
static void T()
{
string s = string.Empty;
for (int i = 0; i < smallLoop; i++)
s = i.ToString();
cnt.Decrement();
}
static void T2()
{
string s = string.Empty;
for (int i = 0; i < largeLoop; i++)
s = i.ToString();
cnt.Decrement();
}
static void TestThreads()
{
Thread[] threads = new Thread[100];
Stopwatch w = new Stopwatch();
for (int i = 0; i < 100; i++)
{
if (i % 2 == 0)
threads[i] = new Thread(T);
else
threads[i] = new Thread(T2);
}
w.Start();
for (int i = 0; i < 100; i++)
{
threads[i].Start();
threads[i].Join();
}
cnt.StartCounting();
w.Stop();
Console.WriteLine("Threads took: {0}", w.ElapsedMilliseconds);
}
static void TestFibers()
{
Fiber[] fibers = new Fiber[100];
Stopwatch w = new Stopwatch();
for (int i = 0; i < 100; i++)
{
if (i % 2 == 0)
fibers[i] = new Fiber(F);
else
fibers[i] = new Fiber(F2);
}
w.Start();
for (int i = 0; i < 100; i++)
{
fibers[i].Run();
}
cnt.StartCounting();
w.Stop();
Console.WriteLine("Fibers took: {0}", w.ElapsedMilliseconds);
}
The test code assumes that some tasks will execute longer then others, as you can see we are testing it on a fairly large number of concurrent tasks as for low concurrency systems you can very easily get away with threads (there is little sense to use fibers if the maximum concurrent load is below NumOfCores * (2 | 4), I do speak from my personal experience though). Please do keep in mind that fibers actually can and will be slower then threads in certain class of problems like very few long lived tasks, or huge number of very short tasks ( yes they will probably be slower ), so this test does not conclude that fibers are best in every given scenario rather then that it concludes that in normal concurrency circumstances (mid lived tasks) they will be faster.
The speeds are as follows:
Summary
Fibers can be a very handy tool for solving concurrency performance issues that are based around expensive thread context switches. This article is the first part of the whole series on fibers as there's much more to topics to be touched like work stealing fibers, and fiber resource locking.
Full source code will be available soon, (i will update this article and put a link here).
4 comments:
Isn't this essentially what the threadpool is for? Do you fibers have an advantage over the threadpool?
jeremywho
A threadpool is not made to decrease the number or cost of context switches between threads, it's to decrease the cost of creating and starting new threads, as well as distributing work (4.0 work stealing) so this is a totally different concept.
Fibers can be considered lightweight threads, they are light in a sense that they actually do in code context switches so they don't have to go to kernel scheduler, which is a lot less expensive, their start-up time is also faster. So Fibers cannot be compared to a thread pool, a fiber can only be compared to a thread, you would need to create a FiberPool and that you could compare to a threadpool (which will be implemented in another part about fibers).
I hope that this explanation cleared things up a bit.
Thanks for the comment.
Hi
I think the test for threads it's not valid:
for (int i = 0; i < 100; i++)
{
threads[i].Start();
threads[i].Join();
}
This piece of code, will have only one thread running concurrently. No?
@José Luis Pedrosa.
Yes you are 100% correct the join code should be in it's own loop.
I did the tests again (on a different machine) and times are as follows:
Threads: 4451
Fibers: 3203
So fibers run faster and everything in the blog post along with conclusions still holds true.
Thank you for your contribution :) I'll update the post some time later.
Post a Comment