added socket communication

EventVisualisation/DataConverter/CMakeLists.txt

@@ -18,6 +18,7 @@ o2_add_library(EventVisualisationDataConverter
                        src/VisualisationEventJSONSerializer.cxx
                        src/VisualisationEventROOTSerializer.cxx
                        src/VisualisationEventOpenGLSerializer.cxx
+                       src/Location.cxx
                PUBLIC_LINK_LIBRARIES RapidJSON::RapidJSON
                         O2::ReconstructionDataFormats
                         O2::DataFormatsParameters
@@ -33,6 +34,7 @@ o2_add_executable(eve-convert
                 src/VisualisationTrack.cxx
                 src/VisualisationCluster.cxx
                 src/VisualisationCalo.cxx
+                src/Location.cxx
         PUBLIC_LINK_LIBRARIES
                 O2::EventVisualisationView
                 RapidJSON::RapidJSON

EventVisualisation/DataConverter/include/EventVisualisationDataConverter/Location.h

@@ -0,0 +1,70 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file    Location.h
+/// \author  Julian Myrcha
+///
+
+#ifndef O2EVE_LOCATION_H
+#define O2EVE_LOCATION_H
+
+#include <string>
+#include <fstream>
+#include <iosfwd>
+
+namespace o2::event_visualisation
+{
+struct LocationParams {
+  std::string fileName;
+  int port = -1;
+  int timeout = 100;
+  std::string host = "localhost";
+  bool toFile = true;
+  bool toSocket = true;
+};
+class Location
+{
+  std::ofstream* mOut;
+  int mClientSocket;
+  bool mToFile;
+  bool mToSocket;
+  std::string mFileName;
+  int mPort;
+  int mTimeout;
+  std::string mHostName;
+
+ public:
+  explicit Location(const LocationParams& params)
+  {
+    this->mFileName = params.fileName;
+    this->mToFile = !params.fileName.empty() && params.toFile;
+    this->mToSocket = params.port != -1 && params.toSocket;
+    this->mOut = nullptr;
+    this->mPort = params.port;
+    this->mHostName = params.host;
+    this->mClientSocket = -1;
+    this->mTimeout = params.timeout;
+  }
+  ~Location()
+  {
+    close();
+  }
+  void open();
+  void close();
+  void write(char* buf, std::streamsize size);
+  [[nodiscard]] std::string fileName() const { return this->mFileName; }
+  [[nodiscard]] std::string hostName() const { return this->mHostName; }
+  [[nodiscard]] int port() const { return this->mPort; }
+};
+} // namespace o2::event_visualisation
+
+#endif // O2EVE_LOCATION_H

EventVisualisation/DataConverter/include/EventVisualisationDataConverter/VisualisationEventJSONSerializer.h

@@ -56,7 +56,7 @@ class VisualisationEventJSONSerializer : public VisualisationEventSerializer
 
  public:
   bool fromFile(VisualisationEvent& event, std::string fileName) override;
-  void toFile(const VisualisationEvent& event, std::string fileName) override;
+  void toFile(const VisualisationEvent& event, Location location) override;
   ~VisualisationEventJSONSerializer() override = default;
 };
 

EventVisualisation/DataConverter/include/EventVisualisationDataConverter/VisualisationEventOpenGLSerializer.h

@@ -38,7 +38,7 @@ class VisualisationEventOpenGLSerializer : public VisualisationEventSerializer
  public:
   const std::string serializerName() const override { return std::string("VisualisationEventOpenGLSerializer"); }
   bool fromFile(VisualisationEvent& event, std::string fileName) override;
-  void toFile(const VisualisationEvent& event, std::string fileName) override;
+  void toFile(const VisualisationEvent& event, Location location) override;
   ~VisualisationEventOpenGLSerializer() override = default;
 };
 

EventVisualisation/DataConverter/include/EventVisualisationDataConverter/VisualisationEventROOTSerializer.h

@@ -44,7 +44,7 @@ class VisualisationEventROOTSerializer : public VisualisationEventSerializer
  public:
   [[nodiscard]] const std::string serializerName() const override { return std::string("VisualisationEventROOTSerializer"); }
   bool fromFile(VisualisationEvent& event, std::string fileName) override;
-  void toFile(const VisualisationEvent& event, std::string fileName) override;
+  void toFile(const VisualisationEvent& event, Location location) override;
   ~VisualisationEventROOTSerializer() override = default;
 };
 

EventVisualisation/DataConverter/include/EventVisualisationDataConverter/VisualisationEventSerializer.h

@@ -17,6 +17,7 @@
 #define O2EVE_VISUALISATIONEVENTSERIALIZER_H
 
 #include "EventVisualisationDataConverter/VisualisationEvent.h"
+#include "EventVisualisationDataConverter/Location.h"
 #include <string>
 #include <map>
 
@@ -45,7 +46,7 @@ class VisualisationEventSerializer
   static o2::dataformats::GlobalTrackID deserialize(unsigned source, unsigned index, unsigned flags);
   static VisualisationEventSerializer* getInstance(std::string ext) { return instances[ext]; }
   virtual bool fromFile(VisualisationEvent& event, std::string fileName) = 0;
-  virtual void toFile(const VisualisationEvent& event, std::string fileName) = 0;
+  virtual void toFile(const VisualisationEvent& event, Location location) = 0;
   virtual const std::string serializerName() const = 0;
   virtual ~VisualisationEventSerializer() = default;
 };

EventVisualisation/DataConverter/src/Location.cxx

@@ -0,0 +1,182 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file    Location.cxx
+/// \author  Julian Myrcha
+///
+
+#include "EventVisualisationDataConverter/Location.h"
+#include <fairlogger/Logger.h>
+#include <sys/socket.h>
+#include <unistd.h>
+#include <netdb.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <ctime>
+
+using namespace std;
+
+namespace o2::event_visualisation
+{
+
+int connect_with_timeout(const int socket, const struct sockaddr* addr, socklen_t addrlen, const unsigned int timeout_ms)
+{
+  int connection = 0;
+  // Setting O_NONBLOCK
+  int socket_flags_before;
+  if ((socket_flags_before = fcntl(socket, F_GETFL, 0) < 0)) {
+    return -1;
+  }
+  if (fcntl(socket, F_SETFL, socket_flags_before | O_NONBLOCK) < 0) {
+    return -1;
+  }
+  do {
+    if (connect(socket, addr, addrlen) < 0) {
+      if ((errno != EWOULDBLOCK) && (errno != EINPROGRESS)) {
+        connection = -1; // error
+      } else {           // wait for complete
+        // deadline 'timeout' ms from now
+        timespec now; // NOLINT(*-pro-type-member-init)
+        if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
+          connection = -1;
+          break;
+        }
+        const timespec deadline = {.tv_sec = now.tv_sec,
+                                   .tv_nsec = now.tv_nsec + timeout_ms * 1000000l};
+        do {
+          if (clock_gettime(CLOCK_MONOTONIC, &now) < 0) {
+            connection = -1;
+            break;
+          }
+          // compute remaining deadline
+          const int ms_until_deadline = static_cast<int>((deadline.tv_sec - now.tv_sec) * 1000l + (deadline.tv_nsec - now.tv_nsec) / 1000000l);
+          if (ms_until_deadline < 0) {
+            connection = 0;
+            break;
+          }
+          pollfd connectionPool[] = {{.fd = socket, .events = POLLOUT}};
+          connection = poll(connectionPool, 1, ms_until_deadline);
+
+          if (connection > 0) { // confirm the success
+            int error = 0;
+            socklen_t len = sizeof(error);
+            if (getsockopt(socket, SOL_SOCKET, SO_ERROR, &error, &len) == 0) {
+              errno = error;
+            }
+            if (error != 0) {
+              connection = -1;
+            }
+          }
+        } while (connection == -1 && errno == EINTR); // If interrupted, try again.
+        if (connection == 0) {
+          errno = ETIMEDOUT;
+          connection = -1;
+        }
+      }
+    }
+  } while (false);
+  // Restore socket state
+  if (fcntl(socket, F_SETFL, socket_flags_before) < 0) {
+    return -1;
+  }
+  return connection;
+}
+
+void Location::open()
+{
+  if (this->mToFile) {
+    this->mOut = new std::ofstream(mFileName, std::ios::out | std::ios::binary);
+  }
+  if (this->mToSocket) {
+    // resolve host name
+    sockaddr_in serverAddress; // NOLINT(*-pro-type-member-init)
+    serverAddress.sin_family = AF_INET;
+    serverAddress.sin_port = htons(this->mPort); // Port number
+
+    // ask once
+    static auto server = gethostbyname(this->mHostName.c_str());
+    if (server == nullptr) {
+      LOGF(info, "Error no such host %s", this->mHostName.c_str());
+      return;
+    };
+
+    bcopy((char*)server->h_addr,
+          (char*)&serverAddress.sin_addr.s_addr,
+          server->h_length);
+
+    // Connect to the server
+    this->mClientSocket = socket(AF_INET, SOCK_STREAM, 0);
+    if (this->mClientSocket == -1) {
+      LOGF(info, "Error creating socket");
+      return;
+    }
+
+    if (connect_with_timeout(this->mClientSocket, (sockaddr*)&serverAddress,
+                             sizeof(serverAddress), this->mTimeout) == -1) {
+      LOGF(info, "Error connecting to %s:%d", this->mHostName.c_str(), this->mPort);
+      ::close(this->mClientSocket);
+      this->mClientSocket = -1;
+      return;
+    }
+    try {
+      char buf[256] = "SEND:";
+      strncpy(buf + 6, this->mFileName.c_str(), sizeof(buf) - 7);
+      strncpy(buf + sizeof(buf) - 6, "ALICE", 6);
+      auto real = send(this->mClientSocket, buf, sizeof(buf), 0);
+      if (real != sizeof(buf)) {
+        throw real;
+      }
+    } catch (...) {
+      ::close(this->mClientSocket);
+      this->mClientSocket = -1;
+      LOGF(info, "Error sending file name to %s:%d", this->mHostName.c_str(), this->mPort);
+    }
+  }
+}
+
+void Location::close()
+{
+  if (this->mToFile && this->mOut) {
+    this->mOut->close();
+    delete this->mOut;
+    this->mOut = nullptr;
+  }
+  if (this->mToSocket && this->mClientSocket != -1) {
+    ::close(this->mClientSocket);
+    this->mClientSocket = -1;
+  }
+}
+
+void Location::write(char* buf, std::streamsize size)
+{
+  if (size == 0) {
+    return;
+  }
+  if (this->mToFile && this->mOut) {
+    this->mOut->write(buf, size);
+  }
+  if (this->mToSocket && this->mClientSocket != -1) {
+    LOGF(info, "Location::write() socket %s ++++++++++++++++++++++", fileName());
+    try {
+      auto real = send(this->mClientSocket, buf, size, 0);
+      if (real != size) {
+        throw real;
+      }
+    } catch (...) {
+      ::close(this->mClientSocket);
+      this->mClientSocket = -1;
+      LOGF(info, "Error sending data to %s:%d", this->mHostName.c_str(), this->mPort);
+    }
+  }
+}
+
+} // namespace o2::event_visualisation

EventVisualisation/DataConverter/src/VisualisationEventJSONSerializer.cxx

@@ -29,8 +29,9 @@ using namespace rapidjson;
 namespace o2::event_visualisation
 {
 
-void VisualisationEventJSONSerializer::toFile(const VisualisationEvent& event, std::string fileName)
+void VisualisationEventJSONSerializer::toFile(const VisualisationEvent& event, Location location)
 {
+  std::string fileName = location.fileName();
   std::string json = toJson(event);
   std::ofstream out(fileName);
   out << json;