greg/stasis-aries-wal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix bookkeeping, get rid of aborts during recovery of partially flushed log.

Browse source
This commit is contained in:
Sears Russell 2008-02-20 22:51:52 +00:00
parent 0fc93d07d3
commit 17d4f95d03
1 changed files with 16 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/stasis/logger/logWriter.c
View file

@ -244,8 +244,6 @@ int openLogWriter() {
const LogEntry * le; const LogEntry * le;
nextAvailableLSN = sizeof(lsn_t) + global_offset; nextAvailableLSN = sizeof(lsn_t) + global_offset;
flushedLSN_stable = nextAvailableLSN;
flushedLSN_internal = nextAvailableLSN;
unsigned int crc = 0; unsigned int crc = 0;
@ -287,6 +285,8 @@ int openLogWriter() {
// Reset log_crc to zero (nextAvailableLSN immediately follows a crc // Reset log_crc to zero (nextAvailableLSN immediately follows a crc
// entry). // entry).
flushedLSN_stable = nextAvailableLSN;
flushedLSN_internal = nextAvailableLSN;
log_crc = 0; log_crc = 0;
return 0; return 0;
@ -391,11 +391,11 @@ static void syncLogInternal() {
pthread_mutex_lock(&nextAvailableLSN_mutex); pthread_mutex_lock(&nextAvailableLSN_mutex);
newFlushedLSN = nextAvailableLSN; newFlushedLSN = nextAvailableLSN;
pthread_mutex_unlock(&nextAvailableLSN_mutex); if(newFlushedLSN > flushedLSN_internal) {
pthread_mutex_unlock(&nextAvailableLSN_mutex);
fflush(log); fflush(log);
writelock(flushedLSN_lock, 0);
writelock(flushedLSN_lock, 0); }
if(newFlushedLSN > flushedLSN_internal) { if(newFlushedLSN > flushedLSN_internal) {
flushedLSN_internal = newFlushedLSN; flushedLSN_internal = newFlushedLSN;
} }
@ -493,12 +493,11 @@ static LogEntry * readLogEntry() {
return (LogEntry*)LLADD_IO_ERROR; return (LogEntry*)LLADD_IO_ERROR;
} else { } else {
lsn_t newSize = size - bytesRead; lsn_t newSize = size - bytesRead;
lsn_t newBytesRead = read (roLogFD, ((byte*)ret)+bytesRead, newSize); lsn_t newBytesRead = read (roLogFD, ((byte*)&size)+bytesRead, newSize);
fprintf(stdout, "Trying to piece together short read.\n"); fflush(stderr); fprintf(stdout, "Trying to piece together short read.\n"); fflush(stderr);
if(newBytesRead == 0) { if(newBytesRead == 0) {
abort();
return NULL; return NULL;
} }
fprintf(stderr, "short read from log. Expected %lld bytes, got %lld.\nFIXME: This is 'normal', but currently not handled", (long long) sizeof(lsn_t), (long long) bytesRead); fprintf(stderr, "short read from log. Expected %lld bytes, got %lld.\nFIXME: This is 'normal', but currently not handled", (long long) sizeof(lsn_t), (long long) bytesRead);
@ -521,8 +520,8 @@ static LogEntry * readLogEntry() {
if(bytesRead == 0) { if(bytesRead == 0) {
fprintf(stderr, "eof reading entry\n"); fprintf(stderr, "eof reading entry\n");
fflush(stderr); fflush(stderr);
abort(); free(ret);
// return(NULL); return(NULL);
} else if(bytesRead == -1) { } else if(bytesRead == -1) {
perror("error reading log"); perror("error reading log");
abort(); abort();
@ -531,13 +530,13 @@ static LogEntry * readLogEntry() {
lsn_t newSize = size - bytesRead; lsn_t newSize = size - bytesRead;
lsn_t newBytesRead = read (roLogFD, ((byte*)ret)+bytesRead, newSize); lsn_t newBytesRead = read (roLogFD, ((byte*)ret)+bytesRead, newSize);
fprintf(stdout, "Trying to piece together short log entry.\n"); fflush(stderr); if(newBytesRead == 0) {
free(ret);
if(newBytesRead == 0) {
abort();
return NULL; return NULL;
} }
fprintf(stdout, "Trying to piece together short log entry.\n"); fflush(stderr);
fprintf(stderr, "short read from log w/ lsn %lld. Expected %lld bytes, got %lld.\nFIXME: This is 'normal', but currently not handled", debug_lsn, size, bytesRead); fprintf(stderr, "short read from log w/ lsn %lld. Expected %lld bytes, got %lld.\nFIXME: This is 'normal', but currently not handled", debug_lsn, size, bytesRead);
fflush(stderr); fflush(stderr);
fprintf(stderr, "\nattempt to read again produced newBytesRead = %lld, newSize was %lld\n", newBytesRead, newSize); fprintf(stderr, "\nattempt to read again produced newBytesRead = %lld, newSize was %lld\n", newBytesRead, newSize);
@ -548,8 +547,8 @@ static LogEntry * readLogEntry() {
} }
} }
entrySize = sizeofLogEntry(ret); // entrySize = sizeofLogEntry(ret);
assert(size == entrySize); // assert(size == entrySize);
return ret; return ret;
} }