Users/cheqi/pre merge tikv parallelbuild

.gitignore

@@ -464,5 +464,4 @@ FodyWeavers.xsd
 *.sln.iml
 
 # SPTAG benchmark generated artifacts
-/perftest_*
-/evaluation/2026-04-23/output_distributed_hostname_*.json
+*perftest_*

.vscode/launch.json

@@ -101,7 +101,7 @@
         },
         {
           "name": "LD_PRELOAD",
-          "value": "/usr/lib/gcc/x86_64-linux-gnu/11/libasan.so"
+          "value": "/usr/lib/gcc/x86_64-linux-gnu/13/libasan.so"
         },
         {
           "name": "PCI_ALLOWED",
@@ -170,7 +170,7 @@
         },
         {
           "name": "LD_PRELOAD",
-          "value": "/usr/lib/gcc/x86_64-linux-gnu/11/libasan.so"
+          "value": "/usr/lib/gcc/x86_64-linux-gnu/13/libasan.so"
         },
         {
           "name": "PCI_ALLOWED",

AnnService/inc/Core/SPANN/Options.h

@@ -34,6 +34,7 @@ namespace SPTAG {
             std::string m_truthPath;
             TruthFileType m_truthType;
             bool m_generateTruth;
+            std::string m_globalIDPath;
             std::string m_indexDirectory;
             std::string m_headIDFile;
             std::string m_headVectorFile;

AnnService/inc/Core/SPANN/ParameterDefinitionList.h

@@ -23,6 +23,7 @@ DefineBasicParameter(m_warmupDelimiter, std::string, std::string("|"), "WarmupDe
 DefineBasicParameter(m_truthPath, std::string, std::string(""), "TruthPath")
 DefineBasicParameter(m_truthType, SPTAG::TruthFileType, SPTAG::TruthFileType::Undefined, "TruthType")
 DefineBasicParameter(m_generateTruth, bool, false, "GenerateTruth") // Mutable
+DefineBasicParameter(m_globalIDPath, std::string, std::string(""), "GlobalIDPath")
 DefineBasicParameter(m_indexDirectory, std::string, std::string("SPANN"), "IndexDirectory")
 DefineBasicParameter(m_headIDFile, std::string, std::string("SPTAGHeadVectorIDs.bin"), "HeadVectorIDs")
 DefineBasicParameter(m_deleteIDFile, std::string, std::string("DeletedIDs.bin"), "DeletedIDs")

AnnService/inc/Helper/KeyValueIO.h

@@ -34,24 +34,25 @@ namespace SPTAG
 
             virtual ErrorCode Put(const SizeType key, const std::string& value, const std::chrono::microseconds& timeout, std::vector<Helper::AsyncReadRequest>* reqs) = 0;
 
-            // Batched writes/deletes. Default implementations return Undefined so that
-            // backends without native batching (RocksDB, FileIO) can ignore them.
-            // TiKVIO overrides these to issue a single batched RPC per region group,
-            // which dramatically reduces the number of synchronous gRPC round-trips
-            // when callers (e.g. SPANN AddIndex Phase 2 / PutPostingToDB) want to
-            // commit several keys at once.
-            virtual ErrorCode MultiPut(const std::vector<std::string>& keys,
-                                       const std::vector<std::string>& values,
-                                       const std::chrono::microseconds& timeout,
-                                       std::vector<Helper::AsyncReadRequest>* reqs) { return ErrorCode::Undefined; }
-
-            virtual ErrorCode MultiDelete(const std::vector<std::string>& keys,
-                                          const std::chrono::microseconds& timeout) { return ErrorCode::Undefined; }
-
             virtual ErrorCode Merge(const SizeType key, const std::string &value,
                                     const std::chrono::microseconds &timeout,
                                     std::vector<Helper::AsyncReadRequest> *reqs, int& size) = 0;
 
+            virtual ErrorCode MultiMerge(const std::vector<SizeType>& keys, const std::vector<std::string>& values, 
+                                         const std::chrono::microseconds& timeout, std::vector<Helper::AsyncReadRequest>* reqs, std::vector<int>& sizes) {
+                if (keys.size() != values.size()) {
+                    return ErrorCode::Undefined;
+                }
+                sizes.resize(keys.size());
+                for (size_t i = 0; i < keys.size(); i++) {
+                    auto err = Merge(keys[i], values[i], timeout, reqs, sizes[i]);
+                    if (err != ErrorCode::Success) {
+                        return err;
+                    }
+                }
+                return ErrorCode::Success;
+            }
+
             virtual ErrorCode Delete(SizeType key) = 0;
 
             virtual ErrorCode DeleteRange(SizeType start, SizeType end) {return ErrorCode::Undefined;}
@@ -79,6 +80,8 @@ namespace SPTAG
             virtual ErrorCode StartToScan(SizeType& key, std::string* value) {return ErrorCode::Undefined;}
 
             virtual ErrorCode NextToScan(SizeType& key, std::string* value) {return ErrorCode::Undefined;}
+
+            virtual void LogAsyncWaitStatsAndReset(int layer) {}
         };
     }
 }

AnnService/inc/Helper/ThreadPool.h

@@ -5,7 +5,7 @@
 #define _SPTAG_HELPER_THREADPOOL_H_
 
 #include <atomic>
-#include <queue>
+#include <deque>
 #include <vector>
 #include <thread>
 #include <mutex>
@@ -78,7 +78,16 @@ namespace SPTAG
             {
                 {
                     std::lock_guard<std::mutex> lock(m_lock);
-                    m_jobs.push(j);
+                    m_jobs.push_back(j);
+                }
+                m_cond.notify_one();
+            }
+
+            void addfront(Job* j)
+            {
+                {
+                    std::lock_guard<std::mutex> lock(m_lock);
+                    m_jobs.push_front(j);
                 }
                 m_cond.notify_one();
             }
@@ -90,7 +99,7 @@ namespace SPTAG
                 if (!m_abort.ShouldAbort()) {
                     j = m_jobs.front();
                     currentJobs++;
-                    m_jobs.pop();
+                    m_jobs.pop_front();
                     return true;
                 }
                 return false;
@@ -113,7 +122,7 @@ namespace SPTAG
 
         protected:
             std::atomic_uint32_t currentJobs{ 0 };
-            std::queue<Job*> m_jobs;
+            std::deque<Job*> m_jobs;
             Abort m_abort;
             std::mutex m_lock;
             std::condition_variable m_cond;