解决了内存溢出的问题：阻塞队列没有实现push函数的阻塞功能

ErianArcher · ErianArcher · commit 7c8c22d97668 · 2019-07-30T20:39:26.000+08:00
diff --git a/blocking_queue.h b/blocking_queue.h
@@ -32,9 +32,11 @@ class block_queue
         m_back = -1;
 
         m_mutex = new pthread_mutex_t;
-        m_cond = new pthread_cond_t;
+        m_rcond = new pthread_cond_t;
+        m_wcond = new pthread_cond_t;
         pthread_mutex_init(m_mutex, NULL);
-        pthread_cond_init(m_cond, NULL);
+        pthread_cond_init(m_rcond, NULL);
+        pthread_cond_init(m_wcond, NULL);
     }
 
     void clear()
@@ -54,10 +56,11 @@ class block_queue
         pthread_mutex_unlock(m_mutex);
 
         pthread_mutex_destroy(m_mutex);
-        pthread_cond_destroy(m_cond);
+        pthread_cond_destroy(m_rcond);
+        pthread_cond_destroy(m_wcond);
 
         delete m_mutex;
-        delete m_cond;
+        delete m_rcond;
     }
 
     bool full()const
@@ -131,29 +134,34 @@ class block_queue
     bool push(const T& item)
     {
         pthread_mutex_lock(m_mutex);
-        if(m_size >= m_max_size)
+        const bool was_empty = (m_size == 0);
+        while (m_size >= m_max_size)
         {
-            pthread_cond_broadcast(m_cond);
-            pthread_mutex_unlock(m_mutex);
-            return false;
+            if (0 != pthread_cond_wait(m_wcond, m_mutex)) {
+                pthread_cond_broadcast(m_rcond);
+                pthread_mutex_unlock(m_mutex);
+                return false;
+            }
         }
 
         m_back = (m_back + 1) % m_max_size;
         m_array[m_back] = item;
 
         m_size++;
-        pthread_cond_broadcast(m_cond);
+        pthread_cond_broadcast(m_rcond);
         pthread_mutex_unlock(m_mutex);
-
+        // 唤醒push
+        if (was_empty) pthread_cond_broadcast(m_rcond);
         return true;
     }
 
     bool pop(T *item)
     {
         pthread_mutex_lock(m_mutex);
+        const bool was_full = (m_size == m_max_size);
         while(m_size <= 0)
         {
-            if(0 != pthread_cond_wait(m_cond, m_mutex))
+            if(0 != pthread_cond_wait(m_rcond, m_mutex))
             {
                 pthread_mutex_unlock(m_mutex);
                 return false;
@@ -164,6 +172,8 @@ class block_queue
         *item = m_array[m_front];
         m_size--;
         pthread_mutex_unlock(m_mutex);
+        // 唤醒push
+        if (was_full) pthread_cond_broadcast(m_wcond);
         return true;
     }
 
@@ -173,11 +183,12 @@ class block_queue
         struct timeval now = {0,0};
         gettimeofday(&now, NULL);
         pthread_mutex_lock(m_mutex);
+        const bool was_full = (m_size == m_max_size);
         if(m_size <= 0)
         {
             t.tv_sec = now.tv_sec + ms_timeout/1000;
             t.tv_nsec = (ms_timeout % 1000)*1000;
-            if(0 != pthread_cond_timedwait(m_cond, m_mutex, &t))
+            if(0 != pthread_cond_timedwait(m_rcond, m_mutex, &t))
             {
                 pthread_mutex_unlock(m_mutex);
                 return false;
@@ -193,12 +204,15 @@ class block_queue
         m_front = (m_front + 1) % m_max_size;
         *item = m_array[m_front];m_size--;
         pthread_mutex_unlock(m_mutex);
+        // 唤醒push
+        if (was_full) pthread_cond_broadcast(m_wcond);
         return true;
     }
 
 private:
     pthread_mutex_t *m_mutex;
-    pthread_cond_t *m_cond;
+    pthread_cond_t *m_rcond;
+    pthread_cond_t *m_wcond;
     T *m_array;
     int m_size;
     int m_max_size;
diff --git a/communicator.cpp b/communicator.cpp
@@ -16,7 +16,7 @@
 
 using namespace std;
 
-block_queue<Item *> *blockQueue = new block_queue<Item *>(1024);
+block_queue<Item *> *blockQueue = new block_queue<Item *>(1024*1024);
 
 #define MAIN_THREAD_RUNNING 1
 #define MAIN_THREAD_WAITING 0
@@ -41,7 +41,7 @@ void mainThreadTellFinished(int thisRank, int world_size) {
         itemStop->tarRank = i;
         itemStop->pack = buf;
         if (nullptr == blockQueue) cerr << "Cannot create blocking queue." << endl;
-        blockQueue->push(itemStop);
+        if (!blockQueue->push(itemStop)) cerr << "Cannot push stop signal." << endl;
         itemStop = nullptr;
         buf = nullptr;
         packSizeStop = 0;
@@ -95,7 +95,7 @@ void requestOtherRanksToStoreEdge(int currank, int world_size, const char *value
     int packSize = 0;
     int strLen = getK();
     int op = EDGE_STORE_OP;
-    int tarrank = edgeId % world_size;
+    size_t tarrank = idBelongTo(world_size, edgeId);
     if (tarrank == currank) {
         cerr << "Putting a edge belonging to this host to other hosts." << endl;
         return;
@@ -110,19 +110,19 @@ void requestOtherRanksToStoreEdge(int currank, int world_size, const char *value
     // DEBUG
     //MPI_Send(&packSize, 1, MPI_INT, tarrank, TAG(currank, tarrank), MPI_COMM_WORLD);
     //MPI_Send(buf, packSize, MPI_PACKED, tarrank, TAG(currank, tarrank), MPI_COMM_WORLD);
-    item->tarRank = tarrank;
+    item->tarRank = (int) tarrank;
     item->pack = buf;
     item->packSize = packSize;
     if (nullptr == blockQueue) cerr << "Cannot create blocking queue." << endl;
-    blockQueue->push(item);
+    if (!blockQueue->push(item)) cerr << "Cannot request new edge" << endl;
 }
 
 void
 requestOtherRanksToStoreVertex(int currank, int world_size, VertexId vertexId, EdgeId edgeId, VertexMode_t vertexMode) {
     char *buf = new char[BUFFER_SIZE];
     int packSize = 0;
     int op = VERTEX_STORE_OP;
-    int tarrank = vertexId % world_size;
+    size_t tarrank = idBelongTo(world_size, vertexId);
     if (tarrank == currank) {
         cerr << "Putting a vertex belonging to this host to other hosts." << endl;
         return;
@@ -137,11 +137,11 @@ requestOtherRanksToStoreVertex(int currank, int world_size, VertexId vertexId, E
     //MPI_Send(&packSize, 1, MPI_INT, tarrank, TAG(currank, tarrank), MPI_COMM_WORLD);
     //MPI_Send(buf, packSize, MPI_PACKED, tarrank, TAG(currank, tarrank), MPI_COMM_WORLD);
 
-    item->tarRank = tarrank;
+    item->tarRank = (int) tarrank;
     item->pack = buf;
     item->packSize = packSize;
     if (nullptr == blockQueue) cerr << "Cannot create blocking queue." << endl;
-    blockQueue->push(item);
+    if (!blockQueue->push(item)) cerr << "Cannot request new vertex" << endl;
 }
 
 // 该方法负责char *指针的delete
@@ -160,10 +160,10 @@ int processRecvRead(const char *filename, size_t pos) {
 // 该方法负责char *指针的delete
 int processRecvEdge(EdgeList *edgeList, char *value, ReadId readId, KMERPOS_t kmerpos) {
     string *strValue = new string(value, 0, getK()-1);
-    VertexId sourceVertex = getId(*strValue);
+    VertexId sourceVertex = getId(strValue->c_str());
     delete strValue;
     strValue = new string(value, 1, getK()-1);
-    VertexId sinkVertex = getId(*strValue);
+    VertexId sinkVertex = getId(strValue->c_str());
     delete strValue;
     int flag = addNewEdge(edgeList, value, sourceVertex, sinkVertex, readId, kmerpos);
     delete[] value;
@@ -348,7 +348,9 @@ void *receiverRunner(void *args) {
                 int sendMsgSize = 0;
 
                 if (vertexList->count(vertexId) == 0) {
+                    vertexList->at(vertexId);
                     cerr << "Cannot query the vertex #" << vertexId << endl;
+                    cerr << thisRank << endl;
                     queryStatus = FAILED_QUERY;
                 } else {
                     // vertex存在时
@@ -361,13 +363,15 @@ void *receiverRunner(void *args) {
                         // 并行环境下不能使用tangleList->count(vertexId) != 0来判断是否为tangle
                         queryStatus = TANGLE_QUERY;
                         Vertex *thisVertex = vertexList->at(vertexId);
-                        edgeId = *thisVertex->outKMer->begin();
+                        /*edgeId = *thisVertex->outKMer->begin();
                         // 删除vertex的一个出度
-                        removeOutEdge(vertexList, vertexId, edgeId);
+                        removeOutEdge(vertexList, vertexId, edgeId);*/
+                        getAndRemoveOutEdge(vertexList, vertexId, &edgeId); // 保护并行环境下的删除节点
                     }else {
-                        edgeId = *thisVertex->outKMer->begin();
+                        /*edgeId = *thisVertex->outKMer->begin();
                         // 删除vertex的一个出度
-                        removeOutEdge(vertexList, vertexId, edgeId);
+                        removeOutEdge(vertexList, vertexId, edgeId);*/
+                        getAndRemoveOutEdge(vertexList, vertexId, &edgeId); // 保护并行环境下的删除节点
                         queryStatus = SUCCESSFUL_QUERY;
                     }
                 }
diff --git a/entity/idset.cpp b/entity/idset.cpp
@@ -63,3 +63,8 @@ int setIntersectTo(SetOfID *a, const SetOfID &b) {
     delete r;
     return a->size();
 }
+
+
+size_t idBelongTo(int world_size, size_t id) {
+    return id % (size_t) world_size;
+}
diff --git a/entity/idset.h b/entity/idset.h
@@ -18,6 +18,8 @@ inline size_t getId(const char *value) {
     return std::hash<std::string>()(value);
 }
 
+size_t idBelongTo(int world_size, size_t id);
+
 //typedef std::unordered_set<size_t> SetOfID;
 typedef size_t ReadId;
 typedef size_t VertexId;
diff --git a/entity/k_minus_mer.cpp b/entity/k_minus_mer.cpp
@@ -224,15 +224,40 @@ int removeOutEdge(const VertexList *vList, const VertexId vId, const EdgeId eId)
     }
     v->outKMer->erase_item(eId);
     v->outDegree--;
-    pthread_mutex_unlock(&kMinusMutex);
+
     if (v->outKMer->size() != v->outDegree) {
         cerr << "Error occurs when removing out edge #"<< eId << "from vertex #" << vId << ".\n";
         v->outDegree++;
         return 0;
     }
+    pthread_mutex_unlock(&kMinusMutex);
     return 1;
 }
 
+int getAndRemoveOutEdge(const VertexList *vList, VertexId vId, EdgeId *eId) {
+    if (vList->find(vId) == vList->end()) {
+        cerr << "Error occurs when deleting an out edge of an non-exist vertex #" << vId << ".\n";
+        return 0;
+    }
+    Vertex *v = vList->at(vId);
+    pthread_mutex_lock(&kMinusMutex);
+    EdgeId nextOutEdgeToDelete = *v->outKMer->begin();
+    if (v->outKMer->count(nextOutEdgeToDelete) == 0) {
+        cerr << "Error occurs when deleting an non-exist out edge #" << eId<< " of a vertex #" << vId << ".\n";
+        return 0;
+    }
+    v->outKMer->erase_item(nextOutEdgeToDelete);
+    v->outDegree--;
+    *eId = nextOutEdgeToDelete;
+    if (v->outKMer->size() != v->outDegree) {
+        cerr << "Error occurs when removing out edge #"<< eId << "from vertex #" << vId << ".\n";
+        v->outDegree++;
+        return 0;
+    }
+    pthread_mutex_unlock(&kMinusMutex);
+    return 0;
+}
+
 int freeVertex(Vertex *pVertex) {
     delete pVertex;
     return 1;
diff --git a/entity/k_minus_mer.h b/entity/k_minus_mer.h
@@ -57,5 +57,7 @@ int addOutEdge(const VertexList *vList, VertexId vId, EdgeId eId);
 
 int removeOutEdge(const VertexList *vList, VertexId vId, EdgeId eId);
 
+int getAndRemoveOutEdge(const VertexList *vList, VertexId, EdgeId *eId);
+
 int freeVertex(Vertex *pVertex);
 #endif //RANDOMSTRINGASSEMBLY_K_MINUS_MER_H
diff --git a/main.cpp b/main.cpp

Original file line number	Diff line number	Diff line change
`@@ -63,3 +63,8 @@ int setIntersectTo(SetOfID *a, const SetOfID &b) {`
`63`	`63`	`delete r;`
`64`	`64`	`return a->size();`
`65`	`65`	`}`
	`66`	`+`
	`67`	`+`
	`68`	`+size_t idBelongTo(int world_size, size_t id) {`
	`69`	`+ return id % (size_t) world_size;`
	`70`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,8 @@ inline size_t getId(const char *value) {`
`18`	`18`	`return std::hash<std::string>()(value);`
`19`	`19`	`}`
`20`	`20`
	`21`	`+size_t idBelongTo(int world_size, size_t id);`
	`22`	`+`
`21`	`23`	`//typedef std::unordered_set<size_t> SetOfID;`
`22`	`24`	`typedef size_t ReadId;`
`23`	`25`	`typedef size_t VertexId;`