[svn r136] MAJOR UNSTABLE UPDATE!!!

Initial commit after moving to Tango instead of Phobos.

Lots of bugfixes...

This build is not suitable for most things.

tango/example/cluster/Add.d

@@ -0,0 +1,60 @@
+/*******************************************************************************
+
+*******************************************************************************/
+
+public import tango.net.cluster.NetworkCall;
+
+
+/*******************************************************************************
+        
+        a Task function
+
+*******************************************************************************/
+
+real add (real x, real y)
+{
+        return x + y;
+}
+
+
+/*******************************************************************************
+
+        a Task function
+
+*******************************************************************************/
+
+int divide (int x, int y)
+{
+        return x / y;
+}
+
+
+/*******************************************************************************
+
+        a verbose Task message
+
+*******************************************************************************/
+
+class Subtract : NetworkCall
+{
+        double  a,
+                b,
+                result;
+
+        double opCall (double a, double b, IChannel channel = null)
+        {
+                this.a = a;
+                this.b = b;
+                send (channel);
+                return result;
+        }
+
+        override void execute ()
+        {
+                result = a - b;
+        }
+
+        override void read  (IReader input)  {input (a)(b)(result);}
+
+        override void write (IWriter output) {output (a)(b)(result);}
+}

tango/example/cluster/alert.d

@@ -0,0 +1,36 @@
+private import  tango.core.Thread;
+
+private import  tango.util.log.Configurator;
+
+private import  tango.net.cluster.NetworkAlert;
+
+private import  tango.net.cluster.tina.Cluster;
+
+/*******************************************************************************
+
+        How to send and recieve Alert messages using tango.net.cluster
+
+*******************************************************************************/
+
+void main()
+{
+        // hook into the cluster
+        auto cluster = (new Cluster).join;
+
+        // hook into the Alert layer
+        auto alert = new NetworkAlert (cluster, "my.kind.of.alert");
+
+        // listen for the broadcast (on this channel)
+        alert.createConsumer (delegate void (IEvent event)
+                             {event.log.info ("Recieved alert on channel " ~ event.channel.name);}
+                             );
+
+        // say what's going on
+        alert.log.info ("broadcasting alert");
+
+        // and send everyone an empty alert (on this channel)
+        alert.broadcast;
+
+        // wait for it to arrive ...
+        Thread.sleep(1);
+}

tango/example/cluster/cclient.d

@@ -0,0 +1,48 @@
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+import tango.io.Stdout;
+
+import tango.time.StopWatch;
+
+import tango.util.log.Configurator;
+
+import tango.net.cluster.NetworkCache;
+
+import tango.net.cluster.tina.Cluster;
+
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        StopWatch w;
+
+        if (args.length > 1)
+           {
+           auto cluster = (new Cluster).join (args[1..$]);
+           auto cache   = new NetworkCache (cluster, "my.cache.channel");
+
+           while (true)
+                 {
+                 w.start;
+                 for (int i=10000; i--;)
+                      cache.put ("key", cache.EmptyMessage);
+
+                 Stdout.formatln ("{} put/s", 10000/w.stop);
+
+                 w.start;
+                 for (int i=10000; i--;)
+                      cache.get ("key");
+        
+                 Stdout.formatln ("{} get/s", 10000/w.stop);
+                 }
+           }
+        else
+           Stdout.formatln ("usage: cache cachehost:port ...");
+}
+

tango/example/cluster/cserver.d

@@ -0,0 +1,30 @@
+/*******************************************************************************
+
+*******************************************************************************/
+
+import tango.io.Console;
+
+import tango.net.InternetAddress;
+
+import tango.net.cluster.tina.CmdParser,
+       tango.net.cluster.tina.CacheServer;
+
+/*******************************************************************************
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        auto arg = new CmdParser ("cache.server");
+
+        // default number of cache entries
+        arg.size = 8192;
+
+        if (args.length > 1)
+            arg.parse (args[1..$]);
+                        
+        if (arg.help)
+            Cout ("usage: cacheserver -port=number -size=cachesize -log[=trace, info, warn, error, fatal, none]").newline;
+        else
+           (new CacheServer(new InternetAddress(arg.port), arg.log, arg.size)).start;
+}

tango/example/cluster/invalidate.d

@@ -0,0 +1,38 @@
+private import  tango.core.Thread;
+
+private import  tango.util.log.Configurator;
+
+private import  tango.net.cluster.tina.Cluster;
+
+private import  tango.net.cluster.QueuedCache,
+                tango.net.cluster.CacheInvalidatee,
+                tango.net.cluster.CacheInvalidator;
+
+/*******************************************************************************
+
+        Demonstrates how to invalidate cache entries across a cluster
+        via a channel
+
+*******************************************************************************/
+
+void main()
+{
+        // access the cluster
+        auto cluster = (new Cluster).join;
+
+        // wrap a cache instance with a network listener
+        auto dst = new CacheInvalidatee (cluster, "my.cache.channel", new QueuedCache!(char[], IMessage)(101));
+
+        // connect an invalidator to that cache channel
+        auto src = new CacheInvalidator (cluster, "my.cache.channel");
+
+        // stuff something in the local cache
+        dst.cache.put ("key", dst.EmptyMessage);
+
+        // get it removed via a network broadcast
+        src.log.info ("invalidating 'key' across the cluster");
+        src.invalidate ("key");
+
+        // wait for it to arrive ...
+        Thread.sleep (1);
+}

tango/example/cluster/qclient.d

@@ -0,0 +1,44 @@
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+import tango.io.Stdout;
+
+import tango.time.StopWatch;
+
+import tango.util.log.Configurator;
+
+import tango.net.cluster.NetworkQueue;
+
+import tango.net.cluster.tina.Cluster;
+
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        StopWatch w;
+
+        auto cluster = (new Cluster).join;
+        auto queue   = new NetworkQueue (cluster, "my.queue.channel");
+
+        while (true)
+              {
+              w.start;
+              for (int i=10000; i--;)
+                   queue.put (queue.EmptyMessage);
+
+              Stdout.formatln ("{} put/s", 10000/w.stop);
+
+              uint count;
+              w.start;
+              while (queue.get !is null)
+                     ++count;
+        
+              Stdout.formatln ("{} get/s", count/w.stop);
+              }
+}
+

tango/example/cluster/qlisten.d

@@ -0,0 +1,41 @@
+private import  tango.core.Thread;
+
+private import  tango.util.log.Configurator;
+
+private import  tango.net.cluster.NetworkQueue;
+
+private import  tango.net.cluster.tina.Cluster;
+
+/*******************************************************************************
+
+        Illustrates how to setup and use a Queue in asynchronous mode
+
+*******************************************************************************/
+
+void main ()
+{
+        void listen (IEvent event)
+        {
+                while (event.get)
+                       event.log.info ("received asynch msg on channel " ~ event.channel.name);
+        }
+                
+
+        // join the cluster 
+        auto cluster = (new Cluster).join;
+
+        // access a queue of the specified name
+        auto queue = new NetworkQueue (cluster, "my.queue.channel");
+
+        // listen for messages placed in my queue, via a delegate
+        queue.createConsumer (&listen);
+
+        // stuff something into the queue
+        queue.log.info ("sending three messages to the queue");
+        queue.put (queue.EmptyMessage);
+        queue.put (queue.EmptyMessage);
+        queue.put (queue.EmptyMessage);
+
+        // wait for asynchronous msgs to arrive ...
+        Thread.sleep (1);
+}

tango/example/cluster/qrequest.d

@@ -0,0 +1,30 @@
+private import  tango.util.log.Configurator;
+
+private import  tango.net.cluster.NetworkQueue;
+
+private import  tango.net.cluster.tina.Cluster;
+
+/*******************************************************************************
+
+        Illustrates how to setup and use a Queue in synchronous mode
+
+*******************************************************************************/
+
+void main ()
+{
+        // join the cluster 
+        auto cluster = (new Cluster).join;
+
+        // access a queue of the specified name
+        auto queue = new NetworkQueue (cluster, "my.queue.channel");
+
+        // stuff something into the queue
+        queue.log.info ("sending three messages to the queue");
+        queue.put (queue.EmptyMessage);
+        queue.put (queue.EmptyMessage);
+        queue.put (queue.EmptyMessage);
+
+        // retreive synchronously
+        while (queue.get)
+               queue.log.info ("retrieved msg");
+}

tango/example/cluster/qserver.d

@@ -0,0 +1,27 @@
+/*******************************************************************************
+
+*******************************************************************************/
+
+import tango.io.Console;
+
+import tango.net.InternetAddress;
+
+import tango.net.cluster.tina.CmdParser,
+       tango.net.cluster.tina.QueueServer;
+
+/*******************************************************************************
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        auto arg = new CmdParser ("queue.server");
+
+        if (args.length > 1)
+            arg.parse (args[1..$]);
+                
+        if (arg.help)
+            Cout ("usage: queueserver -port=number -log[=trace, info, warn, error, fatal, none]").newline;
+        else
+           (new QueueServer(new InternetAddress(arg.port), arg.log)).start;
+}

tango/example/cluster/reply.d

@@ -0,0 +1,38 @@
+private import  tango.core.Thread;
+
+private import  tango.util.log.Configurator;
+
+private import  tango.net.cluster.tina.Cluster;
+
+private import  tango.net.cluster.NetworkQueue;
+
+/*******************************************************************************
+
+*******************************************************************************/
+
+void main()
+{
+        // open the cluster and a queue channel. Note that the queue has
+        // been configured with a reply listener ...
+        auto cluster = (new Cluster).join;
+        auto queue = new NetworkQueue (cluster, "message.channel", 
+                                      (IEvent event){event.log.info ("Received reply");}
+                                      );
+
+        void recipient (IEvent event)
+        {
+                auto msg = event.get;
+        
+                event.log.info ("Replying to message on channel "~msg.reply);
+                event.reply (event.replyChannel(msg), queue.EmptyMessage);
+        }
+
+        // setup a listener to recieve and reply
+        queue.createConsumer (&recipient);
+
+        // toss a message out to the cluster
+        queue.put (queue.EmptyMessage);
+
+        // wait for completion ...
+        Thread.sleep (1);
+}

tango/example/cluster/task.d

@@ -0,0 +1,15 @@
+/*******************************************************************************
+
+        Illustrates usage of cluster tasks
+
+*******************************************************************************/
+
+import Add, tango.io.Stdout, tango.net.cluster.tina.ClusterTask;
+
+void main (char[][] args)
+{
+        scope add = new NetCall!(add);
+
+        Stdout.formatln ("cluster expression of 3.0 + 4.0 = {}", add(3, 4));
+}
+

tango/example/cluster/tclient.d

@@ -0,0 +1,42 @@
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+import Add;
+
+import tango.io.Stdout;
+
+import tango.time.StopWatch;
+
+import tango.util.log.Configurator;
+
+import tango.net.cluster.tina.ClusterTask;
+
+/*******************************************************************************
+
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        // an implicit task instance
+        auto add = new NetCall!(add);
+
+        // an explicit task instance
+        auto sub = new Subtract;
+
+        StopWatch w;
+        while (true)
+              {
+              w.start;
+              for (int i=10000; i--;)
+                  {
+                  // both task types are used in the same manner
+                  add (1, 2);
+                  sub (3, 4);
+                  }
+              Stdout.formatln ("{} calls/s", 20000/w.stop);
+              }
+}
+

tango/example/cluster/tserver.d

@@ -0,0 +1,34 @@
+/*******************************************************************************
+
+*******************************************************************************/
+
+import tango.io.Console;
+
+import tango.net.InternetAddress;
+
+import tango.net.cluster.tina.CmdParser,
+       tango.net.cluster.tina.TaskServer;
+
+import Add;
+
+/*******************************************************************************
+
+*******************************************************************************/
+
+void main (char[][] args)
+{
+        auto arg = new CmdParser ("task.server");
+
+        if (args.length > 1)
+            arg.parse (args[1..$]);
+
+        if (arg.help)
+            Cout ("usage: taskserver -port=number -log[=trace, info, warn, error, fatal, none]").newline;
+        else
+           {
+           auto server = new TaskServer (new InternetAddress(arg.port), arg.log);
+           server.enroll (new NetCall!(add));
+           server.enroll (new Subtract);
+           server.start;
+           }
+}