controller

git-svn-id: svn://elaine/murooma/trunk@120 d8a302eb-03bc-478d-80e4-98257eca68ef

client.cpp

@@ -8,6 +8,7 @@
 #include <zmq.hpp>
 #include <iostream>
 #include <sstream>
+#include <fstream>
 #include <sys/time.h>
 #include <unistd.h>
 #include <deque>
@@ -15,8 +16,13 @@
 #include <algorithm>
 #include <thread> 
 #include <portaudio.h>
+#include <iterator>
+
 #include "chunk.h"
 #include "stream.h"
+#include "zhelpers.hpp"
+
+using namespace std;
 
 
 int bufferMs;
@@ -43,6 +49,44 @@ void player()
 
 
 
+std::string getMacAddress()
+{
+	std::ifstream t("/sys/class/net/eth0/address");
+	std::string str((std::istreambuf_iterator<char>(t)),
+                 std::istreambuf_iterator<char>());
+	str.erase(std::find_if(str.rbegin(), str.rend(), std::not1(std::ptr_fun<int, int>(std::isspace))).base(), str.end());
+	return str;
+}
+
+
+
+void control(Stream* stream)
+{
+    zmq::context_t context(1);
+    zmq::socket_t worker (context, ZMQ_REQ);
+    srand (time(NULL));
+    //  We use a string identity for ease here
+    s_set_id (worker);
+    worker.connect("tcp://192.168.0.2:123459");
+
+    //  Tell the router we're ready for work
+    s_send (worker, "ready");
+    while (1) {
+        std::string cmd = s_recv (worker);
+		istringstream iss(cmd);
+		vector<std::string> splitCmd;
+		copy(istream_iterator<string>(iss), istream_iterator<string>(), back_inserter<vector<string> >(splitCmd));
+		for (size_t n=0; n<splitCmd.size(); ++n)
+			std::cout << "cmd: " << splitCmd[n] << "\n";
+		if (splitCmd.size() > 1)
+		{
+			if (splitCmd[0] == "buffer")
+				stream->setBufferLen(atoi(cmd.c_str()));
+		}
+		s_send(worker, "ACK " + cmd);
+    }
+}
+
 
 /* This routine will be called by the PortAudio engine when audio is needed.
 ** It may called at interrupt level on some machines so don't do anything
@@ -154,7 +198,7 @@ int main (int argc, char *argv[])
 	initAudio();
 	std::thread playerThread(player);
 	
-	std::string cmd;
+/*	std::string cmd;
 	while (true && (argc > 3))
 	{
 		std::cout << "> ";
@@ -171,7 +215,9 @@ int main (int argc, char *argv[])
 			stream->setBufferLen(atoi(cmd.c_str()));
 		}
 	}
-
+*/
+	std::thread controlThread(control, stream);
+//	controlThread.join();
 	playerThread.join();
 
     return 0;

rtrClient.cpp

@@ -3,39 +3,62 @@
 //
 // Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
 
+
+#include <thread> 
+#include <vector>
+#include <string>
+#include <fstream>
+#include <iterator>
 #include "zhelpers.hpp"
+#include <netinet/in.h>
+#include <sys/ioctl.h>
+#include <net/if.h> 
 
-#define NBR_WORKERS 10
+using namespace std;
 
 
-int main () {
+std::string getMacAddress()
+{
+	std::ifstream t("/sys/class/net/eth0/address");
+	std::string str((std::istreambuf_iterator<char>(t)),
+                 std::istreambuf_iterator<char>());
+	str.erase(std::find_if(str.rbegin(), str.rend(), std::not1(std::ptr_fun<int, int>(std::isspace))).base(), str.end());
+	return str;
+}
+
+
+
+void control()
+{
     zmq::context_t context(1);
     zmq::socket_t worker (context, ZMQ_REQ);
     srand (time(NULL));
     //  We use a string identity for ease here
-    s_set_id (worker);
+	string macAddress = getMacAddress();
+    worker.setsockopt(ZMQ_IDENTITY, macAddress.c_str(), macAddress.length());
     worker.connect("tcp://127.0.0.1:10000");
 
-    int total = 0;
+    //  Tell the router we're ready for work
+    s_send (worker, "ready");
     while (1) {
-        //  Tell the router we're ready for work
+        std::string cmd = s_recv (worker);
-        s_send (worker, "ready");
+		istringstream iss(cmd);
-
+		vector<std::string> splitCmd;
-        //  Get workload from router, until finished
+		copy(istream_iterator<string>(iss), istream_iterator<string>(), back_inserter<vector<string> >(splitCmd));
-        std::string workload = s_recv (worker);
+		for (size_t n=0; n<splitCmd.size(); ++n)
-std::cout << "workload: " << workload << "\n";
+			std::cout << "cmd: " << splitCmd[n] << "\n";
-        int finished = (workload.compare("END") == 0);
+		s_send(worker, "ACK " + cmd);
-        
-        if (finished) {
-            std::cout << "Processed: " << total << " tasks" << std::endl;
-            break;
-        }
-        total++;
-
-        //  Do some random work
-        s_sleep(within (100) + 1);
     }
+}
+
+
+int main () {
+	cout << getMacAddress() << "\n";
+	std::thread controlThread(control);
+	controlThread.join();
     return 0;
 }
 
 
+
+

rtrServer.cpp

@@ -3,32 +3,45 @@
 //
 // Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>
 
+#include <thread> 
+#include <istream>
 #include "zhelpers.hpp"
 
-#define NBR_WORKERS 10
+
+
+void receiver(zmq::socket_t* client)
+{
+	while (true)
+	{
+		std::string address = s_recv (*client);
+		std::cout << "Address: " << address << "\n";
+	    // receiving and discarding'empty' message
+	    s_recv (*client);
+	    // receiving and discarding 'ready' message
+	    std::string msg = s_recv (*client);
+		std::cout << "msg: " << msg << "\n";
+	}
+}
+
 
 
 int main () {
     zmq::context_t context(2);
     zmq::socket_t client (context, ZMQ_ROUTER);
-    client.bind("tcp://127.0.0.1:10000");
+    client.bind("tcp://0.0.0.0:123459");
+
+	std::thread receiveThread(receiver, &client);
 
-    int task_nbr;
 	while (true)
 	{
-        //  LRU worker is next waiting in queue
+		std::string address;
-        std::string address = s_recv (client);
+		std::string cmd;
-		std::cout << "Address: " << address << "\n";
-		
-        // receiving and discarding'empty' message
-        s_recv (client);
-        // receiving and discarding 'ready' message
-        std::string msg = s_recv (client);
-		std::cout << "msg: " << msg << "\n";
 
+		std::getline(std::cin, address);
+		std::getline(std::cin, cmd);
         s_sendmore (client, address);
         s_sendmore (client, "");
-        s_send (client, std::string("hello: ") + address);
+        s_send (client, cmd);
     }
     return 0;
 }