apache / carbondata / 220

Build # 220

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/da2686c5d63af4ecce31283d43581ec10db50726">da2686c5d<a href="https://github.com/apache/carbondata/commit/da2686c5d63af4ecce31283d43581ec10db50726">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/063984b6f26f0863101fb760cb1e6de2527cf78c">063984b6f</a>">063984b6f</a><a href="https://github.com/apache/carbondata/commit/da2686c5d63af4ecce31283d43581ec10db50726"> into e14c8173a">e14c8173a</a>

Run Details

36 of 36 new or added lines in 7 files covered. (100.0%)

65689 of 83736 relevant lines covered (78.45%)

2.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.06

/core/src/main/java/org/apache/carbondata/core/util/TaskMetricsMap.java

/*
 * 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.carbondata.core.util;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.carbondata.common.logging.LogServiceFactory;

import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.Logger;

/**
 * This class maintains task level metrics info for all spawned child threads and parent task thread
 */
public class TaskMetricsMap {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(TaskMetricsMap.class.getName());

  private static final InheritableThreadLocal<Long> threadLocal = new InheritableThreadLocal<>();
  /**
   * In this map we are maintaining all spawned child threads callback info for each parent thread
   * here key = parent thread id & values =  list of spawned child threads callbacks
   */
  public static final Map<Long, List<CarbonFSBytesReadOnThreadCallback>> metricMap =
      new ConcurrentHashMap<>();

  public static final TaskMetricsMap taskMetricsMap = new TaskMetricsMap();

  public static TaskMetricsMap getInstance() {
    return taskMetricsMap;
  }

  public static InheritableThreadLocal<Long> getThreadLocal() {
    return threadLocal;
  }

  /**
   * registers current thread callback using parent thread id
   *
   * @return
   */
  public void registerThreadCallback() {
    // parent thread id should not be null as we are setting the same for all RDDs
    if (null != threadLocal.get()) {
      long parentThreadId = threadLocal.get();
      new CarbonFSBytesReadOnThreadCallback(parentThreadId);
    }
  }

  /**
   * removes parent thread entry from map.
   *
   * @param threadId
   */
  public void removeEntry(long threadId) {
    metricMap.remove(threadId);
  }

  /**
   * returns all spawned child threads callback list of given parent thread
   *
   * @param threadId
   * @return
   */
  public List<CarbonFSBytesReadOnThreadCallback> getCallbackList(long threadId) {
    return metricMap.get(threadId);
  }

  public boolean isCallbackEmpty(long threadId) {
    List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(threadId);
    if (null == callbackList) {
      return true;
    }
    return callbackList.isEmpty();
  }

  /**
   * This function updates read bytes of given thread
   * After completing the task, each spawned child thread should update current read bytes,
   * by calling this function.
   *
   * @param callbackThreadId
   */
  public void updateReadBytes(long callbackThreadId) {
    // parent thread id should not be null as we are setting the same for all RDDs
    if (null != threadLocal.get()) {
      long parentThreadId = threadLocal.get();
      List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(parentThreadId);
      if (null != callbackList) {
        for (CarbonFSBytesReadOnThreadCallback callback : callbackList) {
          if (callback.threadId == callbackThreadId) {
            callback.updatedReadBytes += callback.readbytes();
            break;
          }
        }
      }
    }
  }

  /**
   * returns total task read bytes, by summing all parent & spawned threads readbytes
   *
   * @param threadName
   * @return
   */
  public long getReadBytesSum(long threadName) {
    List<CarbonFSBytesReadOnThreadCallback> callbacks = getCallbackList(threadName);
    long sum = 0;
    if (null != callbacks) {
      for (CarbonFSBytesReadOnThreadCallback callback : callbacks) {
        sum += callback.getReadBytes();
      }
    }
    return sum;
  }

  /**
   * adds spawaned thread callback entry in metricmap using parentThreadId
   *
   * @param parentThreadId
   * @param callback
   */
  private void addEntry(long parentThreadId, CarbonFSBytesReadOnThreadCallback callback) {
    List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(parentThreadId);
    if (null == callbackList) {
      //create new list
      List<CarbonFSBytesReadOnThreadCallback> list = new CopyOnWriteArrayList<>();
      list.add(callback);
      metricMap.put(parentThreadId, list);
    } else {
      // add to existing list
      callbackList.add(callback);
    }
  }

  /**
   * This class maintains getReadBytes info of each thread
   */
  class CarbonFSBytesReadOnThreadCallback {
    long baseline = 0;
    long updatedReadBytes = 0;
    long threadId = Thread.currentThread().getId();

    CarbonFSBytesReadOnThreadCallback(long parentThread) {
      // reads current thread readBytes
      this.baseline = readbytes();
      addEntry(parentThread, this);
    }

    /**
     * returns current thread readbytes from FileSystem Statistics
     *
     * @return
     */
    public long readbytes() {
      List<FileSystem.Statistics> statisticsList = FileSystem.getAllStatistics();
      long sum = 0;
      try {
        for (FileSystem.Statistics statistics : statisticsList) {
          Class statisticsClass = Class.forName(statistics.getClass().getName());
          Method getThreadStatisticsMethod =
              statisticsClass.getDeclaredMethod("getThreadStatistics");
          Class statisticsDataClass =
              Class.forName("org.apache.hadoop.fs.FileSystem$Statistics$StatisticsData");
          Method getBytesReadMethod = statisticsDataClass.getDeclaredMethod("getBytesRead");
          sum += (Long) getBytesReadMethod
              .invoke(statisticsDataClass.cast(getThreadStatisticsMethod.invoke(statistics, null)),
                  null);
        }
      } catch (Exception ex) {
        LOGGER.debug(ex.getLocalizedMessage());
      }
      return sum;
    }

    /**
     * After completing task, each child thread should update corresponding
     * read bytes using updatedReadBytes method.
     * if updatedReadBytes > 0 then return updatedReadBytes (i.e thread read bytes).
     *
     * @return
     */
    public long getReadBytes() {
      return updatedReadBytes - baseline;
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.util;
19
20	import java.lang.reflect.Method;
21	import java.util.List;
22	import java.util.Map;
23	import java.util.concurrent.ConcurrentHashMap;
24	import java.util.concurrent.CopyOnWriteArrayList;
25
26	import org.apache.carbondata.common.logging.LogServiceFactory;
27
28	import org.apache.hadoop.fs.FileSystem;
29	import org.apache.log4j.Logger;
30
31	/**
32	* This class maintains task level metrics info for all spawned child threads and parent task thread
33	*/
34	public class TaskMetricsMap {	4✔
35
36	private static final Logger LOGGER =	4✔
37	LogServiceFactory.getLogService(TaskMetricsMap.class.getName());	4✔
38
39	private static final InheritableThreadLocal<Long> threadLocal = new InheritableThreadLocal<>();	4✔
40	/**
41	* In this map we are maintaining all spawned child threads callback info for each parent thread
42	* here key = parent thread id & values = list of spawned child threads callbacks
43	*/
44	public static final Map<Long, List<CarbonFSBytesReadOnThreadCallback>> metricMap =	4✔
45	new ConcurrentHashMap<>();
46
47	public static final TaskMetricsMap taskMetricsMap = new TaskMetricsMap();	4✔
48
49	public static TaskMetricsMap getInstance() {
50	return taskMetricsMap;	4✔
51	}
52
53	public static InheritableThreadLocal<Long> getThreadLocal() {
54	return threadLocal;	2✔
55	}
56
57	/**
58	* registers current thread callback using parent thread id
59	*
60	* @return
61	*/
62	public void registerThreadCallback() {
63	// parent thread id should not be null as we are setting the same for all RDDs
64	if (null != threadLocal.get()) {	4✔
65	long parentThreadId = threadLocal.get();	2✔
66	new CarbonFSBytesReadOnThreadCallback(parentThreadId);	2✔
67	}
68	}	4✔
69
70	/**
71	* removes parent thread entry from map.
72	*
73	* @param threadId
74	*/
75	public void removeEntry(long threadId) {
76	metricMap.remove(threadId);	2✔
77	}	2✔
78
79	/**
80	* returns all spawned child threads callback list of given parent thread
81	*
82	* @param threadId
83	* @return
84	*/
85	public List<CarbonFSBytesReadOnThreadCallback> getCallbackList(long threadId) {
86	return metricMap.get(threadId);	2✔
87	}
88
89	public boolean isCallbackEmpty(long threadId) {
90	List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(threadId);	2✔
91	if (null == callbackList) {	2✔
92	return true;	2✔
93	}
94	return callbackList.isEmpty();	2✔
95	}
96
97	/**
98	* This function updates read bytes of given thread
99	* After completing the task, each spawned child thread should update current read bytes,
100	* by calling this function.
101	*
102	* @param callbackThreadId
103	*/
104	public void updateReadBytes(long callbackThreadId) {
105	// parent thread id should not be null as we are setting the same for all RDDs
106	if (null != threadLocal.get()) {	4✔
107	long parentThreadId = threadLocal.get();	2✔
108	List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(parentThreadId);	2✔
109	if (null != callbackList) {	2✔
110	for (CarbonFSBytesReadOnThreadCallback callback : callbackList) {	2✔
111	if (callback.threadId == callbackThreadId) {	2✔
112	callback.updatedReadBytes += callback.readbytes();	2✔
113	break;	2✔
114	}
115	}	2✔
116	}
117	}
118	}	4✔
119
120	/**
121	* returns total task read bytes, by summing all parent & spawned threads readbytes
122	*
123	* @param threadName
124	* @return
125	*/
126	public long getReadBytesSum(long threadName) {
127	List<CarbonFSBytesReadOnThreadCallback> callbacks = getCallbackList(threadName);	2✔
128	long sum = 0;	2✔
129	if (null != callbacks) {	2✔
130	for (CarbonFSBytesReadOnThreadCallback callback : callbacks) {	2✔
131	sum += callback.getReadBytes();	2✔
132	}	2✔
133	}
134	return sum;	2✔
135	}
136
137	/**
138	* adds spawaned thread callback entry in metricmap using parentThreadId
139	*
140	* @param parentThreadId
141	* @param callback
142	*/
143	private void addEntry(long parentThreadId, CarbonFSBytesReadOnThreadCallback callback) {
144	List<CarbonFSBytesReadOnThreadCallback> callbackList = getCallbackList(parentThreadId);	2✔
145	if (null == callbackList) {	2✔
146	//create new list
147	List<CarbonFSBytesReadOnThreadCallback> list = new CopyOnWriteArrayList<>();	2✔
148	list.add(callback);	2✔
149	metricMap.put(parentThreadId, list);	2✔
150	} else {	2✔
151	// add to existing list
152	callbackList.add(callback);	2✔
153	}
154	}	2✔
155
156	/**
157	* This class maintains getReadBytes info of each thread
158	*/
159	class CarbonFSBytesReadOnThreadCallback {
160	long baseline = 0;	2✔
161	long updatedReadBytes = 0;	2✔
162	long threadId = Thread.currentThread().getId();	2✔
163
164	CarbonFSBytesReadOnThreadCallback(long parentThread) {	2✔
165	// reads current thread readBytes
166	this.baseline = readbytes();	2✔
167	addEntry(parentThread, this);	2✔
168	}	2✔
169
170	/**
171	* returns current thread readbytes from FileSystem Statistics
172	*
173	* @return
174	*/
175	public long readbytes() {
176	List<FileSystem.Statistics> statisticsList = FileSystem.getAllStatistics();	2✔
177	long sum = 0;	2✔
178	try {
179	for (FileSystem.Statistics statistics : statisticsList) {	2✔
180	Class statisticsClass = Class.forName(statistics.getClass().getName());	2✔
181	Method getThreadStatisticsMethod =	2✔
182	statisticsClass.getDeclaredMethod("getThreadStatistics");	2✔
183	Class statisticsDataClass =	2✔
184	Class.forName("org.apache.hadoop.fs.FileSystem$Statistics$StatisticsData");	2✔
185	Method getBytesReadMethod = statisticsDataClass.getDeclaredMethod("getBytesRead");	2✔
186	sum += (Long) getBytesReadMethod	2✔
187	.invoke(statisticsDataClass.cast(getThreadStatisticsMethod.invoke(statistics, null)),	2✔
188	null);
189	}	2✔
190	} catch (Exception ex) {	×
191	LOGGER.debug(ex.getLocalizedMessage());	×
192	}	2✔
193	return sum;	2✔
194	}
195
196	/**
197	* After completing task, each child thread should update corresponding
198	* read bytes using updatedReadBytes method.
199	* if updatedReadBytes > 0 then return updatedReadBytes (i.e thread read bytes).
200	*
201	* @return
202	*/
203	public long getReadBytes() {
204	return updatedReadBytes - baseline;	2✔
205	}
206	}
207	}

apache / carbondata / 220

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous