Loading...
Searching...
No Matches
XrdEcUtilities.hh
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
// Author: Michal Simon <michal.simon@cern.ch>
//------------------------------------------------------------------------------
// This file is part of the XRootD software suite.
//
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//
// In applying this licence, 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.
//------------------------------------------------------------------------------
#ifndef SRC_XRDEC_XRDECUTILITIES_HH_
#define SRC_XRDEC_XRDECUTILITIES_HH_
#include "XrdEc/XrdEcObjCfg.hh"
#include "XrdCl/XrdClXRootDResponses.hh"
#include "XrdCl/XrdClFileSystem.hh"
#include "XrdCl/XrdClUtils.hh"
#include <exception>
#include <memory>
#include <random>
#include <queue>
#include <mutex>
#include <condition_variable>
namespace XrdEc
{
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
struct stripe_t
{
//-------------------------------------------------------------------------
//-------------------------------------------------------------------------
stripe_t( char *buffer, bool valid ) : buffer( buffer ), valid( valid )
{
}
char *buffer; //< buffer with stripe data
bool valid; //< true if data are valid, otherwise false
};
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
typedef std::vector<stripe_t> stripes_t;
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
typedef std::vector<char> buffer_t;
//----------------------------------------------------------------------------
//----------------------------------------------------------------------------
class IOError : public std::exception
{
public:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
IOError( const XrdCl::XRootDStatus &st ) noexcept : st( st ), msg( st.ToString() )
{
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
IOError( const IOError &err ) noexcept : st( err.st ), msg( err.st.ToString() )
{
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
IOError& operator=( const IOError &err ) noexcept
{
st = err.st;
msg = err.st.ToString();
return *this;
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
virtual ~IOError()
{
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
virtual const char* what() const noexcept
{
return msg.c_str();
}
//------------------------------------------------------------------------
//------------------------------------------------------------------------
const XrdCl::XRootDStatus& Status() const
{
return st;
}
enum
{
ioTooManyErrors
};
private:
//------------------------------------------------------------------------
//------------------------------------------------------------------------
XrdCl::XRootDStatus st;
//------------------------------------------------------------------------
//------------------------------------------------------------------------
std::string msg;
};
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
void ScheduleHandler( uint64_t offset, uint32_t size, void *buffer, XrdCl::ResponseHandler *handler );
//---------------------------------------------------------------------------
//---------------------------------------------------------------------------
void ScheduleHandler( XrdCl::ResponseHandler *handler, const XrdCl::XRootDStatus &st = XrdCl::XRootDStatus() );
//---------------------------------------------------------------------------
// A class implementing synchronous queue
//---------------------------------------------------------------------------
template<typename Element>
struct sync_queue
{
//-------------------------------------------------------------------------
// An internal exception used for interrupting the `dequeue` method
//-------------------------------------------------------------------------
struct wait_interrupted{ };
//-------------------------------------------------------------------------
// Default constructor
//-------------------------------------------------------------------------
sync_queue() : interrupted( false )
{
}
//-------------------------------------------------------------------------
// Enqueue new element into the queue
//-------------------------------------------------------------------------
inline void enqueue( Element && element )
{
std::unique_lock<std::mutex> lck( mtx );
elements.push( std::move( element ) );
cv.notify_all();
}
//-------------------------------------------------------------------------
// Dequeue an element from the front of the queue
// Note: if the queue is empty blocks until a new element is enqueued
//-------------------------------------------------------------------------
inline Element dequeue()
{
std::unique_lock<std::mutex> lck( mtx );
while( elements.empty() )
{
cv.wait( lck );
if( interrupted ) throw wait_interrupted();
}
Element element = std::move( elements.front() );
elements.pop();
return element;
}
//-------------------------------------------------------------------------
// Dequeue an element from the front of the queue
// Note: if the queue is empty returns false, true otherwise
//-------------------------------------------------------------------------
inline bool dequeue( Element &e )
{
std::unique_lock<std::mutex> lck( mtx );
if( elements.empty() ) return false;
e = std::move( elements.front() );
elements.pop();
return true;
}
//-------------------------------------------------------------------------
// Checks if the queue is empty
//-------------------------------------------------------------------------
bool empty()
{
std::unique_lock<std::mutex> lck( mtx );
return elements.empty();
}
//-------------------------------------------------------------------------
// Interrupt all waiting `dequeue` routines
//-------------------------------------------------------------------------
inline void interrupt()
{
interrupted = true;
cv.notify_all();
}
private:
std::queue<Element> elements; //< the queue itself
std::mutex mtx; //< mutex guarding the queue
std::condition_variable cv;
std::atomic<bool> interrupted; //< a flag, true if all `dequeue` routines
//< should be interrupted
};
//---------------------------------------------------------------------------
// Extract the block ID from the chunk file name
//---------------------------------------------------------------------------
inline static size_t fntoblk( const std::string &fn )
{
size_t end = fn.rfind( '.' );
size_t begin = fn.rfind( '.', end - 1 ) + 1;
size_t len = end - begin;
return std::stoul( fn.substr( begin, len ) );
}
}
#endif /* SRC_XRDEC_XRDECUTILITIES_HH_ */
XrdClFileSystem.hh
XrdClUtils.hh
XrdClXRootDResponses.hh
XrdEcObjCfg.hh
XrdCl::ResponseHandler
Handle an async response.
Definition XrdClXRootDResponses.hh:1127
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdEc::IOError::IOError
IOError(const IOError &err) noexcept
Copy constructor.
Definition XrdEcUtilities.hh:90
XrdEc::IOError::Status
const XrdCl::XRootDStatus & Status() const
Definition XrdEcUtilities.hh:122
XrdEc::IOError::IOError
IOError(const XrdCl::XRootDStatus &st) noexcept
Definition XrdEcUtilities.hh:83
XrdEc::IOError::~IOError
virtual ~IOError()
Destructor.
Definition XrdEcUtilities.hh:107
XrdEc::IOError::what
virtual const char * what() const noexcept
overloaded
Definition XrdEcUtilities.hh:114
XrdEc::IOError::ioTooManyErrors
@ ioTooManyErrors
Definition XrdEcUtilities.hh:129
XrdEc::IOError::operator=
IOError & operator=(const IOError &err) noexcept
Assigment operator.
Definition XrdEcUtilities.hh:97
XrdEc
Definition XrdClZipArchive.hh:45
XrdEc::stripes_t
std::vector< stripe_t > stripes_t
All stripes in a block.
Definition XrdEcUtilities.hh:64
XrdEc::fntoblk
static size_t fntoblk(const std::string &fn)
Definition XrdEcUtilities.hh:251
XrdEc::ScheduleHandler
void ScheduleHandler(uint64_t offset, uint32_t size, void *buffer, XrdCl::ResponseHandler *handler)
Definition XrdEcUtilities.cc:64
XrdEc::buffer_t
std::vector< char > buffer_t
a buffer type
Definition XrdEcReader.hh:49
XrdEc::stripe_t::buffer
char * buffer
Definition XrdEcUtilities.hh:57
XrdEc::stripe_t::valid
bool valid
Definition XrdEcUtilities.hh:58
XrdEc::stripe_t::stripe_t
stripe_t(char *buffer, bool valid)
Definition XrdEcUtilities.hh:53
XrdEc::sync_queue::dequeue
bool dequeue(Element &e)
Definition XrdEcUtilities.hh:213
XrdEc::sync_queue::empty
bool empty()
Definition XrdEcUtilities.hh:225
XrdEc::sync_queue::enqueue
void enqueue(Element &&element)
Definition XrdEcUtilities.hh:185
XrdEc::sync_queue::sync_queue
sync_queue()
Definition XrdEcUtilities.hh:178
XrdEc::sync_queue::dequeue
Element dequeue()
Definition XrdEcUtilities.hh:196
XrdEc::sync_queue::interrupt
void interrupt()
Definition XrdEcUtilities.hh:234
XrdEc::sync_queue::wait_interrupted
Definition XrdEcUtilities.hh:173