使用C#的Generator编写并发程序
前面说地Generator编写并发程序的优势，当然它本身没有这种能力，需要为它编写调度程序。今天抽点时间写了个简单的，还是满好玩的，它可以调度多个“友好”的并发任务，包括：
1、用户自己编写的适时交出控制权的过程
2、网络IO
由于对C#不是很熟，所以写的可能比较难看，而且IO也只支持网络，有兴趣的可以研究改进一下，比如把它改成SMP版本，增加Actor模型，处理更多的IO模式等。
实现方式基本上是从IoLanguage里面抄过来的，我已经用它编写过Ruby/C++/D/C#版本，当然目前仅限于测试它的切换性能。
好消息是如果你实现得比较好，C#版本性能完全可以超过Erlang，Ruby的Fiber就比较差一些，C++/D里面使用Fiber/ucontext性能也不是很好，Generator性能是最好的。
由于时间短，只测试了ReadEvent，WriteEvent没有测试，这个也是有多种实现方式可以研究的。这里实现的调度器性能不是很好，可能是List用法性能太差，没有仔细测试。
using System;using System.Threading;using System.Collections;using System.Collections.Generic;using System.Net.Sockets;using System.Net;namespace ConsoleApplication1{ public enum EventType { ReadEvent = 1, WriteEvent = 2 } public class Event { private Socket sock; private EventType eventType; private IEnumerator<int> task; public Event(Socket sock, EventType eventType) { this.sock = sock; this.eventType = eventType; } public Socket getSocket() { return sock; } public EventType getEventType() { return eventType; } public void setTask(IEnumerator<int> task) { this.task = task; } public IEnumerator<int> getTask() { return task; } public void onEvent() { Console.WriteLine("onEvent"); Scheduler.scheduleNow(task); } } public class ReadEvent : Event { public ReadEvent(Socket sock) :base(sock, EventType.ReadEvent) { } } public class WriteEvent : Event { public WriteEvent(Socket sock) : base(sock, EventType.WriteEvent) { } } class EventManager { private IDictionary<Socket, Event> readEvents = new Dictionary<Socket, Event>(); private IDictionary<Socket, Event> writeEvents = new Dictionary<Socket, Event>(); private IEnumerator<int> myTask = null; public int registerHandler(Event ev) { Console.WriteLine("registerHandler"); if ((ev.getEventType() & EventType.ReadEvent) > 0) { readEvents.Add(ev.getSocket(), ev); } else { writeEvents.Add(ev.getSocket(), ev); } if (myTask == null) myTask = runLoop(); Scheduler.scheduleNow(myTask); return 0; } public IEnumerator<int> runLoop() { Console.WriteLine("enter run"); do { Console.WriteLine("run loop"); yield return 1; IList rlist = new List<Socket>(readEvents.Keys); IList wlist = new List<Socket>(writeEvents.Keys); int timeoutMS = 1000000; if (Scheduler.hasEvents()) timeoutMS = 0; //Console.WriteLine("begin select: " + rlist.Count + ", " + wlist.Count); Socket.Select(rlist, wlist, null, timeoutMS); //Console.WriteLine("end select: " + rlist.Count + ", " + wlist.Count); Socket sock = null; Event ev = null; if (rlist.Count > 0) { sock = (Socket)rlist; readEvents.TryGetValue(sock, out ev); readEvents.Remove(sock); } else if (wlist.Count > 0) { sock = (Socket)wlist; writeEvents.TryGetValue(sock, out ev); writeEvents.Remove(sock); } if (ev != null) ev.onEvent(); } while (readEvents.Count != 0 || writeEvents.Count != 0); myTask = null; } } public static class Scheduler { public delegate IEnumerable<int> T(); private static IList<IEnumerator<int>> tasks = new List<IEnumerator<int>>(); private static EventManager eventManager = new EventManager(); private static Event lastEvent; public static bool hasEvents() { return tasks.Count > 0; } public static void spawn(T del) { var task = del().GetEnumerator(); schedule(task); } public static void schedule(IEnumerator<int> task) { tasks.Remove(task); if (lastEvent != null) lastEvent.setTask(task); else tasks.Add(task); lastEvent = null; } public static void scheduleNow(IEnumerator<int> task) { tasks.Remove(task); tasks.Insert(0, task); } public static void remove(IEnumerator<int> task) { Console.WriteLine("remove task"); tasks.Remove(task); } public static void run() { while (tasks.Count > 0) { IEnumerator<int> task = tasks; tasks.RemoveAt(0); bool ret = task.MoveNext(); if (ret) schedule(task); } } public static int registerAndWaitEvent(Event ev) { lastEvent = ev; return eventManager.registerHandler(ev); } } // 测试程序： class Program { public static IEnumerable<int> test(string id) { for (int i = 0; i < 5; i++) { System.Console.WriteLine("id: " + id + ", " +i); yield return i; } } public static IEnumerable<int> clientLoop(Socket client) { while (true) { Console.WriteLine("client loop"); yield return Scheduler.registerAndWaitEvent(new ReadEvent(client)); byte[] buffer = new byte; int ret = client.Receive(buffer); if (ret > 0) { //Console.WriteLine(buffer); client.Send(buffer); } else { client.Close(); break; } } } public static IEnumerable<int> serverLoop() { Socket sock = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); IPHostEntry IPHost = Dns.GetHostEntry("localhost"); sock.Bind(new IPEndPoint(IPHost.AddressList, 23456)); sock.Listen(128); while(true) { Console.WriteLine("server loop"); yield return Scheduler.registerAndWaitEvent(new ReadEvent(sock)); Console.WriteLine("accept event"); Socket client = sock.Accept(); Scheduler.spawn(() => clientLoop(client)); } } static void Main(string[] args) { Scheduler.spawn(() => test("a")); Scheduler.spawn(() => test("b")); Scheduler.spawn(() => serverLoop()); Scheduler.run(); } }}
页:
[1]