1988

Build # 1988

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/21e50122ca24f592020236d85dcd1453cb1cc2fa">21e50122c<a href="https://github.com/apache/carbondata/commit/21e50122ca24f592020236d85dcd1453cb1cc2fa">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/124ab41cf3ef9dcbd977200f35435669190c3074">124ab41cf</a>">124ab41cf</a><a href="https://github.com/apache/carbondata/commit/21e50122ca24f592020236d85dcd1453cb1cc2fa"> into 0e467eb06">0e467eb06</a>

Run Details

1511 of 1511 new or added lines in 118 files covered. (100.0%)

62642 of 78664 relevant lines covered (79.63%)

1.06 hits per line

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

73.27

/core/src/main/java/org/apache/carbondata/core/memory/UnsafeSortMemoryManager.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.memory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.util.CarbonProperties;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

/**
 * Memory manager to keep track of
 * all memory for storing the sorted data
 */
public class UnsafeSortMemoryManager {

  /**
   * logger
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(UnsafeSortMemoryManager.class.getName());

  /**
   * offheap is enabled
   */
  private static boolean offHeap = Boolean.parseBoolean(CarbonProperties.getInstance()
      .getPropertyOrDefault(CarbonCommonConstants.ENABLE_OFFHEAP_SORT));

  /**
   * map to keep taskid to memory blocks
   */
  private static Map<String, Set<MemoryBlock>> taskIdToMemoryBlockMap;

  /**
   * singleton instance
   */
  public static final UnsafeSortMemoryManager INSTANCE;

  /**
   * total memory available for sort data storage
   */
  private long totalMemory;

  /**
   * current memory used
   */
  private long memoryUsed;

  /**
   * current memory allocator
   */
  private MemoryAllocator allocator;

  static {
    long size;
    long value = CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB
        .getDefaultValueInt();
    try {
      size = Long.parseLong(CarbonProperties.getInstance().getProperty(
          CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB));
    } catch (Exception e) {
      size = value;
      LOGGER.info("Wrong memory size given, so setting default value to " + size);
    }
    if (size < value) {
      size = value;
      LOGGER.warn(String.format(
          "It is not recommended to set unsafe sort memory size less than %dMB,"
              + " so setting default value to %d",
          value, size));
    }

    long takenSize = size * 1024 * 1024;
    MemoryAllocator allocator;
    if (offHeap) {
      allocator = MemoryAllocator.UNSAFE;
    } else {
      long maxMemory = Runtime.getRuntime().maxMemory() * 60 / 100;
      if (takenSize > maxMemory) {
        takenSize = maxMemory;
      }
      allocator = MemoryAllocator.HEAP;
    }
    INSTANCE = new UnsafeSortMemoryManager(takenSize, allocator);
    taskIdToMemoryBlockMap = new HashMap<>();
  }

  private UnsafeSortMemoryManager(long totalMemory, MemoryAllocator allocator) {
    this.totalMemory = totalMemory;
    this.allocator = allocator;
    LOGGER.info("Sort Memory manager is created with size " + totalMemory + " with " + allocator);
  }

  /**
   * Below method will be used to check whether memory required is
   * available or not
   *
   * @param required
   * @return if memory available
   */
  public synchronized boolean isMemoryAvailable(long required) {
    return memoryUsed + required < totalMemory;
  }

  /**
   * total usable memory for sort memory manager
   * @return size in bytes
   */
  public long getUsableMemory() {
    return totalMemory;
  }

  /**
   * Below method will be used to allocate dummy memory
   * this will be used to allocate first and then used when u need
   *
   * @param size
   */
  public synchronized void allocateDummyMemory(long size) {
    memoryUsed += size;
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(String.format(
          "Sort Memory block is created with size %d. Total memory used %d Bytes, left %d Bytes",
          size, memoryUsed, totalMemory - memoryUsed));
    }
  }

  public synchronized void freeMemory(String taskId, MemoryBlock memoryBlock) {
    if (taskIdToMemoryBlockMap.containsKey(taskId)) {
      taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);
    }
    if (!memoryBlock.isFreedStatus()) {
      allocator.free(memoryBlock);
      memoryUsed -= memoryBlock.size();
      memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(String.format(
            "Freeing sort memory block (%s) with size: %d, current available memory is: %d",
            memoryBlock.toString(), memoryBlock.size(), totalMemory - memoryUsed));
      }
    }
  }

  /**
   * Below method will be used to free all the
   * memory occupied for a task, this will be useful
   * when in case of task failure we need to clear all the memory occupied
   * @param taskId
   */
  public synchronized void freeMemoryAll(String taskId) {
    Set<MemoryBlock> memoryBlockSet = null;
    memoryBlockSet = taskIdToMemoryBlockMap.remove(taskId);
    long occuppiedMemory = 0;
    if (null != memoryBlockSet) {
      Iterator<MemoryBlock> iterator = memoryBlockSet.iterator();
      MemoryBlock memoryBlock = null;
      while (iterator.hasNext()) {
        memoryBlock = iterator.next();
        if (!memoryBlock.isFreedStatus()) {
          occuppiedMemory += memoryBlock.size();
          allocator.free(memoryBlock);
        }
      }
    }
    memoryUsed -= occuppiedMemory;
    memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(
          String.format("Freeing sort memory of size: %d, current available memory is: %d",
              occuppiedMemory, totalMemory - memoryUsed));
    }
    LOGGER.info(String.format(
        "Total sort memory used after task %s is %d. Current running tasks are: %s",
        taskId, memoryUsed, StringUtils.join(taskIdToMemoryBlockMap.keySet(), ", ")));
  }

  /**
   * Before calling this method caller should call allocateMemoryDummy
   * This method will be used to allocate the memory, this can be used
   * when caller wants to allocate memory first and used it anytime
   * @param taskId
   * @param memoryRequested
   * @return memory block
   */
  public synchronized MemoryBlock allocateMemoryLazy(String taskId, long memoryRequested) {
    MemoryBlock allocate = allocator.allocate(memoryRequested);
    Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);
    if (null == listOfMemoryBlock) {
      listOfMemoryBlock = new HashSet<>();
      taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);
    }
    listOfMemoryBlock.add(allocate);
    return allocate;
  }

  /**
   * It tries to allocate memory of `size` bytes, keep retry until it allocates successfully.
   */
  public static MemoryBlock allocateMemoryWithRetry(String taskId, long size)
          throws MemoryException {
    MemoryBlock baseBlock = null;
    int tries = 0;
    while (tries < 100) {
      baseBlock = INSTANCE.allocateMemory(taskId, size);
      if (baseBlock == null) {
        try {
          Thread.sleep(50);
        } catch (InterruptedException e) {
          throw new MemoryException(e);
        }
      } else {
        break;
      }
      tries++;
    }
    if (baseBlock == null) {
      throw new MemoryException("Not enough sort memory, please increase "
          + CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB);
    }
    return baseBlock;
  }

  private synchronized MemoryBlock allocateMemory(String taskId, long memoryRequested) {
    if (memoryUsed + memoryRequested <= totalMemory) {
      MemoryBlock allocate = allocator.allocate(memoryRequested);
      memoryUsed += allocate.size();
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug(String.format(
            "Sort Memory block is created with size %d. Total memory used %d Bytes, left %d Bytes",
            allocate.size(), memoryUsed, totalMemory - memoryUsed));
      }
      Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);
      if (null == listOfMemoryBlock) {
        listOfMemoryBlock = new HashSet<>();
        taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);
      }
      listOfMemoryBlock.add(allocate);
      return allocate;
    }
    return null;
  }

  public static boolean isOffHeap() {
    return offHeap;
  }
}

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	package org.apache.carbondata.core.memory;
18
19	import java.util.HashMap;
20	import java.util.HashSet;
21	import java.util.Iterator;
22	import java.util.Map;
23	import java.util.Set;
24
25	import org.apache.carbondata.common.logging.LogServiceFactory;
26	import org.apache.carbondata.core.constants.CarbonCommonConstants;
27	import org.apache.carbondata.core.util.CarbonProperties;
28
29	import org.apache.commons.lang3.StringUtils;
30	import org.apache.log4j.Logger;
31
32	/**
33	* Memory manager to keep track of
34	* all memory for storing the sorted data
35	*/
36	public class UnsafeSortMemoryManager {
37
38	/**
39	* logger
40	*/
41	private static final Logger LOGGER =	1✔
42	LogServiceFactory.getLogService(UnsafeSortMemoryManager.class.getName());	1✔
43
44	/**
45	* offheap is enabled
46	*/
47	private static boolean offHeap = Boolean.parseBoolean(CarbonProperties.getInstance()	1✔
48	.getPropertyOrDefault(CarbonCommonConstants.ENABLE_OFFHEAP_SORT));	1✔
49
50	/**
51	* map to keep taskid to memory blocks
52	*/
53	private static Map<String, Set<MemoryBlock>> taskIdToMemoryBlockMap;
54
55	/**
56	* singleton instance
57	*/
58	public static final UnsafeSortMemoryManager INSTANCE;
59
60	/**
61	* total memory available for sort data storage
62	*/
63	private long totalMemory;
64
65	/**
66	* current memory used
67	*/
68	private long memoryUsed;
69
70	/**
71	* current memory allocator
72	*/
73	private MemoryAllocator allocator;
74
75	static {
76	long size;
77	long value = CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB	1✔
78	.getDefaultValueInt();	1✔
79	try {
80	size = Long.parseLong(CarbonProperties.getInstance().getProperty(	1✔
81	CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB));
82	} catch (Exception e) {	×
83	size = value;	×
84	LOGGER.info("Wrong memory size given, so setting default value to " + size);	×
85	}	1✔
86	if (size < value) {	1✔
87	size = value;	×
88	LOGGER.warn(String.format(	×
89	"It is not recommended to set unsafe sort memory size less than %dMB,"
90	+ " so setting default value to %d",
91	value, size));	×
92	}
93
94	long takenSize = size * 1024 * 1024;	1✔
95	MemoryAllocator allocator;
96	if (offHeap) {	1✔
97	allocator = MemoryAllocator.UNSAFE;	1✔
98	} else {
99	long maxMemory = Runtime.getRuntime().maxMemory() * 60 / 100;	×
100	if (takenSize > maxMemory) {	×
101	takenSize = maxMemory;	×
102	}
103	allocator = MemoryAllocator.HEAP;	×
104	}
105	INSTANCE = new UnsafeSortMemoryManager(takenSize, allocator);	1✔
106	taskIdToMemoryBlockMap = new HashMap<>();	1✔
107	}	1✔
108
109	private UnsafeSortMemoryManager(long totalMemory, MemoryAllocator allocator) {	1✔
110	this.totalMemory = totalMemory;	1✔
111	this.allocator = allocator;	1✔
112	LOGGER.info("Sort Memory manager is created with size " + totalMemory + " with " + allocator);	1✔
113	}	1✔
114
115	/**
116	* Below method will be used to check whether memory required is
117	* available or not
118	*
119	* @param required
120	* @return if memory available
121	*/
122	public synchronized boolean isMemoryAvailable(long required) {
123	return memoryUsed + required < totalMemory;	1✔
124	}
125
126	/**
127	* total usable memory for sort memory manager
128	* @return size in bytes
129	*/
130	public long getUsableMemory() {
131	return totalMemory;	1✔
132	}
133
134	/**
135	* Below method will be used to allocate dummy memory
136	* this will be used to allocate first and then used when u need
137	*
138	* @param size
139	*/
140	public synchronized void allocateDummyMemory(long size) {
141	memoryUsed += size;	1✔
142	if (LOGGER.isDebugEnabled()) {	1✔
143	LOGGER.debug(String.format(	1✔
144	"Sort Memory block is created with size %d. Total memory used %d Bytes, left %d Bytes",
145	size, memoryUsed, totalMemory - memoryUsed));	1✔
146	}
147	}	1✔
148
149	public synchronized void freeMemory(String taskId, MemoryBlock memoryBlock) {
150	if (taskIdToMemoryBlockMap.containsKey(taskId)) {	1✔
151	taskIdToMemoryBlockMap.get(taskId).remove(memoryBlock);	1✔
152	}
153	if (!memoryBlock.isFreedStatus()) {	1✔
154	allocator.free(memoryBlock);	1✔
155	memoryUsed -= memoryBlock.size();	1✔
156	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	1✔
157	if (LOGGER.isDebugEnabled()) {	1✔
158	LOGGER.debug(String.format(	1✔
159	"Freeing sort memory block (%s) with size: %d, current available memory is: %d",
160	memoryBlock.toString(), memoryBlock.size(), totalMemory - memoryUsed));	1✔
161	}
162	}
163	}	1✔
164
165	/**
166	* Below method will be used to free all the
167	* memory occupied for a task, this will be useful
168	* when in case of task failure we need to clear all the memory occupied
169	* @param taskId
170	*/
171	public synchronized void freeMemoryAll(String taskId) {
172	Set<MemoryBlock> memoryBlockSet = null;	1✔
173	memoryBlockSet = taskIdToMemoryBlockMap.remove(taskId);	1✔
174	long occuppiedMemory = 0;	1✔
175	if (null != memoryBlockSet) {	1✔
176	Iterator<MemoryBlock> iterator = memoryBlockSet.iterator();	1✔
177	MemoryBlock memoryBlock = null;	1✔
178	while (iterator.hasNext()) {	1✔
179	memoryBlock = iterator.next();	×
180	if (!memoryBlock.isFreedStatus()) {	×
181	occuppiedMemory += memoryBlock.size();	×
182	allocator.free(memoryBlock);	×
183	}
184	}
185	}
186	memoryUsed -= occuppiedMemory;	1✔
187	memoryUsed = memoryUsed < 0 ? 0 : memoryUsed;	1✔
188	if (LOGGER.isDebugEnabled()) {	1✔
189	LOGGER.debug(	×
190	String.format("Freeing sort memory of size: %d, current available memory is: %d",	×
191	occuppiedMemory, totalMemory - memoryUsed));	×
192	}
193	LOGGER.info(String.format(	1✔
194	"Total sort memory used after task %s is %d. Current running tasks are: %s",
195	taskId, memoryUsed, StringUtils.join(taskIdToMemoryBlockMap.keySet(), ", ")));	1✔
196	}	1✔
197
198	/**
199	* Before calling this method caller should call allocateMemoryDummy
200	* This method will be used to allocate the memory, this can be used
201	* when caller wants to allocate memory first and used it anytime
202	* @param taskId
203	* @param memoryRequested
204	* @return memory block
205	*/
206	public synchronized MemoryBlock allocateMemoryLazy(String taskId, long memoryRequested) {
207	MemoryBlock allocate = allocator.allocate(memoryRequested);	1✔
208	Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);	1✔
209	if (null == listOfMemoryBlock) {	1✔
210	listOfMemoryBlock = new HashSet<>();	1✔
211	taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);	1✔
212	}
213	listOfMemoryBlock.add(allocate);	1✔
214	return allocate;	1✔
215	}
216
217	/**
218	* It tries to allocate memory of `size` bytes, keep retry until it allocates successfully.
219	*/
220	public static MemoryBlock allocateMemoryWithRetry(String taskId, long size)
221	throws MemoryException {
222	MemoryBlock baseBlock = null;	1✔
223	int tries = 0;	1✔
224	while (tries < 100) {	1✔
225	baseBlock = INSTANCE.allocateMemory(taskId, size);	1✔
226	if (baseBlock == null) {	1✔
227	try {
228	Thread.sleep(50);	×
229	} catch (InterruptedException e) {	×
230	throw new MemoryException(e);	×
231	}	×
232	} else {
233	break;
234	}
235	tries++;	×
236	}
237	if (baseBlock == null) {	1✔
238	throw new MemoryException("Not enough sort memory, please increase "	×
239	+ CarbonCommonConstants.CARBON_SORT_STORAGE_INMEMORY_IN_MB);
240	}
241	return baseBlock;	1✔
242	}
243
244	private synchronized MemoryBlock allocateMemory(String taskId, long memoryRequested) {
245	if (memoryUsed + memoryRequested <= totalMemory) {	1✔
246	MemoryBlock allocate = allocator.allocate(memoryRequested);	1✔
247	memoryUsed += allocate.size();	1✔
248	if (LOGGER.isDebugEnabled()) {	1✔
249	LOGGER.debug(String.format(	1✔
250	"Sort Memory block is created with size %d. Total memory used %d Bytes, left %d Bytes",
251	allocate.size(), memoryUsed, totalMemory - memoryUsed));	1✔
252	}
253	Set<MemoryBlock> listOfMemoryBlock = taskIdToMemoryBlockMap.get(taskId);	1✔
254	if (null == listOfMemoryBlock) {	1✔
255	listOfMemoryBlock = new HashSet<>();	×
256	taskIdToMemoryBlockMap.put(taskId, listOfMemoryBlock);	×
257	}
258	listOfMemoryBlock.add(allocate);	1✔
259	return allocate;	1✔
260	}
261	return null;	×
262	}
263
264	public static boolean isOffHeap() {
265	return offHeap;	×
266	}
267	}

apache / carbondata / 1988

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