timoninmaxim commented on code in PR #11826: URL: https://github.com/apache/ignite/pull/11826#discussion_r2027185098
########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/AbstractFilePerformanceStatisticsWriter.java: ########## @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FILE_MAX_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; + +/** */ +abstract class AbstractFilePerformanceStatisticsWriter { + /** Directory to store performance statistics files. Placed under Ignite work directory. */ + public static final String PERF_STAT_DIR = "perf_stat"; + + /** Default maximum file size in bytes. Performance statistics will be stopped when the size exceeded. */ + public static final long DFLT_FILE_MAX_SIZE = 32 * U.GB; + + /** Default off heap buffer size in bytes. */ + public static final int DFLT_BUFFER_SIZE = (int)(32 * U.MB); + + /** Default minimal batch size to flush in bytes. */ + public static final int DFLT_FLUSH_SIZE = (int)(8 * U.MB); + + /** Default maximum cached strings threshold. String caching will stop on threshold excess. */ + public static final int DFLT_CACHED_STRINGS_THRESHOLD = 10 * 1024; + + /** + * File format version. This version should be incremented each time when format of existing events are + * changed (fields added/removed) to avoid unexpected non-informative errors on deserialization. + */ + public static final short FILE_FORMAT_VERSION = 1; + + /** Hashcodes of cached strings. */ + protected final Set<Integer> knownStrs = new GridConcurrentHashSet<>(); + + /** Minimal batch size to flush in bytes. */ + protected final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** Maximum cached strings threshold. String caching will stop on threshold excess. */ + protected final int cachedStrsThreshold = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD, + DFLT_CACHED_STRINGS_THRESHOLD); + + /** Performance statistics file. */ + protected final File file; + + /** Factory to provide I/O interface. */ + protected final FileIOFactory fileIoFactory = new RandomAccessFileIOFactory(); + + /** Performance statistics file I/O. */ + protected final FileIO fileIo; + + /** Node id. */ + protected final UUID nodeId; + + /** Count of cached strings. */ + protected volatile int knownStrsSz; + + /** */ + protected long fileMaxSize = IgniteSystemProperties.getLong(IGNITE_PERF_STAT_FILE_MAX_SIZE, DFLT_FILE_MAX_SIZE); Review Comment: private ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/AbstractFilePerformanceStatisticsWriter.java: ########## @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FILE_MAX_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; + +/** */ +abstract class AbstractFilePerformanceStatisticsWriter { + /** Directory to store performance statistics files. Placed under Ignite work directory. */ + public static final String PERF_STAT_DIR = "perf_stat"; + + /** Default maximum file size in bytes. Performance statistics will be stopped when the size exceeded. */ + public static final long DFLT_FILE_MAX_SIZE = 32 * U.GB; + + /** Default off heap buffer size in bytes. */ + public static final int DFLT_BUFFER_SIZE = (int)(32 * U.MB); + + /** Default minimal batch size to flush in bytes. */ + public static final int DFLT_FLUSH_SIZE = (int)(8 * U.MB); + + /** Default maximum cached strings threshold. String caching will stop on threshold excess. */ + public static final int DFLT_CACHED_STRINGS_THRESHOLD = 10 * 1024; + + /** + * File format version. This version should be incremented each time when format of existing events are + * changed (fields added/removed) to avoid unexpected non-informative errors on deserialization. + */ + public static final short FILE_FORMAT_VERSION = 1; + + /** Hashcodes of cached strings. */ + protected final Set<Integer> knownStrs = new GridConcurrentHashSet<>(); + + /** Minimal batch size to flush in bytes. */ + protected final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** Maximum cached strings threshold. String caching will stop on threshold excess. */ + protected final int cachedStrsThreshold = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD, + DFLT_CACHED_STRINGS_THRESHOLD); + + /** Performance statistics file. */ + protected final File file; + + /** Factory to provide I/O interface. */ + protected final FileIOFactory fileIoFactory = new RandomAccessFileIOFactory(); + + /** Performance statistics file I/O. */ + protected final FileIO fileIo; + + /** Node id. */ + protected final UUID nodeId; + + /** Count of cached strings. */ + protected volatile int knownStrsSz; + + /** */ + protected long fileMaxSize = IgniteSystemProperties.getLong(IGNITE_PERF_STAT_FILE_MAX_SIZE, DFLT_FILE_MAX_SIZE); + + /** */ + protected int bufSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_BUFFER_SIZE, DFLT_BUFFER_SIZE); + + /** + * @param ctx Context. + * @param fileName File name to write. + */ + protected AbstractFilePerformanceStatisticsWriter(GridKernalContext ctx, String fileName) throws IgniteCheckedException, IOException { + nodeId = ctx.localNodeId(); + file = resolveStatisticsFile(ctx, fileName); + fileIo = fileIoFactory.create(file); + } + + /** Writes {@link UUID} to buffer. */ + static void writeUuid(ByteBuffer buf, UUID uuid) { + buf.putLong(uuid.getMostSignificantBits()); + buf.putLong(uuid.getLeastSignificantBits()); + } + + /** Writes {@link IgniteUuid} to buffer. */ + static void writeIgniteUuid(ByteBuffer buf, IgniteUuid uuid) { + buf.putLong(uuid.globalId().getMostSignificantBits()); + buf.putLong(uuid.globalId().getLeastSignificantBits()); + buf.putLong(uuid.localId()); + } + + /** + * @param buf Buffer to write to. + * @param str String to write. + * @param cached {@code True} if string cached. + */ + static void writeString(ByteBuffer buf, String str, boolean cached) { + buf.put(cached ? (byte)1 : 0); + + if (cached) + buf.putInt(str.hashCode()); + else { + byte[] bytes = str.getBytes(); + + buf.putInt(bytes.length); + buf.put(bytes); + } + } + + /** @return Performance statistics file. */ + private static File resolveStatisticsFile(GridKernalContext ctx, String fileName) throws IgniteCheckedException { + String igniteWorkDir = U.workDirectory(ctx.config().getWorkDirectory(), ctx.config().getIgniteHome()); + + File fileDir = U.resolveWorkDirectory(igniteWorkDir, PERF_STAT_DIR, false); + + File file = new File(fileDir, fileName + ".prf"); + + int idx = 0; + + while (file.exists()) { + idx++; + + file = new File(fileDir, fileName + '-' + idx + ".prf"); + } + + return file; + } + + /** @return {@code True} if string was cached and can be written as hashcode. */ + protected boolean cacheIfPossible(String str) { + if (knownStrsSz >= cachedStrsThreshold) + return false; + + int hash = str.hashCode(); + + // We can cache slightly more strings then threshold value. + // Don't implement solution with synchronization here, because our primary goal is avoid any contention. + if (knownStrs.contains(hash) || !knownStrs.add(hash)) + return true; + + knownStrsSz = knownStrs.size(); + + return false; + } + + protected void cleanup() { + U.closeQuiet(fileIo); + } + + /** */ + String fileAbsolutePath() { Review Comment: lets inline ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java: ########## @@ -287,6 +294,36 @@ else if (opType == QUERY) { return true; } + else if (opType == SYSTEM_VIEW_SCHEMA) { + ForwardableString viewName = readString(buf); + if (viewName == null) + return false; + + ForwardableString walkerName = readString(buf); + if (walkerName == null) + return false; + + assert viewName.str != null : "Views are written by single thread, no string cache misses are possible"; + assert walkerName.str != null : "Views are written by single thread, no string cache misses are possible"; + + try { + sysViewEntry = new SystemViewEntry(viewName.str, walkerName.str); + } + catch (ReflectiveOperationException e) { + throw new IOException("Could not find walker: " + walkerName); + } + return true; + } + else if (opType == SYSTEM_VIEW_ROW) { + List<Object> row = sysViewEntry.nextRow(); + + if (row == null) + return false; + + for (PerformanceStatisticsHandler hnd : curHnd) + hnd.systemView(nodeId, sysViewEntry.viewName, sysViewEntry.schema, row); + return true; Review Comment: Add NL before this line ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/SystemViewFileWriter.java: ########## @@ -0,0 +1,260 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.BufferOverflowException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Predicate; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.systemview.GridSystemViewManager; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.util.worker.GridWorker; +import org.apache.ignite.spi.systemview.view.SystemView; +import org.apache.ignite.spi.systemview.view.SystemViewRowAttributeWalker; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.DFLT_BUFFER_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.DFLT_FLUSH_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.FILE_FORMAT_VERSION; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.resolveStatisticsFile; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.writeString; +import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW_ROW; +import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW_SCHEMA; + +/** Worker to write system views to performance statistics file. */ +class SystemViewFileWriter extends GridWorker { + /** File writer thread name. */ + private static final String SYSTEM_VIEW_WRITER_THREAD_NAME = "performance-statistics-system-view-writer"; + + /** Performance statistics system view file. */ + private final String filePath; + + /** Performance statistics file I/O. */ + private final FileIO fileIo; + + /** */ + private final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** */ + private final GridSystemViewManager sysViewMgr; + + /** System view predicate to filter recorded views. */ + private final Predicate<SystemView<?>> sysViewPredicate; + + /** Writes system view attributes to {@link SystemViewFileWriter#buf}. */ + private final SystemViewRowAttributeWalker.AttributeWithValueVisitor valWriterVisitor = new AttributeWithValueWriterVisitor(); + + /** */ + private StringCache strCache = new StringCache(); + + /** Buffer. */ + private ByteBuffer buf; + + /** + * @param ctx Kernal context. + */ + public SystemViewFileWriter(GridKernalContext ctx) throws IgniteCheckedException, IOException { + super(ctx.igniteInstanceName(), SYSTEM_VIEW_WRITER_THREAD_NAME, ctx.log(SystemViewFileWriter.class)); + + sysViewMgr = ctx.systemView(); + + File file = resolveStatisticsFile(ctx, "node-" + ctx.localNodeId() + "-system-views"); + + fileIo = new RandomAccessFileIOFactory().create(file); + + filePath = file.getPath(); + + int bufSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_BUFFER_SIZE, DFLT_BUFFER_SIZE); + buf = ByteBuffer.allocateDirect(bufSize); + buf.order(ByteOrder.LITTLE_ENDIAN); + + // System views that won't be recorded. They may be large or copy another PerfStat values. + Set<String> ignoredViews = Set.of("baseline.node.attributes", + "metrics", + "caches", + "sql.queries", + "nodes"); + sysViewPredicate = view -> !ignoredViews.contains(view.name()); + + doWrite(buf -> { + buf.put(OperationType.VERSION.id()); + buf.putShort(FILE_FORMAT_VERSION); + }); + } + + /** {@inheritDoc} */ + @Override protected void body() { + if (log.isInfoEnabled()) + log.info("Started writing system views to " + filePath + "."); + + try { + for (SystemView<?> view : sysViewMgr) { + try { + if (sysViewPredicate.test(view)) + systemView(view); + } + catch (RuntimeException | Error e) { Review Comment: BufferOverflowException is RuntimeException. How do you check it? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java: ########## @@ -287,6 +294,36 @@ else if (opType == QUERY) { return true; } + else if (opType == SYSTEM_VIEW_SCHEMA) { + ForwardableString viewName = readString(buf); + if (viewName == null) + return false; + + ForwardableString walkerName = readString(buf); + if (walkerName == null) + return false; + + assert viewName.str != null : "Views are written by single thread, no string cache misses are possible"; + assert walkerName.str != null : "Views are written by single thread, no string cache misses are possible"; + + try { + sysViewEntry = new SystemViewEntry(viewName.str, walkerName.str); + } + catch (ReflectiveOperationException e) { + throw new IOException("Could not find walker: " + walkerName); Review Comment: IgniteException ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java: ########## @@ -112,6 +116,9 @@ public class FilePerformanceStatisticsReader { /** Forward read mode. */ private ForwardRead forwardRead; + /** Walkers for reading system view recors. */ Review Comment: recors -> records ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java: ########## @@ -287,6 +294,36 @@ else if (opType == QUERY) { return true; } + else if (opType == SYSTEM_VIEW_SCHEMA) { + ForwardableString viewName = readString(buf); + if (viewName == null) + return false; + + ForwardableString walkerName = readString(buf); + if (walkerName == null) + return false; + + assert viewName.str != null : "Views are written by single thread, no string cache misses are possible"; + assert walkerName.str != null : "Views are written by single thread, no string cache misses are possible"; + + try { + sysViewEntry = new SystemViewEntry(viewName.str, walkerName.str); + } + catch (ReflectiveOperationException e) { + throw new IOException("Could not find walker: " + walkerName); + } + return true; Review Comment: add NL before ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsWriter.java: ########## @@ -202,8 +192,6 @@ public synchronized void stop() { U.closeQuiet(fileIo); - knownStrs.clear(); Review Comment: Let's clear cache here ########## modules/core/src/test/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsSystemViewTest.java: ########## @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.managers.systemview.walker.AtomicReferenceViewWalker; +import org.apache.ignite.internal.util.lang.gridfunc.PredicateCollectionView; +import org.apache.ignite.spi.systemview.view.datastructures.AtomicReferenceView; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.junit.Test; + +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** + * Tests performance start with system views. + */ +public class PerformanceStatisticsSystemViewTest extends AbstractPerformanceStatisticsTest { + /** */ + private static final List<String> IGNORED_VIEWS = List.of( + "baseline.node.attributes", + "metrics", + "caches", + "sql.queries", + "nodes"); + + /** */ + private final ListeningTestLogger listeningLog = new ListeningTestLogger(); + + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setCacheConfiguration(defaultCacheConfiguration()); + cfg.setGridLogger(listeningLog); + + return cfg; + } + + /** @throws Exception If failed. */ + @Test + public void testSystemViewCaches() throws Exception { + LogListener lsnr = LogListener.matches("Finished writing system views to performance statistics file:").build(); + listeningLog.registerListener(lsnr); + + try (IgniteEx igniteEx = startGrid(0)) { + Set<String> viewsExpected = new HashSet<>(); + igniteEx.context().systemView().forEach(view -> { + if (view.size() > 0 && !IGNORED_VIEWS.contains(view.name())) + viewsExpected.add(view.name()); + }); + + String invalidViewName = "invalidView"; + igniteEx.context().systemView().registerView( Review Comment: Let's add a separate test for checking invalid views ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/SystemViewFileWriter.java: ########## @@ -0,0 +1,260 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.BufferOverflowException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Set; +import java.util.function.Consumer; +import java.util.function.Predicate; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.managers.systemview.GridSystemViewManager; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.util.worker.GridWorker; +import org.apache.ignite.spi.systemview.view.SystemView; +import org.apache.ignite.spi.systemview.view.SystemViewRowAttributeWalker; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.DFLT_BUFFER_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.DFLT_FLUSH_SIZE; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.FILE_FORMAT_VERSION; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.resolveStatisticsFile; +import static org.apache.ignite.internal.processors.performancestatistics.FilePerformanceStatisticsWriter.writeString; +import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW_ROW; +import static org.apache.ignite.internal.processors.performancestatistics.OperationType.SYSTEM_VIEW_SCHEMA; + +/** Worker to write system views to performance statistics file. */ +class SystemViewFileWriter extends GridWorker { + /** File writer thread name. */ + private static final String SYSTEM_VIEW_WRITER_THREAD_NAME = "performance-statistics-system-view-writer"; + + /** Performance statistics system view file. */ + private final String filePath; + + /** Performance statistics file I/O. */ + private final FileIO fileIo; + + /** */ + private final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** */ + private final GridSystemViewManager sysViewMgr; + + /** System view predicate to filter recorded views. */ + private final Predicate<SystemView<?>> sysViewPredicate; + + /** Writes system view attributes to {@link SystemViewFileWriter#buf}. */ + private final SystemViewRowAttributeWalker.AttributeWithValueVisitor valWriterVisitor = new AttributeWithValueWriterVisitor(); + + /** */ + private StringCache strCache = new StringCache(); + + /** Buffer. */ + private ByteBuffer buf; + + /** + * @param ctx Kernal context. + */ + public SystemViewFileWriter(GridKernalContext ctx) throws IgniteCheckedException, IOException { + super(ctx.igniteInstanceName(), SYSTEM_VIEW_WRITER_THREAD_NAME, ctx.log(SystemViewFileWriter.class)); + + sysViewMgr = ctx.systemView(); + + File file = resolveStatisticsFile(ctx, "node-" + ctx.localNodeId() + "-system-views"); + + fileIo = new RandomAccessFileIOFactory().create(file); + + filePath = file.getPath(); + + int bufSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_BUFFER_SIZE, DFLT_BUFFER_SIZE); + buf = ByteBuffer.allocateDirect(bufSize); + buf.order(ByteOrder.LITTLE_ENDIAN); + + // System views that won't be recorded. They may be large or copy another PerfStat values. + Set<String> ignoredViews = Set.of("baseline.node.attributes", + "metrics", + "caches", + "sql.queries", + "nodes"); + sysViewPredicate = view -> !ignoredViews.contains(view.name()); + + doWrite(buf -> { + buf.put(OperationType.VERSION.id()); + buf.putShort(FILE_FORMAT_VERSION); + }); + } + + /** {@inheritDoc} */ + @Override protected void body() { + if (log.isInfoEnabled()) + log.info("Started writing system views to " + filePath + "."); + + try { + for (SystemView<?> view : sysViewMgr) { + try { + if (sysViewPredicate.test(view)) + systemView(view); + } + catch (RuntimeException | Error e) { Review Comment: Error is fatality :) ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/AbstractFilePerformanceStatisticsWriter.java: ########## @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FILE_MAX_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; + +/** */ +abstract class AbstractFilePerformanceStatisticsWriter { + /** Directory to store performance statistics files. Placed under Ignite work directory. */ + public static final String PERF_STAT_DIR = "perf_stat"; + + /** Default maximum file size in bytes. Performance statistics will be stopped when the size exceeded. */ + public static final long DFLT_FILE_MAX_SIZE = 32 * U.GB; + + /** Default off heap buffer size in bytes. */ + public static final int DFLT_BUFFER_SIZE = (int)(32 * U.MB); + + /** Default minimal batch size to flush in bytes. */ + public static final int DFLT_FLUSH_SIZE = (int)(8 * U.MB); + + /** Default maximum cached strings threshold. String caching will stop on threshold excess. */ + public static final int DFLT_CACHED_STRINGS_THRESHOLD = 10 * 1024; + + /** + * File format version. This version should be incremented each time when format of existing events are + * changed (fields added/removed) to avoid unexpected non-informative errors on deserialization. + */ + public static final short FILE_FORMAT_VERSION = 1; + + /** Hashcodes of cached strings. */ + protected final Set<Integer> knownStrs = new GridConcurrentHashSet<>(); + + /** Minimal batch size to flush in bytes. */ + protected final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** Maximum cached strings threshold. String caching will stop on threshold excess. */ + protected final int cachedStrsThreshold = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD, + DFLT_CACHED_STRINGS_THRESHOLD); + + /** Performance statistics file. */ + protected final File file; + + /** Factory to provide I/O interface. */ + protected final FileIOFactory fileIoFactory = new RandomAccessFileIOFactory(); + + /** Performance statistics file I/O. */ + protected final FileIO fileIo; + + /** Node id. */ + protected final UUID nodeId; + + /** Count of cached strings. */ + protected volatile int knownStrsSz; + + /** */ + protected long fileMaxSize = IgniteSystemProperties.getLong(IGNITE_PERF_STAT_FILE_MAX_SIZE, DFLT_FILE_MAX_SIZE); + + /** */ + protected int bufSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_BUFFER_SIZE, DFLT_BUFFER_SIZE); + + /** + * @param ctx Context. + * @param fileName File name to write. + */ + protected AbstractFilePerformanceStatisticsWriter(GridKernalContext ctx, String fileName) throws IgniteCheckedException, IOException { + nodeId = ctx.localNodeId(); + file = resolveStatisticsFile(ctx, fileName); + fileIo = fileIoFactory.create(file); + } + + /** Writes {@link UUID} to buffer. */ + static void writeUuid(ByteBuffer buf, UUID uuid) { Review Comment: Here and below - should not be static ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/FilePerformanceStatisticsReader.java: ########## @@ -621,4 +658,154 @@ private ForwardRead(int hash, long curRecPos, long nextRecPos, int bufPos) { this.bufPos = bufPos; } } + + /** Reads views from buf. */ + private class SystemViewEntry { + /** */ + private final String viewName; + + /** Attribute names of system view. */ + private final List<String> schema; + + /** */ + private final SystemViewRowAttributeWalker<?> walker; + + /** */ + private final RowReaderVisitor rowVisitor; + + /** + * @param viewName System view name. + * @param walkerName Name of walker to visist system view attributes. + */ + public SystemViewEntry(String viewName, String walkerName) throws ReflectiveOperationException { + walker = (SystemViewRowAttributeWalker<?>)Class.forName(walkerName).getConstructor().newInstance(); + + this.viewName = viewName; + + List<String> schemaList = new ArrayList<>(); + + walker.visitAll(new SystemViewRowAttributeWalker.AttributeVisitor() { + @Override public <T> void accept(int idx, String name, Class<T> clazz) { + schemaList.add(name); + } + }); + + schema = Collections.unmodifiableList(schemaList); + + rowVisitor = new RowReaderVisitor(schema.size()); + } + + /** + * @return System view row. + */ + public List<Object> nextRow() { + rowVisitor.clear(); + walker.visitAll(rowVisitor); + return rowVisitor.row(); + } + } + + /** */ + private class RowReaderVisitor implements SystemViewRowAttributeWalker.AttributeVisitor { + /** Number of system view attributes. */ + private final int size; + + /** Row. */ + private List<Object> row; + + /** Not enough bytes flag. */ + private boolean notEnoughBytes; + + /** + * @param size Size of row. + */ + public RowReaderVisitor(int size) { + this.size = size; + + row = new ArrayList<>(size); + } + + /** */ + public List<Object> row() { + if (notEnoughBytes) + return null; + + return Collections.unmodifiableList(row); + } + + /** */ + public void clear() { + row = new ArrayList<>(size); + + notEnoughBytes = false; + } + + /** {@inheritDoc} */ + @Override public <T> void accept(int idx, String name, Class<T> clazz) { + if (notEnoughBytes) + return; + + if (clazz == int.class) { + if (buf.remaining() < 4) { + notEnoughBytes = true; + return; + } + row.add(buf.getInt()); + } + else if (clazz == byte.class) { + if (buf.remaining() < 1) { + notEnoughBytes = true; + return; + } + row.add(buf.get()); + } + else if (clazz == short.class) { + if (buf.remaining() < 2) { + notEnoughBytes = true; + return; + } + row.add(buf.getShort()); + } + else if (clazz == long.class) { + if (buf.remaining() < 8) { + notEnoughBytes = true; + return; + } + row.add(buf.getLong()); + } + else if (clazz == float.class) { + if (buf.remaining() < 4) { + notEnoughBytes = true; + return; + } + row.add(buf.getFloat()); + } + else if (clazz == double.class) { + if (buf.remaining() < 8) { + notEnoughBytes = true; + return; + } + row.add(buf.getDouble()); + } + else if (clazz == char.class) { + if (buf.remaining() < 2) { + notEnoughBytes = true; + return; + } + row.add(buf.getChar()); + } + else if (clazz == boolean.class) { + if (buf.remaining() < 1) { + notEnoughBytes = true; + return; + } + row.add(buf.get() != 0); + } + else { + ForwardableString str = readString(buf); + if (str != null) Review Comment: Otherwise `notEnoughBytes = true?` ########## modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/AbstractFilePerformanceStatisticsWriter.java: ########## @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.performancestatistics; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Set; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteSystemProperties; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; + +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_BUFFER_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FILE_MAX_SIZE; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_PERF_STAT_FLUSH_SIZE; + +/** */ +abstract class AbstractFilePerformanceStatisticsWriter { + /** Directory to store performance statistics files. Placed under Ignite work directory. */ + public static final String PERF_STAT_DIR = "perf_stat"; + + /** Default maximum file size in bytes. Performance statistics will be stopped when the size exceeded. */ + public static final long DFLT_FILE_MAX_SIZE = 32 * U.GB; + + /** Default off heap buffer size in bytes. */ + public static final int DFLT_BUFFER_SIZE = (int)(32 * U.MB); + + /** Default minimal batch size to flush in bytes. */ + public static final int DFLT_FLUSH_SIZE = (int)(8 * U.MB); + + /** Default maximum cached strings threshold. String caching will stop on threshold excess. */ + public static final int DFLT_CACHED_STRINGS_THRESHOLD = 10 * 1024; + + /** + * File format version. This version should be incremented each time when format of existing events are + * changed (fields added/removed) to avoid unexpected non-informative errors on deserialization. + */ + public static final short FILE_FORMAT_VERSION = 1; + + /** Hashcodes of cached strings. */ + protected final Set<Integer> knownStrs = new GridConcurrentHashSet<>(); + + /** Minimal batch size to flush in bytes. */ + protected final int flushSize = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_FLUSH_SIZE, DFLT_FLUSH_SIZE); + + /** Maximum cached strings threshold. String caching will stop on threshold excess. */ + protected final int cachedStrsThreshold = IgniteSystemProperties.getInteger(IGNITE_PERF_STAT_CACHED_STRINGS_THRESHOLD, + DFLT_CACHED_STRINGS_THRESHOLD); + + /** Performance statistics file. */ + protected final File file; + + /** Factory to provide I/O interface. */ + protected final FileIOFactory fileIoFactory = new RandomAccessFileIOFactory(); + + /** Performance statistics file I/O. */ + protected final FileIO fileIo; + + /** Node id. */ + protected final UUID nodeId; + + /** Count of cached strings. */ + protected volatile int knownStrsSz; Review Comment: private -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
