send/resp:通信增加了返回确认

Author: miaobyte

excuter/common/src/client/udpserver.cpp

@@ -42,7 +42,7 @@ namespace client
         while (true)
         {
             len = sizeof(cliaddr);
-            n = recvfrom(sockfd, (char *)buffer, 1024, 0, (struct sockaddr *)&cliaddr, &len);
+            n = recvfrom(sockfd, (char *)buffer, sizeof(buffer), 0, (struct sockaddr *)&cliaddr, &len);
             buffer[n] = '\0';
 
             // 新增换行拆分逻辑
@@ -52,7 +52,9 @@ namespace client
                 if (!line.empty()) {
                     cout << "~" << line << endl;
                     char *IR = const_cast<char *>(line.c_str());
-                    func(IR);
+                    string strresp=func(IR);
+                    sendto(sockfd, strresp.c_str(), strresp.size(), 0,
+                   (const struct sockaddr*)&cliaddr, len);
                 }
             }
         }

excuter/common/src/client/udpserver.hpp

@@ -22,7 +22,7 @@ namespace client{
         udpserver(int port);
         ~udpserver();
         void start();
-        using handlefunc = std::function<void(char *buffer)>;
+        using handlefunc = std::function<std::string(const char *buffer)>;
         handlefunc func;
     };
 

excuter/common/src/deepx/op/op.cpp

@@ -2,83 +2,123 @@
 
 namespace deepx::op
 {
-    void Op::load(const char *str)
-    {
-        // 新格式示例：mul@float32 a(a_grad) b(b_grad) -> a(a_grad) requires_grad=true
+    //与deepx/front/py/deepx/nn/deepxir.py对应
+
+     // 新格式示例：mul@float32 a(a_grad) b(b_grad) -> a(a_grad) //id=1 create_time=1714512000 send_time=1714512000 recv_time=1714512000
+    void Op::load(const char* str) {
         string input(str);
-        size_t arrow_pos = input.find("->");
-        string head = input.substr(0, arrow_pos);
-        string tail = arrow_pos != string::npos ? input.substr(arrow_pos + 2) : "";
+        
+        // 分割元数据部分
+        size_t meta_pos = input.find("//");
+        string body = input.substr(0, meta_pos);
+        string meta = (meta_pos != string::npos) ? input.substr(meta_pos + 2) : "";
+
+        // 解析操作主体
+        size_t arrow_pos = body.find("->");
+        if (arrow_pos == string::npos) {
+            arrow_pos = body.find("<-");
+            if (arrow_pos != string::npos) {
+                grad = true;  // 反向传播标记
+            }
+        }
+        
+        if (arrow_pos == string::npos) {
+            throw runtime_error("Invalid IR format: missing arrow");
+        }
 
-        // 解析操作名和类型
+        string head = body.substr(0, arrow_pos);
+        string tail = body.substr(arrow_pos + 2);
+
+        // 解析操作名和数据类型
         size_t at_pos = head.find('@');
-        if (at_pos != string::npos)
-        {
+        if (at_pos != string::npos) {
             name = head.substr(0, at_pos);
-            dtype = head.substr(at_pos + 1, head.find(' ') - at_pos - 1);
-            head = head.substr(head.find(' ') + 1);
-        }
-        else
-        {
-            name = head.substr(0, head.find(' '));
-            dtype = "any";
-            head = head.substr(name.size() + 1);
+            size_t space_pos = head.find(' ', at_pos);
+            if (space_pos != string::npos) {
+                dtype = head.substr(at_pos + 1, space_pos - at_pos - 1);
+                head = head.substr(space_pos + 1);
+            } else {
+                dtype = head.substr(at_pos + 1);
+                head.clear();
+            }
+        } else {
+            size_t space_pos = head.find(' ');
+            if (space_pos != string::npos) {
+                name = head.substr(0, space_pos);
+                head = head.substr(space_pos + 1);
+                dtype = "any";
+            } else {
+                name = head;
+                head.clear();
+                dtype = "any";
+            }
         }
 
-        // 解析输入参数（支持带括号的梯度名）
+        // 解析输入参数
         stringstream head_ss(head);
         string token;
-        while (head_ss >> token)
-        {
-            size_t bracket = token.find('(');
-            if (bracket != string::npos)
-            {
+        while (head_ss >> token) {
+             size_t bracket = token.find('(');
+            if (bracket != string::npos && token.back() == ')') {
                 args.push_back(token.substr(0, bracket));
-                args_grad.push_back(token.substr(bracket + 1, token.find(')') - bracket - 1));
-                require_grad = true;
-            }
-            else
-            {
+                args_grad.push_back(token.substr(bracket + 1, token.size() - bracket - 2));
+            } else {
                 args.push_back(token);
+                args_grad.emplace_back("");  // 保持梯度与参数数量一致
             }
         }
 
-        // 解析输出参数和标志
+        // 解析输出参数
         stringstream tail_ss(tail);
-        while (tail_ss >> token)
-        {
-            if (token.find('(') != string::npos)
-            {
-                size_t bracket = token.find('(');
+        while (tail_ss >> token) {
+            size_t bracket = token.find('(');
+            if (bracket != string::npos && token.back() == ')') {
                 returns.push_back(token.substr(0, bracket));
-                returns_grad.push_back(token.substr(bracket + 1, token.find(')') - bracket - 1));
-            }
-            else if (token == "requires_grad=true")
-            {
-                require_grad = true;
-            }
-            else
-            {
+                returns_grad.push_back(token.substr(bracket + 1, token.size() - bracket - 2));
+            } else {
                 returns.push_back(token);
+                returns_grad.emplace_back("");  // 保持梯度与参数数量一致
+            }
+        }
+
+        // 解析元数据
+        if (!meta.empty()) {
+            stringstream meta_ss(meta);
+            string key, value;
+            while (meta_ss >> key) {
+                size_t eq_pos = key.find('=');
+                if (eq_pos != string::npos) {
+                    value = key.substr(eq_pos + 1);
+                    key = key.substr(0, eq_pos);
+                    
+                    if (key == "id") {
+                        id = stoi(value);
+                    } else if (key == "created_at") {
+                        created_at = system_clock::from_time_t(stod(value));
+                    } else if (key == "sent_at") {
+                        sent_at = system_clock::from_time_t(stod(value));
+                    }
+                }
             }
         }
     }
 
+ 
     void Op::init(const string &opname,
                   const string &dtype,
                   const vector<string> &args,
                   const vector<string> &returns,
-                  bool require_grad,
+                  bool grad,
                   const vector<string> &args_grad,
                   const vector<string> &returns_grad)
     {
         this->name = opname;
         this->dtype = dtype;
         this->args = args;
         this->returns = returns;
-        this->require_grad = require_grad;
+        this->grad = grad;
 
-        if (require_grad)
+        if (grad)
         {
             // 如果提供了梯度变量名,就使用提供的名字
             if (!args_grad.empty())

excuter/common/src/deepx/op/op.hpp

@@ -7,7 +7,7 @@
 #include <string>
 #include <iostream>
 #include <sstream>
-
+#include <chrono>
 #include "deepx/tensor.hpp"
 #include "deepx/mem/mem.hpp"
 #include "deepx/dtype.hpp"
@@ -16,18 +16,21 @@ namespace deepx::op
 {
     using deepx::mem::Mem;
     using namespace std;
-
+    using namespace std::chrono;
     class Op
     {
     public:
         string name;
         string dtype;
         vector<string> args;
         vector<string> args_grad;
-        bool require_grad;
+        bool grad=false;
         vector<string> returns;
         vector<string> returns_grad;
-
+        int id;
+        system_clock::time_point created_at;
+        system_clock::time_point sent_at;
+        system_clock::time_point recv_at;
     public:
         Op() = default;
         Op(const Op &) = default;
@@ -56,7 +59,7 @@ namespace deepx::op
                   const string &dtype,
                   const vector<string> &args,
                   const vector<string> &returns,
-                  bool require_grad,
+                  bool  grad,
                   const vector<string> &args_grad,
                   const vector<string> &returns_grad);
     };

excuter/op-mem-ompsimd/src/client/main.cpp

@@ -1,9 +1,6 @@
 #include <mutex>
 
 #include <deepx/tensorfunc/init.hpp>
-#include <deepx/tensorfunc/new.hpp>
-#include <deepx/tensorfunc/print.hpp>
-
 #include "deepx/op/op.hpp"
 #include "deepx/op/opfactory.hpp"
 #include "deepx/mem/mem.hpp"
@@ -21,25 +18,31 @@ int main()
     deepx::op::OpFactory opfactory;
     register_all(opfactory);
 
-    server.func = [&mem, &opfactory, &memmutex](const char *buffer)
+    server.func = [&mem, &opfactory, &memmutex](const char *buffer)->std::string
     {
         deepx::op::Op op;
+        op.recv_at = chrono::system_clock::now();
         op.load(buffer);
         if (opfactory.ops.find(op.name)==opfactory.ops.end()){
             cout<<"<op> "<<op.name<<" not found"<<endl;
-            return;
+            return "error";
         }
         auto &type_map = opfactory.ops.find(op.name)->second;
         if (type_map.find(op.dtype)==type_map.end()){
             cout<<"<op>"<<op.name<<" "<<op.dtype<<" not found"<<endl;
-            return;
+            return "error";
         }
         auto src = type_map.find(op.dtype)->second;
- 
-        (*src).init(op.name, op.dtype, op.args, op.returns, op.require_grad, op.args_grad, op.returns_grad);
+
+        (*src).init(op.name, op.dtype, op.args, op.returns, op.grad, op.args_grad, op.returns_grad);
         memmutex.lock();
-        (*src).forward(mem);
+        if (op.grad) {
+            (*src).backward(mem);
+        }else {
+           (*src).forward(mem);
+        }
         memmutex.unlock();
+        return to_string(op.id);
     };
     server.start();
     return 0;

excuter/op-mem-ompsimd/src/client/udpsend.py

@@ -6,7 +6,6 @@
 
 
 namespace deepx::op{
- 
     template <typename T>
     class Transpose : public OpT<T> {
     public:

excuter/op-mem-ompsimd/src/deepx/op/transpose.hpp

@@ -1,5 +1,7 @@
 from typing import Tuple, List, Optional
+import time
 
+_id_counter=0
 class DeepxIR:
     def __init__(self, 
                 name:str,
@@ -37,12 +39,9 @@ def __init__(self,
         self._grad = grad
         self._args_grad = args_grad if grad else []
         self._returns_grad = returns_grad if grad else []
-
-    def forward(self, *input) -> Tuple:
-        raise NotImplementedError
- 
-    def backward(self, *grad_outputs) -> Tuple:
-        raise NotImplementedError
+        self._id=++_id_counter
+        self._created_at=time.time()
+        self._sent_at=None
 
     def __str__(self):
         if self._dtype == None or self._dtype == '':
@@ -67,6 +66,10 @@ def __str__(self):
             if self._grad and self._returns_grad[i]:
                 ret_part += f"({self._returns_grad[i]})"
             parts.append(ret_part)
-        
+
+        parts.append("//")
+        parts.append(f"id={self._id}")
+        parts.append(f"created_at={self._created_at}")
+        parts.append(f"sent_at={self._sent_at}")
         return ' '.join(parts)
 

front/py/deepx/nn/deepxir.py

@@ -7,6 +7,6 @@ def relu(t: Tensor,inplace:bool=False)->Tensor:
     if not inplace:
         out=Tensor(shape=t.shape, dtype=t.dtype, device=t.device)
     ir=DeepxIR("max_scalar",t.dtype,[t._node.name,0], [out._node.name])
-    send(str(ir))
+    send(ir)
     return out
 

front/py/deepx/nn/functional/activite.py

@@ -15,7 +15,7 @@ def _A_B_elementwiseop_C(
     out.node.add_input(opnode)
     if a.graph.eager:
         ir=DeepxIR(op, a.dtype, [a.node.name, b.node.name], [out.node.name])
-        send(str(ir))
+        send(ir)
 def _A_b_elementwiseop_C(
         a:Tensor,
         b: Optional[Union[ float, int]] = None, 
@@ -30,7 +30,7 @@ def _A_b_elementwiseop_C(
         varir=DeepxIR("argset", a.dtype, [b], [varnode.name])
         send(str(varir))
         ir=DeepxIR(op, a.dtype, [a.node.name,varnode.name], [out.node.name])
-        send(str(ir))
+        send(ir)
 #add
 OpNode.register("add")
 OpNode.register("add_scalar")