#include <cassert>
#include <iostream>
#include <algorithm>
#include <netdb.h>
#include "XrdCl/XrdClFile.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClFileSystem.hh"
#include "FWCore/Utilities/interface/CPUTimer.h"
#include "FWCore/Utilities/interface/EDMException.h"
#include "FWCore/Utilities/interface/Likely.h"
#include "FWCore/MessageLogger/interface/MessageLogger.h"
#include "FWCore/ServiceRegistry/interface/Service.h"
#include "FWCore/Utilities/interface/thread_safety_macros.h"
#include "Utilities/StorageFactory/interface/StatisticsSenderService.h"
#include "XrdStatistics.h"
#include "Utilities/XrdAdaptor/src/XrdRequestManager.h"
#include "Utilities/XrdAdaptor/src/XrdHostHandler.hh"

Classes
class	SendMonitoringInfoHandler

Macros
#define	GET_CLOCK_MONOTONIC(ts) clock_gettime(CLOCK_MONOTONIC, &ts);

#define	XRD_ADAPTOR_CHUNK_THRESHOLD 1000

#define	XRD_ADAPTOR_LONG_OPEN_DELAY 2 * 60

#define	XRD_ADAPTOR_OPEN_PROBE_PERCENT 10

#define	XRD_ADAPTOR_SHORT_OPEN_DELAY 5

#define	XRD_ADAPTOR_SOURCE_QUALITY_FUDGE 100

#define	XRD_CL_MAX_CHUNK 512 * 1024

Functions
static void	consumeChunkBack (size_t front, std::vector< IOPosBuffer > &input, std::vector< IOPosBuffer > &output, IOSize chunksize)

static void	consumeChunkFront (size_t &front, std::vector< IOPosBuffer > &input, std::vector< IOPosBuffer > &output, IOSize chunksize)

static void	SendMonitoringInfo (XrdCl::File &file)

long long	timeDiffMS (const timespec &a, const timespec &b)

static IOSize	validateList (const std::vector< IOPosBuffer > req)

Variables
SendMonitoringInfoHandler	nullHandler

Macro Definition Documentation

◆ GET_CLOCK_MONOTONIC

#define GET_CLOCK_MONOTONIC ( ts ) clock_gettime(CLOCK_MONOTONIC, &ts);

Definition at line 54 of file XrdRequestManager.cc.

◆ XRD_ADAPTOR_CHUNK_THRESHOLD

#define XRD_ADAPTOR_CHUNK_THRESHOLD 1000

Definition at line 38 of file XrdRequestManager.cc.

◆ XRD_ADAPTOR_LONG_OPEN_DELAY

#define XRD_ADAPTOR_LONG_OPEN_DELAY 2 * 60

Definition at line 34 of file XrdRequestManager.cc.

◆ XRD_ADAPTOR_OPEN_PROBE_PERCENT

#define XRD_ADAPTOR_OPEN_PROBE_PERCENT 10

Definition at line 33 of file XrdRequestManager.cc.

◆ XRD_ADAPTOR_SHORT_OPEN_DELAY

#define XRD_ADAPTOR_SHORT_OPEN_DELAY 5

Definition at line 25 of file XrdRequestManager.cc.

◆ XRD_ADAPTOR_SOURCE_QUALITY_FUDGE

#define XRD_ADAPTOR_SOURCE_QUALITY_FUDGE 100

Definition at line 35 of file XrdRequestManager.cc.

◆ XRD_CL_MAX_CHUNK

#define XRD_CL_MAX_CHUNK 512 * 1024

Definition at line 23 of file XrdRequestManager.cc.

Function Documentation

◆ consumeChunkBack()

static void consumeChunkBack	(	size_t	front,
		std::vector< IOPosBuffer > &	input,
		std::vector< IOPosBuffer > &	output,
		IOSize	chunksize
	)

static

Definition at line 836 of file XrdRequestManager.cc.

                                                {
   while ((chunksize > 0) && (front < input.size()) && (output.size() <= XRD_ADAPTOR_CHUNK_THRESHOLD)) {
     IOPosBuffer &io = input.back();
     IOPosBuffer &outio = output.back();
     if (io.size() > chunksize) {
       IOSize consumed;
       if (!output.empty() && (outio.size() < XRD_CL_MAX_CHUNK) &&
           (outio.offset() + static_cast<IOOffset>(outio.size()) == io.offset())) {
         if (outio.size() + chunksize > XRD_CL_MAX_CHUNK) {
           consumed = (XRD_CL_MAX_CHUNK - outio.size());
           outio.set_size(XRD_CL_MAX_CHUNK);
         } else {
           consumed = chunksize;
           outio.set_size(outio.size() + consumed);
         }
       } else {
         consumed = chunksize;
         output.emplace_back(IOPosBuffer(io.offset(), io.data(), chunksize));
       }
       chunksize -= consumed;
       IOSize newsize = io.size() - consumed;
       IOOffset newoffset = io.offset() + consumed;
       void *newdata = static_cast<char *>(io.data()) + consumed;
       io.set_offset(newoffset);
       io.set_data(newdata);
       io.set_size(newsize);
     } else if (io.size() == 0) {
       input.pop_back();
     } else {
       output.push_back(io);
       chunksize -= io.size();
       input.pop_back();
     }
   }
 }

References IOPosBuffer::data(), input, IOPosBuffer::offset(), convertSQLitetoXML_cfg::output, IOPosBuffer::set_data(), IOPosBuffer::set_offset(), IOPosBuffer::set_size(), IOPosBuffer::size(), XRD_ADAPTOR_CHUNK_THRESHOLD, and XRD_CL_MAX_CHUNK.

Referenced by XrdAdaptor::RequestManager::splitClientRequest().

◆ consumeChunkFront()

static void consumeChunkFront	(	size_t &	front,
		std::vector< IOPosBuffer > &	input,
		std::vector< IOPosBuffer > &	output,
		IOSize	chunksize
	)

static

Definition at line 797 of file XrdRequestManager.cc.

                                                 {
   while ((chunksize > 0) && (front < input.size()) && (output.size() <= XRD_ADAPTOR_CHUNK_THRESHOLD)) {
     IOPosBuffer &io = input[front];
     IOPosBuffer &outio = output.back();
     if (io.size() > chunksize) {
       IOSize consumed;
       if (!output.empty() && (outio.size() < XRD_CL_MAX_CHUNK) &&
           (outio.offset() + static_cast<IOOffset>(outio.size()) == io.offset())) {
         if (outio.size() + chunksize > XRD_CL_MAX_CHUNK) {
           consumed = (XRD_CL_MAX_CHUNK - outio.size());
           outio.set_size(XRD_CL_MAX_CHUNK);
         } else {
           consumed = chunksize;
           outio.set_size(outio.size() + consumed);
         }
       } else {
         consumed = chunksize;
         output.emplace_back(IOPosBuffer(io.offset(), io.data(), chunksize));
       }
       chunksize -= consumed;
       IOSize newsize = io.size() - consumed;
       IOOffset newoffset = io.offset() + consumed;
       void *newdata = static_cast<char *>(io.data()) + consumed;
       io.set_offset(newoffset);
       io.set_data(newdata);
       io.set_size(newsize);
     } else if (io.size() == 0) {
       front++;
     } else {
       output.push_back(io);
       chunksize -= io.size();
       front++;
     }
   }
 }

References IOPosBuffer::data(), input, IOPosBuffer::offset(), convertSQLitetoXML_cfg::output, IOPosBuffer::set_data(), IOPosBuffer::set_offset(), IOPosBuffer::set_size(), IOPosBuffer::size(), XRD_ADAPTOR_CHUNK_THRESHOLD, and XRD_CL_MAX_CHUNK.

Referenced by XrdAdaptor::RequestManager::splitClientRequest().

◆ SendMonitoringInfo()

static void SendMonitoringInfo ( XrdCl::File & file )

static

Definition at line 85 of file XrdRequestManager.cc.

                                                 {
   // Do not send this to a dCache data server as they return an error.
   // In some versions of dCache, sending the monitoring information causes
   // the server to close the connection - resulting in failures.
   if (Source::isDCachePool(file)) {
     return;
   }
  
   // Send the monitoring info, if available.
   const char *jobId = edm::storage::StatisticsSenderService::getJobID();
   std::string lastUrl;
   file.GetProperty("LastURL", lastUrl);
   if (jobId && !lastUrl.empty()) {
     XrdCl::URL url(lastUrl);
     XrdCl::FileSystem fs(url);
     if (!(fs.SendInfo(jobId, &nullHandler, 30).IsOK())) {
       edm::LogWarning("XrdAdaptorInternal")
           << "Failed to send the monitoring information, monitoring ID is " << jobId << ".";
     }
     edm::LogInfo("XrdAdaptorInternal") << "Set monitoring ID to " << jobId << ".";
   }
 }

References FrontierConditions_GlobalTag_cff::file, edm::storage::StatisticsSenderService::getJobID(), XrdAdaptor::Source::isDCachePool(), nullHandler, AlCaHLTBitMon_QueryRunRegistry::string, hcal_runs::URL, and relmon_authenticated_wget::url.

Referenced by XrdAdaptor::RequestManager::OpenHandler::HandleResponseWithHosts(), and XrdAdaptor::RequestManager::initialize().

◆ timeDiffMS()

long long timeDiffMS	(	const timespec &	a,
		const timespec &	b
	)

Definition at line 59 of file XrdRequestManager.cc.

                                                            {
   long long diff = (a.tv_sec - b.tv_sec) * 1000;
   diff += (a.tv_nsec - b.tv_nsec) / 1e6;
   return diff;
 }

References a, b, and change_name::diff.

Referenced by XrdAdaptor::RequestManager::checkSources(), and XrdAdaptor::RequestManager::checkSourcesImpl().

◆ validateList()

static IOSize validateList ( const std::vector< IOPosBuffer > req )

static

Definition at line 875 of file XrdRequestManager.cc.

                                                              {
   IOSize total = 0;
   off_t last_offset = -1;
   for (const auto &it : req) {
     total += it.size();
     assert(it.offset() > last_offset);
     last_offset = it.offset();
     assert(it.size() <= XRD_CL_MAX_CHUNK);
     assert(it.offset() < 0x1ffffffffff);
   }
   assert(req.size() <= 1024);
   return total;
 }