[BugFix] replace ftok with custom_ftok in get_output/save_output ops (#6822)

* [BugFix] replace ftok with custom_ftok in get_output/save_output ops

* [Test] add unit test for custom_ftok

* [Chore] create custom_ftok.h

* [Chore] reorganize header file

* [Fix] fix cache messager msg_queue_id+rank_id conflict

custom_ops/gpu_ops/custom_ftok.h

@@ -0,0 +1,37 @@
+// Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <errno.h>
+#include <stdio.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+
+// Custom ftok that uses the low 20 bits of id instead of only 8 bits.
+// This avoids dependency on filesystem paths while preserving queue separation.
+inline key_t custom_ftok(const char* path, int id) {
+  struct stat st;
+  if (stat(path, &st) < 0) {
+    fprintf(stderr,
+            "[custom_ftok] stat(\"%s\") failed (errno=%d), "
+            "msg queue key will be invalid!\n",
+            path,
+            errno);
+    return static_cast<key_t>(-1);
+  }
+  // low 4 bits of st_dev | low 8 bits of st_ino | low 20 bits of id
+  return static_cast<key_t>(((st.st_dev & 0x0f) << 28) |
+                            ((st.st_ino & 0xff) << 20) | (id & 0xfffff));
+}

custom_ops/gpu_ops/get_output.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -25,6 +26,7 @@
 
 #define MAX_BSZ 512
 // #define GET_OUTPUT_DEBUG
+
 struct msgdata {
   long mtype;
   int mtext[MAX_BSZ + 2];  // stop_flag, bsz, tokens
@@ -49,7 +51,7 @@ void GetOutput(const paddle::Tensor& x,
 #endif
     msg_queue_id = inference_msg_queue_id_from_env;
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
 #ifdef GET_OUTPUT_DEBUG

custom_ops/gpu_ops/get_output_ep.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "msg_utils.h"
 #include "paddle/extension.h"
 
@@ -33,9 +34,9 @@ void GetOutputKVSignal(const paddle::Tensor& x,
     std::string msg_que_str(msg_que_str_tmp);
     msg_queue_id = std::stoi(msg_que_str);
   }
-  msg_queue_id += rank_id;
+  msg_queue_id = msg_queue_id << 4 + rank_id;
   static struct msgdatakv msg_rcv;
-  static key_t key = ftok("/opt/", msg_queue_id);
+  static key_t key = custom_ftok("/opt/", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   int* out_data = const_cast<int*>(x.data<int>());
@@ -77,10 +78,7 @@ void GetOutputEp(const paddle::Tensor& x,
 #ifdef GET_OUTPUT_DEBUG
   std::cout << "msg_queue_id is: " << msg_queue_id << std::endl;
 #endif
-  // static key_t key = ftok("/dev/shm", msg_queue_id);
-  // static int msgid = msgget(key, IPC_CREAT | 0666);
-
-  key_t key = ftok("/dev/shm", msg_queue_id);
+  key_t key = custom_ftok("/dev/shm", msg_queue_id);
   int msgid = msgget(key, IPC_CREAT | 0666);
 
 #ifdef GET_OUTPUT_DEBUG

custom_ops/gpu_ops/get_output_msg_with_topk.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -53,7 +54,7 @@ void GetOutputTopK(const paddle::Tensor& x,
 #endif
     msg_queue_id = inference_msg_queue_id_from_env;
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef GET_OUTPUT_DEBUG

custom_ops/gpu_ops/remote_cache_kv_ipc.h

@@ -25,6 +25,7 @@
 #include <sys/types.h>
 #include <unistd.h>
 
+#include "custom_ftok.h"
 #include "driver_types.h"
 #include "msg_utils.h"
 #include "paddle/extension.h"
@@ -79,8 +80,8 @@ struct RemoteCacheKvIpc {
           std::string msg_que_str(msg_que_str_tmp);
           msg_queue_id = std::stoi(msg_que_str);
         }
-        msg_queue_id += rank;
-        key_t key = ftok("/opt/", msg_queue_id);
+        msg_queue_id = msg_queue_id << 4 + rank;
+        key_t key = custom_ftok("/opt/", msg_queue_id);
         msgid = msgget(key, IPC_CREAT | 0666);
         inited = true;
       }

custom_ops/gpu_ops/save_output_msg_with_topk.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -96,7 +97,7 @@ void SaveOutMmsgTopK(const paddle::Tensor& x,
               << std::endl;
 #endif
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SAVE_WITH_OUTPUT_DEBUG
   std::cout << "save_output_key: " << key << std::endl;

custom_ops/gpu_ops/save_with_output_msg.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "save_with_output_msg.h"
+#include "custom_ftok.h"
 
 void save_kernel(const paddle::Tensor& x,
                  const paddle::Tensor& not_need_stop,
@@ -68,7 +69,7 @@ void save_kernel(const paddle::Tensor& x,
 #ifdef SAVE_WITH_OUTPUT_DEBUG
   std::cout << "msg_queue_id is: " << msg_queue_id << std::endl;
 #endif
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SAVE_WITH_OUTPUT_DEBUG

custom_ops/gpu_ops/speculate_decoding/draft_model/mtp_save_first_token.cc

@@ -18,6 +18,7 @@
 #include <sys/msg.h>
 #include <sys/types.h>
 #include "../speculate_msg.h"
+#include "../../custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -66,7 +67,7 @@ void MTPSaveFirstToken(const paddle::Tensor& x,
     msg_queue_id = inference_msg_queue_id_from_env;
   }
 
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   msg_sed.mtype = 1;

custom_ops/gpu_ops/speculate_decoding/speculate_get_output.cc

@@ -18,6 +18,7 @@
 #include <sys/msg.h>
 #include <sys/types.h>
 #include "paddle/extension.h"
+#include "../custom_ftok.h"
 
 #ifndef PD_BUILD_STATIC_OP
 #define PD_BUILD_STATIC_OP(name) PD_BUILD_OP(static_op_##name)
@@ -48,7 +49,7 @@ void SpeculateGetOutput(const paddle::Tensor& x,
 
   static struct speculate_msgdata msg_rcv;
 
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 

custom_ops/gpu_ops/speculate_decoding/speculate_get_output_with_topk.cc

@@ -18,6 +18,7 @@
 #include <sys/msg.h>
 #include <sys/types.h>
 #include "paddle/extension.h"
+#include "../custom_ftok.h"
 
 #ifndef PD_BUILD_STATIC_OP
 #define PD_BUILD_STATIC_OP(name) PD_BUILD_OP(static_op_##name)
@@ -59,7 +60,7 @@ void SpeculateGetOutMmsgTopK(const paddle::Tensor& output_tokens,
 #endif
     msg_queue_id = inference_msg_queue_id_from_env;
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SPECULATE_GET_WITH_OUTPUT_DEBUG

custom_ops/gpu_ops/speculate_decoding/speculate_save_output.cc

@@ -18,6 +18,7 @@
 #include <sys/msg.h>
 #include <sys/types.h>
 #include "paddle/extension.h"
+#include "../custom_ftok.h"
 
 #ifndef PD_BUILD_STATIC_OP
 #define PD_BUILD_STATIC_OP(name) PD_BUILD_OP(static_op_##name)
@@ -66,7 +67,7 @@ void SpeculateSaveWithOutputMsg(const paddle::Tensor& accept_tokens,
     msg_queue_id = inference_msg_queue_id_from_env;
   }
   static struct speculate_msgdata msg_sed;
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   msg_sed.mtype = 1;

custom_ops/gpu_ops/speculate_decoding/speculate_save_output_with_topk.cc

@@ -18,6 +18,7 @@
 #include <sys/msg.h>
 #include <sys/types.h>
 #include "paddle/extension.h"
+#include "../custom_ftok.h"
 
 #ifndef PD_BUILD_STATIC_OP
 #define PD_BUILD_STATIC_OP(name) PD_BUILD_OP(static_op_##name)
@@ -124,7 +125,7 @@ void SpeculateSaveOutMmsgTopK(const paddle::Tensor& sampled_token_ids,
               << std::endl;
 #endif
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SPECULATE_SAVE_WITH_OUTPUT_DEBUG
   std::cout << "save_output_key: " << key << std::endl;

custom_ops/gpu_ops/speculate_decoding/speculate_step_reschedule.cu

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "../custom_ftok.h"
 #include "helper.h"
 #include "speculate_msg.h"
 #include "../cccl_compat.h"  // CCCL 3.0 compatibility
@@ -314,8 +315,7 @@ void SpeculateStepSchedule(
 
     } else {
     }
-    // static key_t key = ftok("/dev/shm", msg_queue_id);
-    static key_t key = ftok("./", msg_queue_id);
+    static key_t key = custom_ftok("./", msg_queue_id);
 
     static int msgid = msgget(key, IPC_CREAT | 0666);
     msg_sed.mtype = 1;

custom_ops/gpu_ops/step_reschedule.cu

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "custom_ftok.h"
 #include "helper.h"
 #include "save_with_output_msg.h"
 #include "cccl_compat.h"  // CCCL 3.0 compatibility
@@ -300,7 +301,7 @@ void Schedule(const paddle::Tensor &stop_flags,
             "number.");
       }
     }
-    static key_t key = ftok("/dev/shm", msg_queue_id);
+    static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
     static int msgid = msgget(key, IPC_CREAT | 0666);
     msg_sed.mtype = 1;

custom_ops/xpu_ops/src/ops/custom_ftok.h

@@ -0,0 +1,37 @@
+// Copyright (c) 2025 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include <errno.h>
+#include <stdio.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+
+// Custom ftok that uses the low 20 bits of id instead of only 8 bits.
+// This avoids dependency on filesystem paths while preserving queue separation.
+inline key_t custom_ftok(const char* path, int id) {
+  struct stat st;
+  if (stat(path, &st) < 0) {
+    fprintf(stderr,
+            "[custom_ftok] stat(\"%s\") failed (errno=%d), "
+            "msg queue key will be invalid!\n",
+            path,
+            errno);
+    return static_cast<key_t>(-1);
+  }
+  // low 4 bits of st_dev | low 8 bits of st_ino | low 20 bits of id
+  return static_cast<key_t>(((st.st_dev & 0x0f) << 28) |
+                            ((st.st_ino & 0xff) << 20) | (id & 0xfffff));
+}

custom_ops/xpu_ops/src/ops/get_output.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "msg_utils.h"
 #include "paddle/extension.h"
 
@@ -29,9 +30,9 @@ void GetOutputKVSignal(const paddle::Tensor& x,
     std::string msg_que_str(msg_que_str_tmp);
     msg_queue_id = std::stoi(msg_que_str);
   }
-  msg_queue_id += rank_id;
+  msg_queue_id = msg_queue_id << 4 + rank_id;
   static struct msgdatakv msg_rcv;
-  static key_t key = ftok("/opt/", msg_queue_id);
+  static key_t key = custom_ftok("/opt/", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   int* out_data = const_cast<int*>(x.data<int>());
@@ -70,7 +71,7 @@ void GetOutput(const paddle::Tensor& x,
 #endif
     msg_queue_id = inference_msg_queue_id_from_env;
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
 #ifdef GET_OUTPUT_DEBUG

custom_ops/xpu_ops/src/ops/get_output_msg_with_topk.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -53,7 +54,7 @@ void GetOutputTopK(const paddle::Tensor& x,
 #endif
     msg_queue_id = inference_msg_queue_id_from_env;
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef GET_OUTPUT_DEBUG

custom_ops/xpu_ops/src/ops/mtp/mtp_save_first_token.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "../custom_ftok.h"
 #include "paddle/extension.h"
 #include "speculate_msg.h"
 
@@ -60,7 +61,7 @@ void MTPSaveFirstToken(const paddle::Tensor& x,
     msg_queue_id = inference_msg_queue_id_from_env;
   }
 
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   msg_sed.mtype = 1;

custom_ops/xpu_ops/src/ops/mtp/speculate_get_output.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "../custom_ftok.h"
 #include "paddle/extension.h"
 #include "speculate_msg.h"
 
@@ -43,7 +44,7 @@ void SpeculateGetOutput(const paddle::Tensor& x,
 
   static struct speculate_msgdata msg_rcv;
 
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 

custom_ops/xpu_ops/src/ops/mtp/speculate_save_output.cc

@@ -18,6 +18,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "../custom_ftok.h"
 #include "paddle/extension.h"
 #include "speculate_msg.h"
 #include "xpu/plugin.h"
@@ -67,7 +68,7 @@ void SpeculateSaveWithOutputMsg(const paddle::Tensor& accept_tokens,
     msg_queue_id = inference_msg_queue_id_from_env;
   }
   static struct speculate_msgdata msg_sed;
-  static key_t key = ftok("./", msg_queue_id);
+  static key_t key = custom_ftok("./", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 
   msg_sed.mtype = 1;

custom_ops/xpu_ops/src/ops/mtp/speculate_step_reschedule.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include <paddle/phi/backends/xpu/xpu_context.h>
+#include "../custom_ftok.h"
 #include "paddle/phi/core/enforce.h"
 #include "speculate_msg.h"  // NOLINT
 #include "xpu/plugin.h"
@@ -138,8 +139,7 @@ void SpeculateStepSchedule(
 
     } else {
     }
-    // static key_t key = ftok("/dev/shm", msg_queue_id);
-    static key_t key = ftok("./", msg_queue_id);
+    static key_t key = custom_ftok("./", msg_queue_id);
 
     static int msgid = msgget(key, IPC_CREAT | 0666);
     msg_sed.mtype = 1;

custom_ops/xpu_ops/src/ops/remote_cache_kv_ipc.h

@@ -20,7 +20,8 @@
 #include <unistd.h>
 #include <cstdint>
 #include <vector>
-#include "msg_utils.h"  // NOLINT
+#include "custom_ftok.h"  // NOLINT
+#include "msg_utils.h"    // NOLINT
 
 struct RemoteCacheKvIpc {
   struct save_cache_kv_complete_signal_layerwise_meta_data {
@@ -70,8 +71,8 @@ struct RemoteCacheKvIpc {
           std::string msg_que_str(msg_que_str_tmp);
           msg_queue_id = std::stoi(msg_que_str);
         }
-        msg_queue_id += rank;
-        key_t key = ftok("/opt/", msg_queue_id);
+        msg_queue_id = msg_queue_id << 4 + rank;
+        key_t key = custom_ftok("/opt/", msg_queue_id);
         msgid = msgget(key, IPC_CREAT | 0666);
         inited = true;
       }

custom_ops/xpu_ops/src/ops/save_output_msg_with_topk.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "paddle/extension.h"
 
 #ifndef PD_BUILD_STATIC_OP
@@ -96,7 +97,7 @@ void SaveOutMmsgTopK(const paddle::Tensor& x,
               << std::endl;
 #endif
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SAVE_WITH_OUTPUT_DEBUG
   std::cout << "save_output_key: " << key << std::endl;

custom_ops/xpu_ops/src/ops/save_with_output_msg.cc

@@ -17,6 +17,7 @@
 #include <sys/ipc.h>
 #include <sys/msg.h>
 #include <sys/types.h>
+#include "custom_ftok.h"
 #include "msg_utils.h"
 #include "paddle/extension.h"
 
@@ -80,7 +81,7 @@ void SaveOutMmsg(const paddle::Tensor &x,
               << std::endl;
 #endif
   }
-  static key_t key = ftok("/dev/shm", msg_queue_id);
+  static key_t key = custom_ftok("/dev/shm", msg_queue_id);
 
   static int msgid = msgget(key, IPC_CREAT | 0666);
 #ifdef SAVE_WITH_OUTPUT_DEBUG