Changeset 944

Timestamp:

10/20/06 12:10:17 (7 years ago)

Author:

leo

Message:

using packed message

Files:

• sandbox/trunk/pdp/src/Replication.cpp (modified) (6 diffs)

sandbox/trunk/pdp/src/Replication.cpp

@@ -117 +117 @@
 {
         const int serverProcess = 0;
-        int rank = MPI::COMM_WORLD.Get_rank();
-        int numProcess = MPI::COMM_WORLD.Get_size();
+
+        MPI::Comm& comm = MPI::COMM_WORLD;
+
+        int rank = comm.Get_rank();
+        int numProcess = comm.Get_size();
 
         if(numProcess < 2)
                 XERIAL_THROW(Exception, "numProcess must be more than 2");
 
-        vector<uint8> msgBuffer(_bufferSize, 0);
-        uint8* buffer = &(msgBuffer[0]);
+        size_t maximumPackedDataSize = MPI::INT.Pack_size(1, comm) + MPI::BYTE.Pack_size(_bufferSize, comm);
+        XERIAL_DEBUG(_logger, "message size max = " << maximumPackedDataSize);
+
+        vector<uint8> fileBuffer(_bufferSize, 0);
+        vector<uint8> packedDataBuffer(maximumPackedDataSize, 0);
 
     if(rank == serverProcess)
@@ -134 +140 @@
                 uint64 fileSize = FileUtil::fileSize(_fileName);
                 XERIAL_DEBUG(_logger, "file size = " << fileSize);
-                MPI::COMM_WORLD.Bcast(&fileSize, 1, MPI::INTEGER8, serverProcess);
-
+                comm.Bcast(&fileSize, 1, MPI::INTEGER8, serverProcess);
+
+                
                 uint64 sendBytes = 0;
                 while(sendBytes < fileSize && !feof(fileHandle))
                 {
-                        int readBytes = static_cast<int>(fread(buffer, 1, _bufferSize, fileHandle));
+                        int readBytes = static_cast<int>(fread(&(fileBuffer[0]), 1, _bufferSize, fileHandle));
                         if(readBytes < static_cast<int>(_bufferSize))
                         {
@@ -145 +152 @@
                                         XERIAL_THROW(Exception, format("actual read bytes = %1, specified read size = %2") % readBytes % int(_bufferSize));
                         }
-                        MPI::COMM_WORLD.Bcast(&readBytes, 1, MPI::INT, serverProcess);
-                        MPI::COMM_WORLD.Bcast(buffer, readBytes, MPI::BYTE, serverProcess);
+                        int sendBufferPosition = 0;
+                        MPI::INT.Pack(&readBytes, 1, &(packedDataBuffer[0]), maximumPackedDataSize, sendBufferPosition, comm);
+                        MPI::BYTE.Pack(&(fileBuffer[0]), readBytes, &(packedDataBuffer[0]), maximumPackedDataSize, sendBufferPosition, comm);
+                        
+                        comm.Bcast(&(packedDataBuffer[0]), sendBufferPosition, MPI::PACKED, serverProcess);
                         sendBytes += readBytes;
                         XERIAL_DEBUG(_logger, "send bytes = " << sendBytes);
@@ -154 +164 @@
         else
         {
+
                 uint64 fileSize = 0;
-                MPI::COMM_WORLD.Bcast(&fileSize, 1, MPI::INTEGER8, serverProcess);
+                comm.Bcast(&fileSize, 1, MPI::INTEGER8, serverProcess);
 
                 XERIAL_DEBUG(_logger, "incoming file size = " << fileSize);
@@ -165 +176 @@
                         XERIAL_THROW(Exception, "cannot open file " << outputFileName);
 
+
                 uint64 recievedBytes = 0;
                 while(recievedBytes < fileSize)
                 {
+                        comm.Bcast(&(packedDataBuffer[0]), maximumPackedDataSize, MPI::PACKED, serverProcess);
                         int incomingDataSize = 0;
-                        MPI::COMM_WORLD.Bcast(&incomingDataSize, 1, MPI::INT, serverProcess);
+                        int packedBufferPosition = 0;
+                        MPI::INT.Unpack(&(packedDataBuffer[0]), 1, &incomingDataSize, 1, packedBufferPosition, comm);
                         XERIAL_DEBUG(_logger, "incoming bytes = " << incomingDataSize);
-                        MPI::COMM_WORLD.Bcast(buffer, incomingDataSize, MPI::BYTE, serverProcess);
-
-                        size_t writesize = fwrite(buffer, 1, incomingDataSize, fileHandle);
+                        //MPI::BYTE.Unpack(&(packedDataBuffer[0]), incomingDataSize, &(fileBuffer[0]), _bufferSize, packedBufferPosition, comm);
+                        size_t writesize = fwrite(&(packedDataBuffer[packedBufferPosition]), 1, incomingDataSize, fileHandle);
                         if(writesize != incomingDataSize)
                                 XERIAL_THROW(Exception, "write failed. write size =  " << writesize);
@@ -182 +195 @@
                 fclose(fileHandle);
         }
-        XERIAL_INFO(_logger, "done.");
+        XERIAL_INFO(_logger, format("process %1%: done.") % rank);
 }
 void