XrdEc::StrmWriter Class Reference

#include <XrdEcStrmWriter.hh>

Collaboration diagram for XrdEc::StrmWriter:

Public Member Functions
	StrmWriter (const ObjCfg &objcfg)
	Constructor.
virtual	~StrmWriter ()
	Destructor.
void	Close (XrdCl::ResponseHandler *handler, time_t timeout=0)
uint64_t	GetSize ()
void	Open (XrdCl::ResponseHandler *handler, time_t timeout=0)
void	Write (uint32_t size, const void *buff, XrdCl::ResponseHandler *handler)

Detailed Description

The Stream Writer objects, responsible for writing erasure coded data into selected placement group.

Definition at line 52 of file XrdEcStrmWriter.hh.

Constructor & Destructor Documentation

StrmWriter()

XrdEc::StrmWriter::StrmWriter ( const ObjCfg & objcfg )

inline

Constructor.

Definition at line 64 of file XrdEcStrmWriter.hh.

                                         : objcfg( objcfg ),
                                           writer_thread_stop( false ),
                                           writer_thread( writer_routine, this ),
                                           next_blknb( 0 ),
                                           global_status( this )
      {
      }

~StrmWriter()

virtual XrdEc::StrmWriter::~StrmWriter ( )

inlinevirtual

Destructor.

Definition at line 75 of file XrdEcStrmWriter.hh.

      {
        writer_thread_stop = true;
        buffers.interrupt();
        writer_thread.join();
      }

Member Function Documentation

Close()

void XrdEc::StrmWriter::Close	(	XrdCl::ResponseHandler *	handler,
		time_t	timeout = 0 )

Close the data object

Parameters

handler

: user callback

Definition at line 108 of file XrdEcStrmWriter.cc.

  {
    //-------------------------------------------------------------------------
    // First, check the global status, if we are in an error state just
    // fail the request.
    //-------------------------------------------------------------------------
    XrdCl::XRootDStatus gst = global_status.get();
    if( !gst.IsOK() ) return ScheduleHandler( handler, gst );
    //-------------------------------------------------------------------------
    // Take care of the left-over data ...
    //-------------------------------------------------------------------------
    if( wrtbuff && !wrtbuff->Empty() ) EnqueueBuff( std::move( wrtbuff ) );
    //-------------------------------------------------------------------------
    // Let the global status handle the close
    //-------------------------------------------------------------------------
    global_status.issue_close( handler, timeout );
  }

References XrdCl::Status::IsOK(), and XrdEc::ScheduleHandler().

Here is the call graph for this function:

GetSize()

uint64_t XrdEc::StrmWriter::GetSize ( )

inline

Returns

: get file size

Definition at line 108 of file XrdEcStrmWriter.hh.

      {
        return global_status.get_btswritten();
      }

Open()

void XrdEc::StrmWriter::Open	(	XrdCl::ResponseHandler *	handler,
		time_t	timeout = 0 )

Open the data object for writting

Parameters

handler

: user callback

Definition at line 44 of file XrdEcStrmWriter.cc.

  {
    const size_t size = objcfg.plgr.size();
 
    std::vector<XrdCl::Pipeline> opens;
    opens.reserve( size );
    // initialize all zip archive objects
    for( size_t i = 0; i < size; ++i )
      dataarchs.emplace_back( std::make_shared<XrdCl::ZipArchive>(
          Config::Instance().enable_plugins ) );
 
    for( size_t i = 0; i < size; ++i )
    {
      std::string url = objcfg.GetDataUrl( i );
      XrdCl::Ctx<XrdCl::ZipArchive> zip( *dataarchs[i] );
      opens.emplace_back( XrdCl::OpenArchive( zip, url, XrdCl::OpenFlags::New | XrdCl::OpenFlags::Write ) );
    }
 
    XrdCl::Async( XrdCl::Parallel( opens ).AtLeast( objcfg.nbchunks ) >>
                  [handler,this]( XrdCl::XRootDStatus &st )
                  {
                    if( !st.IsOK() ) this->global_status.report_open( st );
                    handler->HandleResponse( new XrdCl::XRootDStatus( st ), nullptr );
                  }, timeout );
  }

References XrdCl::Async(), XrdEc::Config::Instance(), XrdCl::OpenFlags::New, XrdCl::OpenArchive(), XrdCl::Parallel(), and XrdCl::OpenFlags::Write.

Here is the call graph for this function:

Write()

void XrdEc::StrmWriter::Write	(	uint32_t	size,
		const void *	buff,
		XrdCl::ResponseHandler *	handler )

Write data to the data object

Parameters

size	: number of bytes to be written
buff	: buffer with data to be written
handler	: user callback

Definition at line 73 of file XrdEcStrmWriter.cc.

  {
    //-------------------------------------------------------------------------
    // First, check the global status, if we are in an error state just
    // fail the request.
    //-------------------------------------------------------------------------
    XrdCl::XRootDStatus gst = global_status.get();
    if( !gst.IsOK() ) return ScheduleHandler( handler, gst );
 
    //-------------------------------------------------------------------------
    // Update the number of bytes left to be written
    //-------------------------------------------------------------------------
    global_status.issue_write( size );
 
    const char* buffer = reinterpret_cast<const char*>( buff );
    uint32_t wrtsize = size;
    while( wrtsize > 0 )
    {
      if( !wrtbuff ) wrtbuff.reset( new WrtBuff( objcfg ) );
      uint64_t written = wrtbuff->Write( wrtsize, buffer );
      buffer  += written;
      wrtsize -= written;
      if( wrtbuff->Complete() ) EnqueueBuff( std::move( wrtbuff ) );
    }
 
    //-------------------------------------------------------------------------
    // We can tell the user it's done as we have the date cached in the
    // buffer
    //-------------------------------------------------------------------------
    ScheduleHandler( handler );
  }

References XrdCl::Status::IsOK(), and XrdEc::ScheduleHandler().

Here is the call graph for this function:

---

The documentation for this class was generated from the following files:

• XrdEcStrmWriter.hh
• XrdEcStrmWriter.cc