journal.cc #4

#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 8191
	char buf[ BUFSIZE + 1 ];
	char *bufptr;
	ssize_t n;

	bool seenvv;

	int version;

	buf[ BUFSIZE ] = '\0';
	seenvv = false;

	/*
	 * Read the @vv@ record near 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
	 * we might try to read before the @vv@ record is written after the
	 * journal has been truncated or after the @nx@ record has been
	 * written.
	 */
	for( attempts = 0; attempts < 10; attempts++ )
	{
	    if( !Seek( 0, e ) )
		return false;

	    if( !Read( buf, BUFSIZE, &n, e ) )
		return false;

	    if( n > 31 )
	    {
		if( bufptr = strstr( buf, "@vv@ " ) )
		{
		    if( n > bufptr - &buf[ 0 ] + 31 )
		    {
			/*
			 * Found an @vv@ record with enough characters
			 * that it should be be complete.
			 */
			seenvv = true;
			break;
		    }
		}
	    }

	    sleep( 1 );		// wait for @vv@ record to be written
	}

	if( !seenvv )
	{
	    e->Set( MsgJrep::VVNotSeen );
	    return false;
	}

	version = strtol( bufptr + 5, &bufptr, 10 );
	if( version > 1 )
	{
	    e->Set( MsgJrep::VVUnknownVersion ) << version;
	    return false;
	}

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

	bufptr += 25;

	if( version == 1 )
	{
	    if( strncmp( bufptr, "@", 1 ) )
	    {
		e->Set( MsgJrep::VVInvalidFormat );
		return false;
	    }

	    bufptr += 1;
	}

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