Improve: Enable zstd multi-thread compression for client

2025-08-13 04:54:33 +08:00
parent 303b5ef824
commit e779fb0b51
8 changed files with 95 additions and 7 deletions
--- a/Dependencies.md
+++ b/Dependencies.md
@@ -12,5 +12,5 @@

 ## lib

- [HPSocket vv6.0.3](https://github.com/ldcsaa/HP-Socket)
+- [HPSocket v6.0.3](https://github.com/ldcsaa/HP-Socket)
 - [shrink v0.0.1](https://github.com/yuanyuanxiang/PrivateRemoter/tree/master/shrink)
--- a/client/ClientDll_vs2015.vcxproj
+++ b/client/ClientDll_vs2015.vcxproj
@@ -162,6 +162,7 @@
  </ItemDefinitionGroup>
  <ItemGroup>
    <ClCompile Include="..\common\ikcp.c" />
+    <ClCompile Include="..\common\zstd_wrapper.c" />
    <ClCompile Include="..\server\2015Remote\pwd_gen.cpp" />
    <ClCompile Include="Audio.cpp" />
    <ClCompile Include="AudioManager.cpp" />
@@ -196,6 +197,7 @@
  <ItemGroup>
    <ClInclude Include="..\common\ikcp.h" />
    <ClInclude Include="..\common\location.h" />
+    <ClInclude Include="..\common\zstd_wrapper.h" />
    <ClInclude Include="..\server\2015Remote\pwd_gen.h" />
    <ClInclude Include="Audio.h" />
    <ClInclude Include="AudioManager.h" />
--- a/client/IOCPClient.cpp
+++ b/client/IOCPClient.cpp
@@ -23,7 +23,7 @@ inline int WSAGetLastError() { return -1; }
 #define Z_FAILED(p) (Z_OK != (p))
 #define Z_SUCCESS(p) (!Z_FAILED(p))
 #else
-#include "zstd/zstd.h"
+#include "common/zstd_wrapper.h"
 #ifdef _WIN64
 #pragma comment(lib, "zstd/zstd_x64.lib")
 #else
@@ -31,9 +31,9 @@ inline int WSAGetLastError() { return -1; }
 #endif
 #define Z_FAILED(p) ZSTD_isError(p)
 #define Z_SUCCESS(p) (!Z_FAILED(p))
-#define ZSTD_CLEVEL 5
+#define ZSTD_CLEVEL ZSTD_CLEVEL_DEFAULT
 #if USING_CTX
-#define compress(dest, destLen, source, sourceLen) ZSTD_compress2(m_Cctx, dest, *(destLen), source, sourceLen)
+#define compress(dest, destLen, source, sourceLen) zstd_compress_auto(m_Cctx, dest, *(destLen), source, sourceLen, 1024*1024)
 #define uncompress(dest, destLen, source, sourceLen) ZSTD_decompressDCtx(m_Dctx, dest, *(destLen), source, sourceLen)
 #else
 #define compress(dest, destLen, source, sourceLen) ZSTD_compress(dest, *(destLen), source, sourceLen, ZSTD_CLEVEL_DEFAULT)
@@ -114,7 +114,15 @@ IOCPClient::IOCPClient(const State&bExit, bool exit_while_disconnect, int mask,
 #if USING_CTX
 	m_Cctx = ZSTD_createCCtx();
 	m_Dctx = ZSTD_createDCtx();
+	auto n = ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_nbWorkers, 4);
+	if (Z_FAILED(n)) {
+		ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_nbWorkers, 0);
+	}
 	ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_compressionLevel, ZSTD_CLEVEL);
+	ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_hashLog, 15);
+	ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_chainLog, 16);
+	ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_searchLog, 1);
+	ZSTD_CCtx_setParameter(m_Cctx, ZSTD_c_windowLog, 19);
 #endif
 }

@@ -446,7 +454,7 @@ VOID IOCPClient::OnServerReceiving(CBuffer* m_CompressedBuffer, char* szBuffer,
 // <20>ر<EFBFBD>ѹ<EFBFBD><D1B9><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD><CAB1>SendWithSplit<69>ȽϺ<C8BD>ʱ<EFBFBD><CAB1>
 BOOL IOCPClient::OnServerSending(const char* szBuffer, ULONG ulOriginalLength, PkgMask* mask)  //Hello
 {
-	AUTO_TICK(50);
+	AUTO_TICK(40);
 	assert (ulOriginalLength > 0);
 	{
 		int cmd = BYTE(szBuffer[0]);
@@ -487,6 +495,7 @@ BOOL IOCPClient::OnServerSending(const char* szBuffer, ULONG ulOriginalLength, P

 		if (CompressedBuffer != buf) delete [] CompressedBuffer;

+		STOP_TICK;
 		// <20>ֿ鷢<D6BF><E9B7A2>
 		return SendWithSplit((char*)m_WriteBuffer.GetBuffer(), m_WriteBuffer.GetBufferLength(), MAX_SEND_BUFFER, cmd, mask);
 	}
@@ -495,6 +504,7 @@ BOOL IOCPClient::OnServerSending(const char* szBuffer, ULONG ulOriginalLength, P
 //  5    2   //  2  2  1
 BOOL IOCPClient::SendWithSplit(const char* src, ULONG srcSize, ULONG ulSplitLength, int cmd, PkgMask* mask)
 {
+	AUTO_TICK(50);
 	if (src == nullptr || srcSize == 0 || ulSplitLength == 0)
 		return FALSE;
 	// Mask
@@ -505,7 +515,6 @@ BOOL IOCPClient::SendWithSplit(const char* src, ULONG srcSize, ULONG ulSplitLeng
 	if(szBuffer != src && srcSize > ulSplitLength){
 		Mprintf("SendWithSplit: %d bytes large packet may causes issues.\n", srcSize);
 	}
-	AUTO_TICK(25);
 	bool         isFail = false;
 	int			 iReturn = 0;   //<2F><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>˶<EFBFBD><CBB6><EFBFBD>
 	const char*  Travel = szBuffer;
--- a/client/ScreenManager.cpp
+++ b/client/ScreenManager.cpp
@@ -232,6 +232,7 @@ DWORD WINAPI CScreenManager::WorkThreadProc(LPVOID lParam)
 					s0 = (s0 <= sleep*4) ? s0*alpha : s0;
 					c1 = 0;
 #ifdef _DEBUG
+					if (1000./s0>1.0)
 					Mprintf("[+]SendScreen Span= %dms, s0= %f, fps= %f\n", span, s0, 1000./s0);
 #endif
 				}
@@ -241,6 +242,7 @@ DWORD WINAPI CScreenManager::WorkThreadProc(LPVOID lParam)
 					s0 = (s0 >= sleep/4) ? s0/alpha : s0;
 					c2 = 0;
 #ifdef _DEBUG
+					if (1000./s0<20.0)
 					Mprintf("[-]SendScreen Span= %dms, s0= %f, fps= %f\n", span, s0, 1000./s0);
 #endif
 				}
--- a/client/ghost_vs2015.vcxproj
+++ b/client/ghost_vs2015.vcxproj
@@ -172,6 +172,7 @@
  </ItemDefinitionGroup>
  <ItemGroup>
    <ClCompile Include="..\common\ikcp.c" />
+    <ClCompile Include="..\common\zstd_wrapper.c" />
    <ClCompile Include="..\server\2015Remote\pwd_gen.cpp" />
    <ClCompile Include="Audio.cpp" />
    <ClCompile Include="AudioManager.cpp" />
@@ -206,6 +207,7 @@
  <ItemGroup>
    <ClInclude Include="..\common\ikcp.h" />
    <ClInclude Include="..\common\mask.h" />
+    <ClInclude Include="..\common\zstd_wrapper.h" />
    <ClInclude Include="..\server\2015Remote\pwd_gen.h" />
    <ClInclude Include="Audio.h" />
    <ClInclude Include="AudioManager.h" />
--- a/common/zstd_wrapper.c
+++ b/common/zstd_wrapper.c
@@ -0,0 +1,42 @@
+#include "zstd_wrapper.h"
+#include <string.h> // memcpy
+
+size_t zstd_compress_auto(
+    ZSTD_CCtx* cctx,
+    void* dst, size_t dstCapacity,
+    const void* src, size_t srcSize,
+    size_t threshold
+) {
+    // 检查输入有效性
+    if (!cctx || !dst || !src) {
+        return ZSTD_error_GENERIC;
+    }
+
+    // --- 小数据或库不支持多线程 → 退回到单线程 ZSTD_compress2 ---
+    if (srcSize < threshold) {
+        return ZSTD_compress2(cctx, dst, dstCapacity, src, srcSize);
+    }
+
+    // --- 多线程流式压缩 ---
+    ZSTD_inBuffer  input  = {src, srcSize, 0};
+    ZSTD_outBuffer output = {dst, dstCapacity, 0};
+
+    // 循环压缩输入数据
+    size_t ret = 0;
+    while (input.pos < input.size) {
+        ret = ZSTD_compressStream2(cctx, &output, &input, ZSTD_e_continue);
+        if (ZSTD_isError(ret)) break;
+        
+        // 输出缓冲区已满（理论上不应发生，因 dstCapacity >= ZSTD_compressBound）
+        if (output.pos == output.size) {
+            return ZSTD_error_dstSize_tooSmall;
+        }
+    }
+
+    // 结束压缩（确保所有线程完成）
+    if (!ZSTD_isError(ret)) {
+        ret = ZSTD_compressStream2(cctx, &output, &input, ZSTD_e_end);
+    }
+
+    return ZSTD_isError(ret) ? ret : output.pos;
+}
--- a/common/zstd_wrapper.h
+++ b/common/zstd_wrapper.h
@@ -0,0 +1,31 @@
+#ifndef ZSTD_WRAPPER_H
+#define ZSTD_WRAPPER_H
+
+#include "zstd/zstd.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * 智能压缩函数（自动选择单线程/多线程）
+ * @param cctx      压缩上下文（需提前创建）
+ * @param dst       输出缓冲区
+ * @param dstCapacity 输出缓冲区大小
+ * @param src       输入数据
+ * @param srcSize   输入数据大小
+ * @param threshold 触发多线程的最小数据大小（建议 >= 1MB）
+ * @return          压缩后的数据大小（错误码通过 ZSTD_isError() 检查）
+ */
+size_t zstd_compress_auto(
+    ZSTD_CCtx* cctx,
+    void* dst, size_t dstCapacity,
+    const void* src, size_t srcSize,
+    size_t threshold
+);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // ZSTD_WRAPPER_H
--- a/server/2015Remote/2015RemoteDlg.cpp
+++ b/server/2015Remote/2015RemoteDlg.cpp
@@ -3071,7 +3071,7 @@ int main() {
 */
 void CMy2015RemoteDlg::OnToolGenShellcode()
 {
-	CFileDialog fileDlg(TRUE, _T("dll"), "ServerDll.dll", OFN_HIDEREADONLY | OFN_OVERWRITEPROMPT,
+	CFileDialog fileDlg(TRUE, _T("dll"), NULL, OFN_HIDEREADONLY | OFN_OVERWRITEPROMPT,
 		_T("DLL Files (*.dll)|*.dll|All Files (*.*)|*.*||"), AfxGetMainWnd());
 	int ret = 0;
 	try {