journal.cc #2

#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>

#define NEED_ERRNO

#include <stdhdrs.h>
#include <strbuf.h>
#include <error.h>

#include "fileio.h"
#include "journal.h"
#include "msgjrep.h"
#include "err.h"

void Journal::SetName( StrPtr *Jval )
{
	if( Jval )
	    name.Set( Jval );

	else
	    name.Set( "journal" );
}

char *Journal::GetName()
{
	return name.Text();
}

dev_t Journal::GetDev()
{
	return statbuf.st_dev;
}

ino_t Journal::GetIno()
{
	return statbuf.st_ino;
}

off_tL Journal::GetSize()
{
	return statbuf.st_size;
}

int Journal::GetSequence()
{
	return sequence;
}

bool Journal::Init( StrPtr *Jval, Error *e )
{
	SetName( Jval );

	return true;
}

bool Journal::Open( Error *e )
{
	if( ( journalfd = openL( GetName(), O_RDONLY ) ) == -1 )
	    return syserr( "open", GetName(), e );

	return true;
}

bool Journal::Seek( off_tL offset, Error *e )
{
	if( lseekL( journalfd, offset, SEEK_SET ) == -1 )
	    return syserr( "seek", GetName(), e );

	return true;
}

bool Journal::Read( char *buf, size_t count, ssize_t *n, Error *e )
{
	if( ( *n = read( journalfd, buf, count ) ) == -1 )
	    return syserr( "read", GetName(), e );

	return true;
}

bool Journal::ReadSequence( Error *e )
{
	int attempts;

#define BUFSIZE 63
	char buf[ BUFSIZE + 1 ];
	char *bufptr;
	ssize_t n;

	int version;

	if( !Seek( 0, e ) )
	    return false;

	/*
	 * Read the @vv@ record at the beginning of the journal. Since the
	 * server truncates the journal without locking it first during
	 * journal rotation, several attempts at reading the @vv@ record
	 * are needed. The retry logic is needed since it is possible that
	 * a single read here could occur between the inode updated as a
	 * result of truncating the journal and the inode updated as a
	 * result of writing of the @vv@ record.
	 */
	for( attempts = 0; attempts < 10; attempts++ )
	{
	    if( !Read( buf, BUFSIZE, &n, e ) )
		return false;

	    if( n )
		break;

	    sleep( 1 );
	}

	if( n < 32 )
	{
	    e->Set( MsgJrep::VVTooShort );
	    return false;
	}

	if( strncmp( buf, "@vv@ ", 5 ) )
	{
	    e->Set( MsgJrep::VVNotFirst );
	    return false;
	}

	buf[ BUFSIZE ] = '\0';

	version = strtol( &buf[ 5 ], &bufptr, 10 );
	if( version != 0 )
	{
	    e->Set( MsgJrep::VVUnknownVersion ) << version;
	    return false;
	}

	if( strncmp( bufptr, " @db.counters@ @journal@ ", 25 ) )
	{
	    e->Set( MsgJrep::VVFirstNotJournal );
	    return false;
	}

	bufptr += 25;

	if( !sscanf( bufptr, "%d", &sequence ) )
	{
	    e->Set( MsgJrep::VVSscanfFail );
	    return false;
	}

	return true;
}

bool Journal::Stat( Error *e )
{
	if( fstatL( journalfd, &statbuf ) == -1 )
	    return syserr( "stat", GetName(), e );

	return true;
}

bool Journal::StatWait( Error *e )
{
	/*
	 * After the journal has been renamed as part of journal rotation
	 * in 2007.3 and later servers, it's possible that the new journal
	 * has not yet been created by the server. Give the server a chance
	 * to create the new journal.
	 */
	int attempts;

	for( attempts = 0; attempts < 10; attempts++ )
	{
	    if( !statL( GetName(), &statbuf ) )
		return true;

	    if( errno != ENOENT )
		return syserr( "statwait", GetName(), e );

	    sleep( 1 );
	}

	e->Set( MsgJrep::NotRecreated ) << GetName();
	return false;
}

bool Journal::Close( Error *e )
{
	if( close( journalfd ) == -1 )
	    return syserr( "close", GetName(), e );

	return true;
}