|
132
|
1 /*******************************************************************************
|
|
|
2
|
|
|
3 copyright: Copyright (c) 2004 Kris Bell. All rights reserved
|
|
|
4
|
|
|
5 license: BSD style: $(LICENSE)
|
|
|
6
|
|
|
7 version: July 2004: Initial release
|
|
|
8
|
|
|
9 author: Kris
|
|
|
10
|
|
|
11 *******************************************************************************/
|
|
|
12
|
|
|
13 module tango.net.cluster.tina.Cluster;
|
|
|
14
|
|
|
15 private import tango.math.Random;
|
|
|
16
|
|
|
17 private import tango.core.Thread,
|
|
|
18 tango.core.Runtime,
|
|
|
19 tango.core.Exception;
|
|
|
20
|
|
|
21 private import tango.util.log.Log,
|
|
|
22 tango.util.log.Logger;
|
|
|
23
|
|
|
24 private import tango.time.Clock;
|
|
|
25
|
|
|
26 private import tango.io.Buffer,
|
|
|
27 tango.io.GrowBuffer;
|
|
|
28
|
|
|
29 private import tango.io.model.IConduit;
|
|
|
30
|
|
|
31 private import tango.net.Socket,
|
|
|
32 tango.net.SocketConduit,
|
|
|
33 tango.net.SocketListener,
|
|
|
34 tango.net.InternetAddress,
|
|
|
35 tango.net.MulticastConduit;
|
|
|
36
|
|
|
37 private import tango.net.cluster.NetworkClient;
|
|
|
38
|
|
|
39 public import tango.net.cluster.model.ICluster;
|
|
|
40
|
|
|
41 private import tango.net.cluster.tina.RollCall,
|
|
|
42 tango.net.cluster.tina.ProtocolReader,
|
|
|
43 tango.net.cluster.tina.ProtocolWriter;
|
|
|
44
|
|
|
45 private import Integer = tango.text.convert.Integer;
|
|
|
46
|
|
|
47 /*******************************************************************************
|
|
|
48
|
|
|
49 QOS implementation for sockets. All cluster-client activity is
|
|
|
50 gated through here by the higher level classes; NetworkQueue &
|
|
|
51 NetworkCache for example. You gain access to the cluster by
|
|
|
52 creating an instance of the QOS (quality of service) you desire
|
|
|
53 and either mapping client classes onto it, or usign it directly.
|
|
|
54 For example:
|
|
|
55 ---
|
|
|
56 import tango.net.cluster.tina.Cluster;
|
|
|
57
|
|
|
58 auto cluster = new Cluster;
|
|
|
59 cluster.join;
|
|
|
60
|
|
|
61 auto channel = cluster.createChannel (...);
|
|
|
62 channel.putQueue (...);
|
|
|
63 channel.getQueue ();
|
|
|
64 ---
|
|
|
65
|
|
|
66 Please see the cluster clients for additional details. Currently
|
|
|
67 these include CacheInvalidator, CacheInvalidatee, NetworkMessage,
|
|
|
68 NetworkTask, NetworkQueue, NetworkCache, NetworkCombo, plus the
|
|
|
69 Client base-class.
|
|
|
70
|
|
|
71 *******************************************************************************/
|
|
|
72
|
|
|
73 class Cluster : Broadcaster, ICluster
|
|
|
74 {
|
|
|
75 private FlexNodeSet task,
|
|
|
76 queue;
|
|
|
77 private FixedNodeSet cache;
|
|
|
78 private Logger logger;
|
|
|
79
|
|
|
80 /***********************************************************************
|
|
|
81
|
|
|
82 Create a cluster instance with a default logger and Nagle
|
|
|
83 caching disabled
|
|
|
84
|
|
|
85 ***********************************************************************/
|
|
|
86
|
|
|
87 this ()
|
|
|
88 {
|
|
|
89 this (Log.getLogger ("cluster.generic"), true);
|
|
|
90 }
|
|
|
91
|
|
|
92 /***********************************************************************
|
|
|
93
|
|
|
94 Create a cluster instance with the provided logger. Option
|
|
|
95 noDelay controls the settting of the Nagle algorithm on an
|
|
|
96 active connection to a server, which should be disabled by
|
|
|
97 default (noDelay == true)
|
|
|
98
|
|
|
99 ***********************************************************************/
|
|
|
100
|
|
|
101 this (Logger log, bool noDelay = true)
|
|
|
102 {
|
|
|
103 assert (log);
|
|
|
104 logger = log;
|
|
|
105
|
|
|
106 task = new FlexNodeSet (log, noDelay);
|
|
|
107 queue = new FlexNodeSet (log, noDelay);
|
|
|
108 cache = new FixedNodeSet (log, noDelay);
|
|
|
109 }
|
|
|
110
|
|
|
111 /***********************************************************************
|
|
|
112
|
|
|
113 Join the cluster as a client, discovering servers. Client
|
|
|
114 applications should invoke this before making requests so
|
|
|
115 that there are some servers to address.
|
|
|
116
|
|
|
117 If cache facilities will be used, then the join(cacheHosts)
|
|
|
118 variation should be used instead
|
|
|
119
|
|
|
120 ***********************************************************************/
|
|
|
121
|
|
|
122 final Cluster join ()
|
|
|
123 {
|
|
|
124 // listen for cluster servers
|
|
|
125 auto channel = createChannel ("cluster.server.advertise");
|
|
|
126 channel.createBulletinConsumer (¬ify);
|
|
|
127
|
|
|
128 // ask who's currently running
|
|
|
129 channel.broadcast (new RollCall);
|
|
|
130 logger.trace ("discovering cluster nodes");
|
|
|
131
|
|
|
132 // wait for enabled servers to respond ...
|
|
|
133 Thread.sleep (0.250);
|
|
|
134 return this;
|
|
|
135 }
|
|
|
136
|
|
|
137 /***********************************************************************
|
|
|
138
|
|
|
139 Join the cluster as a client, discovering servers. Client
|
|
|
140 applications should invoke this before making requests so
|
|
|
141 that there are some servers to address.
|
|
|
142
|
|
|
143 If cache facilities will be used, use this method to set
|
|
|
144 the group of valid cache hosts. Each cache host should be
|
|
|
145 described as an array of machine-name and port pairs e.g.
|
|
|
146 ---
|
|
|
147 ["lucy:1234", "daisy:3343", "daisy:3344"]
|
|
|
148 ---
|
|
|
149
|
|
|
150 This sets up a fixed set of cache hosts, which should be
|
|
|
151 identical for all cache clients. Cache hosts not included
|
|
|
152 in this list will be ignored when they come online.
|
|
|
153
|
|
|
154 ***********************************************************************/
|
|
|
155
|
|
|
156 final Cluster join (char[][] cacheHosts)
|
|
|
157 {
|
|
|
158 foreach (addr; cacheHosts)
|
|
|
159 cache.addNode (new Node (log, addr, "cache"));
|
|
|
160 return join;
|
|
|
161 }
|
|
|
162
|
|
|
163 /***********************************************************************
|
|
|
164
|
|
|
165 Return the logger instance provided during construction.
|
|
|
166
|
|
|
167 ***********************************************************************/
|
|
|
168
|
|
|
169 final Logger log ()
|
|
|
170 {
|
|
|
171 return logger;
|
|
|
172 }
|
|
|
173
|
|
|
174 /***********************************************************************
|
|
|
175
|
|
|
176 Create a channel instance. Our channel implementation
|
|
|
177 includes a number of cached IO helpers (ProtocolWriter
|
|
|
178 and so on) which simplifies and speeds up execution.
|
|
|
179
|
|
|
180 ***********************************************************************/
|
|
|
181
|
|
|
182 final IChannel createChannel (char[] channel)
|
|
|
183 {
|
|
|
184 return new Channel (this, channel);
|
|
|
185 }
|
|
|
186
|
|
|
187 /***********************************************************************
|
|
|
188
|
|
|
189 ChannelListener method for listening to RollCall responses.
|
|
|
190 These are sent out by cluster servers both when they get a
|
|
|
191 RollCall request, and when they heartbeat.
|
|
|
192
|
|
|
193 ***********************************************************************/
|
|
|
194
|
|
|
195 private void notify (IEvent event)
|
|
|
196 {
|
|
|
197 scope rollcall = new RollCall;
|
|
|
198 event.get (rollcall);
|
|
|
199
|
|
|
200 switch (rollcall.type)
|
|
|
201 {
|
|
|
202 default:
|
|
|
203 break;
|
|
|
204
|
|
|
205 case RollCall.Task:
|
|
|
206 task.enable (rollcall.addr, "task");
|
|
|
207 break;
|
|
|
208
|
|
|
209 case RollCall.Cache:
|
|
|
210 cache.enable (rollcall.addr);
|
|
|
211 break;
|
|
|
212
|
|
|
213 case RollCall.Queue:
|
|
|
214 queue.enable (rollcall.addr, "queue");
|
|
|
215 break;
|
|
|
216 }
|
|
|
217 }
|
|
|
218 }
|
|
|
219
|
|
|
220
|
|
|
221 /*******************************************************************************
|
|
|
222
|
|
|
223 Basic multicast support across the cluster. Multicast is used
|
|
|
224 for broadcasting messages to all nodes in the cluster. We use
|
|
|
225 it for cache-invalidation, heartbeat, rollcall and notification
|
|
|
226 of queue activity
|
|
|
227
|
|
|
228 *******************************************************************************/
|
|
|
229
|
|
|
230 private class Broadcaster
|
|
|
231 {
|
|
|
232 private static InternetAddress[char[]] groups;
|
|
|
233 private Buffer mBuffer;
|
|
|
234 private ProtocolWriter mWriter;
|
|
|
235 private MulticastConduit mSocket;
|
|
|
236
|
|
|
237 private int groupPort = 3333;
|
|
|
238 private int groupPrefix = 225;
|
|
|
239
|
|
|
240 /***********************************************************************
|
|
|
241
|
|
|
242 Setup a Cluster instance. Currently the buffer & writer
|
|
|
243 are shared for all bulletin serialization; this should
|
|
|
244 probably change at some point such that we can support
|
|
|
245 multiple threads broadcasting concurrently to different
|
|
|
246 output ports.
|
|
|
247
|
|
|
248 ***********************************************************************/
|
|
|
249
|
|
|
250 this ()
|
|
|
251 {
|
|
|
252 mBuffer = new Buffer (1024 * 4);
|
|
|
253 mSocket = new MulticastConduit;
|
|
|
254 mWriter = new ProtocolWriter (mBuffer);
|
|
|
255 }
|
|
|
256
|
|
|
257 /***********************************************************************
|
|
|
258
|
|
|
259 Setup the multicast options. Port is used as the sole
|
|
|
260 address port for multicast usage, prefix is prepended
|
|
|
261 to each fabricated multicast address (should be a valid
|
|
|
262 class-D prefix), and ttl is the number of hops
|
|
|
263
|
|
|
264 ***********************************************************************/
|
|
|
265
|
|
|
266 final MulticastConduit conduit ()
|
|
|
267 {
|
|
|
268 return mSocket;
|
|
|
269 }
|
|
|
270
|
|
|
271 /***********************************************************************
|
|
|
272
|
|
|
273 Setup the multicast options. Port is used as the sole
|
|
|
274 address port for multicast usage & prefix is prepended
|
|
|
275 to each fabricated multicast address (should be a valid
|
|
|
276 class-D prefix: 225 through 239 inclusive)
|
|
|
277
|
|
|
278 ***********************************************************************/
|
|
|
279
|
|
|
280 final void multicast (int port, int prefix=225)
|
|
|
281 {
|
|
|
282 groupPort = port;
|
|
|
283 groupPrefix = prefix;
|
|
|
284 }
|
|
|
285
|
|
|
286 /***********************************************************************
|
|
|
287
|
|
|
288 Broadcast a message on the specified channel. This uses
|
|
|
289 IP/Multicast to scatter the payload to all registered
|
|
|
290 listeners (on the same multicast group). Note that the
|
|
|
291 maximum message size is limited to that of an Ethernet
|
|
|
292 data frame, minus the IP/UDP header size (1472 bytes).
|
|
|
293
|
|
|
294 Also note that we are synchronized to avoid contention
|
|
|
295 on the otherwise shared output buffer.
|
|
|
296
|
|
|
297 ***********************************************************************/
|
|
|
298
|
|
|
299 final synchronized void broadcast (char[] channel, IMessage message=null)
|
|
|
300 {
|
|
|
301 // clear buffer and serialize content
|
|
|
302 mWriter.put (ProtocolWriter.Command.OK, channel, null, message);
|
|
|
303
|
|
|
304 // Ethernet data-frame size minus the 28 byte UDP/IP header:
|
|
|
305 if (mBuffer.position > 1472)
|
|
|
306 throw new ClusterException ("message is too large to broadcast");
|
|
|
307
|
|
|
308 // send it to the appropriate multicast group
|
|
|
309 mSocket.write (mBuffer.slice, getGroup (channel));
|
|
|
310 }
|
|
|
311
|
|
|
312 /***********************************************************************
|
|
|
313
|
|
|
314 Return an internet address representing the multicast
|
|
|
315 group for the specified channel. We use three of the
|
|
|
316 four address segments to represent the channel itself
|
|
|
317 (via a hash on the channel name), and set the primary
|
|
|
318 segment to be that of the broadcast prefix (above).
|
|
|
319
|
|
|
320 ***********************************************************************/
|
|
|
321
|
|
|
322 final synchronized InternetAddress getGroup (char[] channel)
|
|
|
323 {
|
|
|
324 auto p = channel in groups;
|
|
|
325 if (p)
|
|
|
326 return *p;
|
|
|
327
|
|
|
328 // construct a group address from the prefix & channel-hash,
|
|
|
329 // where the hash is folded down to 24 bits
|
|
|
330 uint hash = jhash (channel.ptr, channel.length);
|
|
|
331 hash = (hash >> 24) ^ (hash & 0x00ffffff);
|
|
|
332
|
|
|
333 auto address = Integer.toString (groupPrefix) ~ "." ~
|
|
|
334 Integer.toString ((hash >> 16) & 0xff) ~ "." ~
|
|
|
335 Integer.toString ((hash >> 8) & 0xff) ~ "." ~
|
|
|
336 Integer.toString (hash & 0xff);
|
|
|
337
|
|
|
338 // insert InternetAddress into hashmap
|
|
|
339 auto group = new InternetAddress (address, groupPort);
|
|
|
340 groups [channel] = group;
|
|
|
341 return group;
|
|
|
342 }
|
|
|
343 }
|
|
|
344
|
|
|
345
|
|
|
346 /*******************************************************************************
|
|
|
347
|
|
|
348 A channel represents something akin to a publish/subscribe topic,
|
|
|
349 or a radio station. These are used to segregate cluster operations
|
|
|
350 into a set of groups, where each group is represented by a channel.
|
|
|
351 Channel names are whatever you want then to be: use of dot notation
|
|
|
352 has proved useful in the past.
|
|
|
353
|
|
|
354 Channel maintain internal state in order to avoid heap activity. So
|
|
|
355 they should not be shared across threads without appropriate synchs
|
|
|
356 in place. One remedy is create another channel instance
|
|
|
357
|
|
|
358 *******************************************************************************/
|
|
|
359
|
|
|
360 private class Channel : IChannel
|
|
|
361 {
|
|
|
362 private char[] name_;
|
|
|
363 private Buffer buffer;
|
|
|
364 private ProtocolReader reader;
|
|
|
365 private ProtocolWriter writer;
|
|
|
366 private Cluster cluster_;
|
|
|
367
|
|
|
368 /***********************************************************************
|
|
|
369
|
|
|
370 Construct a channel with the specified name. We cache
|
|
|
371 a number of session-related constructs here also, in
|
|
|
372 order to eliminate runtime overhead
|
|
|
373
|
|
|
374 ***********************************************************************/
|
|
|
375
|
|
|
376 this (Cluster cluster, char[] name)
|
|
|
377 in {
|
|
|
378 assert (cluster);
|
|
|
379 assert (name.length);
|
|
|
380 }
|
|
|
381 body
|
|
|
382 {
|
|
|
383 name_ = name;
|
|
|
384 cluster_ = cluster;
|
|
|
385
|
|
|
386 // this buffer will grow as required to house larger messages
|
|
|
387 buffer = new GrowBuffer (1024 * 2);
|
|
|
388 writer = new ProtocolWriter (buffer);
|
|
|
389
|
|
|
390 // make the reader slice directly from the buffer content
|
|
|
391 reader = new ProtocolReader (buffer);
|
|
|
392 }
|
|
|
393
|
|
|
394 /***********************************************************************
|
|
|
395
|
|
|
396 Return the name of this channel. This is the name provided
|
|
|
397 when the channel was constructed.
|
|
|
398
|
|
|
399 ***********************************************************************/
|
|
|
400
|
|
|
401 final char[] name ()
|
|
|
402 {
|
|
|
403 return name_;
|
|
|
404 }
|
|
|
405
|
|
|
406 /***********************************************************************
|
|
|
407
|
|
|
408 Return the assigned cluster
|
|
|
409
|
|
|
410 ***********************************************************************/
|
|
|
411
|
|
|
412 final Cluster cluster ()
|
|
|
413 {
|
|
|
414 return cluster_;
|
|
|
415 }
|
|
|
416
|
|
|
417 /***********************************************************************
|
|
|
418
|
|
|
419 Return the assigned logger
|
|
|
420
|
|
|
421 ***********************************************************************/
|
|
|
422
|
|
|
423 final Logger log ()
|
|
|
424 {
|
|
|
425 return cluster_.log;
|
|
|
426 }
|
|
|
427
|
|
|
428 /***********************************************************************
|
|
|
429
|
|
|
430 Output this channel via the provided IWriter
|
|
|
431
|
|
|
432 ***********************************************************************/
|
|
|
433
|
|
|
434 final void write (IWriter writer)
|
|
|
435 {
|
|
|
436 writer.put (name_);
|
|
|
437 }
|
|
|
438
|
|
|
439 /***********************************************************************
|
|
|
440
|
|
|
441 Input this channel via the provided IReader
|
|
|
442
|
|
|
443 ***********************************************************************/
|
|
|
444
|
|
|
445 final void read (IReader reader)
|
|
|
446 {
|
|
|
447 reader.get (name_);
|
|
|
448 }
|
|
|
449
|
|
|
450 /***********************************************************************
|
|
|
451
|
|
|
452 deserialize a message into a provided host, or via
|
|
|
453 the registered instance of the incoming message
|
|
|
454
|
|
|
455 ***********************************************************************/
|
|
|
456
|
|
|
457 final IMessage thaw (IMessage host = null)
|
|
|
458 {
|
|
|
459 return reader.thaw (host);
|
|
|
460 }
|
|
|
461
|
|
|
462 /***********************************************************************
|
|
|
463
|
|
|
464 Create a listener of the specified type. Listeners are
|
|
|
465 run within their own thread, since they spend the vast
|
|
|
466 majority of their time blocked on a Socket read. Would
|
|
|
467 be good to support multiplexed reading instead, such
|
|
|
468 that a thread pool could be applied instead.
|
|
|
469
|
|
|
470 ***********************************************************************/
|
|
|
471
|
|
|
472 final IConsumer createConsumer (ChannelListener notify)
|
|
|
473 {
|
|
|
474 cluster_.log.trace ("creating message consumer for '" ~ name_ ~ "'");
|
|
|
475 return new MessageConsumer (this, notify);
|
|
|
476 }
|
|
|
477
|
|
|
478 /***********************************************************************
|
|
|
479
|
|
|
480 Create a listener of the specified type. Listeners are
|
|
|
481 run within their own thread, since they spend the vast
|
|
|
482 majority of their time blocked on a Socket read. Would
|
|
|
483 be good to support multiplexed reading instead, such
|
|
|
484 that a thread pool could be applied instead.
|
|
|
485
|
|
|
486 ***********************************************************************/
|
|
|
487
|
|
|
488 final IConsumer createBulletinConsumer (ChannelListener notify)
|
|
|
489 {
|
|
|
490 cluster_.log.trace ("creating bulletin consumer for '" ~ name_ ~ "'");
|
|
|
491 return new BulletinConsumer (this, notify);
|
|
|
492 }
|
|
|
493
|
|
|
494 /***********************************************************************
|
|
|
495
|
|
|
496 Return a entry from the network cache, and optionally
|
|
|
497 remove it. This is a synchronous operation as opposed
|
|
|
498 to the asynchronous nature of an invalidate broadcast.
|
|
|
499
|
|
|
500 ***********************************************************************/
|
|
|
501
|
|
|
502 final IMessage getCache (char[] key, bool remove, IMessage host = null)
|
|
|
503 {
|
|
|
504 void send (IConduit conduit)
|
|
|
505 {
|
|
|
506 buffer.setConduit (conduit);
|
|
|
507 writer.put (remove ? ProtocolWriter.Command.Remove :
|
|
|
508 ProtocolWriter.Command.Copy, name_, key).flush;
|
|
|
509 }
|
|
|
510
|
|
|
511 if (cluster_.cache.request (&send, reader, key))
|
|
|
512 return reader.thaw (host);
|
|
|
513 return null;
|
|
|
514 }
|
|
|
515
|
|
|
516 /***********************************************************************
|
|
|
517
|
|
|
518 Place an entry into the network cache, replacing the
|
|
|
519 entry with the identical key. Where message.time is
|
|
|
520 set, it will be used to test for newer cache entries
|
|
|
521 than the one being sent i.e. if someone else placed
|
|
|
522 a newer entry into the cache, that one will remain.
|
|
|
523
|
|
|
524 Note that this may cause the oldest entry in the cache
|
|
|
525 to be displaced if the cache is already full.
|
|
|
526
|
|
|
527 ***********************************************************************/
|
|
|
528
|
|
|
529 final bool putCache (char[] key, IMessage message)
|
|
|
530 {
|
|
|
531 void send (IConduit conduit)
|
|
|
532 {
|
|
|
533 buffer.setConduit (conduit);
|
|
|
534 writer.put (ProtocolWriter.Command.Add, name_, key, message).flush;
|
|
|
535 }
|
|
|
536
|
|
|
537 // return false if the cache server said there's
|
|
|
538 // already something newer
|
|
|
539 if (cluster_.cache.request (&send, reader, key))
|
|
|
540 return false;
|
|
|
541 return true;
|
|
|
542 }
|
|
|
543
|
|
|
544 /***********************************************************************
|
|
|
545
|
|
|
546 Load a network cache entry remotely. This sends the given
|
|
|
547 IMessage over a network to the cache host, where it will
|
|
|
548 be executed locally. The benefit of doing so it that the
|
|
|
549 host may deny access to the cache entry for the duration
|
|
|
550 of the load operation. This, in turn, provides a mechanism
|
|
|
551 for gating/synchronizing multiple network clients over a
|
|
|
552 given cache entry; quite handy for those entries that are
|
|
|
553 relatively expensive to construct or access.
|
|
|
554
|
|
|
555 ***********************************************************************/
|
|
|
556
|
|
|
557 final bool loadCache (char[] key, IMessage message)
|
|
|
558 {
|
|
|
559 void send (IConduit conduit)
|
|
|
560 {
|
|
|
561 buffer.setConduit (conduit);
|
|
|
562 writer.put (ProtocolWriter.Command.Load, name_, key, message).flush;
|
|
|
563 }
|
|
|
564
|
|
|
565 return cluster_.cache.request (&send, reader, key);
|
|
|
566 }
|
|
|
567
|
|
|
568 /***********************************************************************
|
|
|
569
|
|
|
570 Query the cluster for queued entries on the corresponding
|
|
|
571 channel. Returns, and removes, the first matching entry
|
|
|
572 from the cluster. Note that this sweeps the cluster for
|
|
|
573 matching entries, and is synchronous in nature. The more
|
|
|
574 common approach is to setup a queue listener, which will
|
|
|
575 grab and dispatch queue entries asynchronously.
|
|
|
576
|
|
|
577 ***********************************************************************/
|
|
|
578
|
|
|
579 final IMessage getQueue (IMessage host = null)
|
|
|
580 {
|
|
|
581 if (scanQueue)
|
|
|
582 return reader.thaw (host);
|
|
|
583 return null;
|
|
|
584 }
|
|
|
585
|
|
|
586 /***********************************************************************
|
|
|
587
|
|
|
588 Query the cluster for queued entries on the corresponding
|
|
|
589 channel. Returns, and removes, the first matching entry
|
|
|
590 from the cluster. Note that this sweeps the cluster for
|
|
|
591 matching entries, and is synchronous in nature. The more
|
|
|
592 common approach is to setup a queue listener, which will
|
|
|
593 grab and dispatch queue entries asynchronously.
|
|
|
594
|
|
|
595 ***********************************************************************/
|
|
|
596
|
|
|
597 private bool scanQueue ()
|
|
|
598 {
|
|
|
599 void send (IConduit conduit)
|
|
|
600 {
|
|
|
601 buffer.setConduit (conduit);
|
|
|
602 writer.put (ProtocolWriter.Command.RemoveQueue, name_).flush;
|
|
|
603 }
|
|
|
604
|
|
|
605 bool scan (Node node)
|
|
|
606 {
|
|
|
607 bool message;
|
|
|
608 node.request (&send, reader, message);
|
|
|
609 return message;
|
|
|
610 }
|
|
|
611
|
|
|
612 // make a pass over each Node, looking for channel entries
|
|
|
613 return cluster_.queue.scan (&scan);
|
|
|
614 }
|
|
|
615
|
|
|
616 /***********************************************************************
|
|
|
617
|
|
|
618 Add an entry to the specified network queue. May throw a
|
|
|
619 QueueFullException if there's no room available.
|
|
|
620
|
|
|
621 ***********************************************************************/
|
|
|
622
|
|
|
623 final IMessage putQueue (IMessage message)
|
|
|
624 {
|
|
|
625 void send (IConduit conduit)
|
|
|
626 {
|
|
|
627 buffer.setConduit (conduit);
|
|
|
628 writer.put (ProtocolWriter.Command.AddQueue, name_, null, message).flush;
|
|
|
629 }
|
|
|
630
|
|
|
631 cluster_.queue.request (&send, reader);
|
|
|
632 return message;
|
|
|
633 }
|
|
|
634
|
|
|
635 /***********************************************************************
|
|
|
636
|
|
|
637 Send a remote call request to a server, and place the result
|
|
|
638 back into the provided message
|
|
|
639
|
|
|
640 ***********************************************************************/
|
|
|
641
|
|
|
642 final bool execute (IMessage message)
|
|
|
643 {
|
|
|
644 void send (IConduit conduit)
|
|
|
645 {
|
|
|
646 buffer.setConduit (conduit);
|
|
|
647 writer.put (ProtocolWriter.Command.Call, name_, null, message).flush;
|
|
|
648 }
|
|
|
649
|
|
|
650 if (cluster_.task.request (&send, reader))
|
|
|
651 {
|
|
|
652 // place result back into the provided message
|
|
|
653 reader.thaw (message);
|
|
|
654 return true;
|
|
|
655 }
|
|
|
656 return false;
|
|
|
657 }
|
|
|
658
|
|
|
659 /***********************************************************************
|
|
|
660
|
|
|
661 Broadcast a message on the specified channel. This uses
|
|
|
662 IP/Multicast to scatter the message to all registered
|
|
|
663 listeners (on the same multicast group). Note that the
|
|
|
664 maximum message size is limited to that of an Ethernet
|
|
|
665 data frame, minus the IP/UDP header size (1472 bytes).
|
|
|
666
|
|
|
667 ***********************************************************************/
|
|
|
668
|
|
|
669 final void broadcast (IMessage message = null)
|
|
|
670 {
|
|
|
671 cluster_.broadcast (name_, message);
|
|
|
672 }
|
|
|
673 }
|
|
|
674
|
|
|
675
|
|
|
676 /*******************************************************************************
|
|
|
677
|
|
|
678 A listener for multicast channel traffic. These are currently used
|
|
|
679 for cache coherency, queue publishing, and node discovery activity;
|
|
|
680 though could be used for direct messaging also.
|
|
|
681
|
|
|
682 Be careful when using the retained channel, since it is shared with
|
|
|
683 the calling thread. Thus a race condition could arise between the
|
|
|
684 client and this thread, were both to use the channel for transfers
|
|
|
685 at the same instant. Note that MessageConsumer makes a copy of the
|
|
|
686 channel for this purpose
|
|
|
687
|
|
|
688 *******************************************************************************/
|
|
|
689
|
|
|
690 private class BulletinConsumer : SocketListener, IConsumer, IEvent
|
|
|
691 {
|
|
|
692 private bool hasMore; // incoming message?
|
|
|
693 private Buffer buffer; // input buffer
|
|
|
694 private ProtocolReader reader; // input decoder
|
|
|
695 private Channel channel_; // associated channel
|
|
|
696 private Cluster cluster; // associated cluster
|
|
|
697 private MulticastConduit consumer; // broadcast listener
|
|
|
698 private ChannelListener listener; // user-level callback
|
|
|
699
|
|
|
700 /***********************************************************************
|
|
|
701
|
|
|
702 Construct a multicast consumer for the specified event. The
|
|
|
703 event handler will be invoked whenever a message arrives for
|
|
|
704 the associated channel.
|
|
|
705
|
|
|
706 ***********************************************************************/
|
|
|
707
|
|
|
708 this (Channel channel, ChannelListener listener)
|
|
|
709 {
|
|
|
710 this.channel_ = channel;
|
|
|
711 this.listener = listener;
|
|
|
712 this.cluster = channel.cluster;
|
|
|
713
|
|
|
714 // buffer doesn't need to be larger than Ethernet data-frame
|
|
|
715 buffer = new Buffer (1500);
|
|
|
716
|
|
|
717 // make the reader slice directly from the buffer content
|
|
|
718 reader = new ProtocolReader (buffer);
|
|
|
719
|
|
|
720 // configure a listener socket
|
|
|
721 consumer = new MulticastConduit (cluster.getGroup (channel_.name), true);
|
|
|
722 consumer.join;
|
|
|
723
|
|
|
724 super (consumer, buffer);
|
|
|
725
|
|
|
726 // fire up this listener
|
|
|
727 super.execute;
|
|
|
728 }
|
|
|
729
|
|
|
730 /***********************************************************************
|
|
|
731
|
|
|
732 Notification callback invoked when we receive a multicast
|
|
|
733 packet. Note that we check the packet channel-name against
|
|
|
734 the one we're consuming, to check for cases where the group
|
|
|
735 address had a hash collision.
|
|
|
736
|
|
|
737 ***********************************************************************/
|
|
|
738
|
|
|
739 override void notify (IBuffer buffer)
|
|
|
740 {
|
|
|
741 ProtocolWriter.Command cmd;
|
|
|
742 char[] channel;
|
|
|
743 char[] element;
|
|
|
744
|
|
|
745 // read the incoming header, along with the object guid
|
|
|
746 // where available
|
|
|
747 hasMore = reader.getHeader (cmd, channel, element);
|
|
|
748
|
|
|
749 // check it's really for us first (might be a hash collision)
|
|
|
750 if (channel == this.channel_.name)
|
|
|
751 invoke (this);
|
|
|
752 }
|
|
|
753
|
|
|
754 /***********************************************************************
|
|
|
755
|
|
|
756 ***********************************************************************/
|
|
|
757
|
|
|
758 IMessage get (IMessage host = null)
|
|
|
759 {
|
|
|
760 if (hasMore)
|
|
|
761 return reader.thaw (host);
|
|
|
762
|
|
|
763 throw new ClusterException ("attempting to thaw a non-existant message");
|
|
|
764 }
|
|
|
765
|
|
|
766 /***********************************************************************
|
|
|
767
|
|
|
768 Return the assigned logger
|
|
|
769
|
|
|
770 ***********************************************************************/
|
|
|
771
|
|
|
772 final Logger log ()
|
|
|
773 {
|
|
|
774 return cluster.log;
|
|
|
775 }
|
|
|
776
|
|
|
777 /***********************************************************************
|
|
|
778
|
|
|
779 Handle error conditions from the listener thread.
|
|
|
780
|
|
|
781 ***********************************************************************/
|
|
|
782
|
|
|
783 override void exception (char [] msg)
|
|
|
784 {
|
|
|
785 cluster.log.error ("BulletinConsumer: "~msg);
|
|
|
786 }
|
|
|
787
|
|
|
788 /***********************************************************************
|
|
|
789
|
|
|
790 Overridable mean of notifying the client code.
|
|
|
791
|
|
|
792 ***********************************************************************/
|
|
|
793
|
|
|
794 protected void invoke (IEvent event)
|
|
|
795 {
|
|
|
796 listener (event);
|
|
|
797 }
|
|
|
798
|
|
|
799 /***********************************************************************
|
|
|
800
|
|
|
801 Return the cluster instance we're associated with.
|
|
|
802
|
|
|
803 ***********************************************************************/
|
|
|
804
|
|
|
805 final Channel channel ()
|
|
|
806 {
|
|
|
807 return channel_;
|
|
|
808 }
|
|
|
809
|
|
|
810 /***********************************************************************
|
|
|
811
|
|
|
812 Temporarily halt listening. This can be used to ignore
|
|
|
813 multicast messages while, for example, the consumer is
|
|
|
814 busy doing other things.
|
|
|
815
|
|
|
816 ***********************************************************************/
|
|
|
817
|
|
|
818 final void pauseGroup ()
|
|
|
819 {
|
|
|
820 consumer.leave;
|
|
|
821 }
|
|
|
822
|
|
|
823 /***********************************************************************
|
|
|
824
|
|
|
825 Resume listening, post-pause.
|
|
|
826
|
|
|
827 ***********************************************************************/
|
|
|
828
|
|
|
829 final void resumeGroup ()
|
|
|
830 {
|
|
|
831 consumer.join;
|
|
|
832 }
|
|
|
833
|
|
|
834 /***********************************************************************
|
|
|
835
|
|
|
836 Cancel this consumer. The listener is effectively disabled
|
|
|
837 from this point forward. The listener thread does not halt
|
|
|
838 at this point, but waits until the socket-read returns.
|
|
|
839 Note that the D Interface implementation requires us to
|
|
|
840 "reimplement and dispatch" trivial things like this ~ it's
|
|
|
841 a pain in the neck to maintain.
|
|
|
842
|
|
|
843 ***********************************************************************/
|
|
|
844
|
|
|
845 final void cancel ()
|
|
|
846 {
|
|
|
847 super.cancel;
|
|
|
848 }
|
|
|
849
|
|
|
850 /***********************************************************************
|
|
|
851
|
|
|
852 Send a message back to the producer
|
|
|
853
|
|
|
854 ***********************************************************************/
|
|
|
855
|
|
|
856 void reply (IChannel channel, IMessage message)
|
|
|
857 {
|
|
|
858 assert (channel);
|
|
|
859 assert (message);
|
|
|
860
|
|
|
861 channel.broadcast (message);
|
|
|
862 }
|
|
|
863
|
|
|
864
|
|
|
865 /***********************************************************************
|
|
|
866
|
|
|
867 Return an appropriate reply channel for the given message,
|
|
|
868 or return null if no reply is expected
|
|
|
869
|
|
|
870 ***********************************************************************/
|
|
|
871
|
|
|
872 IChannel replyChannel (IMessage message)
|
|
|
873 {
|
|
|
874 if (message.reply.length)
|
|
|
875 return cluster.createChannel (message.reply);
|
|
|
876 return null;
|
|
|
877 }
|
|
|
878 }
|
|
|
879
|
|
|
880
|
|
|
881 /*******************************************************************************
|
|
|
882
|
|
|
883 A listener for queue events. These events are produced by the
|
|
|
884 queue host on a periodic bases when it has available entries.
|
|
|
885 We listen for them (rather than constantly scanning) and then
|
|
|
886 begin a sweep to process as many as we can. Note that we will
|
|
|
887 be in competition with other nodes to process these entries.
|
|
|
888
|
|
|
889 Also note that we create a copy of the channel in use, so that
|
|
|
890 race-conditions with the requesting client are avoided.
|
|
|
891
|
|
|
892 *******************************************************************************/
|
|
|
893
|
|
|
894 private class MessageConsumer : BulletinConsumer
|
|
|
895 {
|
|
|
896 /***********************************************************************
|
|
|
897
|
|
|
898 Construct a multicast consumer for the specified event
|
|
|
899
|
|
|
900 ***********************************************************************/
|
|
|
901
|
|
|
902 this (Channel channel, ChannelListener listener)
|
|
|
903 {
|
|
|
904 super (channel, listener);
|
|
|
905
|
|
|
906 // create private channel instance to use in our thread
|
|
|
907 this.channel_ = new Channel (channel.cluster, channel.name);
|
|
|
908 }
|
|
|
909
|
|
|
910 /***********************************************************************
|
|
|
911
|
|
|
912 Handle error conditions from the listener thread.
|
|
|
913
|
|
|
914 ***********************************************************************/
|
|
|
915
|
|
|
916 override void exception (char [] msg)
|
|
|
917 {
|
|
|
918 cluster.log.error ("MessageConsumer: "~msg);
|
|
|
919 }
|
|
|
920
|
|
|
921 /***********************************************************************
|
|
|
922
|
|
|
923 Overrides the default processing to sweep the cluster for
|
|
|
924 queued entries. Each server node is queried until one is
|
|
|
925 found that contains a message. Note that it is possible
|
|
|
926 to set things up where we are told exactly which node to
|
|
|
927 go to; however given that we won't be listening whilst
|
|
|
928 scanning, and that there's likely to be a group of new
|
|
|
929 entries in the cluster, it's just as effective to scan.
|
|
|
930 This will be far from ideal for all environments, so we
|
|
|
931 should make the strategy pluggable instead.
|
|
|
932
|
|
|
933 Note also that the content is retrieved via a duplicate
|
|
|
934 channel to avoid potential race-conditions on the original
|
|
|
935
|
|
|
936 ***********************************************************************/
|
|
|
937
|
|
|
938 override IMessage get (IMessage host = null)
|
|
|
939 {
|
|
|
940 if (channel.scanQueue)
|
|
|
941 return channel.thaw (host);
|
|
|
942 return null;
|
|
|
943 }
|
|
|
944
|
|
|
945 /***********************************************************************
|
|
|
946
|
|
|
947 Send a message back to the producer
|
|
|
948
|
|
|
949 ***********************************************************************/
|
|
|
950
|
|
|
951 override void reply (IChannel channel, IMessage message)
|
|
|
952 {
|
|
|
953 assert (channel);
|
|
|
954 assert (message);
|
|
|
955
|
|
|
956 channel.putQueue (message);
|
|
|
957 }
|
|
|
958
|
|
|
959 /***********************************************************************
|
|
|
960
|
|
|
961 Override the default notification handler in order to
|
|
|
962 disable multicast reciepts while the application does
|
|
|
963 what it needs to
|
|
|
964
|
|
|
965 ***********************************************************************/
|
|
|
966
|
|
|
967 override protected void invoke (IEvent event)
|
|
|
968 {
|
|
|
969 // temporarily pause listening while processing
|
|
|
970 pauseGroup;
|
|
|
971 try {
|
|
|
972 listener (event);
|
|
|
973 } finally resumeGroup;
|
|
|
974 }
|
|
|
975 }
|
|
|
976
|
|
|
977
|
|
|
978 /*******************************************************************************
|
|
|
979
|
|
|
980 An abstraction of a socket connection. Used internally by the
|
|
|
981 socket-based Cluster.
|
|
|
982
|
|
|
983 *******************************************************************************/
|
|
|
984
|
|
|
985 private class Connection
|
|
|
986 {
|
|
|
987 abstract bool reset();
|
|
|
988
|
|
|
989 abstract void done (Time time);
|
|
|
990
|
|
|
991 abstract SocketConduit conduit ();
|
|
|
992 }
|
|
|
993
|
|
|
994
|
|
|
995 /*******************************************************************************
|
|
|
996
|
|
|
997 A pool of socket connections for accessing cluster nodes. Note
|
|
|
998 that the entries will timeout after a period of inactivity, and
|
|
|
999 will subsequently cause a connected host to drop the supporting
|
|
|
1000 session.
|
|
|
1001
|
|
|
1002 *******************************************************************************/
|
|
|
1003
|
|
|
1004 private class ConnectionPool
|
|
|
1005 {
|
|
|
1006 private Logger log;
|
|
|
1007 private int count;
|
|
|
1008 private bool noDelay;
|
|
|
1009 private InternetAddress address;
|
|
|
1010 private PoolConnection freelist;
|
|
|
1011 private TimeSpan timeout = TimeSpan.seconds(60);
|
|
|
1012
|
|
|
1013 /***********************************************************************
|
|
|
1014
|
|
|
1015 Utility class to provide the basic connection facilities
|
|
|
1016 provided by the connection pool.
|
|
|
1017
|
|
|
1018 ***********************************************************************/
|
|
|
1019
|
|
|
1020 static class PoolConnection : Connection
|
|
|
1021 {
|
|
|
1022 Time time;
|
|
|
1023 PoolConnection next;
|
|
|
1024 ConnectionPool parent;
|
|
|
1025 SocketConduit conduit_;
|
|
|
1026
|
|
|
1027 /***************************************************************
|
|
|
1028
|
|
|
1029 Construct a new connection and set its parent
|
|
|
1030
|
|
|
1031 ***************************************************************/
|
|
|
1032
|
|
|
1033 this (ConnectionPool pool)
|
|
|
1034 {
|
|
|
1035 parent = pool;
|
|
|
1036 reset;
|
|
|
1037 }
|
|
|
1038
|
|
|
1039 /***************************************************************
|
|
|
1040
|
|
|
1041 Create a new socket and connect it to the specified
|
|
|
1042 server. This will cause a dedicated thread to start
|
|
|
1043 on the server. Said thread will quit when an error
|
|
|
1044 occurs.
|
|
|
1045
|
|
|
1046 ***************************************************************/
|
|
|
1047
|
|
|
1048 final bool reset ()
|
|
|
1049 {
|
|
|
1050 try {
|
|
|
1051 conduit_ = new SocketConduit;
|
|
|
1052
|
|
|
1053 // apply Nagle settings
|
|
|
1054 conduit.socket.setNoDelay (parent.noDelay);
|
|
|
1055
|
|
|
1056 // set a 500ms timeout for read operations
|
|
|
1057 conduit_.setTimeout (TimeSpan.millis(500));
|
|
|
1058
|
|
|
1059 // open a connection to this server
|
|
|
1060 // parent.log.trace ("connecting to server");
|
|
|
1061 conduit_.connect (parent.address);
|
|
|
1062 return true;
|
|
|
1063
|
|
|
1064 } catch (Object o)
|
|
|
1065 {
|
|
|
1066 if (! Runtime.isHalting)
|
|
|
1067 parent.log.warn ("server is unavailable :: "~o.toString);
|
|
|
1068 }
|
|
|
1069 return false;
|
|
|
1070 }
|
|
|
1071
|
|
|
1072 /***************************************************************
|
|
|
1073
|
|
|
1074 Return the socket belonging to this connection
|
|
|
1075
|
|
|
1076 ***************************************************************/
|
|
|
1077
|
|
|
1078 final SocketConduit conduit ()
|
|
|
1079 {
|
|
|
1080 return conduit_;
|
|
|
1081 }
|
|
|
1082
|
|
|
1083 /***************************************************************
|
|
|
1084
|
|
|
1085 Close the socket. This will cause any host session
|
|
|
1086 to be terminated.
|
|
|
1087
|
|
|
1088 ***************************************************************/
|
|
|
1089
|
|
|
1090 final void close ()
|
|
|
1091 {
|
|
|
1092 conduit_.detach;
|
|
|
1093 }
|
|
|
1094
|
|
|
1095 /***************************************************************
|
|
|
1096
|
|
|
1097 Return this connection to the free-list. Note that
|
|
|
1098 we have to synchronize on the parent-pool itself.
|
|
|
1099
|
|
|
1100 ***************************************************************/
|
|
|
1101
|
|
|
1102 final void done (Time time)
|
|
|
1103 {
|
|
|
1104 synchronized (parent)
|
|
|
1105 {
|
|
|
1106 next = parent.freelist;
|
|
|
1107 parent.freelist = this;
|
|
|
1108 this.time = time;
|
|
|
1109 }
|
|
|
1110 }
|
|
|
1111 }
|
|
|
1112
|
|
|
1113
|
|
|
1114 /***********************************************************************
|
|
|
1115
|
|
|
1116 Create a connection-pool for the specified address.
|
|
|
1117
|
|
|
1118 ***********************************************************************/
|
|
|
1119
|
|
|
1120 this (InternetAddress address, Logger log, bool noDelay)
|
|
|
1121 {
|
|
|
1122 this.log = log;
|
|
|
1123 this.address = address;
|
|
|
1124 this.noDelay = noDelay;
|
|
|
1125 }
|
|
|
1126
|
|
|
1127 /***********************************************************************
|
|
|
1128
|
|
|
1129 Allocate a Connection from a list rather than creating a
|
|
|
1130 new one. Reap old entries as we go.
|
|
|
1131
|
|
|
1132 ***********************************************************************/
|
|
|
1133
|
|
|
1134 final synchronized Connection borrow (Time time)
|
|
|
1135 {
|
|
|
1136 if (freelist)
|
|
|
1137 do {
|
|
|
1138 auto c = freelist;
|
|
|
1139
|
|
|
1140 freelist = c.next;
|
|
|
1141 if (freelist && (time - c.time > timeout))
|
|
|
1142 c.close;
|
|
|
1143 else
|
|
|
1144 return c;
|
|
|
1145 } while (true);
|
|
|
1146
|
|
|
1147 return new PoolConnection (this);
|
|
|
1148 }
|
|
|
1149
|
|
|
1150 /***********************************************************************
|
|
|
1151
|
|
|
1152 Close this pool and drop all existing connections.
|
|
|
1153
|
|
|
1154 ***********************************************************************/
|
|
|
1155
|
|
|
1156 final synchronized void close ()
|
|
|
1157 {
|
|
|
1158 auto c = freelist;
|
|
|
1159 freelist = null;
|
|
|
1160 while (c)
|
|
|
1161 {
|
|
|
1162 c.close;
|
|
|
1163 c = c.next;
|
|
|
1164 }
|
|
|
1165 }
|
|
|
1166 }
|
|
|
1167
|
|
|
1168
|
|
|
1169 /*******************************************************************************
|
|
|
1170
|
|
|
1171 Class to represent a cluster node. Each node supports both cache
|
|
|
1172 and queue functionality. Note that the set of available nodes is
|
|
|
1173 configured at startup, simplifying the discovery process in some
|
|
|
1174 significant ways, and causing less thrashing of cache-keys.
|
|
|
1175
|
|
|
1176 *******************************************************************************/
|
|
|
1177
|
|
|
1178 private class Node
|
|
|
1179 {
|
|
|
1180 private Logger log;
|
|
|
1181 private char[] name,
|
|
|
1182 addr;
|
|
|
1183 private ConnectionPool pool;
|
|
|
1184 private bool enabled;
|
|
|
1185
|
|
|
1186 alias void delegate (IConduit conduit) Requestor;
|
|
|
1187
|
|
|
1188 /***********************************************************************
|
|
|
1189
|
|
|
1190 Construct a node with the provided name. This name should
|
|
|
1191 be the network name of the hosting device.
|
|
|
1192
|
|
|
1193 ***********************************************************************/
|
|
|
1194
|
|
|
1195 this (Logger log, char[] addr, char[] name)
|
|
|
1196 {
|
|
|
1197 this.log = log;
|
|
|
1198 this.addr = addr;
|
|
|
1199 this.name = name ~ ':' ~ addr;
|
|
|
1200 }
|
|
|
1201
|
|
|
1202 /***********************************************************************
|
|
|
1203
|
|
|
1204 Add a cache/queue reference for the remote node
|
|
|
1205
|
|
|
1206 ***********************************************************************/
|
|
|
1207
|
|
|
1208 final void setPool (InternetAddress address, bool noDelay)
|
|
|
1209 {
|
|
|
1210 this.pool = new ConnectionPool (address, log, noDelay);
|
|
|
1211 }
|
|
|
1212
|
|
|
1213 /***********************************************************************
|
|
|
1214
|
|
|
1215 Return the name of this node
|
|
|
1216
|
|
|
1217 ***********************************************************************/
|
|
|
1218
|
|
|
1219 override char[] toString ()
|
|
|
1220 {
|
|
|
1221 return name;
|
|
|
1222 }
|
|
|
1223
|
|
|
1224 /***********************************************************************
|
|
|
1225
|
|
|
1226 Return the network address of this node
|
|
|
1227
|
|
|
1228 ***********************************************************************/
|
|
|
1229
|
|
|
1230 final char[] address ()
|
|
|
1231 {
|
|
|
1232 return addr;
|
|
|
1233 }
|
|
|
1234
|
|
|
1235 /***********************************************************************
|
|
|
1236
|
|
|
1237 Remove this Node from the cluster. The node is disabled
|
|
|
1238 until it is seen to recover.
|
|
|
1239
|
|
|
1240 ***********************************************************************/
|
|
|
1241
|
|
|
1242 final void fail ()
|
|
|
1243 {
|
|
|
1244 setEnabled (false);
|
|
|
1245 pool.close;
|
|
|
1246 }
|
|
|
1247
|
|
|
1248 /***********************************************************************
|
|
|
1249
|
|
|
1250 Get the current state of this node
|
|
|
1251
|
|
|
1252 ***********************************************************************/
|
|
|
1253
|
|
|
1254 final bool isEnabled ()
|
|
|
1255 {
|
|
|
1256 volatile
|
|
|
1257 return enabled;
|
|
|
1258 }
|
|
|
1259
|
|
|
1260 /***********************************************************************
|
|
|
1261
|
|
|
1262 Set the enabled state of this node
|
|
|
1263
|
|
|
1264 ***********************************************************************/
|
|
|
1265
|
|
|
1266 final void setEnabled (bool enabled)
|
|
|
1267 {
|
|
|
1268 if (enabled)
|
|
|
1269 log.trace ("enabling "~name);
|
|
|
1270 else
|
|
|
1271 log.trace ("disabling "~name);
|
|
|
1272
|
|
|
1273 volatile
|
|
|
1274 this.enabled = enabled;
|
|
|
1275 }
|
|
|
1276
|
|
|
1277 /***********************************************************************
|
|
|
1278
|
|
|
1279 request data; fail this Node if we can't connect. Note
|
|
|
1280 that we make several attempts to connect before writing
|
|
|
1281 the node off as a failure. We use a delegate to perform
|
|
|
1282 the request output since it may be invoked on more than
|
|
|
1283 one iteration, where the current attempt fails.
|
|
|
1284
|
|
|
1285 We return true if the cluster node responds, and false
|
|
|
1286 otherwise. Exceptions are thrown if they occured on the
|
|
|
1287 server. Parameter 'message' is set true if a message is
|
|
|
1288 available from the server response
|
|
|
1289
|
|
|
1290 ***********************************************************************/
|
|
|
1291
|
|
|
1292 final bool request (Requestor dg, ProtocolReader reader, out bool message)
|
|
|
1293 {
|
|
|
1294 ProtocolWriter.Command cmd;
|
|
|
1295 Time time;
|
|
|
1296 char[] channel;
|
|
|
1297 char[] element;
|
|
|
1298
|
|
|
1299 // it's possible that the pool may have failed between
|
|
|
1300 // the point of selecting it, and the invocation itself
|
|
|
1301 if (pool is null)
|
|
|
1302 return false;
|
|
|
1303
|
|
|
1304 // get a connection to the server
|
|
|
1305 auto connect = pool.borrow (time = Clock.now);
|
|
|
1306
|
|
|
1307 // talk to the server (try a few times if necessary)
|
|
|
1308 for (int attempts=3; attempts--;)
|
|
|
1309 try {
|
|
|
1310 // attach connection to writer and send request
|
|
|
1311 dg (connect.conduit);
|
|
|
1312
|
|
|
1313 // attach connection to reader
|
|
|
1314 reader.buffer.setConduit (connect.conduit);
|
|
|
1315
|
|
|
1316 // load the returned object. Don't retry on
|
|
|
1317 // failed reads, since the server is either
|
|
|
1318 // really really busy, or near death. We must
|
|
|
1319 // assume it is offline until it tells us
|
|
|
1320 // otherwise (via a heartbeat)
|
|
|
1321 attempts = 0;
|
|
|
1322 message = reader.getHeader (cmd, channel, element);
|
|
|
1323
|
|
|
1324 // return borrowed connection
|
|
|
1325 connect.done (time);
|
|
|
1326
|
|
|
1327 } catch (RegistryException x)
|
|
|
1328 {
|
|
|
1329 connect.done (time);
|
|
|
1330 throw x;
|
|
|
1331 }
|
|
|
1332 catch (IOException x)
|
|
|
1333 {
|
|
|
1334 log.trace ("IOException on server request :: "~x.toString);
|
|
|
1335
|
|
|
1336 // attempt to reconnect?
|
|
|
1337 if (attempts is 0 || !connect.reset)
|
|
|
1338 {
|
|
|
1339 // that server is offline
|
|
|
1340 fail;
|
|
|
1341
|
|
|
1342 // state that we failed
|
|
|
1343 return false;
|
|
|
1344 }
|
|
|
1345 }
|
|
|
1346
|
|
|
1347 // is message an exception?
|
|
|
1348 if (cmd !is ProtocolWriter.Command.OK)
|
|
|
1349 {
|
|
|
1350 // is node full?
|
|
|
1351 if (cmd is ProtocolWriter.Command.Full)
|
|
|
1352 throw new ClusterFullException (channel);
|
|
|
1353
|
|
|
1354 // did node barf?
|
|
|
1355 if (cmd is ProtocolWriter.Command.Exception)
|
|
|
1356 throw new ClusterException (channel);
|
|
|
1357
|
|
|
1358 // bogus response
|
|
|
1359 throw new ClusterException ("invalid response from cluster server");
|
|
|
1360 }
|
|
|
1361
|
|
|
1362 // ok, our server responded
|
|
|
1363 return true;
|
|
|
1364 }
|
|
|
1365 }
|
|
|
1366
|
|
|
1367
|
|
|
1368 /*******************************************************************************
|
|
|
1369
|
|
|
1370 Models a generic set of cluster nodes. This is intended to be
|
|
|
1371 thread-safe, with no locking on a lookup operation
|
|
|
1372
|
|
|
1373 *******************************************************************************/
|
|
|
1374
|
|
|
1375 private class NodeSet
|
|
|
1376 {
|
|
|
1377 private Node[char[]] map;
|
|
|
1378 private Logger log;
|
|
|
1379 private Set set;
|
|
|
1380 private bool noDelay;
|
|
|
1381
|
|
|
1382 /***********************************************************************
|
|
|
1383
|
|
|
1384 ***********************************************************************/
|
|
|
1385
|
|
|
1386 this (Logger log, bool noDelay)
|
|
|
1387 {
|
|
|
1388 this.log = log;
|
|
|
1389 this.set = new Set;
|
|
|
1390 this.noDelay = noDelay;
|
|
|
1391 }
|
|
|
1392
|
|
|
1393 /***********************************************************************
|
|
|
1394
|
|
|
1395 ***********************************************************************/
|
|
|
1396
|
|
|
1397 final Logger logger ()
|
|
|
1398 {
|
|
|
1399 return log;
|
|
|
1400 }
|
|
|
1401
|
|
|
1402 /***********************************************************************
|
|
|
1403
|
|
|
1404 Add a node to the list of servers
|
|
|
1405
|
|
|
1406 ***********************************************************************/
|
|
|
1407
|
|
|
1408 synchronized final Node addNode (Node node)
|
|
|
1409 {
|
|
|
1410 auto addr = node.address;
|
|
|
1411 if (addr in map)
|
|
|
1412 throw new ClusterException ("Attempt to add cluster node '"~addr~"' more than once");
|
|
|
1413
|
|
|
1414 map[addr] = node;
|
|
|
1415
|
|
|
1416 // note that this creates a new Set instance. We do this
|
|
|
1417 // so that selectNode() can avoid synchronization
|
|
|
1418 set = set.add (node);
|
|
|
1419 return node;
|
|
|
1420 }
|
|
|
1421
|
|
|
1422 /***********************************************************************
|
|
|
1423
|
|
|
1424 Select a cluster server based on a starting index. If the
|
|
|
1425 selected server is not currently enabled, we just try the
|
|
|
1426 next one. This behaviour should be consistent across each
|
|
|
1427 cluster client.
|
|
|
1428
|
|
|
1429 ***********************************************************************/
|
|
|
1430
|
|
|
1431 final Node selectNode (uint index)
|
|
|
1432 {
|
|
|
1433 auto hosts = set.nodes;
|
|
|
1434 uint count = hosts.length;
|
|
|
1435
|
|
|
1436 if (count)
|
|
|
1437 {
|
|
|
1438 index %= count;
|
|
|
1439
|
|
|
1440 while (count--)
|
|
|
1441 {
|
|
|
1442 auto node = hosts [index];
|
|
|
1443 if (node.isEnabled)
|
|
|
1444 return node;
|
|
|
1445
|
|
|
1446 if (++index >= hosts.length)
|
|
|
1447 index = 0;
|
|
|
1448 }
|
|
|
1449 }
|
|
|
1450 throw new ClusterEmptyException ("No appropriate cluster nodes are available");
|
|
|
1451 }
|
|
|
1452
|
|
|
1453 /***********************************************************************
|
|
|
1454
|
|
|
1455 Host class for the set of nodes. We utilize this to enable
|
|
|
1456 atomic read/write where it would not be otherwise possible
|
|
|
1457 -- D arrays are organized as ptr+length pairs and are thus
|
|
|
1458 inherently non-atomic for assignment purposes
|
|
|
1459
|
|
|
1460 ***********************************************************************/
|
|
|
1461
|
|
|
1462 private static class Set
|
|
|
1463 {
|
|
|
1464 Node[] nodes,
|
|
|
1465 random;
|
|
|
1466
|
|
|
1467 final Set add (Node node)
|
|
|
1468 {
|
|
|
1469 auto s = new Set;
|
|
|
1470 s.nodes = nodes ~ node;
|
|
|
1471 s.randomize;
|
|
|
1472 return s;
|
|
|
1473 }
|
|
|
1474
|
|
|
1475 private final void randomize ()
|
|
|
1476 {
|
|
|
1477 // copy the node list
|
|
|
1478 random = nodes.dup;
|
|
|
1479
|
|
|
1480 // muddle up the duplicate list. This randomized list
|
|
|
1481 // is used when scanning the cluster for queued entries
|
|
|
1482 foreach (i, n; random)
|
|
|
1483 {
|
|
|
1484 auto j = Random.shared.next (random.length);
|
|
|
1485 auto tmp = random[i];
|
|
|
1486 random[i] = random[j];
|
|
|
1487 random[j] = tmp;
|
|
|
1488 }
|
|
|
1489 }
|
|
|
1490 }
|
|
|
1491 }
|
|
|
1492
|
|
|
1493
|
|
|
1494 /*******************************************************************************
|
|
|
1495
|
|
|
1496 Models a fixed set of cluster nodes. Used for Cache
|
|
|
1497
|
|
|
1498 *******************************************************************************/
|
|
|
1499
|
|
|
1500 private class FixedNodeSet : NodeSet
|
|
|
1501 {
|
|
|
1502 /***********************************************************************
|
|
|
1503
|
|
|
1504 ***********************************************************************/
|
|
|
1505
|
|
|
1506 this (Logger log, bool noDelay)
|
|
|
1507 {
|
|
|
1508 super (log, noDelay);
|
|
|
1509 }
|
|
|
1510
|
|
|
1511 /***********************************************************************
|
|
|
1512
|
|
|
1513 ***********************************************************************/
|
|
|
1514
|
|
|
1515 final synchronized void enable (char[] addr)
|
|
|
1516 {
|
|
|
1517 auto p = addr in map;
|
|
|
1518 if (p)
|
|
|
1519 {
|
|
|
1520 auto node = *p;
|
|
|
1521 if (! node.isEnabled)
|
|
|
1522 {
|
|
|
1523 node.setPool (new InternetAddress(addr), noDelay);
|
|
|
1524 node.setEnabled (true);
|
|
|
1525 }
|
|
|
1526 }
|
|
|
1527 else
|
|
|
1528 // don't throw when no cache hosts have been configured at all
|
|
|
1529 if (set.nodes.length)
|
|
|
1530 throw new ClusterException ("Attempt to enable unregistered cache node '"~addr~"'");
|
|
|
1531 }
|
|
|
1532
|
|
|
1533 /***********************************************************************
|
|
|
1534
|
|
|
1535 Select a cluster server based on the specified key. If the
|
|
|
1536 selected server is not currently enabled, we just try the
|
|
|
1537 next one. This behaviour should be consistent across each
|
|
|
1538 cluster client.
|
|
|
1539
|
|
|
1540 ***********************************************************************/
|
|
|
1541
|
|
|
1542 final bool request (Node.Requestor dg, ProtocolReader reader, char[] key)
|
|
|
1543 {
|
|
|
1544 Node node;
|
|
|
1545 bool message;
|
|
|
1546
|
|
|
1547 do {
|
|
|
1548 node = selectNode (jhash (key.ptr, key.length));
|
|
|
1549 } while (! node.request (dg, reader, message));
|
|
|
1550
|
|
|
1551 return message;
|
|
|
1552 }
|
|
|
1553 }
|
|
|
1554
|
|
|
1555
|
|
|
1556 /*******************************************************************************
|
|
|
1557
|
|
|
1558 Models a flexible set of cluster nodes. Used for queue and task
|
|
|
1559
|
|
|
1560 *******************************************************************************/
|
|
|
1561
|
|
|
1562 private class FlexNodeSet : NodeSet
|
|
|
1563 {
|
|
|
1564 private uint rollover;
|
|
|
1565
|
|
|
1566 /***********************************************************************
|
|
|
1567
|
|
|
1568 ***********************************************************************/
|
|
|
1569
|
|
|
1570 this (Logger log, bool noDelay)
|
|
|
1571 {
|
|
|
1572 super (log, noDelay);
|
|
|
1573 }
|
|
|
1574
|
|
|
1575 /***********************************************************************
|
|
|
1576
|
|
|
1577 ***********************************************************************/
|
|
|
1578
|
|
|
1579 final synchronized void enable (char[] addr, char[] name)
|
|
|
1580 {
|
|
|
1581 auto p = addr in map;
|
|
|
1582 auto node = p ? *p : addNode (new Node (log, addr, name));
|
|
|
1583
|
|
|
1584 if (! node.isEnabled)
|
|
|
1585 {
|
|
|
1586 node.setPool (new InternetAddress(addr), noDelay);
|
|
|
1587 node.setEnabled (true);
|
|
|
1588 }
|
|
|
1589 }
|
|
|
1590
|
|
|
1591 /***********************************************************************
|
|
|
1592
|
|
|
1593 Select a cluster server based on the specified key. If the
|
|
|
1594 selected server is not currently enabled, we just try the
|
|
|
1595 next one. This behaviour should be consistent across each
|
|
|
1596 cluster client.
|
|
|
1597
|
|
|
1598 ***********************************************************************/
|
|
|
1599
|
|
|
1600 final bool request (Node.Requestor dg, ProtocolReader reader)
|
|
|
1601 {
|
|
|
1602 Node node;
|
|
|
1603 bool message;
|
|
|
1604
|
|
|
1605 do {
|
|
|
1606 node = selectNode (++rollover);
|
|
|
1607 } while (! node.request (dg, reader, message));
|
|
|
1608
|
|
|
1609 return message;
|
|
|
1610 }
|
|
|
1611
|
|
|
1612 /***********************************************************************
|
|
|
1613
|
|
|
1614 Sweep the cluster servers. Returns true if the delegate
|
|
|
1615 returns true, false otherwise. The sweep is halted when
|
|
|
1616 the delegate returns true. Note that this scans nodes in
|
|
|
1617 a randomized pattern, which should tend to avoid 'bursty'
|
|
|
1618 activity by a set of clients upon any one cluster server.
|
|
|
1619
|
|
|
1620 ***********************************************************************/
|
|
|
1621
|
|
|
1622 final bool scan (bool delegate(Node) dg)
|
|
|
1623 {
|
|
|
1624 auto hosts = set.random;
|
|
|
1625 auto index = hosts.length;
|
|
|
1626
|
|
|
1627 while (index)
|
|
|
1628 {
|
|
|
1629 // lookup the randomized set of server nodes
|
|
|
1630 auto node = hosts [--index];
|
|
|
1631
|
|
|
1632 // callback on each enabled node
|
|
|
1633 if (node.isEnabled)
|
|
|
1634 if (dg (node))
|
|
|
1635 return true;
|
|
|
1636 }
|
|
|
1637 return false;
|
|
|
1638 }
|
|
|
1639 }
|
|
|
1640
|
|
|
1641
|
|
|
1642 /******************************************************************************
|
|
|
1643
|
|
|
1644 The Bob Jenkins lookup2 algorithm. This should be relocated
|
|
|
1645 to somewhere common
|
|
|
1646
|
|
|
1647 ******************************************************************************/
|
|
|
1648
|
|
|
1649 private static uint jhash (void* k, uint len, uint init = 0)
|
|
|
1650 {
|
|
|
1651 uint a = 0x9e3779b9,
|
|
|
1652 b = 0x9e3779b9,
|
|
|
1653 c = init,
|
|
|
1654 i = len;
|
|
|
1655
|
|
|
1656 // handle most of the key
|
|
|
1657 while (i >= 12)
|
|
|
1658 {
|
|
|
1659 a += *cast(uint*)(k+0);
|
|
|
1660 b += *cast(uint*)(k+4);
|
|
|
1661 c += *cast(uint*)(k+8);
|
|
|
1662
|
|
|
1663 a -= b; a -= c; a ^= (c>>13);
|
|
|
1664 b -= c; b -= a; b ^= (a<<8);
|
|
|
1665 c -= a; c -= b; c ^= (b>>13);
|
|
|
1666 a -= b; a -= c; a ^= (c>>12);
|
|
|
1667 b -= c; b -= a; b ^= (a<<16);
|
|
|
1668 c -= a; c -= b; c ^= (b>>5);
|
|
|
1669 a -= b; a -= c; a ^= (c>>3);
|
|
|
1670 b -= c; b -= a; b ^= (a<<10);
|
|
|
1671 c -= a; c -= b; c ^= (b>>15);
|
|
|
1672 k += 12; i -= 12;
|
|
|
1673 }
|
|
|
1674
|
|
|
1675 // handle the last 11 bytes
|
|
|
1676 c += len;
|
|
|
1677 switch (i)
|
|
|
1678 {
|
|
|
1679 case 11: c+=(cast(uint)(cast(ubyte*)k)[10]<<24);
|
|
|
1680 case 10: c+=(cast(uint)(cast(ubyte*)k)[9]<<16);
|
|
|
1681 case 9 : c+=(cast(uint)(cast(ubyte*)k)[8]<<8);
|
|
|
1682 case 8 : b+=(cast(uint)(cast(ubyte*)k)[7]<<24);
|
|
|
1683 case 7 : b+=(cast(uint)(cast(ubyte*)k)[6]<<16);
|
|
|
1684 case 6 : b+=(cast(uint)(cast(ubyte*)k)[5]<<8);
|
|
|
1685 case 5 : b+=(cast(uint)(cast(ubyte*)k)[4]);
|
|
|
1686 case 4 : a+=(cast(uint)(cast(ubyte*)k)[3]<<24);
|
|
|
1687 case 3 : a+=(cast(uint)(cast(ubyte*)k)[2]<<16);
|
|
|
1688 case 2 : a+=(cast(uint)(cast(ubyte*)k)[1]<<8);
|
|
|
1689 case 1 : a+=(cast(uint)(cast(ubyte*)k)[0]);
|
|
|
1690 default:
|
|
|
1691 }
|
|
|
1692
|
|
|
1693 a -= b; a -= c; a ^= (c>>13);
|
|
|
1694 b -= c; b -= a; b ^= (a<<8);
|
|
|
1695 c -= a; c -= b; c ^= (b>>13);
|
|
|
1696 a -= b; a -= c; a ^= (c>>12);
|
|
|
1697 b -= c; b -= a; b ^= (a<<16);
|
|
|
1698 c -= a; c -= b; c ^= (b>>5);
|
|
|
1699 a -= b; a -= c; a ^= (c>>3);
|
|
|
1700 b -= c; b -= a; b ^= (a<<10);
|
|
|
1701 c -= a; c -= b; c ^= (b>>15);
|
|
|
1702
|
|
|
1703 return c;
|
|
|
1704 }
|
|
|
1705
|
