Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(525)

Side by Side Diff: runtime/lib/isolate_patch.dart

Issue 27215002: Very simple version of Isolates. (Closed) Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart
Patch Set: Address comments. Created 7 years, 1 month ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch | Annotate | Revision Log
« no previous file with comments | « runtime/lib/isolate.cc ('k') | runtime/tests/vm/dart/isolate_mirror_local_test.dart » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file
2 // for details. All rights reserved. Use of this source code is governed by a 2 // for details. All rights reserved. Use of this source code is governed by a
3 // BSD-style license that can be found in the LICENSE file. 3 // BSD-style license that can be found in the LICENSE file.
4 4
5 class _CloseToken { 5 patch class ReceivePort {
6 /// This token is sent from [IsolateSink]s to [IsolateStream]s to ask them to 6 /* patch */ factory ReceivePort() = _ReceivePortImpl;
7 /// close themselves. 7
8 const _CloseToken(); 8 /* patch */ factory ReceivePort.fromRawReceivePort(RawReceivePort rawPort) =
9 _ReceivePortImpl.fromRawReceivePort;
9 } 10 }
10 11
11 patch bool _isCloseToken(var object) { 12 patch class RawReceivePort {
12 // TODO(floitsch): can we compare against const _CloseToken()? 13 /**
13 return object is _CloseToken; 14 * Opens a long-lived port for receiving messages.
14 } 15 *
15 16 * A [RawReceivePort] is low level and does not work with [Zone]s. It
16 patch class MessageBox { 17 * can not be paused. The data-handler must be set before the first
17 /* patch */ MessageBox.oneShot() : this._oneShot(new ReceivePort()); 18 * event is received.
18 MessageBox._oneShot(ReceivePort receivePort) 19 */
19 : stream = new IsolateStream._fromOriginalReceivePortOneShot(receivePort), 20 /* patch */ factory RawReceivePort([void handler(event)]) {
20 sink = new _IsolateSink._fromPort(receivePort.toSendPort()); 21 _RawReceivePortImpl result = new _RawReceivePortImpl();
21 22 result.handler = handler;
22 /* patch */ MessageBox() : this._(new ReceivePort()); 23 return result;
23 MessageBox._(ReceivePort receivePort)
24 : stream = new IsolateStream._fromOriginalReceivePort(receivePort),
25 sink = new _IsolateSink._fromPort(receivePort.toSendPort());
26 }
27
28 class _IsolateSink implements IsolateSink {
29 bool _isClosed = false;
30 final SendPort _port;
31 _IsolateSink._fromPort(this._port);
32
33 void add(dynamic message) {
34 _port.send(message);
35 }
36
37 void addError(Object errorEvent) {
38 throw new UnimplementedError("addError on isolate streams");
39 }
40
41 void close() {
42 if (_isClosed) return;
43 add(const _CloseToken());
44 _isClosed = true;
45 }
46
47 bool operator==(var other) {
48 return other is IsolateSink && _port == other._port;
49 }
50
51 int get hashCode => _port.hashCode + 499;
52 }
53
54 patch IsolateSink streamSpawnFunction(
55 void topLevelFunction(),
56 [bool unhandledExceptionCallback(IsolateUnhandledException e)]) {
57 SendPort sendPort = spawnFunction(topLevelFunction,
58 unhandledExceptionCallback);
59 return new _IsolateSink._fromPort(sendPort);
60 }
61
62 patch class ReceivePort {
63 /* patch */ factory ReceivePort() {
64 return new _ReceivePortImpl();
65 } 24 }
66 } 25 }
67 26
68 class _ReceivePortImpl implements ReceivePort { 27 class _ReceivePortImpl extends Stream implements ReceivePort {
69 factory _ReceivePortImpl() native "ReceivePortImpl_factory"; 28 _ReceivePortImpl() : this.fromRawReceivePort(new RawReceivePort());
70 29
71 receive(void onMessage(var message, SendPort replyTo)) { 30 _ReceivePortImpl.fromRawReceivePort(this._rawPort) {
72 _onMessage = onMessage; 31 _controller = new StreamController(onCancel: close, sync: true);
32 _rawPort.handler = _controller.add;
73 } 33 }
74 34
35 SendPort get sendPort {
36 return _rawPort.sendPort;
37 }
38
39 StreamSubscription listen(void onData(var message),
40 { Function onError,
41 void onDone(),
42 bool cancelOnError }) {
43 return _controller.stream.listen(onData,
44 onError: onError,
45 onDone: onDone,
46 cancelOnError: cancelOnError);
47 }
48
49 close() {
50 _rawPort.close();
51 _controller.close();
52 }
53
54 final RawReceivePort _rawPort;
55 StreamController _controller;
56 }
57
58 class _RawReceivePortImpl implements RawReceivePort {
59 factory _RawReceivePortImpl() native "RawReceivePortImpl_factory";
60
75 close() { 61 close() {
76 _portMap.remove(_id); 62 _portMap.remove(_id);
77 _closeInternal(_id); 63 _closeInternal(_id);
78 } 64 }
79 65
80 SendPort toSendPort() { 66 SendPort get sendPort {
81 return new _SendPortImpl(_id); 67 return new _SendPortImpl(_id);
82 } 68 }
83 69
84 /**** Internal implementation details ****/ 70 /**** Internal implementation details ****/
85 // Called from the VM to create a new ReceivePort instance. 71 // Called from the VM to create a new RawReceivePort instance.
86 static _ReceivePortImpl _get_or_create(int id) { 72 static _RawReceivePortImpl _get_or_create(int id) {
87 if (_portMap != null) { 73 _RawReceivePortImpl port = _portMap[id];
88 _ReceivePortImpl port = _portMap[id]; 74 if (port != null) {
89 if (port != null) { 75 return port;
90 return port;
91 }
92 } 76 }
93 return new _ReceivePortImpl._internal(id); 77 return new _RawReceivePortImpl._internal(id);
94 } 78 }
95 79
96 _ReceivePortImpl._internal(int id) : _id = id { 80 _RawReceivePortImpl._internal(int id) : _id = id {
97 if (_portMap == null) {
98 _portMap = new Map();
99 }
100 _portMap[id] = this; 81 _portMap[id] = this;
101 } 82 }
102 83
103 // Called from the VM to retrieve the ReceivePort for a message. 84 // Called from the VM to retrieve the RawReceivePort for a message.
104 static _ReceivePortImpl _lookupReceivePort(int id) { 85 static _RawReceivePortImpl _lookupReceivePort(int id) {
105 assert(_portMap != null);
106 return _portMap[id]; 86 return _portMap[id];
107 } 87 }
108 88
109 // Called from the VM to dispatch to the handler. 89 // Called from the VM to dispatch to the handler.
110 static void _handleMessage(_ReceivePortImpl port, int replyId, var message) { 90 static void _handleMessage(
91 _RawReceivePortImpl port, int replyId, var message) {
111 assert(port != null); 92 assert(port != null);
112 SendPort replyTo = (replyId == 0) ? null : new _SendPortImpl(replyId); 93 port._handler(message);
113 (port._onMessage)(message, replyTo);
114 } 94 }
115 95
116 // Call into the VM to close the VM maintained mappings. 96 // Call into the VM to close the VM maintained mappings.
117 static _closeInternal(int id) native "ReceivePortImpl_closeInternal"; 97 static _closeInternal(int id) native "RawReceivePortImpl_closeInternal";
98
99 void set handler(Function newHandler) {
100 this._handler = newHandler;
101 }
118 102
119 final int _id; 103 final int _id;
120 var _onMessage; 104 Function _handler;
121 105
122 // id to ReceivePort mapping. 106 // id to RawReceivePort mapping.
123 static Map _portMap; 107 static final Map _portMap = new HashMap();
124 } 108 }
125 109
126 110
127 class _SendPortImpl implements SendPort { 111 class _SendPortImpl implements SendPort {
128 /*--- public interface ---*/ 112 /*--- public interface ---*/
129 void send(var message, [SendPort replyTo = null]) { 113 void send(var message, [SendPort replyTo = null]) {
130 this._sendNow(message, replyTo); 114 this._sendNow(message, replyTo);
131 } 115 }
132 116
133 void _sendNow(var message, SendPort replyTo) { 117 void _sendNow(var message, SendPort replyTo) {
134 int replyId = (replyTo == null) ? 0 : replyTo._id; 118 int replyId = (replyTo == null) ? 0 : replyTo._id;
135 _sendInternal(_id, replyId, message); 119 _sendInternal(_id, replyId, message);
136 } 120 }
137 121
138 Future call(var message) {
139 final completer = new Completer.sync();
140 final port = new _ReceivePortImpl();
141 send(message, port.toSendPort());
142 port.receive((value, ignoreReplyTo) {
143 port.close();
144 if (value is Exception) {
145 completer.completeError(value);
146 } else {
147 completer.complete(value);
148 }
149 });
150 return completer.future;
151 }
152
153 bool operator==(var other) { 122 bool operator==(var other) {
154 return (other is _SendPortImpl) && _id == other._id; 123 return (other is _SendPortImpl) && _id == other._id;
155 } 124 }
156 125
157 int get hashCode { 126 int get hashCode {
158 return _id; 127 const int MASK = 0x3FFFFFFF;
128 int hash = _id;
129 hash = (hash + ((hash & (MASK >> 10)) << 10)) & MASK;
130 hash ^= (hash >> 6);
131 hash = (hash + ((hash & (MASK >> 3)) << 3)) & MASK;
132 hash ^= (hash >> 11);
133 hash = (hash + ((hash & (MASK >> 15)) << 15)) & MASK;
134 return hash;
159 } 135 }
160 136
161 /*--- private implementation ---*/ 137 /*--- private implementation ---*/
162 const _SendPortImpl(int id) : _id = id; 138 const _SendPortImpl(int id) : _id = id;
163 139
164 // _SendPortImpl._create is called from the VM when a new SendPort instance is 140 // _SendPortImpl._create is called from the VM when a new SendPort instance is
165 // needed by the VM code. 141 // needed by the VM code.
166 static SendPort _create(int id) { 142 static SendPort _create(int id) {
167 return new _SendPortImpl(id); 143 return new _SendPortImpl(id);
168 } 144 }
169 145
170 // Forward the implementation of sending messages to the VM. Only port ids 146 // Forward the implementation of sending messages to the VM. Only port ids
171 // are being handed to the VM. 147 // are being handed to the VM.
172 static _sendInternal(int sendId, int replyId, var message) 148 static _sendInternal(int sendId, int replyId, var message)
173 native "SendPortImpl_sendInternal_"; 149 native "SendPortImpl_sendInternal_";
174 150
175 final int _id; 151 final int _id;
176 } 152 }
177 153
178 _getPortInternal() native "isolate_getPortInternal"; 154 _getPortInternal() native "isolate_getPortInternal";
179 155
180 ReceivePort _portInternal; 156 typedef _MainFunction();
157 typedef _MainFunctionArgs(args);
158 typedef _MainFunctionArgsMessage(args, message);
181 159
182 patch class _Isolate { 160 /**
183 /* patch */ static ReceivePort get port { 161 * Takes the real entry point as argument and invokes it with the initial
184 if (_portInternal == null) { 162 * message.
185 _portInternal = _getPortInternal(); 163 *
164 * The initial message is (currently) received through the global port variable.
165 */
166 void _startIsolate(Function entryPoint, bool isSpawnUri) {
167 Isolate._port.first.then((message) {
168 SendPort replyTo = message[0];
169 // TODO(floitsch): don't send ok-message if we can't find the entry point.
170 replyTo.send("started");
171 if (isSpawnUri) {
172 assert(message.length == 3);
173 List<String> args = message[1];
174 var isolateMessage = message[2];
175 if (entryPoint is _MainFunctionArgsMessage) {
176 entryPoint(args, isolateMessage);
177 } else if (entryPoint is _MainFunctionArgs) {
178 entryPoint(args);
179 } else {
180 entryPoint();
181 }
182 } else {
183 assert(message.length == 2);
184 var entryMessage = message[1];
185 entryPoint(entryMessage);
186 } 186 }
187 return _portInternal; 187 });
188 }
189
190 patch class Isolate {
191 /* patch */ static Future<Isolate> spawn(
192 void entryPoint(message), var message) {
193 Completer completer = new Completer<Isolate>.sync();
194 try {
195 // The VM will invoke [_startIsolate] with entryPoint as argument.
196 SendPort controlPort = _spawnFunction(entryPoint);
197 RawReceivePort readyPort = new RawReceivePort();
198 controlPort.send([readyPort.sendPort, message]);
199 readyPort.handler = (readyMessage) {
200 assert(readyMessage == 'started');
201 readyPort.close();
202 completer.complete(new Isolate._fromControlPort(controlPort));
203 };
204 } catch(e, st) {
205 // TODO(floitsch): we want errors to go into the returned future.
206 rethrow;
207 };
208 return completer.future;
188 } 209 }
189 210
190 /* patch */ static SendPort spawnFunction(void topLevelFunction(), 211 /* patch */ static Future<Isolate> spawnUri(
191 [bool unhandledExceptionCallback(IsolateUnhandledException e)]) 212 Uri uri, List<String> args, var message) {
213 Completer completer = new Completer<Isolate>.sync();
214 try {
215 // The VM will invoke [_startIsolate] and not `main`.
216 SendPort controlPort = _spawnUri(uri.path);
217 RawReceivePort readyPort = new RawReceivePort();
218 controlPort.send([readyPort.sendPort, args, message]);
219 readyPort.handler = (readyMessage) {
220 assert(readyMessage == 'started');
221 readyPort.close();
222 completer.complete(new Isolate._fromControlPort(controlPort));
223 };
224 } catch(e, st) {
225 // TODO(floitsch): we want errors to go into the returned future.
226 rethrow;
227 };
228 return completer.future;
229 }
230
231 static final ReceivePort _port =
232 new ReceivePort.fromRawReceivePort(_getPortInternal());
233
234 static SendPort _spawnFunction(Function topLevelFunction)
192 native "isolate_spawnFunction"; 235 native "isolate_spawnFunction";
193 236
194 /* patch */ static SendPort spawnUri(String uri) native "isolate_spawnUri"; 237 static SendPort _spawnUri(String uri) native "isolate_spawnUri";
195 } 238 }
OLDNEW
« no previous file with comments | « runtime/lib/isolate.cc ('k') | runtime/tests/vm/dart/isolate_mirror_local_test.dart » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698