Now PutDataArray slabs in an async mode.

Author: philippeVerney

src/etp/fesapi/FesapiHdfProxy.cpp

@@ -326,6 +326,33 @@ void FesapiHdfProxy::writeArrayNdSlab(
 	const uint64_t* numValuesInEachDimension,
 	const uint64_t* offsetInEachDimension,
 	unsigned int numDimensions)
+{
+	std::set<int64_t>stillProcessingMsgIds = async_writeArrayNdSlab(groupName, datasetName,
+		datatype, values, numValuesInEachDimension,
+		offsetInEachDimension, numDimensions);
+
+	auto t_start = std::chrono::high_resolution_clock::now();
+	while (!stillProcessingMsgIds.empty()) {
+		for (int64_t msgId : stillProcessingMsgIds) {
+			if (!session_->isMessageStillProcessing(msgId)) {
+				stillProcessingMsgIds.erase(msgId);
+			}
+		}
+		if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > session_->getTimeOut()) {
+			throw std::runtime_error("Time out waiting for a writeArrayNdSlab response");
+		}
+	}
+	std::cerr << "writeArrayNdSlab Response got in " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() << " ms" << std::endl;
+}
+
+std::set<int64_t> FesapiHdfProxy::async_writeArrayNdSlab(
+	const string& groupName,
+	const string& datasetName,
+	COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
+	const void* values,
+	const uint64_t* numValuesInEachDimension,
+	const uint64_t* offsetInEachDimension,
+	unsigned int numDimensions)
 {
 	if (!isOpened())
 		open();
@@ -370,6 +397,7 @@ void FesapiHdfProxy::writeArrayNdSlab(
 			"You need to give a COMMON_NS::AbstractObject::numericalDatatypeEnum as the datatype");
 	}
 
+	std::set<int64_t> sentMessageIds;
 	if (totalCount * valueSize <= maxArraySize_) {
 		std::vector<int64_t> counts;
 		std::vector<int64_t> starts;
@@ -389,7 +417,7 @@ void FesapiHdfProxy::writeArrayNdSlab(
 		pdsa.dataSubarrays["0"].data = convertVoidArrayIntoAvroAnyArray(datatype, values, totalCount);
 
 		// Send putDataSubarrays Message
-		session_->sendAndBlock(pdsa, 0, 0x02);
+		sentMessageIds.insert(session_->send(pdsa, 0, 0x02));
 	}
 	else {
 		std::unique_ptr<uint64_t[]> counts(new uint64_t[numDimensions]);
@@ -405,10 +433,11 @@ void FesapiHdfProxy::writeArrayNdSlab(
 			if (numValuesInEachDimension[dimIdx] > 1) {
 				uint64_t previousCount = counts[dimIdx];
 				counts[dimIdx] /= 2;
-				
-				writeArrayNdSlab(groupName, datasetName,
+
+				std::set<int64_t> intermediateResult = async_writeArrayNdSlab(groupName, datasetName,
 					datatype, values, counts.get(),
 					starts.get(), numDimensions);
+				sentMessageIds.insert(intermediateResult.begin(), intermediateResult.end());
 
 				writtenTotalCount = std::accumulate(counts.get(), counts.get() + numDimensions, 1, std::multiplies<size_t>());
 
@@ -424,10 +453,13 @@ void FesapiHdfProxy::writeArrayNdSlab(
 				+ std::to_string(maxArraySize_) + " bytes.");
 		}
 
-		writeArrayNdSlab(groupName, datasetName, datatype,
+		std::set<int64_t> intermediateResult = async_writeArrayNdSlab(groupName, datasetName, datatype,
 			(int8_t*)values + (writtenTotalCount * valueSize), counts.get(),
 			starts.get(), numDimensions);
+		sentMessageIds.insert(intermediateResult.begin(), intermediateResult.end());
 	}
+
+	return sentMessageIds;
 }
 
 void FesapiHdfProxy::readArrayNdOfDoubleValues(

src/etp/fesapi/FesapiHdfProxy.h

@@ -23,6 +23,7 @@ under the License.
 #include "../AbstractSession.h"
 #include "../ProtocolHandlers/GetFullDataArrayHandlers.h"
 
+#include <set>
 #include <type_traits>
 
 namespace ETP_NS
@@ -184,6 +185,26 @@ namespace ETP_NS
 			const uint64_t* offsetValuesInEachDimension,
 			unsigned int numDimensions) final;
 
+		/**
+		* Find the array associated with @p groupName and @p name and write to it asynchronously.
+		* @param groupName                      The name of the group associated with the array.
+		* @param name                           The name of the array (potentially with multi dimensions).
+		* @param datatype						The specific datatype of the values to write.
+		* @param values                         1d array of specific datatype ordered firstly by fastest direction.
+		* @param numValuesInEachDimension       Number of values in each dimension of the array to write. They are ordered from fastest index to slowest index.
+		* @param offsetValuesInEachDimension    Offset values in each dimension of the array to write. They are ordered from fastest index to slowest index.
+		* @param numDimensions                  The number of the dimensions of the array to write.
+		* @return								All message ids which have been sent to the ETP server
+		*/
+		std::set<int64_t> async_writeArrayNdSlab(
+			const std::string& groupName,
+			const std::string& name,
+			COMMON_NS::AbstractObject::numericalDatatypeEnum datatype,
+			const void* values,
+			const uint64_t* numValuesInEachDimension,
+			const uint64_t* offsetValuesInEachDimension,
+			unsigned int numDimensions);
+
 		/**
 		* Write some string attributes into a group
 		*/