Mercurial > projects > dreactor

comparison dreactor/core/Vat.d @ 5:f875a1f278b8

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
housekeeping
author rick@minifunk
date Tue, 08 Jul 2008 12:16:07 -0400
parents dreactor/core/SelectLoop.d@e3dbc9208822
children 287ba7de97c4
comparison
equal deleted inserted replaced
4:f8b01c9f7114 5:f875a1f278b8
1 /*******************************************************************************
2
3 copyright: Copyright (c) 2008 Rick Richardson. All rights reserved
4
5 license: BSD style: $(LICENSE)
6
7 version: Initial release v0.1 : May 2008
8
9 author: Rick Richardson
10
11 *******************************************************************************/
12
13 module dreactor.core.Vat;
14
15 import tango.io.selector.Selector;
16 import tango.io.selector.model.ISelector;
17 import tango.core.Exception;
18 import tango.core.Thread;
19 import tango.core.Atomic;
20 import tango.util.collection.LinkSeq;
21 import tango.util.log.Log;
22
23 import dreactor.transport.AsyncSocketConduit;
24 import dreactor.core.ConnectionHandler;
25 import dreactor.util.ThreadSafeQueue;
26
27 Logger log;
28
29 enum : int {UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2};
30
31 static char[] version_string = "Vat.d 0.1 2008-05-31";
32
33 class Vat
34 {
35 private
36 Thread thread;
37 bool running;
38 Atomic!(int) pending;
39
40 ThreadSafeQueue!(ConnectionHandler) freshList;
41 ThreadSafeQueue!(ConnectionHandler) remList;
42 public
43 this()
44 {
45 freshList = new ThreadSafeQueue!(ConnectionHandler);
46 remList = new ThreadSafeQueue!(ConnectionHandler);
47 log = Log.lookup("dreactor.core.Vat");
48 }
49
50 void run()
51 {
52 running = true;
53 thread = new Thread(&eventLoop);
54 thread.start();
55 }
56
57 void exit()
58 {
59 running = false;
60 }
61
62 bool addConnection(ConnectionHandler handler)
63 {
64 log.trace("adding handler");
65 return freshList.push(handler);
66 }
67
68 bool remConnection(ConnectionHandler handler)
69 {
70 return remList.push(handler);
71 }
72
73 private
74 void eventLoop()
75 {
76 auto selector = new Selector();
77 selector.open();
78 do
79 {
80 auto eventCount = selector.select(0.01);
81
82 if (eventCount > 0)
83 {
84 // process events
85 foreach (SelectionKey key; selector.selectedSet())
86 {
87 if (key.isReadable())
88 {
89 // incoming data
90 log.trace("Read event fired");
91 auto conn = cast(ConnectionHandler) key.attachment;
92 if ( ConnectionHandler.State.listening == conn.getState() )
93 conn.handleConnection(conn.transport, &addConnection);
94 else
95 processReturn(conn.handleIncoming(), selector, conn);
96 }
97 else if (key.isWritable())
98 {
99 log.trace("Write event fired");
100 auto conn = cast(ConnectionHandler) key.attachment;
101 processReturn(conn.handleOutgoing(), selector, conn);
102 }
103 else if (key.isHangup())
104 {
105 log.trace("Hangup event fired");
106 auto conn = cast(ConnectionHandler) key.attachment;
107 processReturn(conn.handleDisconnect(), selector, conn);
108 }
109 else if (key.isError() || key.isInvalidHandle())
110 {
111 log.trace("Error event fired");
112 // error, close connection
113 auto conn = cast(ConnectionHandler) key.attachment;
114 conn.handleError(&remConnection);
115 }
116 }
117 }
118 else if (eventCount == 0)
119 {
120 /* can't think of anything useful to do here. */
121 }
122 else
123 {
124 log.error("Selector.select returned {}", eventCount);
125 }
126 //add Conduits to listener
127 freshList.processAll( (ref ConnectionHandler h)
128 {
129 selector.reregister(h.transport, h.events(), h);
130 return 1;
131 });
132 remList.processAll( (ref ConnectionHandler h)
133 {
134 selector.unregister(h.transport);
135 return 1;
136 });
137
138 } while (running)
139
140 }
141
142 void processReturn(int result, Selector s, ConnectionHandler h)
143 {
144 switch(result)
145 {
146 case UNREGISTER:
147 s.unregister(h.transport);
148 break;
149 case REMAIN:
150 //this space intentially left blank
151 break;
152 case REGISTER:
153 s.register(h.transport, h.events(), h);
154 break;
155 case REREGISTER:
156 s.reregister(h.transport, h.events(), h);
157 break;
158 default:
159 log.error("processReturn: unknown return value");
160 }
161 }
162 }