XrdRequestManager.cc File Reference

#include <assert.h>
#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/MessageLogger/interface/MessageLogger.h"
#include "FWCore/ServiceRegistry/interface/Service.h"
#include "Utilities/StorageFactory/interface/StatisticsSenderService.h"
#include "XrdStatistics.h"
#include "Utilities/XrdAdaptor/src/XrdRequestManager.h"
#include "Utilities/XrdAdaptor/src/XrdHostHandler.hh"

Go to the source code of this file.

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

#define GET_CLOCK_MONOTONIC

(

ts

)

clock_gettime(CLOCK_MONOTONIC, &ts);

Definition at line 53 of file XrdRequestManager.cc.

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

#define XRD_ADAPTOR_CHUNK_THRESHOLD   1000

Definition at line 36 of file XrdRequestManager.cc.

Referenced by consumeChunkBack(), consumeChunkFront(), and XrdAdaptor::RequestManager::splitClientRequest().

#define XRD_ADAPTOR_LONG_OPEN_DELAY   2*60

Definition at line 32 of file XrdRequestManager.cc.

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

#define XRD_ADAPTOR_OPEN_PROBE_PERCENT   10

Definition at line 31 of file XrdRequestManager.cc.

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

#define XRD_ADAPTOR_SHORT_OPEN_DELAY   5

Definition at line 23 of file XrdRequestManager.cc.

Referenced by XrdAdaptor::RequestManager::checkSourcesImpl(), XrdAdaptor::RequestManager::handleOpen(), and XrdAdaptor::RequestManager::initialize().

#define XRD_ADAPTOR_SOURCE_QUALITY_FUDGE   100

Definition at line 33 of file XrdRequestManager.cc.

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

#define XRD_CL_MAX_CHUNK   512*1024

Definition at line 21 of file XrdRequestManager.cc.

Referenced by consumeChunkBack(), consumeChunkFront(), XrdAdaptor::RequestManager::splitClientRequest(), and validateList().

Function Documentation

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

static

Definition at line 937 of file XrdRequestManager.cc.

References IOPosBuffer::data(), IOPosBuffer::offset(), 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().

{
    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.size() && (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();
        }
    }
}

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

static

Definition at line 888 of file XrdRequestManager.cc.

References IOPosBuffer::data(), IOPosBuffer::offset(), 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().

{
    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.size() && (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++;
        }
    }
}

static void SendMonitoringInfo ( XrdCl::File &  file )

static

Definition at line 91 of file XrdRequestManager.cc.

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

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

{
        // 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.size())
    {
        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 << ".";
    }
}

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

Definition at line 59 of file XrdRequestManager.cc.

References mps_update::diff.

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

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

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

static

Definition at line 985 of file XrdRequestManager.cc.

References assert(), pileupDistInMC::total, and XRD_CL_MAX_CHUNK.

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

{
    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;
}

Variable Documentation

SendMonitoringInfoHandler nullHandler

Definition at line 87 of file XrdRequestManager.cc.

Referenced by SendMonitoringInfo().