newtensor:的小bug

Author: miaobyte

excuter/common/src/client/udpserver.cpp

@@ -16,7 +16,7 @@ namespace client
             close(sockfd);
         }
     }
-    void udpserver::start()
+    void udpserver::start(queue<deepx::op::Op> &queue)
     {
         // 创建UDP套接字
         if ((sockfd = socket(AF_INET, SOCK_DGRAM, 0)) < 0)
@@ -26,8 +26,7 @@ namespace client
         }
 
         memset(&servaddr, 0, sizeof(servaddr));
-        memset(&cliaddr, 0, sizeof(cliaddr));
-
+ 
         // 绑定IP和端口
         servaddr.sin_family = AF_INET; // IPv4
         servaddr.sin_addr.s_addr = INADDR_ANY;
@@ -42,7 +41,7 @@ namespace client
         while (true)
         {
             len = sizeof(cliaddr);
-            n = recvfrom(sockfd, (char *)buffer, sizeof(buffer), 0, (struct sockaddr *)&cliaddr, &len);
+            n = recvfrom(sockfd, buffer, sizeof(buffer), 0, (struct sockaddr *)&cliaddr, &len);
             buffer[n] = '\0';
 
             // 新增换行拆分逻辑
@@ -51,13 +50,17 @@ namespace client
             while (getline(ss, line)) {
                 if (!line.empty()) {
                     cout << "~" << line << endl;
-                    char *IR = const_cast<char *>(line.c_str());
-                    string strresp=func(IR);
-                    sendto(sockfd, strresp.c_str(), strresp.size(), 0,
-                   (const struct sockaddr*)&cliaddr, len);
+                    deepx::op::Op op;
+                    op.recv_at = chrono::system_clock::now();
+                    op.load(line);
+                    queue.push(op);
                 }
             }
         }
         close(sockfd);
     }
+    void udpserver::resp(string str){
+         sendto(sockfd, str.c_str(), str.size(), 0,  // 改为sendto
+          (const struct sockaddr *)&cliaddr, sizeof(cliaddr));
+    }
 }

excuter/common/src/client/udpserver.hpp

@@ -7,25 +7,28 @@
 #include <sys/un.h>
 #include <unistd.h>
 #include <functional>
- 
+#include "deepx/op/op.hpp"
+#include <queue>
+
 namespace client{
+    using namespace std;
     class udpserver
     {
     private:
         int port;
         int sockfd;
-        struct sockaddr_in servaddr, cliaddr;
+        struct sockaddr_in servaddr,cliaddr;
         char buffer[1024];
         socklen_t len;
         ssize_t n;
     public:
         udpserver(int port);
         ~udpserver();
-        void start();
-        using handlefunc = std::function<std::string(const char *buffer)>;
+        void start(queue<deepx::op::Op> &tasks);
+        using handlefunc = std::function<void(const char *buffer)>;
         handlefunc func;
+        void resp(string str);
     };
-   
 }
 
 #endif

excuter/common/src/client/worker.hpp

@@ -0,0 +1,8 @@
+#ifndef __WORKER_HPP__
+#define __WORKER_HPP__
+
+
+namespace client{
+ 
+}
+#endif

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

@@ -5,9 +5,7 @@ namespace deepx::op
     //与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);
-        
+    void Op::load(const string &input) {
         // 分割元数据部分
         size_t meta_pos = input.find("//");
         string body = input.substr(0, meta_pos);

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

@@ -54,7 +54,7 @@ namespace deepx::op
             throw std::runtime_error("backward not implemented");
         }
 
-        void load(const char* str) ; 
+        void load(const string &str) ; 
         void init(const string &opname,
                   const string &dtype,
                   const vector<string> &args,

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

@@ -1,11 +1,12 @@
 #include <mutex>
+#include <thread>
 
 #include <deepx/tensorfunc/init.hpp>
 #include "deepx/op/op.hpp"
 #include "deepx/op/opfactory.hpp"
 #include "deepx/mem/mem.hpp"
 #include "client/udpserver.hpp"
- 
+
 using namespace deepx::tensorfunc;
 using namespace deepx::mem;
 
@@ -18,38 +19,45 @@ int main()
     deepx::op::OpFactory opfactory;
     register_all(opfactory);
 
-    server.func = [&mem, &opfactory, &memmutex](const char *buffer)->std::string
-    {
-
-        
-        deepx::op::Op op;
-        op.recv_at = chrono::system_clock::now();
-        op.load(buffer);
-        std::string resp=to_string(op.id);
-        resp+=" recv_at:";
-        resp+=to_string(op.recv_at.time_since_epoch().count());
-        if (opfactory.ops.find(op.name)==opfactory.ops.end()){
-            cout<<"<op> "<<op.name<<" not found"<<endl;
-            resp+="error op not found";
-        }
-        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;
-            resp+="error dtype not found";
-        }
-        auto src = type_map.find(op.dtype)->second;
-
-        (*src).init(op.name, op.dtype, op.args, op.returns, op.grad, op.args_grad, op.returns_grad);
-        memmutex.lock();
-        if (op.grad) {
-            (*src).backward(mem);
-        }else {
-           (*src).forward(mem);
+    queue<deepx::op::Op> tasks;
+    // 启动一个新线程来运行UDP服务器
+    std::thread server_thread([&server, &tasks]() {
+        server.start(tasks);
+    }); 
+    // 分离线程，让它在后台运行
+    server_thread.detach();
+
+    while (true) {
+        if (!tasks.empty()) {
+            deepx::op::Op op = tasks.front();
+            tasks.pop();
+ 
+            std::string resp=to_string(op.id);
+            resp+="recv_at:";
+            resp+=to_string(op.recv_at.time_since_epoch().count());
+            if (opfactory.ops.find(op.name)==opfactory.ops.end()){
+                cout<<"<op> "<<op.name<<" not found"<<endl;
+                resp+="error op not found";
+            }
+            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;
+                resp+="error dtype not found";
+            }
+            auto src = type_map.find(op.dtype)->second;
+
+            (*src).init(op.name, op.dtype, op.args, op.returns, op.grad, op.args_grad, op.returns_grad);
+            memmutex.lock();
+            if (op.grad) {
+                (*src).backward(mem);
+            }else {
+            (*src).forward(mem);
+            }
+            memmutex.unlock();
+            resp+=" success";
+            server.resp(resp);
         }
-        memmutex.unlock();
-        resp+=" success";
-        return resp;
-    };
-    server.start();
+    }
+
     return 0;
 }

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

@@ -25,7 +25,7 @@ namespace deepx::op{
                 vector<int> shape=mem.getvector<int32_t>(this->args[0]);
                 Tensor<T> t=tensorfunc::New<T>(shape);
                 mem.addtensor(name,t);
-            }else if (this->args.size()>1){
+            }else{
                 vector<int> shape;  
                 for (int i = 0; i < this->args.size(); i++) {
                     shape.push_back(atoi(this->args[i].c_str()));

front/py/deepx/scheduler/client/allclient.py

@@ -12,4 +12,3 @@ def send(ir:DeepxIR) -> Optional[dict]:
     _id_counter=_id_counter+1
     ir._id=_id_counter
     resp=default_client.send(str(ir))
-    print(resp)