Source · fairmq/devices/
Sink.hheader
Described in builtin-devices: a topology of built-in devices, Run a benchmark, Built-in devices.
1/********************************************************************************2 * Copyright (C) 2014-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_SINK_H10#define FAIR_MQ_SINK_H11 12#include <fairmq/Device.h>13#include <fairmq/tools/Strings.h>14 15#include <chrono>16#include <fairlogger/Logger.h>17#include <fstream>18#include <string>19#include <stdexcept>20 21namespace fair::mq22{23 24class Sink : public Device25{26 protected:27 bool fMultipart = false;28 uint64_t fMaxIterations = 0;29 uint64_t fNumIterations = 0;30 uint64_t fMaxFileSize = 0;31 uint64_t fBytesWritten = 0;32 std::string fInChannelName;33 std::string fOutFilename;34 std::fstream fOutputFile;35 36 void InitTask() override37 {38 fMultipart = fConfig->GetProperty<bool>("multipart");39 fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");40 fMaxFileSize = fConfig->GetProperty<uint64_t>("max-file-size");41 fInChannelName = fConfig->GetProperty<std::string>("in-channel");42 fOutFilename = fConfig->GetProperty<std::string>("out-filename");43 44 fBytesWritten = 0;45 }46 47 void Run() override48 {49 // store the channel reference to avoid traversing the map on every loop iteration50 Channel& dataInChannel = GetChannel(fInChannelName, 0);51 52 LOG(info) << "Starting sink and expecting to receive " << fMaxIterations << " messages.";53 auto tStart = std::chrono::high_resolution_clock::now();54 55 if (!fOutFilename.empty()) {56 LOG(debug) << "Incoming messages will be written to file: " << fOutFilename;57 if (fMaxFileSize != 0) {58 LOG(debug) << "File output will stop after " << fMaxFileSize << " bytes";59 } else {60 LOG(debug) << "ATTENTION: --max-file-size is 0 - output file will continue to grow until sink is stopped";61 }62 63 fOutputFile.open(fOutFilename, std::ios::out | std::ios::binary);64 if (!fOutputFile) {65 LOG(error) << "Could not open '" << fOutFilename;66 throw std::runtime_error(fair::mq::tools::ToString("Could not open '", fOutFilename));67 }68 }69 70 while (!NewStatePending()) {71 if (fMultipart) {72 Parts parts;73 if (dataInChannel.Receive(parts) < 0) {74 continue;75 }76 if (fOutputFile.is_open()) {77 for (const auto& part : parts) {78 WriteToFile(static_cast<const char*>(part->GetData()), part->GetSize());79 }80 }81 } else {82 MessagePtr msg(dataInChannel.NewMessage());83 if (dataInChannel.Receive(msg) < 0) {84 continue;85 }86 if (fOutputFile.is_open()) {87 WriteToFile(static_cast<const char*>(msg->GetData()), msg->GetSize());88 }89 }90 91 if (fMaxFileSize > 0 && fBytesWritten >= fMaxFileSize) {92 LOG(info) << "Written " << fBytesWritten << " bytes, stopping...";93 break;94 }95 if (fMaxIterations > 0) {96 if (fNumIterations >= fMaxIterations) {97 LOG(info) << "Configured maximum number of iterations reached.";98 break;99 }100 }101 fNumIterations++;102 }103 104 if (fOutputFile.is_open()) {105 fOutputFile.flush();106 fOutputFile.close();107 }108 109 auto tEnd = std::chrono::high_resolution_clock::now();110 auto ms = std::chrono::duration<double, std::milli>(tEnd - tStart).count();111 LOG(info) << "Received " << fNumIterations << " messages in " << ms << "ms.";112 if (!fOutFilename.empty()) {113 auto sec = std::chrono::duration<double>(tEnd - tStart).count();114 LOG(info) << "Closed '" << fOutFilename << "' after writing " << fBytesWritten << " bytes."115 << "(" << (fBytesWritten / (1000. * 1000.)) / sec << " MB/s)";116 }117 118 LOG(info) << "Leaving RUNNING state.";119 }120 121 void WriteToFile(const char* ptr, size_t size)122 {123 fOutputFile.write(ptr, size);124 if (fOutputFile.bad()) {125 LOG(error) << "failed writing to file";126 throw std::runtime_error("failed writing to file");127 }128 fBytesWritten += size;129 }130};131 132} // namespace fair::mq133 134#endif /* FAIR_MQ_SINK_H */