Aggregator <-> RTCPD protocol started. Got as far as receiving the

ackowledge of a request for a VID from RTCPD.  Still a long, long way to go,
but it's a start.

castor/tape/aggregator/AggregatorDaemon.cpp

@@ -128,9 +128,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
           // Log and throw an exception
           castor::dlf::Param params[] = {
-            castor::dlf::Param("reason", oss.str())};
+            castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+            castor::dlf::Param("Reason"  , oss.str())};
           castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-            AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+            AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
           castor::exception::InvalidArgument ex;
           ex.getMessage() << oss.str();
           throw ex;
@@ -148,9 +149,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
         // Log and throw an exception
         castor::dlf::Param params[] = {
-          castor::dlf::Param("reason", oss.str())};
+          castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+          castor::dlf::Param("Reason"  , oss.str())};
         castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
         castor::exception::InvalidArgument ex;
         ex.getMessage() << oss.str();
         throw ex;
@@ -162,9 +164,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
         // Log and throw an exception
         castor::dlf::Param params[] = {
-          castor::dlf::Param("reason", oss.str())};
+          castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+          castor::dlf::Param("Reason"  , oss.str())};
         castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
         castor::exception::InvalidArgument ex;
         ex.getMessage() << oss.str();
         throw ex;
@@ -177,9 +180,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
         // Log and throw an exception
         castor::dlf::Param params[] = {
-          castor::dlf::Param("reason", oss.str())};
+          castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+          castor::dlf::Param("Reason"  , oss.str())};
         castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+          AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
         castor::exception::Internal ex;
         ex.getMessage() << oss.str();
         throw ex;
@@ -193,9 +197,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
     // Log and throw an exception
     castor::dlf::Param params[] = {
-      castor::dlf::Param("reason", oss.str())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Reason", oss.str())};
     castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+      AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
     castor::exception::Internal ex;
     ex.getMessage() << oss.str();
     throw ex;
@@ -210,9 +215,10 @@ void castor::tape::aggregator::AggregatorDaemon::parseCommandLine(
 
     // Log and throw an exception
     castor::dlf::Param params[] = {
-      castor::dlf::Param("reason", oss.str())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Reason"  , oss.str())};
     castor::dlf::dlf_writep(nullCuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 1, params);
+      AGGREGATOR_FAILED_TO_PARSE_COMMAND_LINE, 2, params);
     castor::exception::InvalidArgument e;
     e.getMessage() << oss.str();
     throw e;

castor/tape/aggregator/AggregatorDlfMessageConstants.hpp

-/* This file was generated by ./AggregatorDlfMessagesCodeGenerator on Tue Dec  2 16:35:03 CET 2008
+/* This file was generated by ./AggregatorDlfMessagesCodeGenerator on Thu Dec  4 11:35:54 CET 2008
  */
 
 /******************************************************************************

castor/tape/aggregator/AggregatorDlfMessageStrings.cpp

-/* This file was generated by ./AggregatorDlfMessagesCodeGenerator on Tue Dec  2 16:35:03 CET 2008
+/* This file was generated by ./AggregatorDlfMessagesCodeGenerator on Thu Dec  4 11:35:54 CET 2008
  */
 
 /******************************************************************************

castor/tape/aggregator/Constants.hpp

@@ -34,7 +34,7 @@ namespace tape {
 namespace aggregator {
   	
 const size_t STRERRORBUFLEN = 256;
-const int    NETRW_TIMEOUT  = 300;
+const int    NETRWTIMEOUT   = 300;
 const size_t HDRBUFSIZ      = 3 * sizeof(uint32_t);
 const size_t MSGBUFSIZ      = 1024;
 

castor/tape/aggregator/Marshaller.cpp

@@ -259,7 +259,7 @@ size_t castor::tape::aggregator::Marshaller::marshallRcpJobRequest(
   const size_t totalLen = 3 * sizeof(uint32_t) + len;
 
   // Check that the message buffer is big enough
-  if(sizeof(dstLen) <  totalLen) {
+  if(totalLen > dstLen) {
     castor::exception::Exception ex(EMSGSIZE);
 
     ex.getMessage() << "Buffer too small for job submission request message: "
@@ -398,7 +398,7 @@ size_t castor::tape::aggregator::Marshaller::marshallRTCPTapeRequest(char *dst,
   const size_t totalLen = 3 * sizeof(uint32_t) + len;
 
   // Check that the message buffer is big enough
-  if(sizeof(dstLen) <  totalLen) {
+  if(totalLen > dstLen) {
     castor::exception::Exception ex(EMSGSIZE);
 
     ex.getMessage() << "Buffer too small for tape request message: "

castor/tape/aggregator/RcpJobSubmitter.cpp

@@ -175,7 +175,7 @@ void castor::tape::aggregator::RcpJobSubmitter::readReply(
   // Read and unmarshall the magic number of the request
   uint32_t magic = 0;
   try {
-    magic = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    magic = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::exception::Exception ex2(EIO);
 
@@ -202,7 +202,7 @@ void castor::tape::aggregator::RcpJobSubmitter::readReply(
   // Read and unmarshall the type of the request
   uint32_t reqtype = 0;
   try {
-    reqtype = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    reqtype = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::exception::Exception ex2(EIO);
 
@@ -229,7 +229,7 @@ void castor::tape::aggregator::RcpJobSubmitter::readReply(
   // Read and unmarshall the type of the request
   uint32_t len = 0;
   try {
-    len = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    len = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::exception::Exception ex2(EIO);
 
@@ -272,7 +272,7 @@ void castor::tape::aggregator::RcpJobSubmitter::readReply(
 
   // Read the message body from the socket
   try {
-    SocketHelper::readBytes(socket, NETRW_TIMEOUT, len, body);
+    SocketHelper::readBytes(socket, NETRWTIMEOUT, len, body);
   } catch (castor::exception::Exception &ex) {
     castor::exception::Exception ex2(EIO);
 

castor/tape/aggregator/RtcpdHandlerThread.cpp

@@ -34,6 +34,9 @@
 #include "castor/tape/aggregator/exception/RTCPDErrorMessage.hpp"
 #include "h/common.h"
 #include "h/rtcp.h"
+#include "h/rtcp_constants.h"
+
+#include <string.h>
 
 
 //-----------------------------------------------------------------------------
@@ -47,7 +50,7 @@ castor::tape::aggregator::RtcpdHandlerThread::RtcpdHandlerThread() throw() {
 // destructor
 //-----------------------------------------------------------------------------
 castor::tape::aggregator::RtcpdHandlerThread::~RtcpdHandlerThread()
-  throw () {
+  throw() {
 }
 
 
@@ -70,8 +73,10 @@ void castor::tape::aggregator::RtcpdHandlerThread::run(void *param)
   Cuuid_create(&cuuid);
 
   if(param == NULL) {
+    castor::dlf::Param params[] = {
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__)};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_RTCPD_HANDLER_SOCKET_IS_NULL);
+      AGGREGATOR_RTCPD_HANDLER_SOCKET_IS_NULL, 1, params);
     return;
   }
 
@@ -93,41 +98,45 @@ void castor::tape::aggregator::RtcpdHandlerThread::run(void *param)
 
   } catch(castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_RTCPD_CONNECTION_WITHOUT_INFO, 2, params);
+      AGGREGATOR_RTCPD_CONNECTION_WITHOUT_INFO, 3, params);
   }
 
   std::string vid;
 
 {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", "Just before getVidFromRtcpd")};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , "Just before getVidFromRtcpd")};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_SYSTEM,
-      AGGREGATOR_NULL, 1, params);
+      AGGREGATOR_NULL, 2, params);
 }
 
   try {
-    vid = getVidFromRtcpd(*socket);
+    vid = getVidFromRtcpd(*socket, NETRWTIMEOUT);
 
 {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", "Just after getVidFromRtcpd")};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , "Just after getVidFromRtcpd")};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_SYSTEM,
-      AGGREGATOR_NULL, 1, params);
+      AGGREGATOR_NULL, 2, params);
 }
 
     castor::dlf::Param params[] = {
-      castor::dlf::Param("VID", vid)};
+      castor::dlf::Param("VID"     , vid)};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_SYSTEM,
       AGGREGATOR_GOT_VID_FROM_RTCPD, 1, params);
   } catch(castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_GET_VID_FROM_RTCPD, 2, params);
+      AGGREGATOR_FAILED_TO_GET_VID_FROM_RTCPD, 3, params);
   }
 
   // Close and de-allocate the socket
@@ -148,13 +157,20 @@ void castor::tape::aggregator::RtcpdHandlerThread::stop()
 // getVidFromRtcpd
 //-----------------------------------------------------------------------------
 std::string castor::tape::aggregator::RtcpdHandlerThread::getVidFromRtcpd(
-  castor::io::AbstractTCPSocket &socket) throw(castor::exception::Exception) {
+  castor::io::AbstractTCPSocket &socket, const int netReadWriteTimeout)
+  throw(castor::exception::Exception) {
 
-  char buffer[MSGBUFSIZ];
+  char buf[MSGBUFSIZ];
   RtcpTapeRequest request;
 
+  memset(&request, '\0', sizeof(request));
+
+  // Marshall the request for VID message
+  size_t totalLen = 0;
+
   try {
-    Marshaller::marshallRTCPTapeRequest(&buffer[0], sizeof(buffer), request);
+    totalLen = Marshaller::marshallRTCPTapeRequest(&buf[0], sizeof(buf),
+      request);
   } catch(castor::exception::Exception &ex) {
     castor::exception::Internal ie;
 
@@ -164,6 +180,42 @@ std::string castor::tape::aggregator::RtcpdHandlerThread::getVidFromRtcpd(
     throw ie;
   }
 
+  // Send the request for a VID to RTCPD
+  const int rc = netwrite_timeout(socket.socket(), buf, totalLen,
+    netReadWriteTimeout);
+
+  if(rc == -1) {
+    castor::exception::Exception ex(SECOMERR);
+    ex.getMessage() << __PRETTY_FUNCTION__
+      << ": Failed netwrite of request for VID to RTCPD: " << neterror();
+    throw ex;
+  } else if(rc == 0) {
+    castor::exception::Exception ex(SECONNDROP);
+    ex.getMessage() << __PRETTY_FUNCTION__
+      << ": Failed netwrite of request for VID to RTCPD: Connection dropped";
+    throw ex;
+  }
+
+  // Receive acknowlege from RTCPD
+  uint32_t status = 0;
+
+  try {
+    status = receiveRtcpdAcknowledge(socket, netReadWriteTimeout);
+  } catch(castor::exception::Exception &ex) {
+    castor::exception::Exception ex2(EPROTO);
+
+    ex2.getMessage() << "Failed to receive acknowlege from RTCPD: "
+      << ex.getMessage().str();
+    throw ex2;
+  }
+
+  if(status != 0) {
+    castor::exception::Exception ex(ECANCELED);
+
+    ex.getMessage() << "Recieved negative acknowledge from RTCPD";
+    throw ex;
+  }
+
   return "NOTIMP";
 /*
   SOCKET            s = socket.socket();
@@ -226,3 +278,76 @@ std::string castor::tape::aggregator::RtcpdHandlerThread::getVidFromRtcpd(
   return tapereq.vid;
 */
 }
+
+
+//-----------------------------------------------------------------------------
+// receiveRtcpdAcknowledge
+//-----------------------------------------------------------------------------
+uint32_t castor::tape::aggregator::RtcpdHandlerThread::receiveRtcpdAcknowledge(
+  castor::io::AbstractTCPSocket &socket, const int netReadWriteTimeout)
+  throw(castor::exception::Exception) {
+
+  // Read and unmarshall the magic number
+  uint32_t magic = 0;
+  try {
+    magic = SocketHelper::readUint32(socket, NETRWTIMEOUT);
+  } catch (castor::exception::Exception &ex) {
+    castor::exception::Internal ie;
+
+    ie.getMessage() << __PRETTY_FUNCTION__
+      << ": Failed to read and unmarshall magic number: "
+      << ex.getMessage().str();
+
+    throw ie;
+  }
+
+  if(magic != RTCOPY_MAGIC) {
+    castor::exception::Exception ex(EINVAL);
+
+    ex.getMessage() << __PRETTY_FUNCTION__
+      << ": Wrong magic number: Expected: " << std::hex << RTCOPY_MAGIC
+      << ": Received: " << magic;
+
+    throw ex;
+  }
+
+  // Read and unmarshall the request type
+  uint32_t reqtype = 0;
+  try {
+    reqtype = SocketHelper::readUint32(socket, NETRWTIMEOUT);
+  } catch (castor::exception::Exception &ex) {
+    castor::exception::Internal ie;
+
+    ie.getMessage() << __PRETTY_FUNCTION__
+      << ": Failed to read and unmarshall request type: "
+      << ex.getMessage().str();
+
+    throw ie;
+  }
+
+  if(reqtype != RTCP_TAPEERR_REQ) {
+    castor::exception::Exception ex(EINVAL);
+
+    ex.getMessage() << __PRETTY_FUNCTION__
+      << ": Wrong request type: Expected: " << std::hex << RTCP_TAPEERR_REQ
+      << ": Received: " << reqtype;
+
+    throw ex;
+  }
+
+  // Read and unmarshall the status
+  uint32_t status = 0;
+  try {
+    status = SocketHelper::readUint32(socket, NETRWTIMEOUT);
+  } catch (castor::exception::Exception &ex) {
+    castor::exception::Internal ie;
+
+    ie.getMessage() << __PRETTY_FUNCTION__
+      << ": Failed to read and unmarshall status: "
+      << ex.getMessage().str();
+
+    throw ie;
+  }
+
+  return status;
+}

castor/tape/aggregator/RtcpdHandlerThread.hpp

@@ -76,8 +76,18 @@ namespace aggregator {
      * @param socket The socket of the connection with RTCPD.
      * @return The VID.
      */
-    std::string getVidFromRtcpd(castor::io::AbstractTCPSocket &socket)
-      throw(castor::exception::Exception);
+    std::string getVidFromRtcpd(castor::io::AbstractTCPSocket &socket,
+      const int netReadWriteTimeout) throw(castor::exception::Exception);
+
+    /**
+     * Receives an acknowledge message from RTCPD and returns the status code
+     * embedded within the message.
+     *
+     * @param socket The socket of the connection with RTCPD.
+     * @return The status code embedded within the message.
+     */
+    uint32_t receiveRtcpdAcknowledge(castor::io::AbstractTCPSocket &socket,
+      const int netReadWriteTimeout) throw(castor::exception::Exception);
 
   }; // class RtcpdHandlerThread
 

castor/tape/aggregator/VdqmRequestHandlerThread.cpp

@@ -93,10 +93,11 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::run(void *param)
   } catch(castor::exception::Exception &ex) {
 
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_HANDLE_VDQM_REQUEST_EXCEPT, 2, params);
+      AGGREGATOR_HANDLE_VDQM_REQUEST_EXCEPT, 3, params);
   }
 
   // Close and de-allocate the socket
@@ -123,13 +124,14 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   // Read and unmarshall the magic number of the request
   uint32_t magic = 0;
   try {
-    magic = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    magic = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_READ_MAGIC, 2, params);
+      AGGREGATOR_FAILED_TO_READ_MAGIC, 3, params);
 
     return;
   }
@@ -139,8 +141,9 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   if(handlerMapItor == m_magicToHandlers.end()) {
     // Unknown magic number
     castor::dlf::Param params[] = {
+      castor::dlf::Param("Function"    , __PRETTY_FUNCTION__),
       castor::dlf::Param("Magic Number", magic)};
-    castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR, AGGREGATOR_UNKNOWN_MAGIC, 1,
+    castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR, AGGREGATOR_UNKNOWN_MAGIC, 2,
       params);
 
     return;
@@ -150,13 +153,14 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   // Read and unmarshall the type of the request
   uint32_t reqtype = 0;
   try {
-    reqtype = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    reqtype = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_READ_REQUEST_TYPE, 2, params);
+      AGGREGATOR_FAILED_TO_READ_REQUEST_TYPE, 3, params);
 
     return;
   }
@@ -166,10 +170,11 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   if(handlerItor == handlers->end()) {
     // Unknown request type
     castor::dlf::Param params[] = {
+      castor::dlf::Param("Function"    , __PRETTY_FUNCTION__),
       castor::dlf::Param("Magic Number", magic),
       castor::dlf::Param("Request Type", reqtype)};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_UNKNOWN_REQUEST_TYPE, 2, params);
+      AGGREGATOR_UNKNOWN_REQUEST_TYPE, 3, params);
 
     return;
   }
@@ -178,13 +183,14 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   // Read and unmarshall the length of the message body of the request
   uint32_t len = 0;
   try {
-    len = SocketHelper::readUint32(socket, NETRW_TIMEOUT);
+    len = SocketHelper::readUint32(socket, NETRWTIMEOUT);
   } catch (castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_READ_MESSAGE_BODY_LENGTH, 2, params);
+      AGGREGATOR_FAILED_TO_READ_MESSAGE_BODY_LENGTH, 3, params);
 
     return;
   }
@@ -196,23 +202,25 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::dispatchRequest(
   // If the message body is larger than the message body buffer
   if(len > sizeof(body)) {
     castor::dlf::Param params[] = {
+      castor::dlf::Param("Function"      , __PRETTY_FUNCTION__),
       castor::dlf::Param("Maximum length", sizeof(body)),
-      castor::dlf::Param("Actual length", len)};
+      castor::dlf::Param("Actual length" , len)};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_MESSAGE_BODY_LENGTH_TOO_LARGE, 2, params);
+      AGGREGATOR_MESSAGE_BODY_LENGTH_TOO_LARGE, 3, params);
 
     return;
   }
 
   // Read the message body from the socket
   try {
-    SocketHelper::readBytes(socket, NETRW_TIMEOUT, len, body);
+    SocketHelper::readBytes(socket, NETRWTIMEOUT, len, body);
   } catch (castor::exception::Exception &ex) {
     castor::dlf::Param params[] = {
-      castor::dlf::Param("Message", ex.getMessage().str()),
-      castor::dlf::Param("Code"   , ex.code())};
+      castor::dlf::Param("Function", __PRETTY_FUNCTION__),
+      castor::dlf::Param("Message" , ex.getMessage().str()),
+      castor::dlf::Param("Code"    , ex.code())};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_FAILED_TO_READ_MESSAGE_BODY, 2, params);
+      AGGREGATOR_FAILED_TO_READ_MESSAGE_BODY, 3, params);
 
     return;
   }
@@ -237,28 +245,31 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::handleJobSubmission(
     const int rc = isadminhost(socket.socket(), peerHost);
     if(rc == -1 && serrno != SENOTADMIN) {
       castor::dlf::Param params[] = {
-        castor::dlf::Param("Reason", "Failed to lookup connection"),
+        castor::dlf::Param("Function" , __PRETTY_FUNCTION__),
+        castor::dlf::Param("Reason"   , "Failed to lookup connection"),
         castor::dlf::Param("Peer Host", peerHost)};
       castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 2, params);
+        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 3, params);
       return;
     }
 
     if(*peerHost == '\0' ) {
       castor::dlf::Param params[] = {
-        castor::dlf::Param("Reason", "Peer host name is empty"),
+        castor::dlf::Param("Function" , __PRETTY_FUNCTION__),
+        castor::dlf::Param("Reason"   , "Peer host name is empty"),
         castor::dlf::Param("Peer Host", peerHost)};
       castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 2, params);
+        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 3, params);
       return;
     }
 
     if(rc != 0) {
       castor::dlf::Param params[] = {
-        castor::dlf::Param("Reason", "Unauthorized host"),
+        castor::dlf::Param("Function" , __PRETTY_FUNCTION__),
+        castor::dlf::Param("Reason"   , "Unauthorized host"),
         castor::dlf::Param("Peer Host", peerHost)};
       castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 2, params);
+        AGGREGATOR_RECEIVED_RCP_JOB_FROM_UNAUTHORIZED_HOST, 3, params);
       return;
     }
   }
@@ -266,10 +277,11 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::handleJobSubmission(
   // If the message body is too small
   if(len < 4 * sizeof(uint32_t) + 4) {
     castor::dlf::Param params[] = {
+      castor::dlf::Param("Function"      , __PRETTY_FUNCTION__),
       castor::dlf::Param("Minimum Length", 4 * sizeof(uint32_t) + 4),
-      castor::dlf::Param("Actual Length", len)};
+      castor::dlf::Param("Actual Length" , len)};
     castor::dlf::dlf_writep(cuuid, DLF_LVL_ERROR,
-      AGGREGATOR_JOB_MESSAGE_BODY_LENGTH_TOO_SHORT, 2, params);
+      AGGREGATOR_JOB_MESSAGE_BODY_LENGTH_TOO_SHORT, 3, params);
   }
 
   uint32_t   ui32          = 0; // Used to for casts
@@ -296,17 +308,19 @@ void castor::tape::aggregator::VdqmRequestHandlerThread::handleJobSubmission(
   Marshaller::unmarshallString(p, bodyLen, clientUsername,
     sizeof(clientUsername));
 
-  castor::dlf::Param param[] = {
-    castor::dlf::Param("tapeRequestID" , tapeRequestID ),
-    castor::dlf::Param("clientPort"    , clientPort    ),