Skip to content

Fix schema file I/O failure handling and short writes #19190

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
+565 −54
Open

Fix schema file I/O failure handling and short writes #19190

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,6 +294,11 @@ public final class DataNodeSchemaMessages {
"PBTree File [{}] will be overwritten since already exists.";
public static final String SCHEMA_FILE_WRONG_VERSION =
"SchemaFile with wrong version, please check or upgrade.";
public static final String
EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235 =
"Schema file header write made no progress";
public static final String EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1 =
"Schema page write made no progress";
public static final String NODE_NO_CHILD_IN_PBTREE = "Node [%s] has no child in pbtree file.";
public static final String SCHEMA_FILE_INSPECTED = "SchemaFile[%s] had been inspected.";
public static final String FAILED_TO_CREATE_SCHEMA_FILE_SNAPSHOT =
Expand Down
5 changes: 5 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,11 @@ public final class DataNodeSchemaMessages {
"PBTree 文件 [{}] 已存在，将被覆盖。";
public static final String SCHEMA_FILE_WRONG_VERSION =
"SchemaFile 版本错误，请检查或升级。";
public static final String
EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235 =
"SchemaFile 头部写入没有进展";
public static final String EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1 =
"SchemaPage 写入没有进展";
public static final String NODE_NO_CHILD_IN_PBTREE =
"节点 [%s] 在 pbtree 文件中没有子节点。";
public static final String SCHEMA_FILE_INSPECTED = "SchemaFile[%s] 已被检查。";
Expand Down
123 changes: 99 additions & 24 deletions ...rg/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,10 +273,42 @@ public Iterator<ICachedMNode> getChildren(ICachedMNode parent)

@Override
public void close() throws IOException {
updateHeaderBuffer();
pageManager.close();
forceChannel();
channel.close();
IOException failure = null;
try {
updateHeaderBuffer();
} catch (IOException e) {
failure = e;
}
try {
pageManager.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
try {
forceChannel();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
try {
channel.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
if (failure != null) {
throw failure;
}
}

@Override
Expand Down Expand Up @@ -359,28 +391,34 @@ private void initFileHeader() throws IOException, MetadataException {
if (channel.size() == 0) {
// new pbtree file
lastPageIndex = 0;
ReadWriteIOUtils.write(lastPageIndex, headerContent);
ReadWriteIOUtils.write(dataTTL, headerContent);
ReadWriteIOUtils.write(isEntity, headerContent);
ReadWriteIOUtils.write(sgNodeTemplateIdWithState, headerContent);
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);
lastSGAddr = 0L;
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
} else {
IOUtils.readFully(channel, headerContent);
headerContent.clear();
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
dataTTL = ReadWriteIOUtils.readLong(headerContent);
isEntity = ReadWriteIOUtils.readBool(headerContent);
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);

if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
channel.close();
throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION);
pageManager = null;
try {
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
updateHeaderBuffer();
} catch (IOException | MetadataException e) {
closeAfterInitializationFailure(e);
throw e;
}
} else {
try {
IOUtils.readFully(channel, headerContent);
headerContent.clear();
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
dataTTL = ReadWriteIOUtils.readLong(headerContent);
isEntity = ReadWriteIOUtils.readBool(headerContent);
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);

if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION);
}

pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
} catch (IOException | MetadataException e) {
closeChannelAfterInitializationFailure(e);
throw e;
}
}
}

Expand All @@ -395,7 +433,44 @@ private void updateHeaderBuffer() throws IOException {
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);

headerContent.flip();
channel.write(headerContent, 0);
long position = 0;
while (headerContent.hasRemaining()) {
int written = channel.write(headerContent, position);
if (written <= 0) {
throw new IOException(
DataNodeSchemaMessages.EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235);
}
position += written;
}
}

private void closeAfterInitializationFailure(Exception failure) {
if (pageManager != null) {
try {
pageManager.close();
} catch (IOException closeException) {
failure.addSuppressed(closeException);
}
}
try {
channel.truncate(0);
} catch (IOException cleanupException) {
failure.addSuppressed(cleanupException);
}
try {
channel.force(true);
} catch (IOException cleanupException) {
failure.addSuppressed(cleanupException);
}
closeChannelAfterInitializationFailure(failure);
}

private void closeChannelAfterInitializationFailure(Exception failure) {
try {
channel.close();
} catch (IOException closeException) {
failure.addSuppressed(closeException);
}
}

private void forceChannel() throws IOException {
Expand Down
11 changes: 10 additions & 1 deletion ...rg/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.exception.metadata.schemafile.SegmentNotFoundException;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;

import org.apache.tsfile.utils.ReadWriteIOUtils;

Expand Down Expand Up @@ -106,7 +107,15 @@ public void syncPageBuffer() {
@Override
public void flushPageToChannel(FileChannel channel) throws IOException {
this.pageBuffer.clear();
channel.write(pageBuffer, SchemaFile.getPageAddress(pageIndex));
long position = SchemaFile.getPageAddress(pageIndex);
while (pageBuffer.hasRemaining()) {
int written = channel.write(pageBuffer, position);
if (written <= 0) {
throw new IOException(
DataNodeSchemaMessages.EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1);
}
position += written;
}
dirtyFlag = false;
}

Expand Down
76 changes: 56 additions & 20 deletions ...otdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,19 +52,28 @@ public class PageIOChannel {
this.channel = channel;
this.pmtFile = pmtFile;
this.readChannel = FileChannel.open(pmtFile.toPath(), StandardOpenOption.READ);
if (flushWithLogging) {
// without RATIS, utilize physical logging for integrity
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
this.logWriter = new SchemaFileLogWriter(logPath);
logCounter = new AtomicInteger(pageAcc);
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
singlePageFlushStrategy = this::flushSinglePageWithLogging;
} else {
// with RATIS enabled, integrity is guaranteed by consensus protocol
logCounter = new AtomicInteger();
logWriter = null;
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
try {
if (flushWithLogging) {
// without RATIS, utilize physical logging for integrity
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
this.logWriter = new SchemaFileLogWriter(logPath);
logCounter = new AtomicInteger(pageAcc);
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
singlePageFlushStrategy = this::flushSinglePageWithLogging;
} else {
// with RATIS enabled, integrity is guaranteed by consensus protocol
logCounter = new AtomicInteger();
logWriter = null;
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
}
} catch (IOException | MetadataException e) {
try {
close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
}

Expand All @@ -74,21 +83,48 @@ public void renewLogWriter() throws IOException {
}
}

public void closeLogWriter() throws IOException {
public void close() throws IOException {
IOException failure = null;
if (logWriter != null) {
logWriter.close();
try {
logWriter.close();
} catch (IOException e) {
failure = e;
}
}
try {
readChannel.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
if (failure != null) {
throw failure;
}
}

/** Load bytes from log, deserialize and flush directly into channel, return current length */
private long recoverFromLog(String logPath) throws IOException, MetadataException {
SchemaFileLogReader reader = new SchemaFileLogReader(logPath);
ISchemaPage page;
List<byte[]> res = reader.collectUpdatedEntries();
for (byte[] entry : res) {
// TODO check bytes semantic correctness with CRC32 or other way
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
page.flushPageToChannel(this.channel);
List<byte[]> res;
try {
res = reader.collectUpdatedEntries();
for (byte[] entry : res) {
// TODO check bytes semantic correctness with CRC32 or other way
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
page.flushPageToChannel(this.channel);
}
} catch (IOException | MetadataException e) {
try {
reader.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
reader.close();

Expand Down
27 changes: 18 additions & 9 deletions .../iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,14 +86,23 @@ public abstract class PageManager implements IPageManager {
pageIOChannel = new PageIOChannel(channel, pmtFile, true, logPath);
}

// construct first page if file to init
if (lastPageIndex < 0) {
ISegmentedPage rootPage =
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
pagePool.put(rootPage);
rootPage.syncPageBuffer();
rootPage.flushPageToChannel(channel);
try {
// construct first page if file to init
if (lastPageIndex < 0) {
ISegmentedPage rootPage =
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
pagePool.put(rootPage);
rootPage.syncPageBuffer();
rootPage.flushPageToChannel(channel);
}
} catch (IOException | MetadataException e) {
try {
pageIOChannel.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
}

Expand Down Expand Up @@ -528,7 +537,7 @@ public void inspect(PrintWriter pw) throws IOException, MetadataException {

@Override
public void close() throws IOException {
pageIOChannel.closeLogWriter();
pageIOChannel.close();
}

// endregion
Expand Down
Loading
Loading