[ https://issues.apache.org/jira/browse/HADOOP-17461?focusedWorklogId=790111&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-790111 ]
ASF GitHub Bot logged work on HADOOP-17461: ------------------------------------------- Author: ASF GitHub Bot Created on: 12/Jul/22 18:17 Start Date: 12/Jul/22 18:17 Worklog Time Spent: 10m Work Description: steveloughran commented on code in PR #4352: URL: https://github.com/apache/hadoop/pull/4352#discussion_r919262464 ########## hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/impl/IOStatisticsContext.java: ########## @@ -0,0 +1,238 @@ +/* + * 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.hadoop.fs.statistics.impl; + +import java.lang.ref.WeakReference; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.classification.VisibleForTesting; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.impl.WeakReferenceThreadMap; +import org.apache.hadoop.fs.statistics.IOStatistics; +import org.apache.hadoop.fs.statistics.IOStatisticsAggregator; +import org.apache.hadoop.fs.statistics.IOStatisticsSnapshot; + +import static org.apache.hadoop.fs.CommonConfigurationKeys.THREAD_LEVEL_IOSTATISTICS_ENABLED; +import static org.apache.hadoop.fs.CommonConfigurationKeys.THREAD_LEVEL_IOSTATISTICS_ENABLED_DEFAULT; + +/** + * Implementing the IOStatisticsContext. + * + * A Context defined for IOStatistics collection per thread which captures + * each worker thread's work in FS streams and stores it in the form of + * IOStatisticsSnapshot if thread level aggregation is enabled else returning + * an instance of EmptyIOStatisticsStore for the FS. An active instance of + * the IOStatisticsContext can be used to collect the statistics. + * + * For the current thread the IOStatisticsSnapshot can be used as a way to + * move the IOStatistics data between applications using the Serializable + * nature of the class. + */ +public class IOStatisticsContext { + private static final Logger LOG = + LoggerFactory.getLogger(IOStatisticsContext.class); + private static final boolean IS_THREAD_IOSTATS_ENABLED; + + /** + * Active IOStatistics Context containing different worker thread's + * statistics. Weak Reference so that it gets cleaned up during GC and we + * avoid any memory leak issues due to long lived references. + */ + private static final WeakReferenceThreadMap<IOStatisticsContext> + ACTIVE_IOSTATS_CONTEXT = + new WeakReferenceThreadMap<>(IOStatisticsContext::createNewInstance, + IOStatisticsContext::referenceLostContext + ); + + /** + * Collecting IOStatistics per thread. + */ + private final WeakReferenceThreadMap<IOStatisticsSnapshot> + threadIOStatsContext = + new WeakReferenceThreadMap<>(this::getIOStatisticsSnapshotFactory, + this::referenceLost); + + static { + // Work out if the current context has thread level IOStatistics enabled. + final Configuration configuration = new Configuration(); + IS_THREAD_IOSTATS_ENABLED = + configuration.getBoolean(THREAD_LEVEL_IOSTATISTICS_ENABLED, + THREAD_LEVEL_IOSTATISTICS_ENABLED_DEFAULT); + } + + /** + * Creating a new IOStatisticsContext instance for a FS to be used. + * + * @param key Thread ID that represents which thread the context belongs to. + * @return an instance of IOStatisticsContext. + */ + private static IOStatisticsContext createNewInstance(Long key) { + return new IOStatisticsContext(); + } + + /** + * Get the current IOStatisticsContext. + * + * @return current IOStatisticsContext instance. + */ + public static IOStatisticsContext currentIOStatisticsContext() { + return ACTIVE_IOSTATS_CONTEXT.get(Thread.currentThread().getId()); + } + + /** + * A Method to act as an IOStatisticsSnapshot factory, in a + * WeakReferenceThreadMap. + * + * @param key ThreadID. + * @return an Instance of IOStatisticsSnapshot. + */ + private IOStatisticsSnapshot getIOStatisticsSnapshotFactory(Long key) { + return new IOStatisticsSnapshot(); + } + + /** + * In case of reference loss. + * + * @param key ThreadID. + */ + private void referenceLost(Long key) { + LOG.debug("Reference lost for threadID: {}", key); + } + + /** + * In case of reference loss for IOStatisticsContext. + * + * @param key ThreadID. + */ + private static void referenceLostContext(Long key) { + LOG.debug("Reference lost for threadID for the context: {}", key); + } + + /** + * A Method to get the IOStatisticsAggregator of the currentThread. This + * denotes the aggregated IOStatistics per thread. + * + * @return the instance of IOStatisticsAggregator for the current thread. + */ + public IOStatisticsAggregator getThreadIOStatistics() { Review Comment: let's call this ThreadIOStatisticsAggregator Issue Time Tracking ------------------- Worklog Id: (was: 790111) Time Spent: 3.5h (was: 3h 20m) > Add thread-level IOStatistics Context > ------------------------------------- > > Key: HADOOP-17461 > URL: https://issues.apache.org/jira/browse/HADOOP-17461 > Project: Hadoop Common > Issue Type: Sub-task > Components: fs, fs/azure, fs/s3 > Affects Versions: 3.3.1 > Reporter: Steve Loughran > Assignee: Mehakmeet Singh > Priority: Major > Labels: pull-request-available > Time Spent: 3.5h > Remaining Estimate: 0h > > For effective reporting of the iostatistics of individual worker threads, we > need a thread-level context which IO components update. > * this contact needs to be passed in two background thread forming work on > behalf of a task. > * IO Components (streams, iterators, filesystems) need to update this context > statistics as they perform work > * Without double counting anything. > I imagine a ThreadLocal IOStatisticContext which will be updated in the > FileSystem API Calls. This context MUST be passed into the background threads > used by a task, so that IO is correctly aggregated. > I don't want streams, listIterators &c to do the updating as there is more > risk of double counting. However, we need to see their statistics if we want > to know things like "bytes discarded in backwards seeks". And I don't want to > be updating a shared context object on every read() call. > If all we want is store IO (HEAD, GET, DELETE, list performance etc) then the > FS is sufficient. > If we do want the stream-specific detail, then I propose > * caching the context in the constructor > * updating it only in close() or unbuffer() (as we do from S3AInputStream to > S3AInstrumenation) > * excluding those we know the FS already collects. -- This message was sent by Atlassian Jira (v8.20.10#820010) --------------------------------------------------------------------- To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-issues-h...@hadoop.apache.org