Sloved a serious problem in 7*24 heavy test

ZoomPipeline_FuncSvr/cluster/zp_clusternode.cpp

@@ -57,10 +57,6 @@ namespace ZP_Cluster{
 			//qDebug()<<QString("%1(%2) Node Martked Deleted, return.\n").arg((unsigned int)this).arg(ref());
 			return 0;
 		}
-		//This is important! sometimes without this check, m_list_RawData will crash.
-		//For a single zp_ClusterNode instance, at anytime, there should be only ONE thread in which run() is running.
-		if (ref()>1)
-			return -1;
 		int nCurrSz = -1;
 		int nMessage = m_nMessageBlockSize;
 		while (--nMessage>=0 && nCurrSz!=0  )
@@ -144,10 +140,18 @@ namespace ZP_Cluster{
 			if (m_currentHeader.Mark == 0x1234)
 				//Valid Message
 			{
-				while (m_currentMessageSize< sizeof(CROSS_SVR_MSG::tag_header) && blocklen>offset)
+				//while (m_currentMessageSize< sizeof(CROSS_SVR_MSG::tag_header) && blocklen>offset)
+				if (m_currentMessageSize< sizeof(CROSS_SVR_MSG::tag_header) && blocklen>offset)
 				{
-					m_currentBlock.push_back(dataptr[offset++]);
-					m_currentMessageSize++;
+					int nCpy = sizeof(CROSS_SVR_MSG::tag_header) - m_currentMessageSize;
+					if (nCpy > blocklen - offset)
+						nCpy = blocklen - offset;
+					QByteArray arrCpy(dataptr+offset,nCpy);
+					m_currentBlock.push_back(arrCpy);
+					//m_currentBlock.push_back(dataptr[offset++]);
+					//m_currentMessageSize++;
+					offset += nCpy;
+					m_currentMessageSize += nCpy;
 				}
 				if (m_currentMessageSize < sizeof(CROSS_SVR_MSG::tag_header)) //Header not completed.
 					continue;
@@ -161,11 +165,20 @@ namespace ZP_Cluster{
 					{
 						qint32 bitLeft = m_currentHeader.data_length + sizeof(CROSS_SVR_MSG::tag_header)
 								-m_currentMessageSize ;
-						while (bitLeft>0 && blocklen>offset)
+						//while (bitLeft>0 && blocklen>offset)
+						if (bitLeft>0 && blocklen>offset)
 						{
-							m_currentBlock.push_back(dataptr[offset++]);
-							m_currentMessageSize++;
-							bitLeft--;
+							int nCpy = bitLeft;
+							if (nCpy > blocklen - offset)
+								nCpy = blocklen - offset;
+							QByteArray arrCpy(dataptr+offset,nCpy);
+							m_currentBlock.push_back(arrCpy);
+							offset += nCpy;
+							m_currentMessageSize += nCpy;
+							bitLeft -= nCpy;
+							//m_currentBlock.push_back(dataptr[offset++]);
+							//m_currentMessageSize++;
+							//bitLeft--;
 						}
 						//deal block, may be send data as soon as possible;
 						deal_current_message_block();
@@ -183,11 +196,21 @@ namespace ZP_Cluster{
 					{
 						qint32 bitLeft = m_currentHeader.data_length + sizeof(CROSS_SVR_MSG::tag_header)
 								-m_currentMessageSize ;
-						while (bitLeft>0 && blocklen>offset)
+						//while (bitLeft>0 && blocklen>offset)
+						if (bitLeft>0 && blocklen>offset)
 						{
-							m_currentBlock.push_back(dataptr[offset++]);
-							m_currentMessageSize++;
-							bitLeft--;
+							int nCpy = bitLeft;
+							if (nCpy > blocklen - offset)
+								nCpy = blocklen - offset;
+							QByteArray arrCpy(dataptr+offset,nCpy);
+							m_currentBlock.push_back(arrCpy);
+							offset += nCpy;
+							m_currentMessageSize += nCpy;
+							bitLeft -= nCpy;
+
+							//m_currentBlock.push_back(dataptr[offset++]);
+							//m_currentMessageSize++;
+							//bitLeft--;
 						}
 						//deal block, may be processed as soon as possible;
 						deal_current_message_block();

ZoomPipeline_FuncSvr/pipeline/zp_pltaskbase.cpp

@@ -25,4 +25,14 @@ namespace ZPTaskEngine{
 		QMutexLocker locker(&m_mutex_ref);
 		return refCount;
 	}
+
+	bool zp_plTaskBase::LockRun()
+	{
+		return m_mutex_run.tryLock();
+	}
+
+	void zp_plTaskBase::UnlockRun()
+	{
+		return m_mutex_run.unlock();
+	}
 }

ZoomPipeline_FuncSvr/pipeline/zp_pltaskbase.h

@@ -27,9 +27,13 @@ namespace ZPTaskEngine{
 
 		int ref();
 
+		bool LockRun();
+		void UnlockRun();
+
 	private:
 		int refCount;
 		QMutex m_mutex_ref;
+		QMutex m_mutex_run;
 	signals:
 
 	public slots:

ZoomPipeline_FuncSvr/pipeline/zp_plworkingthread.cpp

@@ -54,11 +54,21 @@ namespace ZPTaskEngine{
 			if (bValid==true && ptr!=NULL)
 			{
 				m_bBusy = true;
-				int res = ptr->run();
-				ptr->delRef();
-				m_bBusy = false;
-				if (res!=0 )
+				if (ptr->LockRun()==true)
+				{
+					int res = ptr->run();
+					ptr->delRef();
+					if (res!=0 )
+						this->m_pipeline->pushTask(ptr,false);
+					ptr->UnlockRun();
+				}
+				else
+				{
+					ptr->delRef();
 					this->m_pipeline->pushTask(ptr,false);
+				}
+				m_bBusy = false;
+
 			}
 
 			emit taskFinished(this);

ZoomPipeline_FuncSvr/smartlink/st_clientnode_basetrans.cpp

@@ -49,10 +49,6 @@ namespace ExampleServer{
 			//qDebug()<<QString("%1(%2) Node Martked Deleted, return.\n").arg((unsigned int)this).arg(ref());
 			return 0;
 		}
-		//This is important! sometimes without this check, m_list_RawData will crash.
-		//For a single zp_ClusterNode instance, at anytime, there should be only ONE thread in which run() is running.
-		if (ref()>1)
-			return -1;
 		int nCurrSz = -1;
 		int nMessage = m_nMessageBlockSize;
 		while (--nMessage>=0 && nCurrSz!=0  )