Changeset 8:acc105036a41 in xplra

Timestamp:

01/04/13 14:24:04 (12 years ago)

Author:

István Váradi <ivaradi@…>

Branch:

default

Phase:

public

Message:

Added the registry of multiple-data queries

Files:

• misc/client.py (modified) (7 diffs)
• misc/test3.sh (added)
• src/xplra/GetDataRefTask.cc (modified) (2 diffs)
• src/xplra/GetMultiDataRefRequest.cc (added)
• src/xplra/GetMultiDataRefRequest.h (added)
• src/xplra/Makefile.am (modified) (1 diff)
• src/xplra/MultiTaskRequest.cc (added)
• src/xplra/MultiTaskRequest.h (added)
• src/xplra/Protocol.h (modified) (3 diffs)
• src/xplra/Request.h (modified) (1 diff)
• src/xplra/ServerThread.cc (modified) (7 diffs)
• src/xplra/ServerThread.h (modified) (5 diffs)

misc/client.py

@@ -39 +39 @@
 
     @staticmethod
+    def _parseGetSpec(words):
+        """Parse a data query specification from the given argument word lisst.
+
+        If there is some failure, it returns None.
+
+        Otherwise it returns a tuple of two items:
+        - the parsed data, which is a tuple of these items:
+          - the name of the dataref to query
+          - the type of the dataref to query (it is a valid type, checked by
+          the function)
+          - the length of the dataref to query
+          - the offset to query the dataref from
+
+        - the remainder of the argument list."""
+        if len(words)<2:
+            print >> sys.stderr, "Missing parameters"
+            return None
+
+        nextIndex = 2
+
+        name = words[0]
+        type = words[1]
+        if type not in CLI._types:
+            print >> sys.stderr, "Invalid type"
+            return None
+
+        length = None
+        offset = None
+        if type in ["ia", "fa", "ba", "s"]:
+            if len(words)<4:
+                print >> sys.stderr, "Missing parameters"
+                return None
+            length = int(words[2])
+            offset = int(words[3])
+            nextIndex = 4
+
+        return ((name, type, length, offset), words[nextIndex:])
+
+    @staticmethod
     def _packLength(x):
         """Pack the given integer as a variable-length value."""
@@ -60 +99 @@
 
         self._stream = stream
+        self._multiGets = {}
 
         self.use_rawinput = True
@@ -78 +118 @@
         """Handle the 'get' command"""
         words = self._splitArgs(args)
-        if len(words)<2:
-            print >> sys.stderr, "Missing parameters"
-            return False
-
-        name = words[0]
-
-        type = words[1]
-        if type not in self._types:
-            print >> sys.stderr, "Invalid type"
-            return False
-
-        length = None
-        offset = None
-        if len(words)>3:
-            length = int(words[2])
-            offset = int(words[3])
+        result = self._parseGetSpec(words)
+        if result is None:
+            return False
+
+        (name, type, length, offset) = result[0]
 
         self._writeU8(0x01)
@@ -107 +136 @@
             print self._readValue(type)
         else:
-            print "Error code:", result
+            print >> sys.stderr, "Error code:", result
 
     def do_set(self, args):
@@ -143 +172 @@
         result = self._readU8()
         if result!=0:
-            print "Error code:", result
+            print >> sys.stderr, "Error code:", result
+
+    def do_reg_multiget(self, args):
+        """Handle the 'reg_multiget' command"""
+        words = self._splitArgs(args)
+        specs = []
+        while words:
+            result = self._parseGetSpec(words)
+            if result is None:
+                return False
+            else:
+                specs.append(result[0])
+                words = result[1]
+
+        if not specs:
+            return False
+
+        self._writeU8(0x11)
+        self._writeU32(len(specs))
+        for (name, type, length, offset) in specs:
+            print name, type
+            self._writeString(name)
+            self._writeU8(self._types[type])
+            if length is not None:
+                self._writeS32(length)
+                self._writeS32(offset)
+        self._flush()
+
+        result = self._readU8()
+        if result==0:
+            id = self._readU32()
+            self._multiGets[id] = \
+                [type for (name, type, length, offset) in specs]
+            print "ID:", id
+        else:
+            print >> sys.stderr, "Error code:", result
+
+    def do_unreg_multiget(self, args):
+        """Handle the 'unreg_multiget' command"""
+        words = self._splitArgs(args)
+        if len(words)<1:
+            print >> sys.stderr, "Missing parameter"
+            return False
+
+        id = int(words[0])
+
+        self._writeU8(0x12)
+        self._writeU32(id)
+        self._flush()
+
+        result = self._readU8()
+        if result!=0:
+            print >> sys.stderr, "Error code:", result
+
+        if id in self._multiGets:
+            del self._multiGets[id]
+
+    def do_exec_multiget(self, args):
+        """Handle the 'unreg_multiget' command"""
+        words = self._splitArgs(args)
+        if len(words)<1:
+            print >> sys.stderr, "Missing parameter"
+            return False
+
+        id = int(words[0])
+        if id not in self._multiGets:
+            print >> sys.stderr, "Invalid ID"
+            return False
+
+        self._writeU8(0x13)
+        self._writeU32(id)
+        self._flush()
+
+        result = self._readU8()
+        if result==0:
+            for type in self._multiGets[id]:
+                value = self._readValue(type)
+                print value
+        elif result==0x02:
+            index = self._readU32()
+            print >> sys.stderr, "Invalid dataref at #%d" % (index,)
+        else:
+            print >> sys.stderr, "Error code:", result
 
     def _writeU8(self, x):
@@ -153 +264 @@
         self._stream.write(struct.pack("i", x))
 
+    def _writeU32(self, x):
+        """Write the given value as an unsigned, 32-bit value."""
+        self._stream.write(struct.pack("I", x))
+
     def _writeFloat(self, x):
         """Write the given value as a single-precision floating point."""
@@ -181 +296 @@
         """Read a signed, 32-bit value from the stream."""
         (value,) = struct.unpack("i", self._stream.read(4))
+        return value
+
+    def _readU32(self):
+        """Read an unsigned, 32-bit value from the stream."""
+        (value,) = struct.unpack("I", self._stream.read(4))
         return value
 

src/xplra/GetDataRefTask.cc

@@ -42 +42 @@
 
 using xplcommon::DataStream;
+using xplcommon::Util;
 
 using std::string;
@@ -54 +55 @@
     uint8_t type = stream.readU8();
     if (!stream) return 0;
+
+    // Util::debug("hu.varadiistvan.xplra.GetDataRefTask::execute: name='%s', type=0x%02x\n",
+    //             name.c_str(), type);
 
     if (type==Protocol::TYPE_INT) {

src/xplra/Makefile.am

@@ -24 +24 @@
 
 libxplra_la_SOURCES= \
-        plugin.cc               \
-        ListenThread.cc         \
-        RequestQueue.cc         \
-        ServerThread.cc         \
-        TaskRequest.cc          \
-        DataRefTask.cc          \
-        GetDataRefTask.cc       \
-        SetDataRefTask.cc
+        plugin.cc                       \
+        ListenThread.cc                 \
+        RequestQueue.cc                 \
+        ServerThread.cc                 \
+        TaskRequest.cc                  \
+        DataRefTask.cc                  \
+        GetDataRefTask.cc               \
+        SetDataRefTask.cc               \
+        MultiTaskRequest.cc             \
+        GetMultiDataRefRequest.cc
 
 libxplra_la_LIBADD=$(LIBXPLCOMMON_LIBS)
 
 noinst_HEADERS= \
-        ListenThread.h          \
-        Request.h               \
-        RequestQueue.h          \
-        Protocol.h              \
-        ServerThread.h          \
-        Task.h                  \
-        TaskRequest.h           \
-        DataRefTask.h           \
-        GetDataRefTask.h        \
-        SetDataRefTask.h
+        ListenThread.h                  \
+        Request.h                       \
+        RequestQueue.h                  \
+        Protocol.h                      \
+        ServerThread.h                  \
+        Task.h                          \
+        TaskRequest.h                   \
+        DataRefTask.h                   \
+        GetDataRefTask.h                \
+        SetDataRefTask.h                \
+        MultiTaskRequest.h              \
+        GetMultiDataRefRequest.h

src/xplra/Protocol.h

@@ -50 +50 @@
      */
     static const uint8_t COMMAND_SET_SINGLE = 0x02;
+
+    /**
+     * Command: register a multiple-data query request
+     */
+    static const uint8_t COMMAND_REGISTER_GET_MULTI = 0x11;
+
+    /**
+     * Command: unregister a multiple-data query request
+     */
+    static const uint8_t COMMAND_UNREGISTER_GET_MULTI = 0x12;
+
+    /**
+     * Command: execute a registered multiple-data query request
+     */
+    static const uint8_t COMMAND_EXECUTE_GET_MULTI = 0x13;
 
     /**
@@ -112 +127 @@
 
     /**
+     * Result code: invalid count
+     */
+    static const uint8_t RESULT_INVALID_COUNT = 0x06;
+
+    /**
+     * Result code: invalid ID
+     */
+    static const uint8_t RESULT_INVALID_ID = 0x07;
+
+    /**
      * Result code: other error
      */
@@ -120 +145 @@
      */
     static const int MAX_LENGTH = 2048;
+
+    /**
+     * The maximal count of requests in a multiple-data query or
+     * update.
+     */
+    static const size_t MAX_MULTI_COUNT = 1024;
 };
 

src/xplra/Request.h

@@ -58 +58 @@
     virtual ~Request();
 
+protected:
     /**
      * Execute the request. Make this very fast!

src/xplra/ServerThread.cc

@@ -36 +36 @@
 #include "SetDataRefTask.h"
 #include "TaskRequest.h"
+#include "GetMultiDataRefRequest.h"
 
 #include <xplcommon/Util.h>
@@ -74 +75 @@
     requestQueue(requestQueue),
     bufferedStream(acceptor.getSocket(&waiter)),
-    stream(*bufferedStream)
+    stream(*bufferedStream),
+    nextGetMultiRequestID(1)
 {
     instancesMutex.lock();
@@ -86 +88 @@
 {
     delete bufferedStream;
+
+    for(getMultiRequests_t::iterator i = getMultiRequests.begin();
+        i!=getMultiRequests.end(); ++i)
+    {
+        delete i->second;
+    }
 
     instancesMutex.lock();
@@ -115 +123 @@
         } else if (command==Protocol::COMMAND_SET_SINGLE) {
             if (!handleSetSingle()) break;
+        } else if (command==Protocol::COMMAND_REGISTER_GET_MULTI) {
+            if (!handleRegisterGetMulti()) break;
+        } else if (command==Protocol::COMMAND_UNREGISTER_GET_MULTI) {
+            if (!handleUnregisterGetMulti()) break;
+        } else if (command==Protocol::COMMAND_EXECUTE_GET_MULTI) {
+            if (!handleExecuteGetMulti()) break;
         } else {
             stream.writeU8(Protocol::RESULT_INVALID_COMMAND);
@@ -131 +145 @@
     GetDataRefTask* task = GetDataRefTask::create(result, stream);
 
-    if (!stream) return false;
-    else if (task==0) {
+    if (!stream) {
+        return false;
+    } else if (task==0) {
         stream.writeU8(result);
         return true;
@@ -157 +172 @@
     SetDataRefTask* task = SetDataRefTask::create(result, stream);
 
-    if (!stream) return false;
-    else if (task==0) {
+    if (!stream) {
+        return false;
+    } else if (task==0) {
         stream.writeU8(result);
         return true;
@@ -168 +184 @@
     stream.writeU8(task->isValid() ? Protocol::RESULT_OK :
                    Protocol::RESULT_UNKNOWN_DATAREF);
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+bool ServerThread::handleRegisterGetMulti()
+{
+    uint32_t numTasks = stream.readU32();
+    if (!stream) {
+        return false;
+    } else if (numTasks==0 || numTasks>Protocol::MAX_MULTI_COUNT) {
+        stream.writeU8(Protocol::RESULT_INVALID_COUNT);
+        return true;
+    }
+
+    uint8_t result = Protocol::RESULT_OK;
+    GetMultiDataRefRequest* request =
+        new GetMultiDataRefRequest(result, numTasks, stream);
+    if (result!=Protocol::RESULT_OK || !stream) {
+        delete request;
+        stream.writeU8(result);
+        return stream;
+    }
+
+    size_t id = nextGetMultiRequestID++;
+    getMultiRequests[id] = request;
+
+    stream.writeU8(Protocol::RESULT_OK);
+    stream.writeU32(id);
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+bool ServerThread::handleUnregisterGetMulti()
+{
+    uint32_t id = stream.readU32();
+    if (!stream) return false;
+
+    getMultiRequests_t::iterator i = getMultiRequests.find(id);
+    if (i==getMultiRequests.end()) {
+        stream.writeU8(Protocol::RESULT_INVALID_ID);
+    } else {
+        GetMultiDataRefRequest* request = i->second;
+        getMultiRequests.erase(i);
+        delete request;
+        stream.writeU8(Protocol::RESULT_OK);
+    }
+
+    return true;
+}
+
+//------------------------------------------------------------------------------
+
+bool ServerThread::handleExecuteGetMulti()
+{
+    uint32_t id = stream.readU32();
+    if (!stream) return false;
+
+    getMultiRequests_t::iterator i = getMultiRequests.find(id);
+    if (i==getMultiRequests.end()) {
+        stream.writeU8(Protocol::RESULT_INVALID_ID);
+    } else {
+        GetMultiDataRefRequest* request = i->second;
+        if (!requestQueue.execute(request)) return false;
+        request->writeResult(stream);
+    }
+
     return true;
 }

src/xplra/ServerThread.h

@@ -41 +41 @@
 
 #include <set>
+#include <map>
 
 //------------------------------------------------------------------------------
@@ -49 +50 @@
 
 class RequestQueue;
+
+class GetMultiDataRefRequest;
 
 //------------------------------------------------------------------------------
@@ -62 +65 @@
      */
     typedef std::set<ServerThread*> instances_t;
+
+    /**
+     * Type for the registered multiple-data requests.
+     */
+    typedef std::map<size_t, GetMultiDataRefRequest*> getMultiRequests_t;
 
     /**
@@ -100 +108 @@
     xplcommon::DataStream stream;
 
+    /**
+     * The ID of the next multiple-data query request.
+     */
+    size_t nextGetMultiRequestID;
+
+    /**
+     * The registered multiple-data query requests.
+     */
+    getMultiRequests_t getMultiRequests;
+
 public:
     /**
@@ -137 +155 @@
      */
     bool handleSetSingle();
+
+    /**
+     * Handle the COMMAND_REGISTER_GET_MULTI command
+     *
+     * @return true, if we can continue, false if the thread should quit
+     */
+    bool handleRegisterGetMulti();
+
+    /**
+     * Handle the COMMAND_UNREGISTER_GET_MULTI command
+     *
+     * @return true, if we can continue, false if the thread should quit
+     */
+    bool handleUnregisterGetMulti();
+
+    /**
+     * Handle the COMMAND_EXECUTE_GET_MULTI command
+     *
+     * @return true, if we can continue, false if the thread should quit
+     */
+    bool handleExecuteGetMulti();
 };